"""Observe two mappings on Linux; flag definitions are pinned to Linux v6.12."""

import ctypes
import mmap
import os
import sys
import tempfile

from pagemap_probe import PFN_MASK, read_u64

KPAGE_FLAGS = (
    ("LOCKED", 0), ("REFERENCED", 2), ("UPTODATE", 3),
    ("DIRTY", 4), ("LRU", 5), ("ACTIVE", 6), ("WRITEBACK", 8),
    ("MMAP", 11), ("SWAPBACKED", 14), ("NOPAGE", 20),
)


def decode_entry(entry):
    result = {
        "present": bool(entry & (1 << 63)),
        "swapped": bool(entry & (1 << 62)),
        "file_or_shared_anon": bool(entry & (1 << 61)),
        "soft_dirty": bool(entry & (1 << 55)),
        "exclusive": bool(entry & (1 << 56)),
        "uffd_wp": bool(entry & (1 << 57)),
    }
    if result["swapped"]:
        result["swap_type"] = entry & 31
        result["swap_offset"] = (entry & PFN_MASK) >> 5
    elif result["present"]:
        result["pfn"] = entry & PFN_MASK
    return result


def decode_kpageflags(value):
    return [name for name, bit in KPAGE_FLAGS if value & (1 << bit)]


def mapping_address(area):
    # Release the exported buffer before the mmap context tries to unmap it.
    exported = ctypes.c_char.from_buffer(area)
    address = ctypes.addressof(exported)
    del exported
    return address


def observe(label, area, page_size):
    address = mapping_address(area)
    entry = read_u64("/proc/self/pagemap", address // page_size)
    fields = decode_entry(entry)
    print(f"\n{label}: address={address:#x} raw={entry:#018x}")
    print(" ".join(f"{name}={value}" for name, value in fields.items()))
    if not fields["present"] or fields["swapped"]:
        return
    pfn = fields["pfn"]
    if not pfn:
        print("PFN zero: possibly hidden; physical-page lookup skipped.")
        return
    for path in ("/proc/kpagecount", "/proc/kpageflags"):
        try:
            value = read_u64(path, pfn)
        except (OSError, EOFError) as error:
            print(f"{path}: unavailable ({error})")
        else:
            if path.endswith("kpageflags"):
                names = ",".join(decode_kpageflags(value)) or "none selected"
                print(f"kpageflags={value:#x}: {names}")
            else:
                print(f"kpagecount={value}")


def run():
    page_size = os.sysconf("SC_PAGE_SIZE")
    protection = mmap.PROT_READ | mmap.PROT_WRITE
    with mmap.mmap(
        -1, page_size, flags=mmap.MAP_PRIVATE | mmap.MAP_ANONYMOUS,
        prot=protection,
    ) as anonymous:
        observe("anonymous: before access", anonymous, page_size)
        anonymous[0] = ord("A")
        observe("anonymous: after write", anonymous, page_size)

    with tempfile.TemporaryFile() as backing:
        backing.truncate(page_size)
        backing.write(b"F")
        backing.flush()
        with mmap.mmap(
            backing.fileno(), page_size, flags=mmap.MAP_SHARED,
            prot=protection,
        ) as shared:
            observe("file: before access", shared, page_size)
            print(f"first byte={shared[0]:#x}")
            observe("file: after read", shared, page_size)
            shared[0] = ord("W")
            observe("file: after shared write", shared, page_size)
            shared.flush()


def main():
    if not sys.platform.startswith("linux"):
        raise SystemExit("This example requires Linux procfs.")
    try:
        run()
    except (OSError, EOFError) as error:
        raise SystemExit(f"Observation could not be completed: {error}") from error


if __name__ == "__main__":
    main()