Motors¶

User-facing motor classes.

Note

Every class below is documented once, at the module where it is defined. from cubemarspycan import MitMotor and from cubemarspycan.motor import MitMotor both reach cubemarspycan.motor.mit.MitMotor. Documenting the re-export sites as well would register each method at two fully qualified names, which is what makes a bare :meth:`hold` ambiguous.

The bus¶

Frame routing.

One MotorBus owns one transport and fans received frames out to registered endpoints. Two properties matter:

Containment. An endpoint that raises is caught and counted; the other motors on the bus keep receiving. One motor’s bug must not silence the rest.
Lock-free reads. The endpoint list is an immutable tuple, replaced wholesale on register. The receive thread does a single attribute read to snapshot it, so registering a motor never blocks reception.

Unmatched frames are counted and sampled, which is what makes cubemars scan able to tell “nothing on the bus” from “something is there but not answering to that id”.

class Endpoint(*args, **kwargs)[source]¶

Bases: Protocol

Something that consumes frames addressed to it.

accepts(frame: Frame) → bool[source]¶: Cheap predicate. Must not raise.

on_frame(frame: Frame, rx_monotonic: float) → None[source]¶: Handle an accepted frame. Runs on the receive thread; must not raise.

class BusStats( rx_matched: int = 0, rx_unmatched: int = 0, endpoint_errors: int = 0, errors: ~collections.deque[str] = <factory>, unmatched_samples: ~collections.deque[str] = <factory>, )[source]¶

Bases: object

Routing counters. Diagnostics, not shared state.

record_endpoint_error(endpoint: object, exc: BaseException) → None[source]¶

Count and sample an exception raised by an endpoint.

Called on the receive thread, from inside the except that keeps one motor’s bug from silencing the others, so it must not raise. The sample list is bounded.

class MotorBus(transport: FrameTransport)[source]¶

Bases: object

Routes frames between a transport and a set of motors.

start() → None[source]¶

Begin receiving. Idempotent - a second call is a no-op.

Registering a motor afterwards is safe and needs no restart, since routing reads an immutable tuple that register() replaces wholesale.

close() → None[source]¶

Close the transport and stop receiving.

Delegates to the transport, whose close is required to be idempotent and non-raising, so this is safe from an error path. Endpoints stay registered: closing the bus ends reception, it does not dismantle the object.

property transport: FrameTransport¶

The injected transport. Exposed for diagnostics, not for sending.

Use send() instead, so frames are counted.

property endpoints: tuple[Endpoint, ...]¶

The registered endpoints, as an immutable snapshot.

Safe to read from any thread: register() and unregister() publish a new tuple rather than mutating this one, which is what lets the receive thread take its snapshot with a single attribute read and no lock.

register(endpoint: Endpoint) → None[source]¶: Add an endpoint. Publishes a new tuple so the receive thread never locks.

unregister(endpoint: Endpoint) → None[source]¶

Remove an endpoint. Unknown endpoints are ignored.

Compares by identity, not equality. A frame already being dispatched may still reach the endpoint: the receive thread snapshots the tuple before iterating, so removal takes effect from the next frame, not the current one.

send(frame: Frame, timeout: float | None = 0.05) → None[source]¶

Put one frame on the bus, blocking up to timeout seconds.

Called on the caller’s thread. Propagates SendFailed unchanged - a command that did not reach the motor must not look like one that did.

describe() → str[source]¶

A multi-line diagnostic summary: channel, endpoint count, and receive stats.

Includes the last few frames that matched no endpoint, which is what distinguishes “nothing is on the bus” from “something is there but not answering to that id”.

Shared endpoint behaviour¶

Shared motor plumbing: latch wiring, the update path, and the control lifecycle.

The one rule that shapes this module: nothing in the receive path raises. A fault arrives as a frame, becomes a FaultEvent in a latch, and only turns into control flow inside MotorEndpoint.update(), on the caller’s thread, after a safe-stop frame has gone out. TMotorCANControl raises RuntimeError straight from the python-can notifier thread, where the exception is swallowed and the motor keeps being commanded while faulted.

