CANopen

Overview

pyrbs includes a dedicated CANopen layer under pyrbs.canbus.canopen and the high-level rbs.canopen handler on the main pyrbs.PyRbs instance. The package provides protocol helpers in Python with runtime bindings delegated to pyrbs_core.

The CANopen layer covers these protocol areas for restbus simulation and device integration:

NMT for node state control
Heartbeat monitoring and boot-up handling
Object Dictionary access
SDO client/server transfers
PDO mapping and transport
EMCY event handling
Sysvar bindings between CANopen objects and PyRbs variables

Protocol Model

CANopen uses a small set of standardized communication objects on top of 11-bit CAN identifiers. The implementation in this repo follows the predefined connection set from CiA 301.

Core object classes:

NMT on COB-ID 0x000 for node control
SYNC on COB-ID 0x080 for synchronous PDO traffic
EMCY on 0x080 + node_id for emergency signaling
TPDO / RPDO for cyclic or event-driven process data
SDO on 0x580 + node_id and 0x600 + node_id for confirmed object access
Heartbeat on 0x700 + node_id for node supervision

NMT states represented in this package:

BOOT_UP
PRE_OPERATIONAL
OPERATIONAL
STOPPED

CANopen object access is centered around the Object Dictionary:

each object is addressed by index and subindex
communication profile objects typically live in 0x1000 to 0x1FFF
manufacturer-specific objects typically live in 0x2000 to 0x5FFF
PDO mapping usually references application objects that fit into the 8-byte PDO payload

PyRbs Integration

PyRbs exposes CANopen through rbs.canopen after pyrbs.PyRbs initialization.

from pyrbs import PyRbs

rbs = PyRbs()
nodes = rbs.canopen.list_nodes("CAN1")

The handler is refreshed from the loaded CAN channel configuration, so CANopen is configured per CAN channel rather than as a separate global subsystem.

Configuration

CANopen configuration is attached to a CAN channel in can.toml via the canopen section. A minimal manager-style setup looks like this:

[can.CAN1]
hw_vendor = "pcan"
hw_channel = 1
alias = "DRIVE_CAN"

[can.CAN1.hw]
clock_Hz = 80000000

[can.CAN1.hw.nominal]
brp = 2
tseg1 = 63
tseg2 = 16
sjw = 16
tsam = 16
bitrate = 500000
sample_point = 8000
tq = 25
bitrate_real = 0

[can.CAN1.canopen]
role = "manager"
strict = true
autostart = true

[can.CAN1.canopen.default_sysvars]
create_default_sysvars = true
create_node_sysvars = true

[[can.CAN1.canopen.nodes]]
node_id = 1
name = "left_motor"
eds = "./motor.eds"
auto_bindings = true

Behavior:

the raw CANopen config is preserved on rbs.canX.config.canopen
rbs.canopen.list_nodes(...) reads from the configured nodes list
object_name=... resolution requires an eds or dcf path on the node config
the EDS resolver maps ParameterName values to index and subindex

Bindings Between CANopen And Sysvars

The main PyRbs workflow is binding CANopen objects to sysvars through rbs.canopen.bind(...).

Typical SDO polling example:

current = rbs.sysvar.add("application_motor_current", 0)

rbs.canopen.bind(
    channel="CAN1",
    node_id=1,
    object_name="Max Motor Current",
    transport="sdo",
    mode="poll",
    interval=0.1,
    sysvar=current,
    codec="u16",
)

Important bind parameters:

channel: channel object, "CAN1", or channel id
node_id: CANopen node id
index and sub_index: explicit OD address
object_name: resolved through EDS or DCF if index is omitted
transport: "sdo" or "pdo"
direction: runtime binding direction, defaults to "rx"
mode: explicit runtime mode, or inferred from interval
interval: polling interval; values below 10 are treated as seconds and converted to milliseconds
codec: decoding or encoding hint such as u8, u16, u32, i16, f32, or utf8
pdo_cob_id: required for PDO bindings

For PDO transport, the runtime binding requires an explicit PDO COB-ID:

rbs.canopen.bind(
    channel="CAN1",
    node_id=1,
    index=0x2003,
    sub_index=0,
    sysvar=current,
    transport="pdo",
    pdo_cob_id=0x181,
    pdo_length=2,
    codec="u16",
)

Callbacks And Selectors

The handler dispatches typed CANopen events. This is the main monitoring and automation surface for application code.

Examples:

@rbs.canopen.on_object("CAN1.1.0x2003.0")
def on_object(update):
    print(update.value)

@rbs.canopen.on_sdo("CAN1.1")
def on_sdo(event):
    print(event.kind, event.abort_code)

@rbs.canopen.on_heartbeat("CAN1.1")
def on_heartbeat(event):
    print(event.kind, event.state)

Supported callback families:

on_object(...) for decoded object updates
on_binding(...) for sysvar-binding updates
on_sdo(...) for SDO events such as aborts
on_pdo(...) for PDO-backed binding updates
on_emcy(...) for emergency frames
on_heartbeat(...) for heartbeat, boot-up, and timeout events
on_nmt(...) for NMT state updates

Selector rules implemented by the handler:

node selectors use "CANx" or "CANx.node_id"
object selectors use "CANx.node_id.0xINDEX[.sub]"
object selectors may also use "CANx.node_id.Object Name" when an EDS or DCF is configured
on_pdo(...) also accepts an integer or hex-string COB-ID directly

Lower-Level Protocol API

The package also exposes lower-level protocol building blocks in pyrbs.canbus.canopen for simulations or unit-style protocol work outside of the high-level rbs.canopen handler.

Notable pieces:

pyrbs.canbus.canopen.node.CanOpenNode
pyrbs.canbus.canopen.node.CanOpenNetwork
pyrbs.canbus.canopen.nmt.NmtStateMachine
pyrbs.canbus.canopen.sdo.SdoClient
pyrbs.canbus.canopen.sdo.SdoServer
pyrbs.canbus.canopen.pdo.PdoManager
pyrbs.canbus.canopen.heartbeat.HeartbeatProducer
pyrbs.canbus.canopen.heartbeat.HeartbeatConsumer
pyrbs.canbus.canopen.od.ObjectDictionary

API Reference

High-level PyRbs handler

Python-facing CANopen handler for PyRbs.

This layer provides the public rbs.canopen.* API expected by PyRbs while delegating runtime protocol work to pyrbs_core.

class pyrbs.canbus.canopen.canopen_handler.CanOpenHandler(_rbs)

