Transport

transport.

The protocols

Transport protocols.

Deliberately CAN-shaped. A serial link is a byte stream with its own framing and CRC, and forcing both under one abstraction now would produce something that fits neither. The seam that keeps servo-over-serial possible for v1.1 is one level up: the motor classes never touch a Frame transport directly, they talk to a link object.

class FrameSink(*args, **kwargs)

Bases: Protocol

Called on the receive thread for every frame. Must never raise.

class FrameTransport(*args, **kwargs)

Bases: Protocol

A bidirectional CAN frame link.

send(

frame: Frame,

timeout: float | None = ...,

) → None

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

Called on the caller’s thread, from inside the control loop, so the blocking window is the loop’s jitter budget: a socketcan send is microseconds, an slcan send over USB serial is 0.5-2 ms.

Raises SendFailed rather than returning a status or swallowing the error. A command that did not reach the motor is not a detail the caller can be left to infer.

add_sink(sink: FrameSink) → None

Register a receiver. Sinks must be added before start().

Every registered sink is called on the receive thread for every frame, so a sink must never raise and must not block - see FrameSink. An implementation is expected to contain and count an exception rather than let it kill reception for the other sinks.

start() → None

Begin receiving. Idempotent: calling it twice must not start a second reader.

Until this is called, frames may be dropped by the underlying driver. Sinks registered afterwards are not guaranteed to see earlier frames.

close() → None

Stop receiving and release the link. Idempotent, and safe after a failed start().

Must not raise: it runs from __exit__ and from error-recovery paths, where another exception is usually already unwinding.

python-can

python-can transport.

The bus is injected. CanTransport(bus) takes any can.BusABC you have already constructed, so socketcan, slcan, gs_usb, PCAN, Kvaser and the virtual backend all work without this module knowing about them. CanTransport.open() is sugar over the common cases and is never the only path.

Nothing here shells out to change anything. There is exactly one subprocess call - a read-only ip -details -json link show used for diagnostics, never on the data path, never with sudo, and tolerant of ip being absent. TMotorCANControl instead runs os.system('sudo /sbin/ip link set can0 up ...') from a singleton’s __new__, which hard-codes the interface and the bitrate, requires root, only works on Linux, and leaves no seam to inject a bus - which is why none of its thirty defects had a test.

DEFAULT_BITRATE

The AK drivers use 1 Mbit/s. The manual says changing it is not recommended.

parse_bitrate(text: str) → int

"1M" -> 1000000, "500K" -> 500000, "1000000" -> 1000000.

class TransportStats(

rx: int = 0,

tx: int = 0,

rx_ignored: int = 0,

tx_failed: int = 0,

errors: ~collections.deque[tuple[str,

str]] = <factory>,

tx_durations: ~collections.deque[float] = <factory>,

)

Bases: object

Counters worth looking at when something is not working.

Mutable on purpose: this is diagnostics, not state that crosses a thread boundary as a snapshot.

rx_ignored: int

Error and remote frames, which carry no payload for us.

record_error(where: str, exc: BaseException) → None

Count and sample an exception, tagged with where it happened.

where is one of "frame-build", "sink", "notifier" or "send". Called from the receive thread’s except blocks, so it must not raise; the sample list is bounded.

tx_percentiles() → dict[str, float]

p50/p95/max of send() wall time, in milliseconds.

Worth watching on slcan: a send is an ASCII line over USB CDC, typically 0.5-2 ms with scheduling spikes into the tens of milliseconds. That is what decides whether a given loop rate is realistic on a given adapter.

class CanTransport(bus: BusABC, *, owns_bus: bool = False)

Bases: Listener

Wraps an injected can.BusABC and fans received frames out to sinks.

classmethod open(

url: str,

**kwargs: Any,

) → CanTransport

Build a transport from scheme:channel[@bitrate].

socketcan:can0 (Linux, production), slcan:/dev/tty.usbmodem1101@1M, gs_usb:0@1M, virtual:test.