class MotorEndpoint( bus: MotorBus, motor_id: int, spec: MotorSpec, *, policy: SafetyPolicy = DEFAULT_POLICY, supply_voltage: float | None = None, )[source]¶

Bases: Generic[StateT]

One motor on one bus.

accepts(frame: Frame) → bool[source]¶

Whether this frame belongs to this motor. Runs on the receive thread.

Called for every frame on the bus, so it must be cheap and must never raise. Subclasses filter on several conditions, not just the payload’s first byte: a library that checks only that decodes any colliding frame as this motor’s state.

on_frame(frame: Frame, rx_monotonic: float) → None[source]¶

Decode one accepted frame and publish it. Runs on the receive thread.

Must never raise: a decode failure is counted, and a fault becomes latched data. Neither turns into control flow until update() runs on the caller’s thread, after a safe-stop frame has gone out.

rx_monotonic is the arrival time from time.monotonic(), which is what staleness is measured against - not the bus timestamp.

update() → StateT[source]¶

Send the staged command and return the state snapshot taken before it.

Subclasses widen this with their own optional arguments. The zero-argument form is the contract the base class relies on, in settle().

property in_control: bool¶

Whether a control() block is currently open.

Commands are refused outside one with NotInControlMode, because the driver is not in the mode that would act on them.

control( wait_s: float = 1.0, ) → Iterator[MotorEndpoint[StateT]][source]¶

Enter control mode, and guarantee leaving it.

The exit path runs even if the body raises, and it sends a safe stop before the exit frame so the motor is never left producing torque.

wait_s blocks until the first feedback frame arrives, which catches a wrong id or a dead bus at the top of the with rather than a hundred silent iterations later. Pass 0.0 to skip it, which is what the stepped simulator needs because nothing advances until the test says so.

settle( seconds: float, period: float = 0.01, ) → StateT | None[source]¶

Hold the current command for seconds, keeping feedback flowing.

Use after zero_here(), or anywhere you need to wait without letting the link go quiet. Returns the last state seen.

A bare time.sleep here sends nothing, so the motor stops replying and the next update() raises. Note that keeping the link alive is not by itself enough when the driver goes quiet - staleness is measured on received frames - which is what expect_silence() is for.

expect_silence(seconds: float) → None[source]¶

Tolerate missing feedback for seconds, starting now.

Some operations stop the driver replying for a while - zero_here() is the one that bites. Staleness is measured against the last frame received, so transmitting through the gap does not help: without this, the first update() after such an operation raises StaleFeedbackError even though nothing is wrong.

This suppresses only the fatal limit. The warning still fires, so a gap that turns out to be permanent is still visible, and a frame received after the window opened ends it early - see _in_stale_grace for why “after” rather than “fresh”.

property fault: FaultEvent | None¶: The latched fault, if any. Reading it does not consume it.

property faulted: bool¶

Whether a fault is latched, whether or not it has been raised yet.

Stays true until clear_fault(), so it survives catching the exception. Reads the latch, so it is safe from either thread and sends nothing.

clear_fault() → None[source]¶: Forget the latched fault. Call only once the cause is addressed.

property clamp_events: int¶: How many command fields have been clamped since construction.

device_info() → str[source]¶: "AK40-10-KV170 id 1" - spec name and CAN id, for messages and logs.

MIT mode¶

The user-facing MIT-mode motor.

class MitReplyMode(*values)[source]¶

Bases: Enum

Which arbitration id the driver answers on.

The manual says “0X00+Drive ID”, which is ambiguous, and firmware builds differ. So we learn it from the first plausible reply and then filter strictly on it, rather than matching on the payload’s first byte alone the way TMotorCANControl does - which lets any 8-byte frame whose first byte collides be decoded as motor state.

