Master Backends

Contents

Master Backends#

Note

MockMasterEx, RecordingMaster, ReplayMaster, and the BusPolicies bundle (barriers/cache/retry) have all shipped and are covered in detail below. SimMaster is alpha — its semantics today match MockMaster, with the behavioural-simulator divergence (RDL side-effects honoured natively) landing in a future batch. A handful of SoC-level helpers (soc.attach_master, soc.barrier, soc.batch, soc.trace, soc.lock) and the async_session surface remain part of the design sketch in docs/IDEAL_API_SKETCH.md §13. The shape and naming are stable; a class that is not yet implemented will not appear in the autogenerated reference at the foot of this page.

Overview#

A master is the bus binding for the generated SoC tree. It is the only component that talks to real hardware (or pretends to, for tests), and it is the piece you swap when you move from a unit test to a JTAG bring-up or from a lab notebook to a CI replay. Three principles drive the design:

  • Replaceable. The same generated SoC tree runs against MockMaster, OpenOCDMaster, SSHMaster, a simulator, or a recorded trace. Switching masters does not require code changes anywhere else.

  • Observable. Every read and every write goes through the master, so it is the right place to attach traces, coverage observers, retry policies, and recording wrappers.

  • Explicit. Bus semantics (barriers, caching, retries, batching) are first-class, named, and discoverable. Nothing important happens by accident.

Built-in masters#

Entries marked (alpha) have shipped a thin implementation but still point at a richer behavioural target — they appear in examples so the intended shape is concrete.

MockMaster#

In-memory master backed by a dict. The default for unit tests, REPL exploration, and notebook examples — every read returns the value last written (or zero), no side effects, no hooks. Reach for MockMaster when the test only needs a place to land bytes.

For tests that need register-level read/write hooks, RDL side-effect semantics (rclr clear-on-read, woclr write-1-to-clear), or bulk memory preloading, use MockMasterEx (described below) instead.

MockMasterEx#

A subclass of MockMaster that adds the surface tests actually need. Lives in peakrdl_pybind11.masters.mock_extensions and is also re-exported at peakrdl_pybind11.masters for convenience.

  • on_read(reg, fn) / on_write(reg, fn) — install per-register callbacks. reg may be a register handle (its info.address is resolved automatically) or a raw integer address.

  • preload(mem, ndarray) — bulk-seed a memory region with an iterable or numpy array of word values; the default word_size of 4 matches a 32-bit memory layout.

  • mark_rclr(reg) / mark_woclr(reg) — explicitly tag an address with read-clear-on-read or write-1-to-clear semantics. When a hook is installed against a register handle whose info.on_read is "rclr" (or info.on_write is "woclr"), the tag is inferred automatically; mark_* is the raw-address equivalent.

Pick MockMasterEx whenever a test needs to react to a write or assert on hook-observed values; pick MockMaster when in-memory storage is enough.

OpenOCDMaster#

Talks to an OpenOCD server over its TCL/RPC port for JTAG- or SWD-attached silicon. Use it for bring-up at a workstation with a debug probe wired to the target.

SSHMaster#

Tunnels register access through an SSH connection. Useful when the bus is exposed by a Linux helper running on the target SoC, on a remote tester, or on a lab machine you reach through a jump host.

CallbackMaster#

Lightweight master that forwards every transaction to user-supplied callables. Handy for ad-hoc bridging (e.g. piping to a custom UART or PCIe shim) without subclassing the base.

SimMaster (alpha)#

The intent is a bus-functional model that drives a co-simulated RTL design (cocotb, verilator, etc.) and natively honours RDL side-effects (rclr, W1C, singlepulse, …). Today the source flags it as alpha-quality: behaviour matches MockMaster plus a constructor that pre-seeds the in-memory store, so a test written for MockMaster runs against SimMaster unchanged. Tests that depend on side-effect simulation should keep using the MockMasterEx mock-with-callbacks pattern until the divergence lands in a future batch.