Bases: object

Parameters:: _rbs (PyRbs)

refresh()

Return type:: None

bind(*, channel, node_id, sysvar, index=None, sub_index=0, object_name=None, transport='sdo', direction='rx', mode=None, interval=None, codec=None, pdo_cob_id=None, pdo_offset=0, pdo_length=None, change_only=True, min_abs_delta=None)

Parameters:

channel (str | int | 'CanChannel')
node_id (int)
sysvar (str | 'SysVar')
index (int | None)
sub_index (int)
object_name (str | None)
transport (str)
direction (str)
mode (str | None)
interval (float | int | None)
codec (str | None)
pdo_cob_id (int | None)
pdo_offset (int)
pdo_length (int | None)
change_only (bool)
min_abs_delta (float | None)

Return type:

CanOpenBindingConfig

list_nodes(channel=None)

Parameters:: channel (str | int | 'CanChannel' | None)
Return type:: list[dict[str, Any]]

list_bindings()

Return type:: list[dict[str, Any]]

on_object(target, callback_name=None, priority=100)

Parameters:

target (str)
callback_name (str | None)
priority (int)

on_binding(target, callback_name=None, priority=100)

Parameters:

target (str)
callback_name (str | None)
priority (int)

on_sdo(target, callback_name=None, priority=100)

Parameters:

target (str)
callback_name (str | None)
priority (int)

on_pdo(target, callback_name=None, priority=100)

Parameters:

target (str | int)
callback_name (str | None)
priority (int)

on_emcy(target, callback_name=None, priority=100)

Parameters:

target (str)
callback_name (str | None)
priority (int)

on_heartbeat(target, callback_name=None, priority=100)

Parameters:

target (str)
callback_name (str | None)
priority (int)

on_nmt(target, callback_name=None, priority=100)

Parameters:

target (str)
callback_name (str | None)
priority (int)

pyrbs.canbus.canopen.canopen_handler.is_canopen_message(arbitration_id)

Parameters:: arbitration_id (int)
Return type:: bool

Node and network model

class pyrbs.canbus.canopen.node.EmergencyError(error_code, error_register, data=b'')

Bases: object

CANopen Emergency Error

Parameters:

error_code (int)
error_register (int)
data (bytes)

classmethod from_data(data)

Parse emergency from CAN data

Parameters:: data (bytes)
Return type:: EmergencyError

to_data()

Convert to CAN data

Return type:: bytes

class pyrbs.canbus.canopen.node.CanOpenNode(node_id, send_func, od=None, is_local=True)

Bases: object

CANopen Node

Main abstraction for a CANopen device, providing unified access to all CANopen protocols.

Example

>>> # Create local node
>>> node = CanOpenNode(node_id=1, send_func=send)
>>> node.nmt.boot_up_complete()
>>> node.heartbeat.enable()

>>> # Read from remote node
>>> value = await node.sdo.upload(0x1000, 0)

>>> # Set up PDO
>>> tpdo = node.pdo.add_tpdo(1)
>>> tpdo.add_mapping(0x6000, 1, 16)

Parameters:

node_id (int)
send_func (Callable[[int, bytes], None])
od (ObjectDictionary | None)
is_local (bool)

property node_id: int: Node ID

property is_local: bool: True if this is a local node

property od: ObjectDictionary: Object Dictionary

property nmt: NmtStateMachine: NMT state machine

property pdo: PdoManager: PDO manager

property heartbeat: HeartbeatProducer: Heartbeat producer

property heartbeat_consumer: HeartbeatConsumer: Heartbeat consumer for monitoring other nodes

property sdo_client: SdoClient | None: SDO client (for remote nodes)

property sdo_server: SdoServer | None: SDO server (for local nodes)

property state: NmtState: Current NMT state

property is_operational: bool: Check if node is operational

async read(index, subindex=0, data_type=None)

Read from Object Dictionary (via SDO for remote nodes).

Parameters:

index (int) – OD index
subindex (int) – OD subindex
data_type (OdDataType | None) – Data type for decoding

Returns:

Read value

Return type:

Any

async write(index, subindex, value, data_type=None)

Write to Object Dictionary (via SDO for remote nodes).

Parameters:

index (int) – OD index
subindex (int) – OD subindex
value (Any) – Value to write
data_type (OdDataType | None) – Data type for encoding

Returns:

True if successful

Return type:

bool

process_message(arbitration_id, data)

Process incoming CAN message.

Routes message to appropriate handler based on COB-ID.

Parameters:

arbitration_id (int) – CAN arbitration ID
data (bytes) – CAN data bytes

Returns:

True if message was handled

Return type:

bool

send_emergency(error_code, error_register, data=b'')

Send emergency message.

Parameters:

error_code (int) – 16-bit error code
error_register (int) – Error register value
data (bytes) – Additional manufacturer-specific data (max 5 bytes)

Return type:

None

on_emergency(callback)

Add emergency callback

Parameters:: callback (Callable[[EmergencyError], None])
Return type:: None

boot_up()

Perform boot-up sequence.

Sends boot-up message and transitions to pre-operational.

Return type:: None

start()

Start node (enter operational state)

Return type:: None

stop()

Stop node (enter stopped state)

Return type:: None

tick()

Periodic tick function.

Call this periodically to handle: - Heartbeat production - Heartbeat consumer timeout checks

Return type:: None

class pyrbs.canbus.canopen.node.CanOpenNetwork(send_func)

Bases: object

CANopen Network Manager

Manages multiple CANopen nodes on a network.

Example

>>> network = CanOpenNetwork(send_func=send)
>>> local = network.add_local_node(1)
>>> remote = network.add_remote_node(2)
>>> network.process_message(cob_id, data)

Parameters:: send_func (Callable[[int, bytes], None])

property nmt: NmtMaster: NMT master for network management

add_local_node(node_id, od=None)

Add a local node to the network.

Parameters:

node_id (int) – Node ID
od (ObjectDictionary | None) – Optional Object Dictionary

Returns:

Created node

Return type:

CanOpenNode

add_remote_node(node_id, od=None)

Add a remote node to the network.

Parameters:

node_id (int) – Node ID
od (ObjectDictionary | None) – Optional Object Dictionary (for expected structure)

Returns:

Created node

Return type:

CanOpenNode

get_node(node_id)

Get node by ID

Parameters:: node_id (int)
Return type:: CanOpenNode | None

remove_node(node_id)

Remove node from network

