-
Notifications
You must be signed in to change notification settings - Fork 1
Usage Python
Python bindings are built with pyo3 and maturin. The Python module name is sedsprintf_rs.
Option 1: use build.py (source | mirror) ( recommended in this repo)
./build.py python
Option 2: maturin
maturin develop
If you want a wheel:
./build.py maturin-build
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.
See python-example/main.py (source | mirror) for a more complete multi-process example. Time sync is demonstrated in python-example/timesync_example.py (source | mirror). 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.
Call router.poll_timesync() from your main loop to let the router queue any due announce/request
traffic, then run the normal queue processing methods. The call is non-blocking and returns
whether it queued a time-sync packet during that poll.
The Python API exposes typed log helpers that mirror the Rust API:
-
log_f32,log_i16,log_u32, etc. -
log_stringfor UTF-8 payloads. -
log_binaryfor raw bytes.
If you already have a packet or bytes, use:
tx_serialized(bytes)rx_serialized(bytes)
Handlers are registered as tuples:
(endpoint_id, handler_fn, user)
handler_fn receives (packet).
The router can queue RX/TX operations. If you use the queue variants, call:
process_rx_queue()process_tx_queue()process_all_queues()
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)
-
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.