ReplayMaster#

Replays a recorded session against the same SoC tree. Used for regression tests, post-mortems, and deterministic CI replays of recorded bring-up runs. Strict mode is the default; mismatches raise ReplayMismatchError (see Tracing and replay).

RecordingMaster#

Wraps any other master and records every transaction to a file. The recorded log is consumed by ReplayMaster and may be either a JSON array (the default for .save("trace.json")) or NDJSON (one event per line, used for crash-safe streaming via file=).

Composing masters#

Most chips do not have a single bus. JTAG owns the debug peripherals, a memory-mapped DMA owns SRAM, and a mock fills in for absent silicon. The SoC supports multiple masters, each scoped to a region.

from peakrdl_pybind11.masters import (
    MockMaster, OpenOCDMaster, SSHMaster, SimMaster,
    ReplayMaster, RecordingMaster,
)

# Single master for the whole SoC
soc = MySoC.create(master=OpenOCDMaster("localhost:6666"))

# Per-region routing
soc = MySoC.create()
soc.attach_master(jtag, where="peripherals.*")
soc.attach_master(mem_master, where="ram")
soc.attach_master(MockMaster(), where=lambda node: node.info.is_external)
soc.attach_master(other, where=(0x4000_0000, 0x4000_FFFF))

The where= selector accepts:

  • a glob string against the node path ("peripherals.*"),

  • a callable (node) -> bool for arbitrary predicates,

  • an address-range tuple (lo, hi) over absolute addresses.

Masters must serve disjoint regions; the SoC raises RoutingError on overlap and on any address that no attached master claims.

Transactions as objects#

The master accepts a list of reified transactions, so you can script the bus without going through the typed tree (handy for tools that have an address but not a node, or for re-issuing a recorded sequence).

from peakrdl_pybind11 import Read, Write, Burst

txns = [
    Read(0x4000_1000),
    Write(0x4000_1004, 0x42),
    Burst(0x4000_2000, count=128, op="read"),
]
results = soc.master.execute(txns)

For typed-tree users, soc.batch() coalesces a sequence of operations into one master command when the backend supports queuing:

with soc.batch() as b:
    b.uart.control.write(1)
    b.uart.data.write(0x55)
# All transactions sent at exit; one round-trip if the master can queue.

The BusPolicies bundle#

Barrier ordering, read caching, and retry/backoff are bundled into a single peakrdl_pybind11.runtime.bus_policies.BusPolicies container that wraps a master’s read / write once and keeps the three policies in a defined order: cache lookup → retry loop → barrier fence → master. The SoC layer attaches a bundle to every master at attach time, but the bundle is also reachable directly when you need to configure a policy from your own code:

from peakrdl_pybind11.runtime import attach_master_extension
from peakrdl_pybind11.masters import MockMaster

master = MockMaster()
bundle = attach_master_extension("bus_policies", master)
# bundle is a BusPolicies instance with three policy objects:
bundle.barriers     # BarrierPolicy — fences and "auto" raw-fencing
bundle.cache        # CachePolicy   — TTL slots + cached_window blocks
bundle.retry        # RetryPolicy   — backoff, on_disconnect, give-up

The bundle can also be imported directly when the registry seam is not needed:

from peakrdl_pybind11.runtime.bus_policies import BusPolicies

bundle = BusPolicies(master=master)

Note that BusPolicies mutates the master in place — it captures the original read / write and replaces them with delegations through the policy chain. Wrapping the same master twice would double-wrap; if you need to re-install, reconstruct the master first. The next three sections walk through what each policy on the bundle does in practice; see The Bus Layer for the conceptual depth on barriers, caching, and retry semantics.

Barriers and fences#

Many masters queue or coalesce writes; some buses post writes asynchronously. A barrier forces all in-flight writes on the relevant master(s) to drain before the next read.

# Per-master scope (default and usually what you want)
soc.uart.barrier()         # masters serving the uart subtree
soc.master.barrier()       # the explicit single-master form
soc.barrier()              # masters used at the current call site