Parameters:: node_id (int)
Return type:: CanOpenNode | None

process_message(arbitration_id, data)

Process incoming CAN message.

Routes to appropriate node(s).

Parameters:

arbitration_id (int) – CAN arbitration ID
data (bytes) – CAN data bytes

Returns:

True if message was handled

Return type:

bool

tick()

Periodic tick for all nodes

Return type:: None

property node_ids: List[int]: List of all node IDs

NMT

CANopen Network Management (NMT) Module

NMT handles the network state management for CANopen nodes.

NMT State Machine:

INITIALISATION -> PRE-OPERATIONAL -> OPERATIONAL
                         |                |
                         v                v
                     STOPPED <--------> OPERATIONAL

NMT Commands (sent via COB-ID 0x000): - Start Remote Node (0x01): Pre-operational/Stopped -> Operational - Stop Remote Node (0x02): Any -> Stopped - Enter Pre-Operational (0x80): Any -> Pre-operational - Reset Node (0x81): Any -> Re-initialization - Reset Communication (0x82): Any -> Reset communication parameters

class pyrbs.canbus.canopen.nmt.NmtState(*values)

Bases: IntEnum

CANopen NMT States

Values correspond to heartbeat state byte encoding.

BOOT_UP = 0

STOPPED = 4

OPERATIONAL = 5

PRE_OPERATIONAL = 127

UNKNOWN = 255

class pyrbs.canbus.canopen.nmt.NmtCommand(*values)

Bases: IntEnum

CANopen NMT Commands (Command Specifier)

These are sent in byte 0 of NMT messages to COB-ID 0x000.

START_REMOTE_NODE = 1

STOP_REMOTE_NODE = 2

ENTER_PRE_OPERATIONAL = 128

RESET_NODE = 129

RESET_COMMUNICATION = 130

class pyrbs.canbus.canopen.nmt.NmtStateMachine(node_id, initial_state=NmtState.BOOT_UP)

Bases: object

CANopen NMT State Machine

Tracks and manages the NMT state for a CANopen node. Supports callbacks for state transitions.

Example

>>> nmt = NmtStateMachine(node_id=1)
>>> nmt.add_state_change_callback(lambda old, new: print(f"{old} -> {new}"))
>>> nmt.apply_command(NmtCommand.START_REMOTE_NODE)

Parameters:

node_id (int)
initial_state (NmtState)

property node_id: int: Node ID

property state: NmtState: Current NMT state

property is_operational: bool: Check if node is in operational state

property is_pre_operational: bool: Check if node is in pre-operational state

property is_stopped: bool: Check if node is in stopped state

add_state_change_callback(callback)

Add callback for state changes.

Parameters:: callback (Callable[[NmtState, NmtState], None]) – Function called with (old_state, new_state)
Return type:: None

remove_state_change_callback(callback)

Remove state change callback

Parameters:: callback (Callable[[NmtState, NmtState], None])
Return type:: None

apply_command(command)

Apply NMT command and transition to new state.

Parameters:: command (NmtCommand) – NMT command to apply
Returns:: New NMT state after transition
Return type:: NmtState

set_state(state)

Directly set NMT state (e.g., from heartbeat message).

Parameters:: state (NmtState) – New NMT state
Return type:: None

boot_up_complete()

Signal that boot-up is complete. Transitions from BOOT_UP to PRE_OPERATIONAL.

Return type:: None

class pyrbs.canbus.canopen.nmt.NmtMessage(command, target_node_id)

Bases: object

CANopen NMT Message structure

NMT messages are sent to COB-ID 0x000 with: - Byte 0: Command specifier (NmtCommand) - Byte 1: Node ID (0 = broadcast to all nodes)

Parameters:

command (NmtCommand)
target_node_id (int)

command: NmtCommand

target_node_id: int

classmethod from_data(data)

Parse NMT message from CAN data.

Parameters:: data (bytes) – 2-byte CAN message data
Returns:: NmtMessage or None if invalid
Return type:: NmtMessage | None

to_data()

Convert to CAN message data

Return type:: bytes

applies_to_node(node_id)

Check if this NMT message applies to the given node

Parameters:: node_id (int)
Return type:: bool

class pyrbs.canbus.canopen.nmt.NmtMaster(send_func)

Bases: object

CANopen NMT Master

Sends NMT commands to control network nodes.

Example

>>> master = NmtMaster(send_func)
>>> master.start_node(1)  # Start node 1
>>> master.stop_all_nodes()  # Stop all nodes

Parameters:: send_func (Callable[[int, bytes], None])

NMT_COB_ID = 0

start_node(node_id)

Send Start Remote Node command to specific node

Parameters:: node_id (int)
Return type:: None

stop_node(node_id)

Send Stop Remote Node command to specific node

Parameters:: node_id (int)
Return type:: None

enter_pre_operational(node_id)

Send Enter Pre-Operational command to specific node

Parameters:: node_id (int)
Return type:: None

reset_node(node_id)

Send Reset Node command to specific node

Parameters:: node_id (int)
Return type:: None

reset_communication(node_id)

Send Reset Communication command to specific node

Parameters:: node_id (int)
Return type:: None

start_all_nodes()

Start all nodes in the network

Return type:: None

stop_all_nodes()

Stop all nodes in the network

Return type:: None

reset_all_nodes()

Reset all nodes in the network

Return type:: None

Heartbeat

CANopen Heartbeat Module

Heartbeat protocol for node monitoring (CiA 301).

Producer: Sends periodic heartbeat messages Consumer: Monitors heartbeat messages from other nodes

Heartbeat Message (COB-ID: 0x700 + Node-ID): | Byte 0 | | NMT State |

The heartbeat message contains the current NMT state: - 0x00: Boot-up - 0x04: Stopped - 0x05: Operational - 0x7F: Pre-operational

class pyrbs.canbus.canopen.heartbeat.HeartbeatProducer(node_id, send_func, interval_ms=0)

Bases: object

Heartbeat Producer

Sends periodic heartbeat messages indicating NMT state.

Example

>>> producer = HeartbeatProducer(node_id=1, send_func=send)
>>> producer.set_interval(1000)  # 1000ms
>>> producer.set_state(NmtState.OPERATIONAL)
>>> producer.send()  # Manual trigger, or use timer

Parameters:

node_id (int)
send_func (Callable[[int, bytes], None])
interval_ms (int)

property node_id: int

property cob_id: int

property interval_ms: int

property state: NmtState

property enabled: bool

