cawlib 0.1.0

A high-performance Rust-powered Python library for CAN bus communication and FOC motor control in robotics

cawlib

A high-performance Rust-powered Python library for CAN bus communication and FOC motor control in robotics. Built with PyO3 and Tokio for blazing-fast async I/O, precise timing, and effortless concurrency.

Features

• CAN Bus Communication — SocketCAN blocking/async APIs; opens the interface at 1 Mbps automatically when needed (ip link, may prompt for sudo)
• FOC Motor Control — Broadcast-style control of up to 4 motors per CAN bus with current/speed/position modes
• High-Precision Timers — Three tiers of timing precision, from ~1 ms down to ~1 μs
• Async Task Execution — Run Python functions as background tasks with cancellation and result retrieval
• Lifecycle Management — Global task manager with graceful shutdown and signal handling (SIGINT/SIGTERM)

Requirements

• OS: Linux with SocketCAN (can / vcan driver)
• Python: >= 3.8
• Hardware: A CAN-capable network device (e.g. USB-CAN); the interface name (e.g. can0) must exist in the system. You do not need to run ip link set … up by hand first — see below.

Installation

pip install cawlib

CAN interface (automatic setup)

When you create cawlib.CanDevice, CanDeviceNonBlocking, or Motor, the library checks whether the named interface is UP at 1 Mbit/s. If not, it runs ip link to bring the interface down, set type can with bitrate 1000000, and bring it up again. If that fails without elevated privileges, it retries with sudo (you may be prompted).

• The interface must already exist (driver loaded, correct name). If ip link show can0 fails, create the device first (udev, modprobe, or your distro’s tools).
• The default bitrate is fixed at 1 Mbps in the library; change the constant in code if you need another rate.

Quick Start

import cawlib

# Create a motor controller on CAN bus
motor = cawlib.Motor("can0", host=0x100)

# Add motor devices (up to 4 per bus)
d1 = motor.add_device(slot_index=0, node_id=0x201, mode="speed")
d2 = motor.add_device(slot_index=1, node_id=0x202, mode="speed")

# Set target speeds and broadcast
d1.set_target(10.0)   # 10 rad/s
d2.set_target(-5.0)   # -5 rad/s
motor.broadcast()

# Run a periodic control loop at 100 Hz
@cawlib.timer(100.0)
def control_loop():
    motor.broadcast()
    feedback = d1.get_watch_data()
    print(f"pos={feedback['position']:.2f}, speed={feedback['speed']:.2f}")

handle = control_loop()

# Block until Ctrl+C, then gracefully shut down
cawlib.block_all(timeout_ms=5000)

API Reference

CAN Devices

CanDevice(interface: str)

Blocking CAN device for synchronous communication.

can = cawlib.CanDevice("can0")
can.send(0x100, [0x01, 0x02, 0x03])
frame_id, data = can.receive()
print(can.interface())

Method	Description
send(id: int, data: list[int])	Send a CAN frame
receive() -> tuple[int, bytes]	Receive a CAN frame (blocking)
interface() -> str	Get CAN interface name

CanDeviceNonBlocking(interface: str)

Non-blocking CAN device with background I/O.

can = cawlib.CanDeviceNonBlocking("can0")
can.send(0x100, [0x01, 0x02])
result = can.try_receive()           # Returns None if no frame available
result = can.receive_timeout(1000)   # Wait up to 1000 ms

Method	Description
send(id: int, data: list[int])	Non-blocking send
try_receive() -> tuple[int, bytes] | None	Non-blocking receive
receive_timeout(timeout_ms: int) -> tuple[int, bytes] | None	Receive with timeout
interface() -> str	Get CAN interface name

Motor Control

Motor(interface: str, host=None, host_id=None)

Broadcast motor controller supporting up to 4 devices per CAN bus. FOC broadcast and DR v1 control frames use the host-side CAN ID (firmware can_host_id). Pass the ID as the second positional argument, or as keyword host or host_id (same meaning; do not pass both).

motor = cawlib.Motor("can0", host=0x100)
motor = cawlib.Motor("can0", 0x100)
motor = cawlib.Motor("can0", host_id=0x100)

# Add devices in different control modes
d1 = motor.add_device(slot_index=0, node_id=0x201, mode="current")
d2 = motor.add_device(slot_index=1, node_id=0x202, mode="speed")
d3 = motor.add_device(slot_index=2, node_id=0x203, mode="position")

# Control
d1.set_target(1.5)     # 1.5 A
d2.set_target(10.0)    # 10 rad/s
d3.set_target(3.14)    # 3.14 rad
motor.broadcast()

# Read feedback
data = d1.get_watch_data()
# {"position": 0.0, "speed": 0.0, "current": 1.5}

Method	Description
add_device(slot_index: int, node_id: int, mode: str) -> Device	Enter DR config session, set FOC, leave session (no Flash, no encoder zeroing). May raise OSError on send failure
set_target(slot_index: int, value: float)	Set target for a slot
broadcast()	Send FOC broadcast frame (same host_id)
get_watch_data(slot_index: int) -> dict | None	Get feedback data for a slot
host_id, host	Host CAN ID for outbound control / DR (same value)
enter_configure_broadcast()	DR v1: enter DR config session (dest 0xFFFF; 0x73 then 0x74, same dest and CTL byte 6; XOR checksum on byte 7)
exit_configure_broadcast()	DR v1: leave DR config session (session group code 0)
calibrate_broadcast()	DR v1: encoder electrical zeroing (must be in DR config session); blocks ~200 ms after send
set_foc_mode_broadcast(mode: str)	DR v1: set FOC mode (broadcast)

