The Python SDK

manhattan-reasoning-gym (imported as mrg) has three surfaces, matching the three ways to use the platform:

Surface What it does Needs
mrg.build local synth / place-and-route reports, no board Docker
mrg.Sandbox run an untrusted agent in a locked container that can promote to silicon Docker
mrg.cloud program + drive a real ECP5 over the cloud API key

mrg.cloud: real hardware

You describe an application with three things: a design (an Amaranth .py file, or a plain Verilog .v file, whose top-level module is a Wishbone B4 slave), an optional register map (the byte offsets your design exposes), and API config.

import manhattan_reasoning_gym as mrg

class Regs(mrg.cloud.RegisterMap):
    CTRL     = 0x0000
    DATA_IN  = 0x0004
    DATA_OUT = 0x0008

app = mrg.cloud.App(
    "my_design",
    design="design.py",
    registers=Regs,
)

@app.local_entrypoint()
def main():
    app.write(Regs.DATA_IN, 0x1234)
    app.write(Regs.CTRL, 1)            # kick off
    print(hex(app.read(Regs.DATA_OUT)))

Run it with the CLI, which programs the FPGA then calls the entrypoint:

mrg run my_design.py

App

mrg.cloud.App(
    name,
    *,
    design,                    # path to the Amaranth .py or plain Verilog .v design
    top=None,                   # Verilog-only top-module disambiguator, see below
    fpga_id=None,               # for --no-program reconnect only, see below
    registers=None,              # a RegisterMap subclass (optional)
    api_key=None,                # explicit arg > $MRG_API_KEY > `mrg login`
    api_url=DEFAULT_API_URL,
    sys_clk_freq=None,            # SoC compute clock override (Hz)
    timing_target_mhz=None,        # PnR/grading timing target (MHz)
)

Creating an App registers it so the CLI can discover it, you don't export anything. If a file defines several, mrg run uses the last one.

Verilog designs. design= also accepts a plain Verilog .v file — the extension picks the language server-side, nothing else about App changes. Since a .v file has no Elaboratable to scan for, its top module is found by matching the one module whose port list has this Wishbone contract by name, width, and direction:

Port Width Direction
clk 1 input
rst 1 input
wb_cyc 1 input
wb_stb 1 input
wb_we 1 input
wb_adr 9 input
wb_dat_w 32 input
wb_sel 4 input
wb_dat_r 32 output
wb_ack 1 output

If a file has more than one module matching this contract, pass top="..." (ignored for an Amaranth design) to disambiguate; with only one match it's auto-detected. See the Verilog Hello example.

fpga_id is normally left unset. A fresh build never picks a board itself — the server claims a build slot (a network identity baked into the bitstream, independent of any physical board) and dispatches the build immediately; whichever board frees up first claims the finished bitstream and flashes it. app.fpga_id is filled in from that completed job once _program() finishes, not chosen up front. Pass fpga_id explicitly only to reconnect to a board you already have a live session on without rebuilding (mrg run --no-program --fpga-id N / App(..., fpga_id=N) with programming skipped) — on any run that does build, whatever you pass here is overwritten with the real assigned board.

Sys clock vs. timing target

A build carries two independent frequencies. Sys clock is what the SoC actually runs at (produced by the ECP5 PLL from a fixed 12 MHz input, default 50 MHz), because the PLL divides that fixed input, only certain output frequencies are realizable. Timing target is the frequency place-and-route is constrained to hit and the build is graded against; it carries no PLL restriction, so it can be any value, and defaults to the sys clock. Keeping them separate lets you ask "can this design close at 90 MHz?" without re-clocking the SoC, and grade against thresholds the PLL can't synthesize exactly. Set both via App(sys_clk_freq=..., timing_target_mhz=...) or the CLI's --sys-clk/--timing-target-mhz flags.

app.read(addr, count=1)

Read count 32-bit words starting at byte address addr. Returns a single int when count == 1, otherwise a list[int]. Programs the FPGA first if it hasn't been programmed yet.

app.write(addr, value, fixed_address=False)

Write one or more 32-bit words to byte address addr. value may be a single int or a list[int] for a burst write, which increments the address by 4 per word (loading a register array, like a RAM). Pass fixed_address=True to instead repeat addr for every word in the burst -- for a FIFO or push-register port where your design keeps its own internal write index (a common streaming-load pattern: writing a sequence of words one at a time to a single register, with the RTL auto-advancing into the next clause/slot). A plain burst would scatter those words across whatever registers happen to sit at addr+4, addr+8, ... instead of pushing them all through the one port.

app.stream()

with app:
    with app.stream() as s:
        for word in words:
            s.write(LITERAL_IN, word, fixed_address=True)
        s.write(REG_CTRL, 1)
        while not (s.read(REG_CTRL) & 1):
            pass