set_interval(interval_ms)

Set heartbeat interval (0 to disable)

Parameters:: interval_ms (int)
Return type:: None

set_state(state)

Set current NMT state for heartbeat

Parameters:: state (NmtState)
Return type:: None

enable()

Enable heartbeat production

Return type:: None

disable()

Disable heartbeat production

Return type:: None

send()

Send heartbeat message

Return type:: None

send_bootup()

Send boot-up message (heartbeat with state 0x00)

Return type:: None

tick()

Check if heartbeat should be sent (call periodically).

Returns:: True if heartbeat was sent
Return type:: bool

time_until_next_ms()

Get time until next heartbeat in milliseconds

Return type:: int | None

class pyrbs.canbus.canopen.heartbeat.ConsumerEntry(node_id, timeout_ms, last_seen=0, state=NmtState.UNKNOWN, timed_out=False)

Bases: object

Entry for a monitored node

Parameters:

node_id (int)
timeout_ms (int)
last_seen (float)
state (NmtState)
timed_out (bool)

node_id: int

timeout_ms: int

last_seen: float = 0

state: NmtState = 255

timed_out: bool = False

class pyrbs.canbus.canopen.heartbeat.HeartbeatConsumer

Bases: object

Heartbeat Consumer

Monitors heartbeat messages from other nodes and detects timeouts.

Example

>>> consumer = HeartbeatConsumer()
>>> consumer.add_node(2, timeout_ms=2000)  # Monitor node 2
>>> consumer.on_timeout(lambda node_id: print(f"Node {node_id} timeout!"))
>>> consumer.process(0x702, data)  # Process heartbeat from node 2

add_node(node_id, timeout_ms)

Add a node to monitor.

Parameters:

node_id (int) – Node ID to monitor
timeout_ms (int) – Timeout in milliseconds

Return type:

None

remove_node(node_id)

Remove a node from monitoring

Parameters:: node_id (int)
Return type:: None

get_node_state(node_id)

Get last known state of a node

Parameters:: node_id (int)
Return type:: NmtState | None

is_timed_out(node_id)

Check if node has timed out

Parameters:: node_id (int)
Return type:: bool

on_heartbeat(callback)

Add callback for heartbeat received

Parameters:: callback (Callable[[int, NmtState], None])
Return type:: None

on_timeout(callback)

Add callback for heartbeat timeout

Parameters:: callback (Callable[[int], None])
Return type:: None

on_recovered(callback)

Add callback for node recovered from timeout

Parameters:: callback (Callable[[int, NmtState], None])
Return type:: None

process(cob_id, data)

Process incoming CAN message.

Parameters:

cob_id (int) – CAN arbitration ID
data (bytes) – CAN data

Returns:

True if message was a heartbeat

Return type:

bool

tick()

Check for timed out nodes (call periodically).

Returns:: List of node IDs that just timed out
Return type:: List[int]

property monitored_nodes: List[int]: List of monitored node IDs

property active_nodes: List[int]: List of nodes that haven’t timed out

property timed_out_nodes: List[int]: List of timed out nodes

SDO

CANopen Service Data Object (SDO) Module

SDO provides confirmed access to the Object Dictionary. Uses client-server model with request/response protocol.

SDO Transfer Types: 1. Expedited Transfer: Data fits in single frame (≤4 bytes) 2. Segmented Transfer: Data spans multiple frames (>4 bytes) 3. Block Transfer: Optimized for large data (not implemented here)

SDO COB-IDs (Predefined Connection Set): - Client -> Server (Request): 0x600 + Node-ID - Server -> Client (Response): 0x580 + Node-ID

class pyrbs.canbus.canopen.sdo.SdoCommand(*values)

Bases: IntEnum

SDO Command Specifier (byte 0 upper nibble)

DOWNLOAD_INIT_REQ = 32

DOWNLOAD_SEG_REQ = 0

UPLOAD_INIT_REQ = 64

UPLOAD_SEG_REQ = 96

ABORT = 128

DOWNLOAD_INIT_RESP = 96

DOWNLOAD_SEG_RESP = 32

UPLOAD_INIT_RESP = 64

UPLOAD_SEG_RESP = 0

class pyrbs.canbus.canopen.sdo.SdoAbortCode(*values)

Bases: IntEnum

SDO Abort Domain Transfer codes (CiA 301)

TOGGLE_BIT_NOT_ALTERNATED = 84082688

SDO_PROTOCOL_TIMED_OUT = 84148224

COMMAND_SPECIFIER_INVALID = 84148225

INVALID_BLOCK_SIZE = 84148226

INVALID_SEQUENCE_NUMBER = 84148227

CRC_ERROR = 84148228

OUT_OF_MEMORY = 84148229

UNSUPPORTED_ACCESS = 100728832

READ_ONLY = 100728833

WRITE_ONLY = 100728834

OBJECT_NOT_EXIST = 100794368

OBJECT_CANNOT_BE_MAPPED = 100925505

PDO_LENGTH_EXCEEDED = 100925506

PARAMETER_INCOMPATIBILITY = 100925507

INTERNAL_INCOMPATIBILITY = 100925511

HARDWARE_ERROR = 101056512

DATA_TYPE_MISMATCH = 101122064

DATA_TYPE_TOO_HIGH = 101122066

DATA_TYPE_TOO_LOW = 101122067

SUBINDEX_NOT_EXIST = 101253137

VALUE_RANGE_EXCEEDED = 101253168

VALUE_TOO_HIGH = 101253169

VALUE_TOO_LOW = 101253170

MAX_LESS_THAN_MIN = 101253174

RESOURCE_NOT_AVAILABLE = 101318691

GENERAL_ERROR = 134217728

DATA_CANNOT_STORE = 134217760

DATA_CANNOT_STORE_LOCAL = 134217761

DATA_CANNOT_STORE_DEVICE_STATE = 134217762

OBJECT_DICTIONARY_DYNAMIC_FAIL = 134217763

NO_DATA_AVAILABLE = 134217764

pyrbs.canbus.canopen.sdo.get_abort_message(abort_code)

Get human-readable message for SDO abort code

Parameters:: abort_code (int)
Return type:: str

class pyrbs.canbus.canopen.sdo.SdoRequest(index, subindex, data=b'', expedited=True, size_indicated=False)

Bases: object

SDO request frame

Parameters:

index (int)
subindex (int)
data (bytes)
expedited (bool)
size_indicated (bool)

index: int

subindex: int

data: bytes = b''