class MitMotor( bus: MotorBus, motor_id: int, spec: MotorSpec, *, policy: SafetyPolicy = DEFAULT_POLICY, supply_voltage: float | None = None, reply_mode: MitReplyMode = MitReplyMode.AUTO, wrap_mode: WrapMode | None = None, )[source]¶

Bases: MotorEndpoint[MitState]

An AK actuator in MIT (impedance) mode.

Commands go out as a single frame carrying five coupled fields, so they are set together through command() rather than one attribute at a time. Reading state and writing a setpoint are deliberately different operations: m.position = x would write a setpoint while m.position read feedback, and the value you read back is never the value you wrote.

accepts(frame: Frame) → bool[source]¶: Four conditions, not one. Learns the reply arbitration id on the first match.

on_frame(frame: Frame, rx_monotonic: float) → None[source]¶: Runs on the receive thread. Must never raise.

control( wait_s: float = 1.0, ) → Iterator[MitMotor][source]¶

Enter MIT mode, and guarantee a safe stop and an exit on the way out.

Narrows the base context manager’s type so with m.control() as m: yields a MitMotor. The behaviour - enter frames, optional wait for first feedback, then a zero-gain zero-torque frame followed by the exit frame in a finally - is the base class’s; see control(). wait_s=0 skips the wait, which is what a stepped simulator needs.

Stage the next command. Unspecified fields hold their previous value.

Returns whatever had to be clamped, so a saturating controller is visible rather than silently trimmed.

hold() → None[source]¶: Free-spin: zero gains, zero feed-forward torque.

brake(kd: float = 1.0) → None[source]¶: Damping only. Resists motion but does not seek a position.

Snapshot the latest state, send the staged command, return the snapshot.

The returned state is taken at the top of the call, so it predates the frame this call puts on the wire. One cycle of causality, stated rather than accidental. Calling with no arguments re-sends the staged command, which is what a fault-recovery path wants.

zero_here(*, grace_s: float = 1.5) → None[source]¶

Set the current position as zero.

The manual does not say whether this survives a power cycle, so do not rely on either behaviour.

The driver needs about a second afterwards before position is trustworthy, and it may stop replying during that time. Because staleness is measured against the last frame received, transmitting through that gap does not keep it at bay - so this calls expect_silence() for grace_s and the wait becomes routine:

m.zero_here()
m.settle(1.5)

grace_s=0 restores the strict behaviour if you would rather see the gap.

The gains are dropped for you. Zeroing moves the coordinate system, so a setpoint staged in the old frame would become an instruction to drive back to where the motor just came from. Staging zero gains is not enough on its own - hold() only stages, it does not transmit - so the zeroed command is put on the wire before the origin moves.

This transmits - a zeroed command frame, then the zero-position frame - so like update() it requires control mode. Methods that only stage (command(), hold(), brake()) do not: that is the line. Outside the with block the driver is not in MIT mode, so both frames would go to a device that is not listening and the caller would never learn.

(set_origin() is a deliberate exception: servo mode has no documented entry handshake, so “control mode” there is a library-side assertion rather than a device state, and the origin packet stands alone.)

property state: MitState | None¶: The most recent state, without sending anything.

property reply_arbitration_id: int | None¶: Which id the driver actually answers on, once learned. Worth writing down.

property staged_command: tuple[float, float, float, float, float]¶

The five staged MIT fields, output-side: (position_rad, velocity_radps, kp, kd, torque_nm).

Staged, not sent - nothing reaches the wire until update(). These are the values after clamping, so they are what will actually go out rather than what was asked for.

property decode_errors: int¶

Frames accepted for this motor that then failed to decode.

Should stay at zero. A climbing count means something is answering on this id that is not this motor, or the link is corrupting payloads.

describe() → str[source]¶

A multi-line report of the fields, the effective limits, and the provenance.

Prints both the wire field range and the effective limit for each quantity, which differ in either direction across the AK line, plus where every constant came from. Worth printing once at startup: it is the fastest way to see that a spec is incomplete before a conversion refuses mid-run.

Servo mode¶

