"""DMA / address-aperture translation seam.

Evidence kind: **hardware address-map header + firmware use + observed
access pattern (through the falcon DMA registers).**

The GSP RISC-V address map (Turing) defines apertures the firmware and its
DMA engine address in GPU physical space:

* ``IMEM``  ``0x0_0000_0000`` (4 GiB window)   -- local instruction memory
* ``DMEM``  ``0x1_0000_0000`` (4 GiB window)   -- local data memory
* ``EMEM``  ``0x2_0000_0000`` (4 GiB window)   -- external memory window
* ``FBGPA`` ``0x6180_0000_0000_0000``          -- framebuffer aperture
* ``SYSGPA````0x8180_0000_0000_0000``          -- system-memory aperture
* ``PRIV``  ``0x2000_0000_0000_0000``          -- register aperture

The simulator keeps local memory in one flat RAM region; framebuffer and
system memory only exist as *profile-declared windows* that map a GPU
physical aperture onto a simulator RAM address.  A window not declared in the
profile is not modelled: the transfer is refused and recorded (never
silently zero-filled).  This class is the single place where that mapping
lives, so a future IOMMU model can replace it without touching devices.
"""
from __future__ import annotations


class DmaTranslator:
    """Translate GPU physical addresses to simulator RAM through windows."""

    def __init__(self, *, ram, ram_base: int, ram_size: int, windows=None):
        self.ram = ram
        self.ram_base = ram_base
        self.ram_size = ram_size
        self.windows: list[tuple[int, int, int]] = []  # (pa_base, sim_base, size)
        self.transfers = 0
        self.refused = 0
        self.events: list[dict] = []
        for win in windows or ():
            self.add_window(int(win["pa"]), int(win["sim"]), int(win["size"]))

    def add_window(self, pa_base: int, sim_base: int, size: int) -> None:
        if size <= 0:
            raise ValueError("DMA window size must be positive")
        self.windows.append((pa_base, sim_base, size))

    def translate(self, pa: int) -> int | None:
        """GPU physical address -> simulator address, or None if unmodelled."""
        for pa_base, sim_base, size in reversed(self.windows):
            if pa_base <= pa < pa_base + size:
                return sim_base + (pa - pa_base)
        return None

    def read(self, pa: int, size: int) -> bytes | None:
        sim = self.translate(pa)
        if sim is None:
            self._refuse(pa, size, "r")
            return None
        return self._read_sim(sim, size)

    def write(self, pa: int, data: bytes) -> bool:
        sim = self.translate(pa)
        if sim is None:
            self._refuse(pa, len(data), "w")
            return False
        self._write_sim(sim, data)
        return True

    def _in_ram(self, addr: int, size: int) -> bool:
        off = addr - self.ram_base
        return 0 <= off and off + size <= self.ram_size

    def _read_sim(self, sim: int, size: int) -> bytes | None:
        if not self._in_ram(sim, size):
            self._refuse(sim, size, "r-sim")
            return None
        off = sim - self.ram_base
        return bytes(self.ram[off:off + size])

    def _write_sim(self, sim: int, data: bytes) -> bool:
        if not self._in_ram(sim, len(data)):
            self._refuse(sim, len(data), "w-sim")
            return False
        off = sim - self.ram_base
        self.ram[off:off + len(data)] = data
        return True

    def copy(self, src_pa: int, dst_pa: int, size: int) -> bool:
        """Copy ``size`` bytes between GPU physical addresses.

        A transfer is counted only when both the source read and the
        destination write complete; a refused destination write leaves the
        transfer in :attr:`refused` and not in :attr:`transfers`.
        """
        data = self.read(src_pa, size)
        if data is None:
            return False
        ok = self.write(dst_pa, data)
        if ok:
            self.transfers += 1
        return ok

    def _refuse(self, pa: int, size: int, kind: str) -> None:
        self.refused += 1
        if len(self.events) < 4096:
            self.events.append({"pa": pa, "size": size, "kind": kind})

    def state(self) -> dict:
        return {
            "windows": [{"pa": p, "sim": s, "size": n} for p, s, n in self.windows],
            "transfers": self.transfers,
            "refused": self.refused,
            "events": list(self.events),
        }

    def reset(self) -> None:
        self.transfers = 0
        self.refused = 0
        self.events.clear()