expedited: bool = True

size_indicated: bool = False

to_download_init_frame()

Create download (write) initiate request frame

Return type:: bytes

to_upload_init_frame()

Create upload (read) initiate request frame

Return type:: bytes

class pyrbs.canbus.canopen.sdo.SdoResponse(index, subindex, data=b'', expedited=True, size_indicated=False, abort_code=None)

Bases: object

SDO response frame

Parameters:

index (int)
subindex (int)
data (bytes)
expedited (bool)
size_indicated (bool)
abort_code (int | None)

index: int

subindex: int

data: bytes = b''

expedited: bool = True

size_indicated: bool = False

abort_code: int | None = None

property is_abort: bool

classmethod from_frame(frame)

Parse SDO response from CAN frame

Parameters:: frame (bytes)
Return type:: SdoResponse

class pyrbs.canbus.canopen.sdo.SdoClient(node_id, send_func, timeout=1.0)

Bases: object

CANopen SDO Client

Performs SDO read/write operations to remote CANopen nodes.

Example

>>> client = SdoClient(node_id=1, send_func=send, timeout=1.0)
>>> # Read device type
>>> value = await client.upload(0x1000, 0)
>>> # Write parameter
>>> await client.download(0x2000, 1, value=0x1234, data_type=OdDataType.UNSIGNED16)

Parameters:

node_id (int)
send_func (Callable[[int, bytes], None])
timeout (float)

property node_id: int

property tx_cob_id: int: COB-ID for requests (Client -> Server)

property rx_cob_id: int: COB-ID for responses (Server -> Client)

handle_response(data)

Handle incoming SDO response. Call this when receiving a CAN message with rx_cob_id.

Parameters:: data (bytes) – CAN message data (8 bytes)
Return type:: None

async upload(index, subindex=0, data_type=None)

Upload (read) from Object Dictionary.

Parameters:

index (int) – OD index
subindex (int) – OD subindex
data_type (OdDataType | None) – Optional data type for decoding

Returns:

Read value (decoded if data_type provided)

Return type:

Any

async download(index, subindex, value, data_type=None)

Download (write) to Object Dictionary.

Parameters:

index (int) – OD index
subindex (int) – OD subindex
value (Any) – Value to write
data_type (OdDataType | None) – Data type for encoding (required if value is not bytes)

Return type:

None

abort_transfer(index, subindex, abort_code)

Send SDO abort transfer

Parameters:

index (int)
subindex (int)
abort_code (SdoAbortCode)

Return type:

None

exception pyrbs.canbus.canopen.sdo.SdoAbortError(abort_code, index, subindex)

Bases: Exception

SDO abort error

Parameters:

abort_code (int)
index (int)
subindex (int)

class pyrbs.canbus.canopen.sdo.SdoServer(node_id, od, send_func)

Bases: object

CANopen SDO Server

Handles SDO requests from clients for local Object Dictionary access.

Example

>>> server = SdoServer(node_id=1, od=od, send_func=send)
>>> # Process incoming SDO request
>>> server.handle_request(can_data)

Parameters:

node_id (int)
od (ObjectDictionary)
send_func (Callable[[int, bytes], None])

property rx_cob_id: int: COB-ID for requests (Client -> Server)

property tx_cob_id: int: COB-ID for responses (Server -> Client)

handle_request(data)

Handle incoming SDO request.

Parameters:: data (bytes) – CAN message data
Return type:: None

PDO

class pyrbs.canbus.canopen.pdo.PdoTransmissionType(*values)

Bases: IntEnum

PDO Transmission Types (CiA 301)

SYNC_ACYCLIC = 0

SYNC_CYCLIC_1 = 1

SYNC_CYCLIC_240 = 240

RTR_SYNC = 252

RTR_ASYNC = 253

ASYNC_MFCT = 254

ASYNC_PROFILE = 255

class pyrbs.canbus.canopen.pdo.PdoMapping(index, subindex, bit_length, bit_offset=0)

Bases: object

Single PDO mapping entry.

Maps an OD entry (or part of it) to PDO data position.

Parameters:

index (int)
subindex (int)
bit_length (int)
bit_offset (int)

index

OD index

Type:: int

subindex

OD subindex

Type:: int

bit_length

Number of bits (typically 8, 16, 32)

Type:: int

bit_offset

Offset in PDO data (calculated automatically)

Type:: int

index: int

subindex: int

bit_length: int

bit_offset: int = 0

classmethod from_mapping_value(value, offset=0)

Create mapping from 32-bit OD mapping entry value.

Format: Index(16) | Subindex(8) | BitLength(8)

Parameters:

value (int)
offset (int)

Return type:

PdoMapping

to_mapping_value()

Convert to 32-bit OD mapping entry value

Return type:: int

property byte_length: int: Length in bytes (rounded up)

property byte_offset: int: Offset in bytes

class pyrbs.canbus.canopen.pdo.PdoConfig(pdo_number, cob_id, transmission_type=PdoTransmissionType.ASYNC_PROFILE, inhibit_time=0, event_timer=0, mappings=<factory>, enabled=True)

Bases: object

PDO Configuration

Contains all parameters for a single PDO.

Parameters:

pdo_number (int)
cob_id (int)
transmission_type (PdoTransmissionType)
inhibit_time (int)
event_timer (int)
mappings (List[PdoMapping])
enabled (bool)

pdo_number

PDO number (1-4 for predefined, 1-512 for full range)

Type:: int

cob_id

COB-ID for this PDO

Type:: int

transmission_type

Transmission type (sync/async)

Type:: pyrbs.canbus.canopen.pdo.PdoTransmissionType

inhibit_time

Minimum time between transmissions (100us units)

Type:: int

event_timer

Event timer period (ms), 0 = disabled

Type:: int

mappings

List of OD mappings

Type:: List[pyrbs.canbus.canopen.pdo.PdoMapping]

enabled

Whether PDO is active

Type:: bool

pdo_number: int

cob_id: int

transmission_type: PdoTransmissionType = 255

inhibit_time: int = 0

event_timer: int = 0

mappings: List[PdoMapping]

enabled: bool = True

property is_valid: bool: Check if PDO is valid (COB-ID bit 31 = 0)

property is_rtr_allowed: bool: Check if RTR is allowed (COB-ID bit 30 = 0)

property data_length: int: Total PDO data length in bytes

add_mapping(index, subindex, bit_length)

Add a mapping entry

Parameters:

index (int)
subindex (int)
bit_length (int)