Opens a persistent, low-latency session for many small read/write ops. It is a context manager exposing the same write(addr, value, fixed_address=False) and read(addr, count=1) methods as App, but skips the per-call job queue: every op on app.write()/app.read() directly dispatches its own job against the cloud API and then polls for completion every 0.5s, so each individual call costs roughly that much wall-clock time no matter how small the payload is. That's fine for a handful of calls, but it dominates for a tight loop -- loading a CNF instance one literal per write, or an RL reward loop that needs to load and grade many episodes per training step. A Stream instead holds one WebSocket open for the whole with block, relayed straight through to the FPGA's Wishbone bus, and pays that connection cost once instead of once per operation.

A stream and app.write()/app.read() (or a second stream) can't be used on the same FPGA at the same time -- the bridge firmware only safely serves one open connection per board, so opening a stream holds an exclusive lock on the FPGA's link for the life of the with block.

Per-write payload limit: a burst write(addr, value, ...) is capped at roughly 2KB (~500 32-bit words). Going over it closes the connection, raised client-side as websockets.exceptions.ConnectionClosedError with the reason "FPGA is busy with a run job" -- ignore that text, it's a payload size limit, not a concurrency/session issue. If you need to send more than that, split value into chunks under ~500 words and call write() once per chunk; fixed_address=True composes fine across chunked calls.

app.release()

Release the active session, returning the board to idle. Returns the reset job_id. App is also a context manager that calls this on exit.

RegisterMap

A marker base class for your design's address map, subclass it and set integer class attributes for the byte offsets. Byte offset = 4 × word offset. Passing registers= to App is optional; it's for your own organization, read/write take raw addresses either way.

@app.local_entrypoint()

Marks the function mrg run calls after programming. Not called on plain import/python, so the same file doubles as an importable module and a runnable app.

secret(env_var)

Reads a required environment variable, raising immediately if it's unset, useful for API keys you want to fail loudly on rather than silently pass through as None.

Module-level session functions

mrg.cloud.get_session(fpga_id, api_key, api_url)      # current session info
mrg.cloud.release_session(fpga_id, api_key, api_url)  # release, returns job_id

mrg.Sandbox: locked agent execution

For running an untrusted agent: the agent gets a locked-down container with no API key and no network. Having vetted a candidate locally with mrg.build, it calls mrg.sandbox.promote(...), a file handoff on the shared workspace, and your trusted process outside the container decides whether to run it on real silicon.

import manhattan_reasoning_gym as mrg

sb = mrg.Sandbox(files=["design.py", "agent.py"])  # mock silicon unless a key is set
result = sb.run("agent.py")
for promotion in result.promotions:
    print(promotion)

mrg.Sandbox(
    files=(),
    *,
    isolation="locked",       # or "dev", or a custom SandboxProfile
    silicon="auto",           # "auto" | "cloud" | "mock" | a callable
    api_key=None,
    api_url=None,
    sys_clk_freq=None,
    guard=None,               # optional (design_bytes, report) -> reject reason | None
    image=None,
    poll_interval=0.2,
)

sb.run(entrypoint, *, timeout=1800) launches entrypoint in the container and brokers its promotes to silicon, returning a SandboxResult(returncode, stdout, stderr, promotions).

No gating by default: the framework doesn't decide whether a promote is "good enough"; the agent gates itself, and you only get a promote check if you pass guard=. The trust boundary is fixed: the container has no key and no network; promote is a file handoff on the shared workspace; only the trusted process outside the container holds the key and touches silicon.

Inside the container: mrg.sandbox.promote

The only function available to a sandboxed agent, it never holds a key or calls the orchestrator directly:

report = mrg.build.pnr("design.py")
if report.fits and report.timing_met:
    mrg.sandbox.promote("design.py", report, agent="demo-agent")

mrg.build: local synth / PnR

Fast, no-cloud, no-board build feedback. The same call works everywhere: run inside the sandbox image (toolchain present) executes in-process; run on a plain host transparently runs the pinned Docker image and parses its JSON report. Either way you get a BuildReport back.

mrg.build.synth(design, *, top=None, work=None) -> BuildReport
mrg.build.pnr(
    design, *,
    top=None,                # Verilog-only top-module disambiguator
    target_mhz=None,        # legacy alias, sets both knobs below
    sys_clk_mhz=None,
    timing_target_mhz=None,
    seed=1,
    work=None,
) -> BuildReport
  • synth: resource utilization only, fast, no timing analysis.
  • pnr: full-SoC place-and-route (Fmax, timing_met, SoC-wide utilization).
  • design accepts an Amaranth .py or plain Verilog .v file; top is the same Verilog-only disambiguator as App(top=...) (ignored for Amaranth), only needed when a .v file exposes more than one module matching the Wishbone contract.

BuildReport fields: mode, ok, scope, fits, fmax_mhz, sys_clk_mhz, target_mhz, timing_met, clock, util, synth_cells, warnings, design_hash, toolchain, log_tail, plus .to_json()/.to_dict().

See also