# SoC-wide scope (rare; needed when read on master B depends on a
# write that went out via master A)
soc.barrier(scope="all")
soc.global_barrier()       # alias of the above; reads better in scripts

Auto-barrier policy is set per SoC. The default flushes the relevant master only before a read-after-write on the same master, since flushing every master on every barrier is expensive when masters serve disjoint regions.

soc.set_barrier_policy("auto")         # default: same-master raw fences
soc.set_barrier_policy("none")         # opt out, faster, you fence yourself
soc.set_barrier_policy("strict")       # barrier before every read AND write
soc.set_barrier_policy("auto-global")  # auto-fences across all masters (paranoid)

Read coalescing and cache policy#

Tight polling loops often hammer the same register. The master can cache recent reads when it is safe to do so.

soc.uart.status.cache_for(50e-3)    # within 50 ms, return cached value
soc.uart.status.invalidate_cache()

with soc.cached(window=10e-3):
    # any read inside this block may serve from a 10 ms cache window
    run_polling_loop()

Caching is refused for side-effecting reads (info.on_read != NONE) and for volatile registers (info.is_volatile). The master does not attach a cache to those nodes even if asked, and an explicit cache_for(...) call on a destructive register raises.

Bus error recovery#

Real hardware misbehaves. The master applies a configurable retry policy and surfaces a structured error when it gives up.

soc.master.set_retry_policy(
    retries=3,
    backoff=0.05,           # exponential backoff base, in seconds
    on=("timeout", "nack"), # underlying errors that trigger retry
    on_giveup="raise",      # also: "log", "panic"
)

# Per-call override
soc.uart.control.read(retries=10)

# Reconnect hook (e.g. re-attach JTAG, replay last N transactions)
soc.master.on_disconnect(lambda m: m.reconnect())

When the policy gives up, the call raises BusError carrying the failed transaction, the retry count, the underlying exception, and the node path / absolute address. That is enough for a CI run to triage why it died.

Tracing and replay#

Every transaction passes through the master, which makes the master the right place to record and replay sessions.

with soc.trace() as t:
    soc.uart.control.write(0x42)
    soc.uart.status.read()
print(t)
# 2 transactions, 8 bytes
#   wr  @0x40001000  0x00000042   (uart.control)
#   rd  @0x40001004  -> 0x00000001 (uart.status)

t.save("session.json")
soc2 = MySoC.create(master=ReplayMaster.from_file("session.json"))

RecordingMaster(inner, file=None)#

For long-running lab sessions, wrap the live master in RecordingMaster so the transcript lands on disk as it happens. Two on-disk shapes are supported and chosen by extension:

  • JSON array — a single document, written by .save("trace.json") (the default for any extension other than .ndjson / .jsonl). Compact and easy to load with one json.loads; not crash-safe (the file is rewritten in one go).

  • NDJSON — one event per line, used by both the streaming file= argument and .save("trace.ndjson") / .save("trace.jsonl"). Each event is flushed per-op so a long-running session that crashes still leaves a usable trace, and concatenating multiple NDJSON traces yields a valid combined log.

from peakrdl_pybind11.masters import RecordingMaster

# Stream NDJSON to disk as the run unfolds.
soc.attach_master(RecordingMaster(jtag, file="run.ndjson"))

# Or capture in memory, then dump as JSON or NDJSON depending on path.
rec = RecordingMaster(jtag)
soc.attach_master(rec)
...
rec.save("trace.json")    # JSON array
rec.save("trace.ndjson")  # NDJSON, one event per line

The recording is a stream of the same transaction objects the master already speaks, so it round-trips exactly under ReplayMaster.

ReplayMaster.from_file(path)#

