Realtime Bridge¶

realtime_bridge — three-tier controller interface with shared buffer and command queue.

Cluster: Core Infrastructure | Type: component | MCP Tools: None

Overview¶

Simulation-only three-tier controller interface that decouples high-level agent commands from low-level joint-control experiments. Maintains a strict 12-DOF state buffer for simulated sensor data, a priority command queue for discrete actions, and per-tier rate configuration (fast loop / slow loop / agent tier). Provides an emergency-stop latch that blocks writes and queued commands until explicitly cleared.

Outputs use the canonical reliability envelope: completion_state is one of verified, qualified-draft, or blocked-escalated; degraded outputs include warning_card, evidence, and top-level backpressure / dropped_count fields where relevant.

When to use:

• Testing how a G6 agent's high-level decisions would flow to a low-level controller at configurable rates
• Buffering simulated joint state from a fast loop while an agent processes at a slower cadence
• Exercising emergency-stop and queue-draining behavior before any hardware integration exists

Example:

from mvp.realtime_bridge import RealtimeBridgeBlock, RealtimeBridgeInput

block = RealtimeBridgeBlock(name="realtime_bridge")
block.infer(RealtimeBridgeInput(op="write_target", joint_positions=[0.0]*12))
block.infer(RealtimeBridgeInput(op="enqueue_command", command_type="position", priority=1))
result = block.infer(RealtimeBridgeInput(op="read_state"))
# result.value.joint_positions → list[float]

MCP exposure is inspection-only: read_state, get_buffer_status, get_tier_config, capabilities, and health. Control-like and mutating ops remain local-only until a separate safety policy exists.

Works well with: motor_control in simulation/test harnesses.

Public API¶

SharedStateBuffer¶

Thread-style shared buffer for 12-DOF joint state.

Constructor:

Methods:

write_positions(values: list[float]) -> None¶

write_velocities(values: list[float]) -> None¶

write_forces(values: list[float]) -> None¶

read_positions() -> list[float]¶

read_velocities() -> list[float]¶

read_forces() -> list[float]¶

set_emergency(flag: bool) -> None¶

is_emergency() -> bool¶

get_status() -> dict¶

reset() -> None¶

CommandQueue¶

Priority-sorted FIFO command queue with backpressure.

Constructor:

Methods:

enqueue(command: dict) -> bool¶

Insert sorted by (-priority, timestamp).

dequeue() -> dict | None¶

Pop highest priority command (first in sorted list).

count() -> int¶

clear() -> None¶

snapshot() -> tuple[dict, ...]¶

contains(command: dict) -> bool¶

dropped_count() -> int¶

register_subscriber(subscriber_id: str) -> None¶

unregister_subscriber(subscriber_id: str) -> bool¶

active_subscribers() -> frozenset[str]¶

RealtimeBridgeBlock(AIBlock[RealtimeBridgeInput, RealtimeBridgeOutput, dict])¶

Methods:

infer(data: RealtimeBridgeInput) -> Result[RealtimeBridgeOutput]¶

RealtimeBridgeInput¶

RealtimeBridgeOutput¶

Methods:

model_dump(**_kwargs: Any) -> dict¶

Serialize to a flat dict.