Simulator

Protocol-accurate fake motors for tests and demos.

Harness

Wiring for tests and demos.

Two independent can.Bus(interface="virtual") instances on one channel: one owned by the library’s transport, one by the simulator. python-can delivers between instances on the same channel, so the real receive path runs - notifier thread, sink dispatch, routing, accepts, codec, latch - rather than a monkeypatched stand-in. That is what makes a green CI run on a laptop mean something.

SteppedSim is driven by the test (pump/step), so nothing depends on wall clock and there are no flaky sleeps.

class SteppedSim(

bus: can.BusABC,

mit_drivers: list[~cubemarspycan.sim.mit.SimMitDriver] = <factory>,

servo_drivers: list[~cubemarspycan.sim.servo.SimServoDriver] = <factory>,

dropped: int = 0,

received: list[~cubemarspycan.frame.Frame] = <factory>,

_drop_every: int = 0,

_seen: int = 0,

frozen: bool = False,

settle_s: float = 0.0003,

)

Bases: object

A fake bus segment the test advances by hand.

received: list[Frame]

Every command frame the sim saw, in order. Lets tests assert on what was sent.

settle_s: float

Yield to the notifier thread after each step.

Replies go out over a real python-can virtual bus and are delivered by a real notifier thread. A test loop that only sends and steps never releases the GIL, so that thread is starved and no feedback ever lands. A short sleep hands it the interpreter, which is also what a real control loop does while it waits for its period.

freeze() → None

Stop replying entirely, as a motor that has lost power would.

thaw() → None

Resume replying after freeze(). Idempotent.

drop_every(n: int) → None

Drop one reply in n. 0 disables.

pump() → int

Consume every pending command and emit replies. Returns frames handled.

step(dt: float) → None

Advance every plant, and emit any periodic servo status frames.

advance(dt: float) → None

One control period: handle what arrived, advance time, let the reader run.

settle() → None

Give the notifier thread a chance to deliver what was just sent.

inject(frame: Frame) → None

Put an arbitrary frame on the bus, for adversarial tests.

close() → None

Stop the notifier and shut down the simulator’s own bus.

The library’s transport is closed separately - the two ends are deliberately distinct buses so that CI exercises the real receive path.

sim_bus(

mit_drivers: list[SimMitDriver] | None = None,

servo_drivers: list[SimServoDriver] | None = None,

channel: str | None = None,

) → tuple[MotorBus, SteppedSim]

Build a started MotorBus wired to a SteppedSim.

Caller closes both; SteppedSim.close() and MotorBus.close are independent.

The plant

A minimal mechanical plant, output-side.

J * omega_dot = tau - b * omega - tau_load, integrated semi-implicitly.

It exists so the simulator produces motion that is plausible rather than scripted: a position step overshoots, velocity is genuinely non-zero, and a bad unwrapper or a sign error has something to fail against. A simulator that always reported zero velocity would pass tests that should fail.

FIRMWARE_LOOP_DT

Inner-loop period of the fake firmware, 10 kHz.

The real driver closes its impedance loop internally at tens of kilohertz off the last setpoint it received - it does not recompute only when a CAN frame arrives. Modelling it the naive way makes the simulator ring at the control period and turns every gain test into a measurement of the integrator instead of the protocol.

class Plant(

inertia: float = 0.001,

damping: float = 0.002,

position: float = 0.0,

velocity: float = 0.0,

load_torque: float = 0.0,

limit_lo: float | None = None,

limit_hi: float | None = None,

)

Bases: object

Rigid single-axis plant at the output shaft.

load_torque: float

Constant external torque, e.g. gravity on a lever.

limit_lo: float | None

Hard stop in the negative direction, rad. None for free travel.

limit_hi: float | None

Hard stop in the positive direction, rad.

step(dt: float, torque: float) → None

One explicit Euler step. Callers choose the rate; see FIRMWARE_LOOP_DT.

property at_limit: bool

Whether the position is resting on either hard stop.

True while a stop is being pushed against, which is the condition a homing routine looks for. limit_lo/limit_hi default to None, so a plant with free travel always reports False.

zero_here() → None

Shift the origin to here, carrying any travel limits with it.

A fake MIT driver

A fake AK driver speaking MIT mode.

Decodes commands with its own hand-written bit arithmetic rather than calling the library codec, so a shared bug cannot cancel itself out.

The knobs exist to test things the manual leaves open, and each one corresponds to a question the bench sequence answers:

• ScalingVariant - the manual’s pack and unpack formulas are not exact inverses, so we cannot know which the firmware uses. Our encoder must be within 1 LSB of either.

• WrapMode - whether position wraps or saturates past the field limit.

• reply_arbitration_id - the manual says “0x00 + Drive ID”, which is ambiguous.

class ScalingVariant(*values)

Bases: Enum

Which float<->uint convention the fake firmware uses.

EXACT

span / ((1<<bits) - 1), as the manual’s uint_to_float documents.

TRUNCATED

span / (1<<bits), the inverse of the manual’s float_to_uint.

class SimMitDriver(

spec: MotorSpec,

motor_id: int = 1,

plant: Plant | None = None,

*,

reply_arbitration_id: int | None = 0x00,

scaling: ScalingVariant = ScalingVariant.EXACT,

wrap_mode: WrapMode = WrapMode.WRAP,

velocity_wraps: bool = False,

temperature_c: int = 30,

fault_code: int = 0,

)

Bases: object

A driver that ignores everything until it is told to enter MIT mode.

handle(

frame: Frame,

) → Frame | None

Process one frame; return a reply, or None if it was not for us.

step(dt: float) → None

Advance by dt, running the inner loop at the firmware’s own rate.

A fake servo driver

A fake AK driver speaking servo mode over CAN.

Like the MIT sim, it decodes with its own struct calls rather than the library codec.

Two behaviours here exist because they are the traps a real bench session hits:

• status_rate_hz=0 reproduces a driver whose CAN status messages are disabled in CubeMarsTool. Nothing is wrong with the wiring, no frames ever arrive, and a naive library reports zeros forever.

• reject_permanent reproduces a single-encoder motor refusing origin mode 1, so the library’s capability check is validated end to end rather than only at its own boundary.

class SimServoDriver(

spec: MotorSpec,

motor_id: int = 1,

plant: Plant | None = None,

*,

status_rate_hz: float = 100.0,

ack_on_first_command: bool = True,

reject_permanent: bool | None = None,

temperature_c: int = 30,

fault_code: int = 0,

)

Bases: object

Servo-mode driver with a periodic status upload.

handle(

frame: Frame,

) → list[Frame]

Decode one command frame and return whatever the driver would reply with.

Decodes with its own hand-written struct unpacking rather than this library’s codec, so a shared bug cannot cancel out. Protocol-accurate for the commands it models; a pass here is not a hardware guarantee.

step(dt: float) → list[Frame]

Advance the plant and emit any due status frames.

status_frame() → Frame

One status upload, as the driver sends at its configured rate.

A rate of 0 is modelled too - see status_rate_hz - because a driver that never uploads is the most common servo-mode misconfiguration and the library has to diagnose it rather than report zeros forever.

bootloader_frame() → Frame

The 0x2C “entered servo mode” reply, function id only.

Handled as an event, never decoded as position. The manual documents this reply but no frame that causes mode entry, which is why servo mode is detected rather than commanded.