Return type:

None

clear_mappings()

Clear all mappings

Return type:: None

class pyrbs.canbus.canopen.pdo.TpdoHandler(node_id, pdo_number, od, send_func, config=None)

Bases: object

Transmit PDO Handler

Manages TPDO transmission based on configured triggers.

Example

>>> tpdo = TpdoHandler(node_id=1, pdo_number=1, od=od, send_func=send)
>>> tpdo.add_mapping(0x6000, 1, 16)  # Map OD entry
>>> tpdo.transmit()  # Send PDO

Parameters:

node_id (int)
pdo_number (int)
od (ObjectDictionary)
send_func (Callable[[int, bytes], None])
config (PdoConfig | None)

property config: PdoConfig

property cob_id: int

property pdo_number: int

add_mapping(index, subindex, bit_length)

Add OD entry mapping

Parameters:

index (int)
subindex (int)
bit_length (int)

Return type:

None

clear_mappings()

Clear all mappings

Return type:: None

on_transmit(callback)

Add transmit callback

Parameters:: callback (Callable[[bytes], None])
Return type:: None

build_data()

Build PDO data from mapped OD entries.

Returns:: PDO data bytes (max 8)
Return type:: bytes

transmit(force=False)

Transmit PDO.

Parameters:: force (bool) – Force transmission even if data unchanged
Returns:: True if transmitted
Return type:: bool

on_sync(sync_counter=0)

Handle SYNC message for synchronous transmission.

Parameters:: sync_counter (int) – SYNC counter value (if enabled)
Return type:: None

class pyrbs.canbus.canopen.pdo.RpdoHandler(node_id, pdo_number, od, config=None)

Bases: object

Receive PDO Handler

Processes incoming PDOs and updates Object Dictionary.

Example

>>> rpdo = RpdoHandler(node_id=1, pdo_number=1, od=od)
>>> rpdo.add_mapping(0x6200, 1, 16)  # Map to OD entry
>>> rpdo.process(can_data)  # Process received PDO

Parameters:

node_id (int)
pdo_number (int)
od (ObjectDictionary)
config (PdoConfig | None)

property config: PdoConfig

property cob_id: int

property pdo_number: int

add_mapping(index, subindex, bit_length)

Add OD entry mapping

Parameters:

index (int)
subindex (int)
bit_length (int)

Return type:

None

clear_mappings()

Clear all mappings

Return type:: None

on_receive(callback)

Add receive callback

Parameters:: callback (Callable[[bytes], None])
Return type:: None

process(data)

Process received PDO data.

Updates mapped OD entries with received values.

Parameters:: data (bytes) – Received PDO data bytes
Return type:: None

class pyrbs.canbus.canopen.pdo.PdoManager(node_id, od, send_func)

Bases: object

PDO Manager

Manages all TPDOs and RPDOs for a node.

Example

>>> manager = PdoManager(node_id=1, od=od, send_func=send)
>>> manager.add_tpdo(1)  # Add TPDO1
>>> manager.add_rpdo(1)  # Add RPDO1
>>> manager.process_message(cob_id, data)  # Handle incoming

Parameters:

node_id (int)
od (ObjectDictionary)
send_func (Callable[[int, bytes], None])

add_tpdo(pdo_number, config=None)

Add TPDO handler

Parameters:

pdo_number (int)
config (PdoConfig | None)

Return type:

TpdoHandler

add_rpdo(pdo_number, config=None)

Add RPDO handler

Parameters:

pdo_number (int)
config (PdoConfig | None)

Return type:

RpdoHandler

get_tpdo(pdo_number)

Get TPDO handler by number

Parameters:: pdo_number (int)
Return type:: TpdoHandler | None

get_rpdo(pdo_number)

Get RPDO handler by number

Parameters:: pdo_number (int)
Return type:: RpdoHandler | None

process_message(cob_id, data)

Process incoming CAN message.

Parameters:

cob_id (int) – CAN arbitration ID
data (bytes) – CAN data bytes

Returns:

True if message was handled as RPDO

Return type:

bool

on_sync(sync_counter=0)

Handle SYNC message.

Triggers synchronous PDO transmission.

Parameters:: sync_counter (int) – SYNC counter value
Return type:: None

transmit_all()

Transmit all enabled TPDOs

Return type:: None

Object Dictionary

CANopen Object Dictionary (OD) Module

The Object Dictionary is the central data structure in CANopen. It contains all communication and application objects.

Object Dictionary Structure: - Index: 16-bit (0x0000 - 0xFFFF) - Subindex: 8-bit (0x00 - 0xFF)

Standard Index Ranges: - 0x0000: Not used - 0x0001-0x001F: Data types - 0x0020-0x003F: Complex data types - 0x0040-0x005F: Manufacturer-specific complex data types - 0x0060-0x007F: Device profile specific data types - 0x0080-0x009F: Reserved - 0x00A0-0x0FFF: Reserved - 0x1000-0x1FFF: Communication profile area - 0x2000-0x5FFF: Manufacturer-specific profile area - 0x6000-0x9FFF: Standardized device profile area - 0xA000-0xBFFF: Standardized interface profile area - 0xC000-0xFFFF: Reserved

class pyrbs.canbus.canopen.od.OdAccessType(*values)

Bases: IntEnum

Object Dictionary access types

READ_ONLY = 1

WRITE_ONLY = 2

READ_WRITE = 3

CONST = 4

class pyrbs.canbus.canopen.od.OdDataType(*values)

Bases: IntEnum

CANopen data types (CiA 301 standard)

Values match the data type indices in the OD (0x0001-0x001F)

BOOLEAN = 1

INTEGER8 = 2

INTEGER16 = 3

INTEGER32 = 4

UNSIGNED8 = 5

UNSIGNED16 = 6

UNSIGNED32 = 7

REAL32 = 8

VISIBLE_STRING = 9

OCTET_STRING = 10

UNICODE_STRING = 11

TIME_OF_DAY = 12

TIME_DIFFERENCE = 13

DOMAIN = 15

INTEGER24 = 16

REAL64 = 17

INTEGER40 = 18

INTEGER48 = 19

INTEGER56 = 20

INTEGER64 = 21

UNSIGNED24 = 22

UNSIGNED40 = 24

UNSIGNED48 = 25

UNSIGNED56 = 26

UNSIGNED64 = 27

pyrbs.canbus.canopen.od.get_data_type_size(data_type)

Get size in bytes for a data type, None for variable-length types

