diff --git a/docs/session-api.md b/docs/session-api.md index e0f3f5e..d4824e5 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -63,7 +63,7 @@ one. `timeout` is wall-clock effort, not virtual time — assert on | method | returns | | --- | --- | | `time` | elapsed virtual seconds | -| `read_uart()` / `uart_records()` | text since the last read / `[{t, machine, label, text}]` | +| `read_uart()` / `read_uart_bytes()` / `uart_records()` | text / raw bytes since the last read / `[{t, machine, label, text, bytes}]` | | `frames()` | `[{t, machine, label, protocol, direction, summary, id, data}]` — SPI/I2C/CAN/BLE/Ethernet | | `logs()` | `[{t, level, source, message}]` — unhandled registers, model warnings | | `interrupts()` | `[{t, machine, direction, exception, name}]` (with `trace_interrupts=True`) | diff --git a/src/simantic/_rust.py b/src/simantic/_rust.py index 8d20522..6815cb6 100644 --- a/src/simantic/_rust.py +++ b/src/simantic/_rust.py @@ -93,7 +93,7 @@ def _advance(self, seconds: float) -> list[dict]: # The engine hands back runs of bytes, not one entry per byte: the # per-object boundary cost is what dominates a chatty UART. fresh = [{"t": t, "machine": self.machines[0], "label": label, - "text": bytes(data).decode("latin-1")} + "text": bytes(data).decode("latin-1"), "bytes": bytes(data)} for t, label, data in self._s.take_uart()] self._records["uart"].extend(fresh) return fresh diff --git a/src/simantic/engine.py b/src/simantic/engine.py index 41521b1..9444e5f 100644 --- a/src/simantic/engine.py +++ b/src/simantic/engine.py @@ -102,6 +102,29 @@ def engine_dir(explicit: str | os.PathLike[str] | None = None, *, fetch: bool = ) +def _resolve_plugin_assemblies_from(d: Path) -> None: + """Let the runtime find Renode plugin assemblies (e.g. Microsoft.Dynamic, + IronPython) that aren't in sim.deps.json. + + Renode loads some assemblies dynamically (a platform's PythonPeripheral, + for one) rather than through Simantic.Core's own dependency graph, so + they never make it into the CLI's deps.json. The `sim` executable finds + them anyway because a framework-dependent apphost falls back to probing + its own directory; pythonnet's manual coreclr host does not get that + fallback, so a plain AssemblyLoadContext.Resolving hook does it here. + """ + from System import AppDomain # type: ignore[import-not-found] + from System.IO import File as NetFile, Path as NetPath # type: ignore[import-not-found] + from System.Reflection import Assembly # type: ignore[import-not-found] + + def handler(_sender, args): + name = str(args.Name).split(",")[0] + candidate = NetPath.Combine(str(d), name + ".dll") + return Assembly.LoadFrom(candidate) if NetFile.Exists(candidate) else None + + AppDomain.CurrentDomain.AssemblyResolve += handler + + @cache def load(explicit: str | os.PathLike[str] | None = None): """Host the .NET runtime and import Simantic.Core. Returns the Session namespace.""" @@ -122,6 +145,7 @@ def load(explicit: str | os.PathLike[str] | None = None): if str(d) not in sys.path: sys.path.append(str(d)) + _resolve_plugin_assemblies_from(d) clr.AddReference("Simantic.Core") import Simantic.Core.Emulation.Session as session_ns # type: ignore[import-not-found] diff --git a/src/simantic/session.py b/src/simantic/session.py index ec7098a..4b1790b 100644 --- a/src/simantic/session.py +++ b/src/simantic/session.py @@ -88,7 +88,7 @@ def _bytes(net_bytes) -> bytes: def _uart(r) -> dict: - return {"t": r.T, "machine": r.Machine, "label": r.Label, "text": r.Text} + return {"t": r.T, "machine": r.Machine, "label": r.Label, "text": r.Text, "bytes": _bytes(r.Bytes)} def _frame(r) -> dict: @@ -282,8 +282,18 @@ def read_uart(self, from_start: bool = False) -> str: self._pending.clear() return "".join(r["text"] for r in recs) + def read_uart_bytes(self, from_start: bool = False) -> bytes: + """Raw bytes the firmware sent since the last read (or ever), byte-accurate. + + Prefer this over `read_uart()` for binary protocols (e.g. UBX, MAVLink): + `read_uart()`'s `str` round-trips through an encoding, this does not. + """ + recs = self.uart_records(from_start) + self._pending.clear() + return b"".join(r["bytes"] for r in recs) + def uart_records(self, from_start: bool = False) -> list[dict]: - """Timestamped UART records: {t, machine, label, text}.""" + """Timestamped UART records: {t, machine, label, text, bytes}.""" return self._records("uart", from_start) def frames(self, from_start: bool = False) -> list[dict]: @@ -423,6 +433,11 @@ def _add_machine(self, spec, name: str, repl, mcu, overlay, elf, symbols_elf) -> if overlay is not None: raise ValueError("overlay= applies to mcu=, not repl=") sm = spec.AddMachine(name, str(self._base / repl), elf_path) + # A ready .repl is loaded as-is, so relative `using` lines resolve + # against its own directory; only a .replx template needs the + # engine's render step (which writes to a temp path and would + # otherwise break those relative references). + sm.RenderPlatform = str(repl).endswith(".replx") elif os.environ.get(MCU_LIB_ENV): platform = platform_path(mcu, self._base / overlay if overlay else None, self._work) sm = spec.AddMachine(name, str(platform), elf_path)