1 files changed, 190 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/memory.py b/debug/risc-v-gsp-rm-vm/greeneye/memory.py
new file mode 100644
index 0000000..23d9ca0
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/memory.py
@@ -0,0 +1,190 @@
+"""Memory inspection helpers: read/write, hexdump, search, snapshots, diffs."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .errors import GreenEyeError
+
+
+@dataclass
+class Diff:
+    addr: int
+    old: bytes
+    new: bytes
+
+    @property
+    def size(self) -> int:
+        return len(self.old)
+
+
+class Memory:
+    """A convenience facade over an RSP client (and optional symbol DB)."""
+
+    def __init__(self, client, symbols=None, recorder=None):
+        self.client = client
+        self.symbols = symbols
+        self.recorder = recorder
+
+    # -- basic access ------------------------------------------------------
+
+    def read(self, addr: int, size: int) -> bytes:
+        data = self.client.read_mem(addr, size)
+        if self.recorder is not None:
+            self.recorder.event("mem_read", addr=addr, size=size)
+        return data
+
+    def write(self, addr: int, data: bytes | int | str) -> None:
+        if isinstance(data, int):
+            data = bytes([data & 0xFF])
+        elif isinstance(data, str):
+            data = data.encode("utf-8")
+        self.client.write_mem(addr, data)
+        if self.recorder is not None:
+            self.recorder.event("mem_write", addr=addr, size=len(data), data=data.hex())
+
+    def read_u8(self, addr: int) -> int:
+        return self.read(addr, 1)[0]
+
+    def read_u16(self, addr: int) -> int:
+        return int.from_bytes(self.read(addr, 2), "little")
+
+    def read_u32(self, addr: int) -> int:
+        return int.from_bytes(self.read(addr, 4), "little")
+
+    def read_u64(self, addr: int) -> int:
+        return int.from_bytes(self.read(addr, 8), "little")
+
+    def write_u32(self, addr: int, value: int) -> None:
+        self.write(addr, value.to_bytes(4, "little"))
+
+    def write_u64(self, addr: int, value: int) -> None:
+        self.write(addr, value.to_bytes(8, "little"))
+
+    def read_cstring(self, addr: int, max_len: int = 256) -> str:
+        return self.read(addr, max_len).split(b"\x00", 1)[0].decode("utf-8", "replace")
+
+    def read_ptr(self, addr: int, width: int = 8) -> int:
+        return int.from_bytes(self.read(addr, width), "little")
+
+    # -- display -----------------------------------------------------------
+
+    def hexdump(self, addr: int, size: int, width: int = 16, annotate: bool = True) -> str:
+        data = self.read(addr, size)
+        lines: list[str] = []
+        if annotate and self.symbols is not None:
+            sym = self.symbols.resolve(addr)
+            if sym is not None:
+                lines.append(f"; {sym.label(addr)} (source={sym.source or '?'})")
+        for offset in range(0, len(data), width):
+            chunk = data[offset : offset + width]
+            hex_part = " ".join(f"{b:02x}" for b in chunk)
+            ascii_part = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
+            lines.append(f"{addr + offset:08x}  {hex_part:<{width * 3 - 1}}  |{ascii_part}|")
+        return "\n".join(lines)
+
+    def hexdump_annotated(
+        self,
+        addr: int,
+        size: int,
+        struct_schema=None,
+        width: int = 16,
+    ) -> str:
+        """Hexdump plus a decoded structure overlay when a schema is given."""
+        lines = [self.hexdump(addr, size, width=width, annotate=True)]
+        if struct_schema is not None:
+            from .structs import decode, render
+
+            values = decode(struct_schema, self.read(addr, struct_schema.size), base=addr)
+            lines.append("")
+            lines.append(render(struct_schema, values))
+        return "\n".join(lines)
+
+    # -- search ------------------------------------------------------------
+
+    def search(
+        self,
+        addr: int,
+        size: int,
+        pattern: bytes | int,
+        *,
+        mask: bytes | None = None,
+        max_hits: int = 64,
+        chunk: int = 0x10000,
+    ) -> list[int]:
+        if isinstance(pattern, int):
+            pattern = bytes([pattern & 0xFF])
+        if not pattern:
+            raise GreenEyeError("empty search pattern")
+        if mask is not None and len(mask) != len(pattern):
+            raise GreenEyeError("mask length must match pattern length")
+        hits: list[int] = []
+        offset = 0
+        overlap = len(pattern) - 1
+        carry = b""
+        while offset < size and len(hits) < max_hits:
+            want = min(chunk, size - offset)
+            blob = carry + self.read(addr + offset - len(carry), want + len(carry)) if carry else self.read(addr + offset, want)
+            base = addr + offset - len(carry)
+            for index in range(0, len(blob) - len(pattern) + 1):
+                candidate = blob[index : index + len(pattern)]
+                if mask is None:
+                    if candidate == pattern:
+                        hits.append(base + index)
+                else:
+                    if all((candidate[i] & mask[i]) == (pattern[i] & mask[i]) for i in range(len(pattern))):
+                        hits.append(base + index)
+                if len(hits) >= max_hits:
+                    break
+            carry = blob[-overlap:] if overlap else b""
+            offset += want
+        return hits
+
+    def search_u32(self, addr: int, size: int, value: int, **kw) -> list[int]:
+        return self.search(addr, size, value.to_bytes(4, "little"), **kw)
+
+    def search_string(self, addr: int, size: int, text: str, **kw) -> list[int]:
+        return self.search(addr, size, text.encode("utf-8"), **kw)
+
+    # -- capture / diff ----------------------------------------------------
+
+    def snapshot(self, addr: int, size: int, name: str = "region") -> bytes:
+        data = self.client.read_mem(addr, size)
+        if self.recorder is not None:
+            path = self.recorder.snapshot(name, addr, data)
+            self.recorder.event("snapshot", label=name, addr=addr, size=size, path=path)
+        return data
+
+    @staticmethod
+    def diff(old: bytes, new: bytes, base: int = 0) -> list[Diff]:
+        length = min(len(old), len(new))
+        changed = [i for i in range(length) if old[i] != new[i]]
+        if not changed:
+            return []
+        # coalesce adjacent bytes into runs
+        runs: list[Diff] = []
+        start = changed[0]
+        prev = changed[0]
+        for index in changed[1:]:
+            if index == prev + 1:
+                prev = index
+                continue
+            runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1]))
+            start = prev = index
+        runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1]))
+        return runs
+
+    @staticmethod
+    def render_diff(diffs: list[Diff], symbols=None, max_lines: int = 64) -> str:
+        lines: list[str] = []
+        for diff in diffs[:max_lines]:
+            old = " ".join(f"{b:02x}" for b in diff.old[:8])
+            new = " ".join(f"{b:02x}" for b in diff.new[:8])
+            label = ""
+            if symbols is not None:
+                sym = symbols.resolve(diff.addr)
+                if sym is not None:
+                    label = f"  ; {sym.label(diff.addr)}"
+            lines.append(f"{diff.addr:08x}  {old:<24} -> {new:<24}{label}")
+        if len(diffs) > max_lines:
+            lines.append(f"... {len(diffs) - max_lines} more changed runs")
+        return "\n".join(lines)