Parameters:: data_type (OdDataType)
Return type:: int | None

class pyrbs.canbus.canopen.od.OdObjectType(*values)

Bases: IntEnum

Object types in the Object Dictionary

NULL = 0

DOMAIN = 2

DEFTYPE = 5

DEFSTRUCT = 6

VAR = 7

ARRAY = 8

RECORD = 9

class pyrbs.canbus.canopen.od.OdEntry(index, subindex, name, data_type, access_type=OdAccessType.READ_WRITE, default_value=None, min_value=None, max_value=None, pdo_mapping=False, object_type=OdObjectType.VAR, _value=None, _callbacks=<factory>)

Bases: object

Object Dictionary Entry

Represents a single entry in the Object Dictionary. For arrays/records, subindex 0 typically contains the number of entries.

Parameters:

index (int)
subindex (int)
name (str)
data_type (OdDataType)
access_type (OdAccessType)
default_value (Any)
min_value (Any)
max_value (Any)
pdo_mapping (bool)
object_type (OdObjectType)
_value (Any)
_callbacks (List[Callable[[Any, Any], None]])

index: int

subindex: int

name: str

data_type: OdDataType

access_type: OdAccessType = 3

default_value: Any = None

min_value: Any = None

max_value: Any = None

pdo_mapping: bool = False

object_type: OdObjectType = 7

property value: Any: Get current value

add_callback(callback)

Add value change callback (old_value, new_value)

Parameters:: callback (Callable[[Any, Any], None])
Return type:: None

remove_callback(callback)

Remove value change callback

Parameters:: callback (Callable[[Any, Any], None])
Return type:: None

property size: int | None: Get size in bytes, None for variable-length

property is_readable: bool: Check if entry is readable

property is_writable: bool: Check if entry is writable

encode_value(value=None)

Encode value to bytes for SDO/PDO transfer.

Parameters:: value (Any) – Value to encode, or current value if None
Returns:: Encoded bytes
Return type:: bytes

decode_value(data)

Decode bytes from SDO/PDO transfer.

Parameters:: data (bytes) – Raw bytes to decode
Returns:: Decoded value
Return type:: Any

pyrbs.canbus.canopen.od.encode_od_value(value, data_type)

Encode a Python value to CANopen bytes (little-endian).

Parameters:

value (Any) – Python value to encode
data_type (OdDataType) – CANopen data type

Returns:

Encoded bytes

Return type:

bytes

pyrbs.canbus.canopen.od.decode_od_value(data, data_type)

Decode CANopen bytes to Python value.

Parameters:

data (bytes) – Raw bytes (little-endian)
data_type (OdDataType) – CANopen data type

Returns:

Decoded Python value

Return type:

Any

class pyrbs.canbus.canopen.od.ObjectDictionary

Bases: object

CANopen Object Dictionary

Container for all OD entries of a node.

Example

>>> od = ObjectDictionary()
>>> od.add_entry(OdEntry(0x1000, 0, "Device Type", OdDataType.UNSIGNED32))
>>> od[0x1000, 0].value = 0x000F0191
>>> print(od[0x1000, 0].value)

add_entry(entry)

Add an entry to the Object Dictionary.

Parameters:: entry (OdEntry) – OdEntry to add
Return type:: None

remove_entry(index, subindex)

Remove an entry from the Object Dictionary.

Parameters:

index (int) – 16-bit index
subindex (int) – 8-bit subindex

Returns:

Removed entry or None

Return type:

OdEntry | None

get_entry(index, subindex=0)

Get an entry by index and subindex.

Parameters:

index (int) – 16-bit index
subindex (int) – 8-bit subindex (default 0)

Returns:

OdEntry or None

Return type:

OdEntry | None

get_subindices(index)

Get all subindices for an index.

Parameters:: index (int) – 16-bit index
Returns:: Dict mapping subindex -> OdEntry
Return type:: Dict[int, OdEntry]

has_entry(index, subindex=0)

Check if entry exists

Parameters:

index (int)
subindex (int)

Return type:

bool

__getitem__(key)

Get entry using [] operator.

Usage:: od[0x1000] # Gets subindex 0 od[0x1000, 0] # Gets specific subindex

Parameters:: key (int | tuple)
Return type:: OdEntry | None

__setitem__(key, value)

Set entry value using [] operator.

Usage:: od[0x1000] = 0x12345678 od[0x1000, 0] = 0x12345678

Parameters:

key (int | tuple)
value (Any)

Return type:

None

__contains__(key)

Check if entry exists using ‘in’ operator

Parameters:: key (int | tuple)
Return type:: bool

__iter__(): Iterate over all entries

__len__()

Number of entries

Return type:: int

property indices: List[int]: List of all indices in the OD

read(index, subindex=0)

Read value from OD entry.

Parameters:

index (int) – 16-bit index
subindex (int) – 8-bit subindex

Returns:

Value or None if entry not found or not readable

Return type:

Any | None

write(index, subindex, value)

Write value to OD entry.

Parameters:

index (int) – 16-bit index
subindex (int) – 8-bit subindex
value (Any) – Value to write

Returns:

True if successful, False if entry not found or not writable

Return type:

bool

pyrbs.canbus.canopen.od.create_standard_od()

Create an Object Dictionary with standard CiA 301 entries.

Returns:: ObjectDictionary with basic communication objects
Return type:: ObjectDictionary

COB-ID Helpers

CANopen COB-ID (Communication Object Identifier) Module

Node ID range: 1-127 (0 is reserved for broadcast)

class pyrbs.canbus.canopen.cob_id.CanOpenObjectType(*values)

Bases: IntEnum

CANopen communication object types based on function code

NMT = 0

SYNC = 128

TIME = 256

EMERGENCY = 1

TPDO1 = 3

RPDO1 = 4

TPDO2 = 5

RPDO2 = 6

TPDO3 = 7

RPDO3 = 8

TPDO4 = 9

RPDO4 = 10

SDO_TX = 11

SDO_RX = 12

HEARTBEAT = 14

UNKNOWN = 255

class pyrbs.canbus.canopen.cob_id.PredefinedCobIds(NMT=0, SYNC=128, TIME=256)

Bases: object

Predefined Connection Set COB-ID formulas

Parameters:

NMT (int)
SYNC (int)
TIME (int)

NMT: int = 0

SYNC: int = 128

TIME: int = 256

static emergency(node_id)

Emergency COB-ID: 0x080 + Node-ID

Parameters:: node_id (int)
Return type:: int

