Usage Python

Python Usage

Python bindings are built with pyo3 and maturin. The Python module name is sedsprintf_rs.

Build and install

Option 1: use build.py (source) ( recommended in this repo)

./build.py python

Option 2: maturin

maturin develop

If you want a wheel:

./build.py maturin-build

Minimal example

import sedsprintf_rs as seds

DT = seds.DataType
EP = seds.DataEndpoint
RM = seds.RouterMode


def tx(bytes_buf):
    # send bytes to transport
    pass


def on_packet(pkt):
    print(pkt)


handlers = [
    (int(EP.SD_CARD), on_packet, None),
]

router = seds.Router(handlers=handlers, mode=RM.Sink)
router.add_side_serialized("RADIO", tx)
router.log_f32(ty=DT.GPS_DATA, values=[1.0, 2.0, 3.0])
router.process_all_queues()

If you need a custom monotonic source for tests or simulation, pass now_ms=.... Otherwise the router uses its internal monotonic clock on std builds. If the extension was built with the timesync feature but you do not want router-managed time sync for a particular instance, construct Router(..., timesync_enabled=False).

See python-example/main.py (source) for a more complete multi-process example. Time sync is demonstrated in python-example/timesync_example.py (source). See Time-Sync for the time sync packet flow and roles.

With timesync enabled, Router keeps an internal network clock. TIME_SYNC packets are handled internally, network_time() / network_time_ms() expose the merged current time, and source/master nodes can set partial or complete local time with set_local_network_time(...), set_local_network_date(...), and the set_local_network_* datetime helpers. For normal application loops, call router.periodic(timeout_ms) to run time sync, discovery, and queue draining together. If you need to skip time sync for a cycle while keeping the feature enabled, call router.periodic_no_timesync(timeout_ms) instead. router.poll_timesync() remains available as a lower-level non-blocking hook when you want to manage the maintenance phases manually.

With discovery enabled, both Router and Relay also expose announce_discovery() and poll_discovery(). poll_discovery() remains available as a lower-level hook when you want to manage discovery separately, while Relay.periodic(timeout_ms) bundles discovery polling and queue draining into one call. announce_discovery() still forces an immediate advertise cycle.

Logging API

The Python API exposes typed log helpers that mirror the Rust API:

• log_f32, log_i16, log_u32, etc.
• log_string for UTF-8 payloads.
• log_binary for raw bytes.

If you already have a packet or bytes, use:

• tx_serialized(bytes)
• rx_serialized(bytes)

Handlers

Handlers are registered as tuples:

(endpoint_id, handler_fn, user)

handler_fn receives (packet).

Queue processing

The router can queue RX/TX operations. If you use the queue variants, call:

• process_rx_queue()
• process_tx_queue()
• process_all_queues()
• periodic(timeout_ms)
• periodic_no_timesync(timeout_ms) for router loops that should skip time sync on that cycle

Sides

Routers use named sides (UART/CAN/RADIO/etc.). Register sides with:

• add_side_serialized(name, tx_cb)
• add_side_packet(name, tx_cb)

As of v3.0.0, side tracking is internal, so most apps call rx_serialized(bytes) without passing a side ID. Use side-aware ingress only when you need to override ingress explicitly.

Side-aware ingress:

• receive_serialized_from_side(side_id, bytes)
• receive_packet_from_side(side_id, packet)

Debugging tips

• print(pkt) uses the packet's string formatter.
• If a log call fails, check the schema for payload size/type mismatches.
• Ensure your Python environment matches the one used by maturin develop.