ReplayMaster.from_file() auto-detects the format (NDJSON vs JSON array) and is strict by default: every requested transaction must match the next recorded event in order (op kind + address + width; writes also compare value). A mismatch raises ReplayMismatchError — a distinct exception class from BusError so tests can catch “the recording doesn’t match what the driver did” without conflating it with “the bus itself errored”. Pass strict=False for the looser mode that serves matching reads from the recording and silently ignores extras.

from peakrdl_pybind11.masters import ReplayMaster, ReplayMismatchError

replay = ReplayMaster.from_file("run.ndjson")  # strict mode
soc = MySoC.create(master=replay)
try:
    run_driver_sequence(soc)
except ReplayMismatchError as exc:
    # exc.expected is the recorded Event (or None if exhausted);
    # exc.actual is a {"op", "address", "width", ["value"]} dict.
    print(f"recording diverged: expected {exc.expected}, got {exc.actual}")

# Loose mode for a quick "what would the driver do here?" probe.
loose = ReplayMaster.from_file("run.ndjson", strict=False)

Mock master with hooks#

The hook-driven, side-effect-aware mock is MockMasterEx — MockMaster itself is a plain dict-backed store and does not have on_read / on_write / preload. Tests that want to express behaviour rather than bookkeeping should import the extension class:

import numpy as np
from peakrdl_pybind11.masters import MockMasterEx

mock = MockMasterEx()
mock.on_read(soc.uart.intr_status, lambda addr: 0b101)
mock.on_write(soc.uart.data,       lambda addr, val: stdout.append(val))
mock.preload(soc.ram, np.arange(1024, dtype=np.uint32))

soc.attach_master(mock)

The mock honours the clear-on-read (rclr) and write-one-to-clear (woclr) semantics declared in RDL. When you pass a register handle whose info.on_read is "rclr" (or info.on_write is "woclr") to on_read / on_write, the side-effect tag is inferred automatically — a register declared rclr clears under the mock just as it would on silicon. When you only have a raw integer address, mock.mark_rclr(addr) and mock.mark_woclr(addr) apply the tags explicitly.

Concurrency#

The master holds a re-entrant lock by default; multi-threaded callers do not tear up shared state. For sequences that must not be interleaved even with other transactions on the same master, take the lock explicitly:

with soc.lock():
    soc.dma.config.write(...)
    soc.dma.start.pulse()
    # nothing else hits this master until the block exits

The async dual surface exposes aread, awrite, and amodify on every node. It is opened with a session context so cancellation, back-pressure, and reconnection are well-defined:

async with soc.async_session():
    await soc.uart.control.awrite(0x1)
    v = await soc.uart.status.aread()

Custom masters#

To add a new backend, inherit from MasterBase and implement the read/write surface. The base class handles the routing, retry, observation, and lock plumbing — the subclass only needs to know how to move bytes.

from peakrdl_pybind11.masters import MasterBase

class MyCustomMaster(MasterBase):
    def read(self, address, width):
        ...  # one bus read; raise BusError on failure
        return value

    def write(self, address, width, value):
        ...  # one bus write; raise BusError on failure

    # Optional: implement burst, peek, barrier, reconnect, on_disconnect
    # if your transport supports them. Default fallbacks loop / raise
    # NotSupportedError as appropriate.

Reference#

Master interfaces for register access

class peakrdl_pybind11.masters.AccessOp(address, value=0, width=4)[source]#

Bases: object

One register-access operation used by batched read_many / write_many.

For reads, value is ignored and conventionally zero; for writes it carries the value to write. Mirrors the C++ AccessOp struct exposed by every generated module.

Parameters:
address: int#
value: int = 0#
width: int = 4#
__init__(address, value=0, width=4)#
Parameters:
Return type:

None

class peakrdl_pybind11.masters.CallbackMaster(read_callback=None, write_callback=None, read_many_callback=None, write_many_callback=None)[source]#

Bases: MasterBase

Callback-based Master

Allows custom read/write functions to be provided. Optionally accepts batched callbacks (read_many_callback / write_many_callback) that receive the full list of AccessOp in one call, amortizing the Python-side dispatch across N ops. If unset, read_many() / write_many() fall back to looping the single-op callbacks.

