1 files changed, 263 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/sim/devices/base.py b/debug/risc-v-gsp-rm-vm/sim/devices/base.py
new file mode 100644
index 0000000..877109c
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/devices/base.py
@@ -0,0 +1,263 @@
+"""Device-model core: address-range registry and device base classes.
+
+The registry decodes a guest MMIO address to at most one modelled device
+window (``base <= addr < base + size``) and asks that device for read/write
+semantics.  Modeled windows take precedence over the lenient device-map /
+unknown fallback in :class:`sim.machine.Machine`; accesses inside a known
+window that the device does not model are *not* silently zeroed here -- they
+return ``unhandled`` so the machine still logs them as unknowns.
+
+Every modelled access is recorded in :attr:`DeviceRegistry.accesses` with the
+PC, width, direction and value so a session can explain exactly what the
+device model served.  All state is exposed as JSON-serialisable data
+(:meth:`Device.state` / :meth:`DeviceRegistry.snapshot`) so snapshot support
+can be layered on later without changing this API.
+
+Evidence rule (SIMULATOR-WORK-BOUNDARY / docs/35): a register is only given
+semantics when it is backed by at least one of
+
+* a hardware register/field header (offset, width, access, reset/field docs);
+* a firmware/source use (the code that reads or writes it);
+* an observed access pattern (address, width, direction, PC, value sequence).
+
+Each device class documents the evidence kind for every register it models.
+Anything else stays ``unhandled`` and is reported by the machine's unknown
+log; we never invent a read-back value.
+"""
+from __future__ import annotations
+
+import json
+
+#: Schema tag for :meth:`DeviceRegistry.snapshot`.
+SNAPSHOT_SCHEMA = "greeneye-sim-devices/1"
+
+
+class Device:
+    """One modelled register block at one or more address windows.
+
+    Subclasses implement :meth:`read`/:meth:`write` (guest side, side effects
+    allowed) and :meth:`peek` (debugger side, no side effects).  Returning
+    ``None`` from ``read``/``peek`` or ``False`` from ``write`` means "not
+    modelled"; the registry then reports the access as unhandled so it is
+    still logged by the caller.
+    """
+
+    name = "device"
+
+    def __init__(self, base: int = 0, size: int = 0):
+        self.base = base
+        self.size = size
+
+    # -- guest-visible semantics ------------------------------------------
+
+    def read(self, offset: int, size: int) -> int | None:
+        return None
+
+    def write(self, offset: int, size: int, value: int) -> bool:
+        return False
+
+    # -- debugger-visible semantics (no side effects) ----------------------
+
+    def peek(self, offset: int, size: int) -> int | None:
+        return self.read(offset, size)
+
+    # -- serialisable state ------------------------------------------------
+
+    def state(self) -> dict:
+        return {}
+
+    def load_state(self, data: dict) -> None:
+        pass
+
+    def reset(self) -> None:
+        self.load_state({})
+
+    # -- helpers -----------------------------------------------------------
+
+    @staticmethod
+    def _mask(size: int) -> int:
+        return (1 << (8 * size)) - 1
+
+    def __repr__(self) -> str:  # pragma: no cover - cosmetic
+        return f"{type(self).__name__}({self.name} @ {self.base:#x}+{self.size:#x})"
+
+
+class Window:
+    """One (base, size) -> :class:`Device` mapping."""
+
+    __slots__ = ("base", "size", "device")
+
+    def __init__(self, base: int, size: int, device: Device):
+        self.base = base
+        self.size = size
+        self.device = device
+
+    def contains(self, addr: int, size: int) -> bool:
+        return self.base <= addr and addr + size <= self.base + self.size
+
+    def to_dict(self) -> dict:
+        return {
+            "base": self.base,
+            "size": self.size,
+            "device": self.device.name,
+        }
+
+
+class DeviceRegistry:
+    """Ordered set of device windows plus an access log.
+
+    Lookup is last-registered-first so a narrow window may override a wider
+    one without hiding it: when the last matching window does not implement the
+    access (``read`` returns ``None`` / ``write`` returns ``False``), the
+    registry continues to the next matching window (safe fallback).  If every
+    matching window refuses, the access is recorded once as unhandled.
+    Accesses are matched by exact containment (``addr + size`` inside the
+    window) so a straddling access is never half-served.
+    """
+
+    def __init__(self, *, max_access_log: int = 100_000):
+        self.windows: list[Window] = []
+        self.devices: list[Device] = []
+        self.services: list[object] = []
+        self.dma = None  # optional DmaTranslator aperture seam
+        self.accesses: list[dict] = []
+        self.dropped_accesses = 0
+        self.unhandled = 0
+        self.max_access_log = max(0, int(max_access_log))
+
+    def add_service(self, service) -> object:
+        """Attach a non-window service (e.g. an IRQ router) for state export."""
+        self.services.append(service)
+        return service
+
+    # -- construction ------------------------------------------------------
+
+    def add(self, device: Device, base: int | None = None, size: int | None = None) -> Device:
+        """Register ``device`` for one window (defaults: device.base/size)."""
+        b = device.base if base is None else base
+        s = device.size if size is None else size
+        if s <= 0:
+            raise ValueError(f"device window size must be positive, got {s}")
+        if device not in self.devices:
+            self.devices.append(device)
+        self.windows.append(Window(b, s, device))
+        return device
+
+    def device_by_name(self, name: str) -> Device | None:
+        for dev in self.devices:
+            if dev.name == name:
+                return dev
+        return None
+
+    # -- lookup ------------------------------------------------------------
+
+    def _candidates(self, addr: int, size: int):
+        """Matching windows, last-registered-first (fallback order)."""
+        for win in reversed(self.windows):
+            if win.contains(addr, size):
+                yield win
+
+    def _record(self, addr: int, size: int, write: bool, value: int | None,
+                handled: bool, pc: int) -> None:
+        entry = {
+            "addr": addr,
+            "size": size,
+            "rw": "w" if write else "r",
+            "handled": bool(handled),
+            "pc": pc,
+        }
+        if write and value is not None:
+            entry["value"] = value & ((1 << (8 * size)) - 1)
+        if handled and not write:
+            entry["value"] = value
+        if len(self.accesses) < self.max_access_log:
+            self.accesses.append(entry)
+        else:
+            self.dropped_accesses += 1
+
+    def read(self, addr: int, size: int, pc: int = 0) -> tuple[bool, int]:
+        """Guest read: ``(handled, value)``.  Unhandled accesses are logged.
+
+        The last matching window is asked first; a window that does not model
+        the offset defers to the next matching window.
+        """
+        matched = False
+        for win in self._candidates(addr, size):
+            matched = True
+            value = win.device.read(addr - win.base, size)
+            if value is not None:
+                self._record(addr, size, False, value, True, pc)
+                return True, value
+        self.unhandled += 1
+        if matched:
+            self._record(addr, size, False, None, False, pc)
+        return False, 0
+
+    def write(self, addr: int, size: int, value: int, pc: int = 0) -> bool:
+        """Guest write: True when handled.  Unhandled accesses are logged.
+
+        The last matching window is asked first; a window that does not model
+        the offset defers to the next matching window.
+        """
+        matched = False
+        for win in self._candidates(addr, size):
+            matched = True
+            if win.device.write(addr - win.base, size, value):
+                self._record(addr, size, True, value, True, pc)
+                return True
+        self.unhandled += 1
+        if matched:
+            self._record(addr, size, True, value, False, pc)
+        return False
+
+    def peek(self, addr: int, size: int) -> bytes | None:
+        """Debugger read of device space: bytes, or None when unmapped."""
+        for win in self._candidates(addr, size):
+            value = win.device.peek(addr - win.base, size)
+            if value is not None:
+                return int(value).to_bytes(size, "little")
+        return None
+
+    # -- state -------------------------------------------------------------
+
+    def state(self) -> dict:
+        return {
+            "schema": SNAPSHOT_SCHEMA,
+            "devices": [{"name": d.name, "base": d.base, "size": d.size,
+                         "state": d.state()} for d in self.devices],
+            "services": [{"name": getattr(s, "name", type(s).__name__),
+                          "state": s.state()} for s in self.services],
+            "counters": {
+                "accesses": len(self.accesses),
+                "dropped_accesses": self.dropped_accesses,
+                "unhandled": self.unhandled,
+            },
+        }
+
+    def load_state(self, data: dict) -> None:
+        if data.get("schema") != SNAPSHOT_SCHEMA:
+            raise ValueError(f"not a {SNAPSHOT_SCHEMA} snapshot")
+        for entry in data.get("devices", []):
+            dev = self.device_by_name(entry.get("name", ""))
+            if dev is not None:
+                dev.load_state(entry.get("state", {}))
+        counters = data.get("counters", {})
+        self.dropped_accesses = int(counters.get("dropped_accesses", 0))
+        self.unhandled = int(counters.get("unhandled", 0))
+
+    def snapshot_json(self) -> str:
+        return json.dumps(self.state(), sort_keys=True)
+
+    def reset(self) -> None:
+        self.accesses.clear()
+        self.dropped_accesses = 0
+        self.unhandled = 0
+        for dev in self.devices:
+            dev.reset()
+        for service in self.services:
+            reset = getattr(service, "reset", None)
+            if callable(reset):
+                reset()
+
+    def describe(self) -> list[dict]:
+        return [w.to_dict() for w in self.windows]