CAN Module

This module is primarily responsible for providing a reliable high-level interface to the CAN Bus as well as describing the format and structure of raw CAN messages according to the CANOpen spec.

class canopen_monitor.can.Interface(if_name: str)[source]

This is a model of a POSIX interface

Used to manage a singular interface and any encoded messages streaming across it

Parameters

name (str) – Name of the interface bound to
last_activity (datetime.datetime) – Timestamp of the last activity on the interface

property age: datetime.timedelta

Deterimes the age of the message, since it was received

Returns: Age of the message
Return type: datetime.timedelta

property duplex: int

Determines the duplex, if there is any

Returns: Duplex value
Return type: int

property is_up: bool

Determines if the interface is in the UP state

Returns: True if in the UP state False if in the DOWN state
Return type: bool

property mtu: int

Maximum Transmission Unit

Returns: Maximum size of a packet
Return type: int

recv() → canopen_monitor.can.message.Message[source]

A wrapper for pyvit.hw.SocketCanDev.recv()

Instead of returning a can.Frame, it intercepts the recv() and converts it to a canopen_monitor.Message at the last minute.

Returns: A loaded canopen_monitor.Message from the interface if a message is recieved within the configured SOCKET_TIMEOUT (default is 0.3 seconds), otherwise returns None
Return type: Message, None

restart() → None[source]

A macro-fuction for restarting the interface connection

This is the same as doing:

>>> iface.stop()
>>> iface.start()

property speed: int

Determines the Baud Rate of the bus, if any

Warning

This will appear as 0 for virtual can interfaces.

Returns: Baud rate
Return type: int

start(block_wait: bool = True) → None[source]

A wrapper for pyvit.hw.SocketCanDev.start()

If block-waiting is enabled, then instead of imediately binding to the interface, it waits for the state to change to UP first before binding.

Parameters: block_wait (bool) – Enables block-waiting

stop() → None[source]: A wrapper for pyvit.hw.SocketCanDev.stop()

class canopen_monitor.can.MagicCANBus(if_names: [str], no_block: bool = False)[source]

This is a macro-manager for multiple CAN interfaces

Parameters: interfaces ([Interface]) – The list of serialized Interface objects the bus is managing

add_interface(interface: str) → None[source]

This will add an interface at runtime

Parameters: interface (string) – The name of the interface to add

handler(iface: canopen_monitor.can.interface.Interface) → None[source]

This is a handler for listening and block-waiting for messages on the CAN bus

It will operate on the condition that the Magic Can Bus is still active, using thread-safe events.

Parameters: iface (Interface) – The interface to bind to when listening for messages

property interface_list: [str]: A list of strings representing all interfaces :return: a list of strings indicating the name of each interface :rtype: [str]

remove_interface(interface: str) → None[source]

This will remove an interface at runtime

Parameters: interface (string) – The name of the interface to remove

start_handler(iface: canopen_monitor.can.interface.Interface) → threading.Thread[source]

This is a wrapper for starting a single interface listener thread: This wrapper also creates a keep alive event for each thread which can be used to kill the thread.

Warning

If for any reason, the interface cannot be listened to, (either it doesn’t exist or there are permission issues in reading from it), then the default behavior is to stop listening for messages, block wait for the interface to come back up, then resume. It is possible that a thread starts but no listener starts due to a failure to bind to the interface.

Parameters: iface (Interface) – The interface to bind to when listening for messages
Returns: The new listener thread spawned
Return type: threading.Thread

property statuses: [tuple]

This property is simply an aggregate of all of the interfaces and whether or not they both exist and are in the UP state

Returns: a list of tuples containing the interface names and a bool indication an UP/DOWN status
Return type: [tuple]

class canopen_monitor.can.Message(arb_id: int, **kwargs)[source]

This class is a wrapper class for the pyvit.can.Frame class

Ref: See this for documentation on a PyVit Frame

It’s primary purpose is to carry all of the same CAN message data as a frame, while adding age and state attributes as well.

property age: datetime.timedelta

The age of the Message since it was received from the CAN bus

Returns: Age of the message
Return type: datetime.timedelta

property node_id: int

The Node ID, otherwise known as the unique device identifier

This is a property that is arbitratily decided in an Object Dictionary and can sometimes have a name attatched to it

Returns: Node ID
Return type: int

property state: canopen_monitor.can.message.MessageState

The state of the message since it was received from the CAN bus

Returns: State of the message
Return type: MessageState

property supertype: canopen_monitor.can.message.MessageType

Super-Type of CAN Message

Returns: CAN Message Super-Type
Return type: MessageType

property type: canopen_monitor.can.message.MessageType

Type of CAN Message

Returns: CAN Message Type
Return type: MessageType

class canopen_monitor.can.MessageState(value)[source]

This enumeration describes all possible states of a CAN Message

State	Age (sec)
ALIVE	x<60
STALE	60<=x<=120
DEAD	120<=x

ALIVE = 'Alive'

DEAD = 'Dead'

STALE = 'Stale'

class canopen_monitor.can.MessageTable(parser=None)[source]

clear() → None[source]: Clear the table to remove all its messages.

filter(types: MessageType, start: int = 0, end: int = None, sort_by: str = 'arb_id', reverse=False) → [Message][source]

class canopen_monitor.can.MessageType(value)[source]

This enumeration describes all of the ranges in the CANOpen spec that defines specific kinds of messages.

See wikipedia for details

EMER = (128, 255)

HEARTBEAT = (1792, 2047)

NMT = (0, 0)

PDO = (384, 1407)

PDO1_RX = (512, 639)

PDO1_TX = (384, 511)

PDO2_RX = (768, 895)

PDO2_TX = (640, 767)

PDO3_RX = (1024, 1151)

PDO3_TX = (896, 1023)

PDO4_RX = (1280, 1407)

PDO4_TX = (1152, 1279)

SDO = (1408, 1664)

SDO_RX = (1536, 1664)

SDO_TX = (1408, 1535)

SYNC = (1, 127)

TIME = (256, 256)

UKNOWN = (-1, -1)

static cob_id_to_type(cob_id: int) → canopen_monitor.can.message.MessageType[source]

Determines the message type based on the COB ID

Parameters: cob_id (int) – The Raw CAN Message COB ID
Returns: The message type (range) the COB ID fits into
Return type: MessageType

static cob_to_node(msg_type: canopen_monitor.can.message.MessageType, cob_id: int) → int[source]

Determines the Node ID based on the given COB ID

The COB ID is the raw ID sent with the CAN message, and the node id is simply the sub-id within the COB ID, which is used as a device identifier.

Example

If the COB ID is 0x621

Then the Type is SDO_RX (an SDO being received)

The start of the SDO_RX range is 0x600

Therefore the Node ID is 0x621 - 0x600 = 0x21

Parameters

mtype (MessageType) – The message type
cob_id (int) – The Raw CAN Message COB ID

Returns

The Node ID

Return type

int

property supertype: canopen_monitor.can.message.MessageType

Determines the “Supertype” of a Message

There are only two supertypes: MessageType.PDO and MessageType.SDO, and they emcompass all of the PDO_T/RX and SDO_T/RX ranges respectively. This simply returns which range the type is in if any, or MessageType.UNKNOWN if it’s in neither supertype range.

Returns: The supertype of this type
Return type: MessageType