Parameters:
__init__(read_callback=None, write_callback=None, read_many_callback=None, write_many_callback=None)[source]#

Initialize with optional callbacks

Parameters:
Return type:

None

read(address, width)[source]#

Read using callback

Return type:

int

Parameters:
write(address, value, width)[source]#

Write using callback

Return type:

None

Parameters:
read_many(ops)[source]#

Batched read. Uses read_many_callback if set, else loops.

Return type:

list[int]

Parameters:

ops (Sequence[AccessOp])

write_many(ops)[source]#

Batched write. Uses write_many_callback if set, else loops.

Return type:

None

Parameters:

ops (Sequence[AccessOp])

class peakrdl_pybind11.masters.Event[source]#

Bases: TypedDict

One recorded bus transaction.

The schema is the same on the wire (NDJSON / JSON) and in memory. Keys mirror sketch §13.6:

  • op — "read" or "write".

  • address — absolute address (int).

  • value — the value read (for read) or written (for write).

  • width — register width in bytes.

  • timestamp — monotonic seconds since the recorder started.

op: str#
address: int#
value: int#
width: int#
timestamp: float#
class peakrdl_pybind11.masters.MasterBase[source]#

Bases: ABC

Base class for Master interfaces

Masters provide the actual communication mechanism for reading/writing registers.

Note

For in-memory test/mock fixtures, prefer the C++ MockMaster and CallbackMaster classes shipped inside every generated module (e.g. my_soc.MockMaster()). They live entirely in C++, skip the pybind11 trampoline, and are noticeably faster on a tight register loop than wrapping a Python subclass of MasterBase via wrap_master. Subclass MasterBase only when the master truly has to be implemented in Python (sockets, REST APIs, exotic hardware glue) — at which point per-access overhead is dominated by I/O anyway.

Extension points (no-op defaults; sibling units of the API overhaul override these in runtime/bus_policies.py and friends):

abstractmethod read(address, width)[source]#

Read a value from the given address

Parameters:

  • address (int) – Absolute address to read from

  • width (int) – Width of the register in bytes

Return type:

int

Returns:

Value read from the address

abstractmethod write(address, value, width)[source]#

Write a value to the given address

Parameters:

  • address (int) – Absolute address to write to

  • value (int) – Value to write

  • width (int) – Width of the register in bytes

Return type:

None

read_many(ops)[source]#

Batched read. Default impl loops single-op read().

Subclasses can override with a fast path that performs one transport round-trip for N ops (e.g. one socket exchange instead of N).

Return type:

list[int]

Parameters:

ops (Sequence[AccessOp])

write_many(ops)[source]#

Batched write. Default impl loops single-op write().

Return type:

None

Parameters:

ops (Sequence[AccessOp])

peek(address, width)[source]#

Non-snooping read.

Default implementation forwards to read(). Sibling unit runtime/bus_policies.py overrides this on caching masters to avoid promoting cache lines.

Return type:

int

Parameters:
barrier()[source]#

Flush any in-flight operations.

Default: no-op. Sibling unit runtime/bus_policies.py provides per-master implementations (e.g. waiting for outstanding write completions on JTAG masters). The soc.barrier(scope="all") SoC-wide form fans this out across every attached master.

Return type:

None

on_disconnect(callback)[source]#

Register callback to fire when the transport drops.

Default: store on a per-instance list. Concrete masters that actually have a transport (SSH, OpenOCD, JTAG) override this to wire the callback into their disconnect detection.

Sibling extension point: runtime/bus_policies.py.

Return type:

None

Parameters:

callback (Callable[[], None])

set_retry_policy(**kwargs)[source]#

Configure transient-error retry behaviour.

Default: stash kwargs on a per-instance dict. Sibling unit runtime/bus_policies.py consumes the dict in its retry/backoff wrapper.

Return type:

None