classmethod virtual(

channel: str = 'cubemars',

) → CanTransport

An in-process bus. Used by the simulator and by CI on every platform.

add_sink(sink: FrameSink) → None

Register a receiver. Sinks must be added before start().

start() → None

Start the python-can notifier thread. Idempotent while open.

Sinks registered after this point are still called - the sink list is read per frame - but they will not see frames that already arrived.

close() → None

Stop the notifier and shut the bus down. Idempotent, and never raises.

Only shuts down the underlying can.BusABC if this transport created it: an injected bus belongs to the caller. Safe to call after a failed start().

property bus: BusABC

The wrapped can.BusABC.

Exposed so an application can reach backend-specific features this wrapper does not model. Send through send() instead, so frames are counted and failures are wrapped.

property channel_info: str

The backend’s own description of the channel, for diagnostics and messages.

Free-form and backend-specific - useful to a human, not something to parse.

send(

frame: Frame,

timeout: float | None = 0.05,

) → None

Put one frame on the bus.

Raises SendFailed rather than swallowing it.

set_filters(filters: Any | None) → None

Install hardware/kernel receive filters.

On socketcan these are applied in the kernel, so the receive thread is not woken for traffic belonging to other nodes. Opt-in: a wrong filter drops frames silently, which is a worse failure than a few wasted wakeups.

on_message_received(msg: Message) → None

python-can’s receive hook. Runs on the notifier thread and never raises.

Stamps arrival from time.monotonic() here, not from msg.timestamp: the bus timestamp is epoch-based and on some backends comes from the driver with an unrelated origin, which would make staleness meaningless. The bus value is passed along for logs.

Error and remote frames are counted and dropped rather than decoded. Every sink is called inside its own except BaseException - a sink raising KeyboardInterrupt must not take the receive thread down and silence every motor on the bus.

on_error(exc: Exception) → None

python-can’s own notifier error hook.

class LinkStatus(up: bool | None, bitrate: int | None, state: str | None)

Bases: object

Everything doctor reports about one interface, from a single ip call.

read_link_status(interface: str) → LinkStatus

All three link facts from one ip invocation.

The three single-fact readers below each run their own probe, which is right for standalone use and wasteful in a loop: doctor was forking ip up to three times per interface for three keys of one JSON object, multiplying the 2 s timeout by three.

read_socketcan_bitrate(interface: str) → int | None

The kernel’s configured bitrate, or None if it cannot be determined.

python-can cannot set a socketcan bitrate - ip link does - so the only honest check is to read back what the interface is actually running at. A virtual interface has no bit timing at all, and returns None.

socketcan_link_flags(interface: str) → list[str]

Link flags for interface, e.g. ["NOARP", "UP", "LOWER_UP"].

Empty if the interface does not exist or cannot be read. Never raises.

socketcan_is_up(interface: str) → bool | None

Whether interface is administratively up. None means “cannot tell”.

Reads the UP flag, not operstate. A virtual CAN interface has no carrier, so it reports state UNKNOWN however healthy it is, while real CAN hardware reports state UP. Anything that keys off operstate will call a working vcan interface down.

False also covers “there is no such interface”. A definitive absence is an answer; reporting it as “cannot tell” hides a typo’d interface name behind a shrug.

None is returned only when neither sysfs nor ip could be consulted at all - no /sys, no iproute2, a slim container, a BusyBox rootfs, macOS. Reporting that as “down” is how a healthy interface gets blamed, which is the failure this replaced.

sysfs is tried first so a host without iproute2 still gets a real answer.

read_socketcan_state(interface: str) → str | None

The CAN controller’s error state, e.g. ERROR-ACTIVE or BUS-OFF.

Worth surfacing: a controller in ERROR-PASSIVE or BUS-OFF usually means nothing is acknowledging its frames - no other powered node, or missing termination - which looks identical to a software fault from the application’s side.