Device

Individual motor unit within a broadcast group.

Method / Property	Description
set_target(value: float)	Set target value (unit depends on mode)
get_target() -> float	Get current target
get_watch_data() -> dict	Get feedback: {"position", "speed", "current"}
slot_index	Slot index (0–3)
node_id	Node / feedback CAN ID (matches firmware can_node_id)
enter_configure()	DR v1: enter DR config session (unicast; 0x73 then 0x74; same dest and CTL byte 6; checksum byte 7)
exit_configure()	DR v1: leave DR config session (session group code 0)
start_calibration()	DR v1: encoder electrical zeroing; must be in DR config session. blocks ~200 ms after send
set_foc_mode(mode: str)	DR v1: set FOC mode

Control Modes and Units:

Mode	Unit	Resolution	Range
current	A	0.01 A	±327.67 A
speed	rad/s	0.1 rad/s	±3276.7 rad/s
position	rad	0.01 rad	±327.67 rad

Timers

@timer(frequency_hz: float)

Standard timer decorator using Tokio async runtime. Resolution ~1–10 ms.

@cawlib.timer(10.0)  # 10 Hz
def update():
    print("tick")

handle = update()
print(handle.actual_frequency)  # Achieved frequency
handle.stop()

@precision_timer(frequency_hz: float, realtime: bool = False)

High-precision timer. Two modes:

• realtime=False (default): Busy-wait loop, ~10–100 μs resolution
• realtime=True: Dedicated OS thread, ~1–10 μs resolution (CPU-intensive)

@cawlib.precision_timer(1000.0)  # 1 kHz high-precision
def fast_loop():
    motor.broadcast()

handle = fast_loop()

TimerHandle

Returned when calling a timer-decorated function.

Method / Property	Description
stop()	Stop the timer
is_running() -> bool	Check if running
frequency	Target frequency (Hz)
actual_frequency	Achieved average frequency (Hz)
tick_count	Number of ticks elapsed
elapsed_secs	Elapsed time in seconds
precision_mode	"normal", "high", or "realtime"
task_id	Task ID

Async Tasks

@run_async

Run a Python function as a background async task.

@cawlib.run_async
def long_task():
    import time
    time.sleep(5)
    return 42

handle = long_task()
handle.wait(timeout_ms=10000)
print(handle.try_result())  # 42

AsyncHandle

Method / Property	Description
is_done() -> bool	Task finished (success or failure)
is_completed() -> bool	Completed successfully
is_cancelled() -> bool	Task was cancelled
cancel()	Request cancellation
try_result() -> Any | None	Get result if done
wait(timeout_ms: int | None = None)	Block until done or timeout
task_id	Task ID

Task Management

get_task_manager() -> TaskManager

Access the global task manager for monitoring and lifecycle control.

tm = cawlib.get_task_manager()
print(tm.status())
# {"running_timers": 2, "running_tasks": 1, "stopped": 0, "total": 3}

tm.stop_all()
tm.wait_all(timeout_ms=5000)

Method / Property	Description
running_count	Number of running tasks
total_count	Total registered tasks
timer_count	Running timers
async_task_count	Running async tasks
stop_all()	Stop all tasks
wait_all(timeout_ms: int | None = None)	Wait for all tasks to finish
shutdown(timeout_ms: int | None = None)	Stop all and wait
cleanup()	Remove finished tasks
status() -> dict	Get status summary

Module-Level Functions

Function	Description
shutdown_all(timeout_ms=None)	Stop all tasks and wait for completion
block_all(timeout_ms=None)	Block until SIGINT/SIGTERM, then gracefully shut down
block_until(wait_ms=None, shutdown_timeout_ms=None)	Block until timeout or signal, then shut down

Example: Complete Motor Control Loop

import cawlib

# Setup
motor = cawlib.Motor("can0", host=0x100)
d1 = motor.add_device(slot_index=0, node_id=0x201, mode="speed")
d2 = motor.add_device(slot_index=1, node_id=0x202, mode="speed")

target_speed = 5.0

# High-frequency control loop
@cawlib.precision_timer(500.0)  # 500 Hz
def control():
    d1.set_target(target_speed)
    d2.set_target(-target_speed)
    motor.broadcast()

# Low-frequency monitoring
@cawlib.timer(2.0)  # 2 Hz
def monitor():
    w1 = d1.get_watch_data()
    w2 = d2.get_watch_data()
    print(f"D1: speed={w1['speed']:.1f} rad/s, current={w1['current']:.2f} A")
    print(f"D2: speed={w2['speed']:.1f} rad/s, current={w2['current']:.2f} A")

# Start everything
ctrl_handle = control()
mon_handle = monitor()

# Block until Ctrl+C
cawlib.block_all(timeout_ms=3000)

License

Proprietary - All rights reserved.