path: root/debug/risc-v-gsp-rm-vm/sim/devices/base.py

"""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"

    #: Opt-in hook: when True the registry forwards the faulting PC to
    #: ``read``/``write`` as ``pc=`` (observer windows record it; every other
    #: device keeps the original two-/three-argument call unchanged).
    wants_pc = False

    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
            if win.device.wants_pc:
                value = win.device.read(addr - win.base, size, pc=pc)
            else:
                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.wants_pc:
                handled = win.device.write(addr - win.base, size, value, pc=pc)
            else:
                handled = win.device.write(addr - win.base, size, value)
            if handled:
                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]