Parameters:

kwargs (Any)

cache_get(address)[source]#

Look up address in the master’s read cache.

Default: None (no cache). Sibling unit runtime/bus_policies.py overrides on caching masters.

Return type:

int | None

Parameters:

address (int)

cache_set(address, value)[source]#

Insert (address, value) into the master’s read cache.

Default: pass-through (no cache).

Return type:

None

Parameters:
class peakrdl_pybind11.masters.MockMaster[source]#

Bases: MasterBase

Mock Master for testing without hardware

Simulates register storage in memory

__init__()[source]#
Return type:

None

read(address, width)[source]#

Read from simulated memory, masked to width bytes.

Return type:

int

Parameters:
write(address, value, width)[source]#

Write to simulated memory

Return type:

None

Parameters:
reset()[source]#

Clear all stored values

Return type:

None

read_many(ops)[source]#

Batched read; touches the dict directly without per-op dispatch.

Return type:

list[int]

Parameters:

ops (Sequence[AccessOp])

write_many(ops)[source]#

Batched write; touches the dict directly without per-op dispatch.

Return type:

None

Parameters:

ops (Sequence[AccessOp])

class peakrdl_pybind11.masters.MockMasterEx[source]#

Bases: MockMaster

A MockMaster with read/write hooks and rclr/woclr semantics.

Existing MockMaster semantics (read / write from the backing dict) are preserved as the fallback path — addresses without a registered hook still hit the in-memory store.

__init__()[source]#
Return type:

None

on_read(reg_or_addr, fn)[source]#

Register a read callback.

fn(address) -> int is invoked whenever this address is read; the returned value is masked to the access width and returned to the caller. If reg_or_addr is a register handle and its info.on_read is "rclr", the address is auto-marked for rclr so the underlying value clears after every read.

Return type:

None

Parameters:
on_write(reg_or_addr, fn)[source]#

Register a write callback.

fn(address, value) -> None is invoked for every write at this address. The write also propagates to the in-memory store (so a subsequent read returns the latest value) unless the address is marked woclr, in which case write-1-to-clear semantics apply.

Return type:

None

Parameters:
mark_rclr(reg_or_addr)[source]#

Mark an address (or register) as read-clear-on-read.

Return type:

None

Parameters:

reg_or_addr (int | object)

mark_woclr(reg_or_addr)[source]#

Mark an address (or register) as write-1-to-clear.

Return type:

None

Parameters:

reg_or_addr (int | object)

preload(mem_or_addr, values, *, word_size=4)[source]#

Seed a memory region with an iterable / ndarray of word values.

mem_or_addr may be a memory handle (info.address is used as the base) or a raw int base address. Each element of values is stored at base + i * word_size. The default word_size of 4 matches the typical 32-bit memory layout in generated code; override for byte- or 64-bit-addressed memories.

Return type:

None

Parameters:
read(address, width)[source]#

Read at address, dispatching to the hook if registered.

After a successful read, if the address is marked rclr, the backing storage is cleared so the next non-hook read sees zero.

Return type:

int

Parameters:
write(address, value, width)[source]#

Write value at address, applying woclr if marked.

Hooks observe the caller’s requested value (not the post-woclr storage), so test capture lists reflect intent rather than derived state.

Return type:

None

Parameters:
read_many(ops)[source]#

Batched read; touches the dict directly without per-op dispatch.

Return type:

list[int]

Parameters:

ops (Sequence[AccessOp])

write_many(ops)[source]#

Batched write; touches the dict directly without per-op dispatch.

Return type:

None

Parameters:

ops (Sequence[AccessOp])

reset()[source]#

Clear stored values. Hook registrations and rclr/woclr marks are intentionally preserved so a fixture can be reused across test phases without re-arming every hook.

Return type:

None

class peakrdl_pybind11.masters.OpenOCDMaster(host='localhost', port=6666, timeout=5.0)[source]#

Bases: MasterBase

Master interface using OpenOCD TCL server