The user-facing servo-mode motor.

class ServoMotor( bus: MotorBus, motor_id: int, spec: MotorSpec, *, policy: SafetyPolicy = DEFAULT_POLICY, supply_voltage: float | None = None, assume_mit: bool = False, )[source]¶

Bases: MotorEndpoint[ServoStatus]

An AK actuator in servo mode over CAN.

Servo mode is configured in CubeMarsTool, not commanded over CAN: the manual documents 0x2C as an “entered servo mode” reply but no frame that causes entry. So this class detects rather than asserts, and says what it observed when detection fails. TMotorCANControl guesses, sending the MIT FF..FC payload as an extended frame, where it lands as a malformed duty-cycle command.

assume_mit¶: Send the MIT exit frame on entry, for a driver left in MIT mode.

accepts(frame: Frame) → bool[source]¶

Whether this frame is servo feedback for this motor. Runs on the receive thread.

Servo framing is unambiguous, so this is exact rather than learned: extended id, the motor id in the low byte, and a function id this library handles. 0x2C and 0x09 are accepted as events and never decoded as position.

on_frame(frame: Frame, rx_monotonic: float) → None[source]¶: Runs on the receive thread. Must never raise.

control( wait_s: float = 1.0, ) → Iterator[ServoMotor][source]¶: Confirm the driver really is in servo mode, then hand control over.

command(setpoint: Setpoint) → None[source]¶: Stage a setpoint. Only one command mode is in flight at a time.

update( setpoint: Setpoint | None = None, ) → ServoStatus[source]¶

Snapshot the latest status, send the staged setpoint, return the snapshot.

As in MIT mode, the returned status is taken at the top of the call and so predates the frame this call sends.

stop() → None[source]¶: Stage zero duty. The motor free-spins.

set_origin( mode: OriginMode = OriginMode.TEMPORARY, ) → None[source]¶

Set the current position as origin.

PERMANENT writes flash and the manual restricts it to dual-encoder models. On a single-encoder motor such as the AK40-10 this raises CapabilityError and no frame is sent. TMotorCANControl sends mode 1 unconditionally, and by default.

erpm_for(output_radps: float) → float[source]¶: Output rad/s to electrical RPM. Refuses if pole pairs or gear ratio are unknown.

property status: ServoStatus | None¶

The most recent status, without sending anything. None before the first.

A persistent None while the wiring is fine almost always means the driver’s CAN status rate is 0, so it never uploads - the single most common servo-mode confusion.

property servo_mode_acknowledged: bool¶: True once a 0x2C handshake has been seen.

property bootloader_seen: bool¶: True if the driver announced a jump to its bootloader (function id 0x09).

property events: list[ServoEvent]¶: Non-status replies, in arrival order.

property staged_setpoint: Setpoint¶

The setpoint that update() will re-send if called with no argument.

Exactly one setpoint is in flight at a time: the six command types are mutually exclusive, which is why they are distinct types rather than fields.

property decode_errors: int¶: Frames accepted for this motor that then failed to decode. Should stay zero.

describe() → str[source]¶

A multi-line report of the scaling, the current limit, and the provenance.

States where the current limit comes from - datasheet peak, policy ceiling, or neither, in which case current commands are refused rather than sent with no bound. The servo current field accepts +/-60 A against this motor’s 7.3 A peak, so an unset limit is a real hazard and is reported as one.

Servo setpoints¶

Servo-mode setpoints, as value objects.

Servo mode has six mutually exclusive commands: a duty cycle, a current, a braking current, a speed, a position, and a position with a speed and acceleration profile. Only one can be in flight, and there is no meaningful way to combine them.

Making each a distinct type puts that exclusivity in the type system rather than in a mode enum the caller has to keep in step. m.update(servo.Position(90.0)) says exactly one thing; m.position = 90; m.current = 2.0 - which is how TMotorCANControl models it - says two contradictory things and silently resolves them by whichever mode flag was set last. For example:

