feathersdk.comms.comms_manager Module

feathersdk.comms.comms_manager.MsgUIDType

Type of the unique identifier of a message.

feathersdk.comms.comms_manager.SingleMsgCheckerFuncType

Type of a function that checks for expected responses in single message mode.

alias of Callable[[CanSocketResult | TcpSocketResult | ErrorSocketResult], str | None]

exception feathersdk.comms.comms_manager.SocketCANLibError

Bases: Exception

Error raised when the socketcan library returns a non-zero error code.

exception feathersdk.comms.comms_manager.EndpointOverloadError

Bases: Exception

Error raised when the CAN bus is overloaded.

exception feathersdk.comms.comms_manager.UnknownEndpointError(*args, **kwargs)

Bases: Exception

Error raised when the interface is not an interface, or is not being tracked by CommsManager.

Parameters:
Return type:

None

exception feathersdk.comms.comms_manager.CommsManagerDisabledError

Bases: Exception

Error raised when the comms manager is disabled.

exception feathersdk.comms.comms_manager.MissingMsgUIDError(*args, **kwargs)

Bases: Exception

Error raised when a message UID is required but not provided.

Parameters:
Return type:

None

exception feathersdk.comms.comms_manager.PendingReceiveMessageError(*args, **kwargs)

Bases: Exception

Error raised when a message is to be sent when we are awaiting a response in single message mode.

Parameters:
Return type:

None

class feathersdk.comms.comms_manager.CommsManagerConfig(*, max_can_rate=3000.0, can_leniency=100.0, max_ip_rate=10000.0, ip_leniency=1000.0, result_operating_frequency=200.0, await_thread_start_timeout=1.0, processing_toggle_timeout=0.3, show_all_sdo_errors=False, disable=False, DANGEROUS_disable_overload_checks=False)

Bases: PyDFrozenBaseModel

Configuration for the comms manager.

Parameters:

  • max_can_rate (float)

  • can_leniency (float)

  • max_ip_rate (float)

  • ip_leniency (float)

  • result_operating_frequency (float)

  • await_thread_start_timeout (float)

  • processing_toggle_timeout (float)

  • show_all_sdo_errors (bool)

  • disable (bool)

  • DANGEROUS_disable_overload_checks (bool)

max_can_rate: float

The maximum messages per second that can be send on a CAN interface. If the rate is exceeded, an error will be raised. See EventThrottler for more information.

can_leniency: float

The amount of leniency allowed when checking for CAN bus overload. We allow max_can_rate / can_leniency messages to be sent in a burst before throttling occurs. See EventThrottler for more information.

max_ip_rate: float

The maximum messages per second that can be send on an IP interface. If the rate is exceeded, an error will be raised. See EventThrottler for more information.

ip_leniency: float

The amount of leniency allowed when checking for IP bus overload. We allow max_ip_rate / ip_leniency messages to be sent in a burst before throttling occurs. See EventThrottler for more information.

result_operating_frequency: float

The frequency at which the comms manager result processing thread will attempt to process results from the socket result buffer.

await_thread_start_timeout: float

The timeout in seconds to wait for the processing thread to start.

processing_toggle_timeout: float

The timeout in seconds to wait for the processing thread to toggle processing when making changes to the processing state.

show_all_sdo_errors: bool

CANOpen often logs unneeded SDO messages, so we can disable them by default. Set to True to re-enable them.

disable: bool

Disable the comms manager. Used internally for clients talking to server to avoid duplicate comms manager.

DANGEROUS_disable_overload_checks: bool

Disable the overload checks for the comms manager. This is dangerous and should only be used for testing!

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class feathersdk.comms.comms_manager.Endpoint(*, name, type='can', socket_idx, iface_uid)

Bases: PyDFrozenBaseModel

An endpoint to send and receive messages from.

Parameters:
name: str

The name of the endpoint.

type: Literal['can', 'ip']

The type of the endpoint.

socket_idx: int

The socket index of the endpoint.

iface_uid: int

The unique identifier of the interface.

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class feathersdk.comms.comms_manager.CanopenNetwork(endpoint, bitrate=1000000)

Bases: object

A CANopen network.

Parameters:

  • endpoint (str)

  • bitrate (int)

endpoint: str

The endpoint of the network.

network: Network

The CANopen network.

nodes: dict[int, RemoteNode]

All currently open canopen nodes on the network.

read(node_id, idx, subidx=None)

Read a CANopen value.

Parameters:

  • node_id (int)

  • idx (int)

  • subidx (int | None)

Return type:

Any

write(node_id, value, idx, subidx=None)

Write a CANopen value.

Parameters:
Return type:

None

add_node(node_id, obj_dict=None)

Add a CANopen node to the network. If the node already exists, returns the existing node.

Parameters:

  • node_id (int)

  • obj_dict (str | ObjectDictionary | None)

Return type:

RemoteNode

remove_node(node_id)

Remove a CANopen node from the network.

Parameters:

node_id (int)

Return type:

None

disconnect()

Disconnect the network.

Return type:

None

feathersdk.comms.comms_manager.initialize_comms_manager(config)

Initialize the CommsManager. Should only be called once.

Parameters:

config (CommsManagerConfig)

Return type:

_CommsManager

feathersdk.comms.comms_manager.get_comms_manager()

Get the CommsManager instance. Should only be called after initialize_comms_manager has been called.

Return type:

_CommsManager

feathersdk.comms.comms_manager.is_comms_manager_initialized()

Check if the CommsManager is initialized.

Return type:

bool

feathersdk.comms.comms_manager.CommsManagerType

Type of the CommsManager instance.

alias of Type[_CommsManager]