Connects to OpenOCD’s TCL interface for reading/writing memory

Parameters:
__init__(host='localhost', port=6666, timeout=5.0)[source]#

Initialize OpenOCD connection

Parameters:

  • host (str) – OpenOCD server host

  • port (int) – OpenOCD TCL server port (default 6666)

  • timeout (float) – Socket timeout in seconds

Return type:

None

read(address, width)[source]#

Read memory via OpenOCD

Parameters:

  • address (int) – Memory address to read

  • width (int) – Width in bytes

Return type:

int

Returns:

Value read from memory

write(address, value, width)[source]#

Write memory via OpenOCD

Parameters:

  • address (int) – Memory address to write

  • value (int) – Value to write

  • width (int) – Width in bytes

Return type:

None

halt()[source]#

Halt the target

Return type:

None

resume()[source]#

Resume the target

Return type:

None

reset(halt=False)[source]#

Reset the target

Return type:

None

Parameters:

halt (bool)

close()[source]#

Close the connection

Return type:

None

class peakrdl_pybind11.masters.RecordingMaster(inner, file=None, flush='event')[source]#

Bases: MasterBase

Master that records every read/write to an event log.

Parameters:

  • inner (MasterBase) – The wrapped master that actually services transactions.

  • file (str | Path | None) – Optional path. When set, every event is appended to the file as a single JSON document on its own line (NDJSON). Streaming this way means a long-running session that crashes still leaves a usable trace on disk. Use save() to dump the in-memory log as a JSON array if you prefer a single-file artefact.

  • flush (Union[Literal['event', 'never'], int]) – When file is set, controls how aggressively the streaming file is flushed. "event" (default) flushes after every recorded op — safest, but ~6.7 us/op overhead on top of the bus op. "never" only flushes on close() / __exit__ — fastest, but a hard crash loses the unflushed tail. A positive int N flushes every N events. close() and __exit__ always flush any remaining buffer regardless of policy.

__init__(inner, file=None, flush='event')[source]#
Parameters:
Return type:

None

events: list[Event]#
close()[source]#

Flush and close the streaming file if one was opened.

Always flushes any buffered events before closing, regardless of the flush policy — otherwise flush="never" would lose the tail of the trace on the cm exit.

Return type:

None

read(address, width)[source]#

Read a value from the given address

Parameters:

  • address (int) – Absolute address to read from

  • width (int) – Width of the register in bytes

Return type:

int

Returns:

Value read from the address

write(address, value, width)[source]#

Write a value to the given address

Parameters:

  • address (int) – Absolute address to write to

  • value (int) – Value to write

  • width (int) – Width of the register in bytes

Return type:

None

read_many(ops)[source]#

Batched read. Default impl loops single-op read().

Subclasses can override with a fast path that performs one transport round-trip for N ops (e.g. one socket exchange instead of N).

Return type:

list[int]

Parameters:

ops (Sequence[AccessOp])

write_many(ops)[source]#

Batched write. Default impl loops single-op write().

Return type:

None

Parameters:

ops (Sequence[AccessOp])

save(path)[source]#

Write the in-memory event log to path.

Format is selected by extension: .ndjson / .jsonl writes one event per line (the same shape as the streaming file= output); anything else writes a single JSON array.

Return type:

None

Parameters:

path (str | Path)

class peakrdl_pybind11.masters.ReplayMaster(events, strict=True)[source]#

Bases: MasterBase

Master that replays a previously recorded trace.

Reads return the recorded value; writes are accepted (and optionally compared in strict mode). The recording is loaded from a JSON array or NDJSON file; both formats produced by RecordingMaster.save() round-trip.

Parameters:

  • events (Sequence[Event]) – The recorded events. Use from_file() for the common case of loading from disk.

  • strict (bool) – When True (default), every transaction must match the next recorded event in order. When False, reads serve matching addresses from the recording and unmatched ops are silently ignored — useful for shorter/longer scripts that want to share a recording.