m.update(servo.Duty(0.05))
m.update(servo.Current(1.5))
m.update(servo.Position(90.0))
m.update(servo.PositionSpeed(90.0, speed_erpm=5000, accel_erpm_s2=30000))

class Current(amps: float)[source]¶

Bases: Setpoint

q-axis current in amps. Output torque is roughly amps * Kt * gear_ratio.

to_frame( motor_id: int, spec: MotorSpec, ) → Frame[source]¶: Packet 1, SET_CURRENT. Amps times the spec’s current scale, int32.

property summary: str¶: +1.50 A, two decimals and always signed.

class CurrentBrake(amps: float)[source]¶

Bases: Setpoint

Braking current in amps, never negative. Holds position; watch the temperature.

to_frame( motor_id: int, spec: MotorSpec, ) → Frame[source]¶

Packet 2, SET_CURRENT_BRAKE. Same scaling as Current.

A negative value has no meaning here - braking current is a magnitude.

property summary: str¶: brake 0.50 A. Unsigned, because the value is a magnitude.

class Duty(value: float)[source]¶

Bases: Setpoint

Open-loop duty cycle, -1.0 to 1.0. Square-wave-like drive; no closed loop at all.

to_frame( motor_id: int, spec: MotorSpec, ) → Frame[source]¶: Packet 0, SET_DUTY. Duty times the spec’s duty scale, as int32.

property summary: str¶: duty +0.050, three decimals and always signed.

class Position(degrees: float)[source]¶

Bases: Setpoint

Target angle in degrees. The driver moves there at its configured maximum speed.

to_frame( motor_id: int, spec: MotorSpec, ) → Frame[source]¶

Packet 4, SET_POS. Degrees scaled by 1e4, as int32.

1e4, not 1e6. TMotorCANControl uses 1e6, which is a 100x error: a commanded 90 degrees arrives as 0.9.

property summary: str¶: +90.00 deg, two decimals and always signed.

class PositionSpeed(degrees: float, speed_erpm: float, accel_erpm_s2: float)[source]¶

Bases: Setpoint

Trapezoidal move: a target angle with speed and acceleration limits.

speed_erpm and accel_erpm_s2 are packed as int16 after dividing by 10, so their resolution is 10 ERPM and 10 ERPM/s^2 respectively.

to_frame( motor_id: int, spec: MotorSpec, ) → Frame[source]¶

Packet 6, SET_POS_SPD. Degrees by 1e4; speed and accel each /10.

The divide-by-ten applies to both limits before they are packed as int16, so the effective resolution is 10 ERPM and 10 ERPM/s^2. Missing it on either field is a 10x error in the move profile.

property summary: str¶: +90.00 deg at 5000 ERPM, accel 30000 - the target and both limits.

class Rpm(erpm: float)[source]¶

Bases: Setpoint

Speed in electrical RPM, not mechanical.

Use erpm_for() to convert from output rad/s, which needs the pole pair count and gear ratio and refuses if either is unknown.

to_frame( motor_id: int, spec: MotorSpec, ) → Frame[source]¶: Packet 3, SET_RPM. Electrical RPM times the spec’s rpm scale.

property summary: str¶: +3000 ERPM, no decimals. Electrical, not mechanical.

class Setpoint[source]¶

Bases: object

Base for the six servo commands.

to_frame( motor_id: int, spec: MotorSpec, ) → Frame[source]¶

Encode this setpoint as one extended CAN frame for motor_id.

The shared contract, which every subclass keeps: pure, allocates exactly one Frame, and performs no safety clamping - the wire fields are far wider than any motor (the current field is +/-60 A against the AK40-10’s 7.3 A peak), so limits belong to SafetyPolicy, not here. Raises SpecIncompleteError if the scaling needed for this command is unknown for spec.

Each subclass documents only what differs: its packet id and its scale factor.

property summary: str¶

A short human-readable form, for logs and CLI output.

Value and unit only - never the motor id, and never enough to reconstruct the frame. Each subclass states its own format.