static tpdo1(node_id)

TPDO1 COB-ID: 0x180 + Node-ID

Parameters:: node_id (int)
Return type:: int

static rpdo1(node_id)

RPDO1 COB-ID: 0x200 + Node-ID

Parameters:: node_id (int)
Return type:: int

static tpdo2(node_id)

TPDO2 COB-ID: 0x280 + Node-ID

Parameters:: node_id (int)
Return type:: int

static rpdo2(node_id)

RPDO2 COB-ID: 0x300 + Node-ID

Parameters:: node_id (int)
Return type:: int

static tpdo3(node_id)

TPDO3 COB-ID: 0x380 + Node-ID

Parameters:: node_id (int)
Return type:: int

static rpdo3(node_id)

RPDO3 COB-ID: 0x400 + Node-ID

Parameters:: node_id (int)
Return type:: int

static tpdo4(node_id)

TPDO4 COB-ID: 0x480 + Node-ID

Parameters:: node_id (int)
Return type:: int

static rpdo4(node_id)

RPDO4 COB-ID: 0x500 + Node-ID

Parameters:: node_id (int)
Return type:: int

static sdo_tx(node_id)

SDO TX (Server->Client) COB-ID: 0x580 + Node-ID

Parameters:: node_id (int)
Return type:: int

static sdo_rx(node_id)

SDO RX (Client->Server) COB-ID: 0x600 + Node-ID

Parameters:: node_id (int)
Return type:: int

static heartbeat(node_id)

Heartbeat/NMT Error Control COB-ID: 0x700 + Node-ID

Parameters:: node_id (int)
Return type:: int

class pyrbs.canbus.canopen.cob_id.CobId(raw)

Bases: object

CANopen COB-ID (Communication Object Identifier)

Parses and generates 11-bit CANopen CAN identifiers.

Example

>>> cob_id = CobId.from_can_id(0x181)  # TPDO1 from node 1
>>> cob_id.node_id
1
>>> cob_id.object_type
<CanOpenObjectType.TPDO1: 3>

Parameters:: raw (int)

classmethod from_can_id(can_id)

Create CobId from CAN arbitration ID

Parameters:: can_id (int)
Return type:: CobId

classmethod from_function_and_node(function_code, node_id)

Create CobId from function code and node ID.

Parameters:

function_code (int) – 4-bit function code (0-15)
node_id (int) – 7-bit node ID (0-127)

Return type:

CobId

property raw: int: Raw 11-bit COB-ID value

property function_code: int: 4-bit function code (upper bits)

property node_id: int: 7-bit node ID (lower bits), 0 if not applicable

property object_type: CanOpenObjectType: CANopen object type

to_can_id()

Convert to CAN arbitration ID

Return type:: int

is_pdo()

Check if this COB-ID is a PDO

Return type:: bool

is_sdo()

Check if this COB-ID is an SDO

Return type:: bool

is_tpdo()

Check if this is a Transmit PDO

Return type:: bool

is_rpdo()

Check if this is a Receive PDO

Return type:: bool

pyrbs.canbus.canopen.cob_id.get_default_cob_id(object_type, node_id)

Get the default COB-ID for a given object type and node ID.

Parameters:

object_type (CanOpenObjectType) – CANopen object type
node_id (int) – Node ID (1-127)

Returns:

Default COB-ID according to predefined connection set

Return type:

int

Typed events

Bases: object

Parameters:

channel (str)
channel_id (int)
node_id (int)
index (int)
sub_index (int)
sysvar (str | None)
value (int | float | str | bytes)
data (bytes)
codec (str | None)
transport (str | None)
object_name (str | None)
pdo_cob_id (int | None)

channel: str

channel_id: int

node_id: int

index: int

sub_index: int

sysvar: str | None

value: int | float | str | bytes

data: bytes

codec: str | None

transport: str | None

object_name: str | None

pdo_cob_id: int | None

Bases: object

Parameters:

channel (str)
channel_id (int)
node_id (int)
index (int)
sub_index (int)
sysvar (str | None)
value (int | float | str | bytes)
data (bytes)
codec (str | None)
transport (str | None)
object_name (str | None)
pdo_cob_id (int | None)

channel: str

channel_id: int

node_id: int

index: int

sub_index: int

sysvar: str | None

value: int | float | str | bytes

data: bytes

codec: str | None

transport: str | None

object_name: str | None

pdo_cob_id: int | None

class pyrbs.canbus.canopen.events.CanOpenSdoEvent(channel: 'str', channel_id: 'int', node_id: 'int', index: 'int', sub_index: 'int', kind: 'str', data: 'bytes | None' = None, abort_code: 'int | None' = None)

Bases: object

Parameters:

channel (str)
channel_id (int)
node_id (int)
index (int)
sub_index (int)
kind (str)
data (bytes | None)
abort_code (int | None)

channel: str

channel_id: int

node_id: int

index: int

sub_index: int

kind: str

data: bytes | None

abort_code: int | None

class pyrbs.canbus.canopen.events.CanOpenHeartbeatEvent(channel: 'str', channel_id: 'int', node_id: 'int', kind: 'str', state: 'str | None' = None, timestamp_ns: 'int | None' = None, last_seen_ns: 'int | None' = None, deadline_ns: 'int | None' = None)

Bases: object

Parameters:

channel (str)
channel_id (int)
node_id (int)
kind (str)
state (str | None)
timestamp_ns (int | None)
last_seen_ns (int | None)
deadline_ns (int | None)

channel: str

channel_id: int

node_id: int

kind: str

state: str | None

timestamp_ns: int | None

last_seen_ns: int | None

deadline_ns: int | None

class pyrbs.canbus.canopen.events.CanOpenNmtEvent(channel: 'str', channel_id: 'int', node_id: 'int', state: 'str')

Bases: object

Parameters:

channel (str)
channel_id (int)
node_id (int)
state (str)

channel: str

channel_id: int

node_id: int

state: str

class pyrbs.canbus.canopen.events.CanOpenEmcyEvent(channel: 'str', channel_id: 'int', node_id: 'int', code: 'int', register: 'int', manufacturer_data: 'bytes', timestamp_ns: 'int')

Bases: object

Parameters:

channel (str)
channel_id (int)
node_id (int)
code (int)
register (int)
manufacturer_data (bytes)
timestamp_ns (int)

channel: str

channel_id: int

node_id: int

code: int

register: int

manufacturer_data: bytes

timestamp_ns: int