__init__(events, strict=True)[source]#
Parameters:
Return type:

None

events: list[Event]#
classmethod from_file(path, strict=True)[source]#

Load a recording from path.

The format is auto-detected: NDJSON (one event per line) or a single JSON array. RecordingMaster.save writes whichever shape matches the path’s extension.

Return type:

ReplayMaster

Parameters:
read(address, width)[source]#

Read a value from the given address

Parameters:

  • address (int) – Absolute address to read from

  • width (int) – Width of the register in bytes

Return type:

int

Returns:

Value read from the address

write(address, value, width)[source]#

Write a value to the given address

Parameters:

  • address (int) – Absolute address to write to

  • value (int) – Value to write

  • width (int) – Width of the register in bytes

Return type:

None

exception peakrdl_pybind11.masters.ReplayMismatchError(expected, actual, message=None)[source]#

Bases: Exception

Raised by ReplayMaster (strict mode) when a requested transaction does not match the next recorded event.

Parameters:
Return type:

None

__init__(expected, actual, message=None)[source]#
Parameters:
Return type:

None

class peakrdl_pybind11.masters.SSHMaster(host, username=None, key_file=None, tool='devmem')[source]#

Bases: MasterBase

Master interface using SSH for remote memory access

Uses devmem or similar tools on the remote system

Parameters:

  • host (str)

  • username (str | None)

  • key_file (str | None)

  • tool (str)

__init__(host, username=None, key_file=None, tool='devmem')[source]#

Initialize SSH connection

Parameters:

  • host (str) – SSH host to connect to

  • username (str | None) – SSH username (optional, uses current user if not specified)

  • key_file (str | None) – Path to SSH private key file (optional)

  • tool (str) – Memory access tool on remote system (default: “devmem”)

Return type:

None

Note

Password authentication is not supported. Use SSH keys via key_file (or your ssh-agent / ~/.ssh/config).

read(address, width)[source]#

Read memory via SSH using devmem

Parameters:

  • address (int) – Memory address to read

  • width (int) – Width in bytes

Return type:

int

Returns:

Value read from memory

write(address, value, width)[source]#

Write memory via SSH using devmem

Parameters:

  • address (int) – Memory address to write

  • value (int) – Value to write

  • width (int) – Width in bytes

Return type:

None

class peakrdl_pybind11.masters.SimMaster(state=None, *, soc=None)[source]#

Bases: MockMaster

In-memory master with optional RDL side-effect simulation.

Parameters:

  • state (Mapping[int, int] | None) – Mapping of address -> value to seed the in-memory store. Copied so the caller can safely mutate the original. None (the default) starts with empty state, which makes SimMaster a drop-in replacement for MockMaster.

  • soc (Any) – When provided, walk the SoC tree at construction time and build the per-register side-effect models. Equivalent to constructing without soc= and calling attach_soc() immediately afterwards.

Without an attached SoC, every read/write goes straight through to the underlying dict (i.e. MockMaster semantics). With an SoC attached, register accesses honour the field-level RDL side effects described in the module docstring.

__init__(state=None, *, soc=None)[source]#
Parameters:
Return type:

None

attach_soc(soc)[source]#

Build (or rebuild) the side-effect map from soc.

Existing memory contents are preserved. A subsequent call replaces the models – handy for tests that swap in different SoC shapes against a single master instance.

Return type:

None

Parameters:

soc (Any)

read(address, width)[source]#

Read from simulated memory, masked to width bytes.

Return type:

int

Parameters:
write(address, value, width)[source]#

Write to simulated memory

Return type:

None

Parameters:
read_many(ops)[source]#

Batched read; touches the dict directly without per-op dispatch.

Return type:

list[int]

Parameters:

ops (Sequence[AccessOp])

write_many(ops)[source]#

Batched write; touches the dict directly without per-op dispatch.

Return type:

None

Parameters:

ops (Sequence[AccessOp])

Contents