Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/session-api.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`) |
Expand Down
2 changes: 1 addition & 1 deletion src/simantic/_rust.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
24 changes: 24 additions & 0 deletions src/simantic/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand All @@ -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]

Expand Down
19 changes: 17 additions & 2 deletions src/simantic/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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]:
Expand Down Expand Up @@ -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)
Expand Down
Loading