API Reference

Codecs

zeloscloud.codecs.ble

Concrete codec/signal implementation classes for ble comms

BleCodec(name, link, config=None)

Bases: CodecBase

BLE Codec for encoding and decoding BLE characteristic data based on a provided schema.

Initializes the BleCodec with a name, a link, and a config.

Parameters:

Name	Type	Description	Default
name	str	Name of the codec.	required
link	BleClientLinkBase	Interface to the BLE link.	required
config	Dict[str, Union[str, Dict[str, Any]]]	Configuration dictionary for the codec.	None

BleMessage(name, service, codec)

Bases: MessageBase

Initialize the BLE message with signals based on the characteristic fields.

Parameters:

Name	Type	Description	Default
name	str	The name of the associated characteristic.	required
service	GattService	The service for the associated GattCharacteristic.	required
codec	CodecBase	The BLE codec to use for communication/updates.	required

properties: str property

Returns:

Type	Description
str	The properties of the characteristic.

service_name: str property

Returns:

Type	Description
str	The name of the service.

service_uuid: str property

Returns:

Type	Description
str	The UUID of the service.

uuid: str property

Returns:

Type	Description
str	The UUID of the characteristic.

recv()

Receive data from the characteristic and update signals.

send()

Encode the data from signals and send it to the characteristic.

zeloscloud.codecs.can

Concrete codec/signal implementation classes for CAN comms

CanCodec(name, link, dbc, config=None, **kwargs)

Bases: CodecBase, Listener

Codec for interfacing with a CAN bus.

Initialize the CanCodec with a name, link, and a configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of the codec.	required
link	CanLink	Link to the physical bus (e.g., CAN link).	required
dbc	str	Path to the DBC file.	required
config	Optional[dict]	Optional configuration dictionary.	None
kwargs		Additional keyword arguments.	{}

allow_duplicates: bool property

Returns the value of the allow_duplicates configuration.

messages_by_id: dict[int, CanMessage] property

Returns the dictionary of messages by their IDs.

software_timestamp: bool property

Returns the value of the software_timestamp configuration.

suppress_errors: bool property

Returns the value of the suppress_errors configuration.

get_messages(nodes)

Returns a tuple of the messages that a given list of nodes sends/receives.

Parameters:

Name	Type	Description	Default
nodes	list[str]	List of node names.	required

Returns:

Type	Description
tuple[list[CanMessage], list[CanMessage]]	a tuple of the messages that a given list of nodes sends/receives

handle_message(message)

Cracks a message and updates internal signals.

on_error(error)

Handles a CAN operation error. Implements can.Listener.on_error abstract method.

on_message_received(message)

Cracks a message and updates internal signals. Implements can.Listener.on_message_received abstract method.

Parameters:

Name	Type	Description	Default
message	Message	The received CAN message.	required

stop()

Stops the CAN bus listener. Implements can.Listener.stop abstract method.

CanMessage(message, codec)

Bases: MessageBase

Helper class that helps manage periodics for a given CAN message.

Initialize the CanMessage.

Parameters:

Name	Type	Description	Default
message	Message	The CAN message definition	required
codec	CodecBase	The codec used for encoding/decoding	required

cycle_time: Optional[float] property

Get the cycle time of the message in seconds.

Returns:

Type	Description
Optional[float]	The cycle time of the message in seconds or None if undefined

name: str property

Returns:

Type	Description
str	the name of the message

get_elapsed_time(timestamp_monotonic=None)

Returns the time elapsed since the last message based on the current monotonic time.

Parameters:

Name	Type	Description	Default
timestamp_monotonic	Optional[float]	The current monotonic time value. Defaults to time.monotonic()	None

Returns:

Type	Description
Optional[float]	The time difference in seconds or None if the timestamp is not available

get_signals_by_mux(mux=None)

Get the signals for a given multiplexer value.

Parameters:

Name	Type	Description	Default
mux		The multiplexer value	None

Returns:

Type	Description
	A list of signals

is_mia(mia_time)

Checks if the system is 'Missing in Action' (MIA) based on the time since the last message.

NOTE: Consider some future default value computation for MIA time like this in the future: DEFAULT_MIA_TIME = some sane number, user provided? DEFAULT_MIA_FACTOR = some sane number, user provided?

if mia_time is None: if self.cycle_time is None: mia_time = DEFAULT_MIA_TIME else: if self.multiplexer_values: mia_time = self.cycle_time / len(self.multiplexer_values) * DEFAULT_MIA_FACTOR else: mia_time = self.cycle_time * DEFAULT_MIA_FACTOR

Parameters:

Name	Type	Description	Default
mia_time	float	The threshold time (in seconds) to consider the system MIA	required

Returns:

Type	Description
bool	True if the message is MIA, False otherwise

send(mux=None, **kwargs)

Takes a snapshot of the signals then constructs + transmits a CAN message.

Parameters:

Name	Type	Description	Default
mux	Optional[Any]	The multiplexer value, set to None if no multiplexer is present in the message or to send all multiplexors values	None
kwargs		Additional keyword arguments to pass to the send method	{}

send_periodic(mux=None, period=None, **kwargs)

Takes a snapshot of the signals then constructs + transmits a CAN message periodically.

Parameters:

Name	Type	Description	Default
mux	Optional[Any]	The multiplexer value, set to None if no multiplexer is present in the message or to send all multiplexors values	None
period	Optional[float]	Periodic transmit rate, if None then it looks it up in the DBC	None
kwargs		Additional keyword arguments to pass to the send_periodic method	{}

set(signal_values, timestamp=None, timestamp_monotonic=None)

Apply signal sets from a dict of signal values.

Parameters:

Name	Type	Description	Default
signal_values	Dict[str, Any]	A dictionary of signal names and their values	required
timestamp	Optional[float]	The timestamp to set for the signals	None
timestamp_monotonic	Optional[float]	The monotonic timestamp to set for the signals	None

stop_all_periodics()

Stop all periodic transmissions.

stop_periodic(mux=None)

Stop periodic transmission of CAN message(s).

Parameters:

Name	Type	Description	Default
mux	Optional[Any]	The multiplexer value, set to None to stop all periodic transmissions	None

update_periodic(mux=None, **kwargs)

Update the periodic transmission of CAN message(s).

Parameters:

Name	Type	Description	Default
mux	Optional[Any]	The multiplexer value, set to None to update all periodic transmissions	None
kwargs		Additional keyword arguments to pass to the send_periodic method	{}

CanSignal(signal, **kwargs)

Bases: ReadWriteSignal

Represents a signal in a message on a CAN bus.

Attributes: signal (cantools.db.Signal): The corresponding CAN signal definition _mux (Optional[int]): The multiplexer ID this signal belongs to, if any

Initialize the CanSignal.

Parameters:

Name	Type	Description	Default
signal	Signal	The signal definition	required
kwargs	Any	Additional keyword arguments to pass to the parent class constructor	{}

choices: OrderedDict[int, str | NamedSignalValue] | None property

Returns:

Type	Description
OrderedDict[int, str | NamedSignalValue] | None	The choices (enumerations) for this signal

mux: Optional[int] property

Get the multiplexer ID this signal belongs to.

Returns:

Type	Description
Optional[int]	The multiplexer ID this signal belongs to, or None if not multiplexed

is_mia(mia_time=None)

Checks if the system is 'Missing in Action' (MIA) based on the time since the last message.

Parameters:

Name	Type	Description	Default
mia_time	Optional[float]	The threshold time (in seconds) to consider the system MIA. Defaults to None	None

Returns:

Type	Description
bool	True if the message is MIA, False otherwise

zeloscloud.codecs.ppk2

PowerProfilerKit2(name, port, poll_rate_s=0.1, **kwargs)

Bases: CodecBase

Codec Wrapper for the Power Profile Kit 2. Manages the communication with the PPK2 device and handles data polling, signal creation, and state management.

Parameters:

Name	Type	Description	Default
name	str	Name of the codec.	required
port	str	The port where PPK2 is connected.	required
poll_rate_s	int	Polling rate in seconds. Default is 0.1.	0.1

modifiers: Dict[str, Any] property

Gets the last set modifiers from the device memory. Updated in open().

zeloscloud.codecs.proto

Proto(name, config=None, **kwargs)

Bases: CodecBase

Codec that wraps user provided protobuf messages.

For map fields with int keys, users can specify a mapping (typically an enum) ex. "mappings": { "path/to/message/key": CustomEnumType, }

Parameters:

Name	Type	Description	Default
name	str	Name of the codec.	required
config	Optional[dict]	The configuration of the Proto Codec.	None

Initializes the Proto Codec instance.

Parameters:

Name	Type	Description	Default
name	str	Name of the Codec.	required
config	Optional[dict]	Configuration for the Proto Codec.	None
kwargs		Additional keyword arguments.	{}

config: dict[str, Any] property

Returns the configuration of the Proto Codec.

Returns:

Type	Description
dict[str, Any]	Configuration dictionary.

mappings: dict[str, TraceSchema] property

Returns the mappings of the Proto Codec.

Returns:

Type	Description
dict[str, TraceSchema]	mappings dictionary.

schemas: dict[str, TraceSchema] property

Returns the schemas of the Proto Codec.

Returns:

Type	Description
dict[str, TraceSchema]	schemas dictionary.

as_dict(message, conversion_strategy=None)

Converts the protobuf message to a dictionary.

Parameters:

Name	Type	Description	Default
message	Message	Protobuf message to convert.	required
conversion_strategy	Optional[dict]	Strategy for converting the message to a dictionary.	None

Returns:

Type	Description
dict[str, Any]	Dictionary representation of the protobuf message.

generate_schema(name, descriptor)

Updates the schema of a given named descriptor.

Returns:

Type	Description
	update a descriptor's schema.

get_schema(name, descriptor)

Returns the schema of a given named message. :note: schemas are lazily generated if not already computed

Returns:

Type	Description
	a message's schema.

has_mapping(message)

Returns true if the message has a mapping key defined

Returns:

Type	Description
bool	true if the message has a mapping key defined

is_map(descriptor) staticmethod

Returns true if the descriptor is for a map field

Returns:

Type	Description
bool	true if the descriptor is for a map field

trace(message, name=None, time_ns=None, conversion_strategy=None)

Emits a trace of the protobuf message in accordance with the conversion strategy.

Parameters:

Name	Type	Description	Default
message	Message	Protobuf message to trace.	required
name	Optional[str]	Name for the trace entry. Defaults to the message's descriptor name.	None
time_ns	Optional[int]	Timestamp for the trace entry in nanoseconds. Defaults to current time.	None
conversion_strategy	Optional[dict]	Strategy for converting the message to a dictionary.	None

zeloscloud.codecs.serial

Concrete codec/signal implementation classes for serial comms

SerialCodec(name, link, config, **kwargs)

Bases: CodecBase

Codec for interacting with a cmd/query like interface

The codec uses a configuration to set up the signals representing commands or queries on a link.

Initialize the SerialCodec with a name, link, and a configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of the codec.	required
link	LinkBase	Interface to the physical bus, (ex: serial link)	required
config	Dict[str, Union[str, Dict[str, Any]]]	command/query signal configurations	required

zeloscloud.codecs.serial.messages

TODO

Classes: - QueryMessageSignal: - QueryMessage:

QueryMessage(name, codec, query, signals, **kwargs)

Bases: MessageBase

Initializes a QueryMessage which represents a single query containing multiple signal values.

Parameters:

Name	Type	Description	Default
name	str	The name of the query message.	required
codec	CodecBase	The codec that this message belongs to.	required
query	Union[str, Dict[str, str]]	The query command or a dict containing the command and pattern.	required
signals	Dict[str, Dict[str, Any]]	A dictionary of the messages signal configurations.	required
kwargs		Additional keyword arguments.	{}

update()

Sends the query command and updates all signals with the received data.

QueryMessageSignal(name, pattern, **kwargs)

Bases: ReadWriteSignal

Initializes the QueryMessageSignal, a specialized ReadWriteSignal that extracts and converts a specific piece of data from a larger message based on a regular expression pattern.

Parameters:

Name	Type	Description	Default
name	str	Name of the signal.	required
pattern	str	Regex pattern to extract the signal value from a message.	required

zeloscloud.codecs.serial.signals

Defines concrete implementations of signal classes for interacting with a serial interface

Classes: - QuerySignal: Represents a read-only signal specific for querying data. - CmdSignal: Represents a write-only signal for setting a command. 'get' will return the last set value - CmdQuerySignal: Represents a read-write signal for both setting a command and querying data

CmdQuerySignal(name, codec, cmd, query, **kwargs)

Bases: CmdSignal, QuerySignal

A signal for commanding and querying data from a link.

Initialize the CmdQuerySignal

Parameters:

Name	Type	Description	Default
name	str	Name of the signal.	required
codec	CodecBase	The codec that this signal belongs to.	required
cmd	str	The command string.	required
query	Union[str, Dict[str, str]]	The query command string.	required

CmdSignal(name, codec, cmd, **kwargs)

Bases: ReadWriteSignal

A Signal that sends a command on a link.

Attributes: - _cmd (str): The command associated with this signal.

Initialize the CmdSignal

Parameters:

Name	Type	Description	Default
name	str	Name of the signal.	required
codec	CodecBase	The codec that this signal belongs to.	required
cmd	str	The command string.	required

QuerySignal(name, codec, query, **kwargs)

Bases: ReadSignal

A signal for querying data from a link.

This Signal only allows reading (i.e., sending a query to the device and getting a response). A query can also contain a regex pattern if applicable for parsing out an exact response.

Attributes: - _query (str): The query command associated with this signal. - _pattern (str): The pattern for extracting the value from the query

Initialize the QuerySignal

Parameters:

Name	Type	Description	Default
name	str	Name of the signal.	required
codec	CodecBase	The codec that this signal belongs to.	required
query	Union[str, Dict[str, str]]	The query command string.	required

zeloscloud.codecs.uds

Concrete codec/signal implementation classes for CAN comms

HexDidCodec()

Bases: DidCodec

Codec to perform simple binary Data Identifier read / writes

to_bytes(data) staticmethod

Convert data to bytes

UdsCodec(name, link, config=UdsConfig.create(), **kwargs)

Bases: CodecBase

Codec for interacing with a CAN bus

Initialize the UdsCodec with a name, link, and a configuration. Forwards **kwargs to super() init as well as udsoncan.client.Client()

TODO(tkeairns): Implement support for .odx, .json, and other UDS definition schemes

Parameters:

Name	Type	Description	Default
name	str	Name of the codec	required
link	UdsLink	UdsLink to facilitate UDS comms	required
config	UdsConfig	UdsConfig containing all UDS config data	create()

Raises:

Type	Description
TypeError	on invalid config

client: udsoncan.client.Client property

WARNING:

Performing UDS operations with the client directly is dangerous.
Prefer client access through the `client_threadsafe` context manager.
Codec methods that depend on the underlying client provide a threadsafe API.

Returns:

Type	Description
Client	The underlying udsoncan.client.Client object.

config: UdsConfig property

Returns:

Type	Description
UdsConfig	The internal UdsConfig object.

task: Task property

Returns:

Type	Description
Task	The internal Task object.

add_data_identifier(definition)

Adds a data identifier to the codec's signal set and data identifier configuration dictionary.

This method facilitates adding a data identifier along with its associated codec definition to the codec's signal set and data identifier configuration. It accepts either a UdsDidDefinition object or a dictionary containing the necessary information. If a dictionary is provided, it will be converted to a UdsDidDefinition object before proceeding with the addition.

Parameters:

Name	Type	Description	Default
definition	Union[UdsDidDefinition, dict]	The definition of the data identifier to be added.	required

Returns:

Type	Description
None	None

add_file(path_and_name, data_format_identifier=None, file_size=None)

Requests the server to create a new file.

Parameters:

Name	Type	Description	Default
path_and_name	str	Path and name of the file to create	required
data_format_identifier	Optional[int]	Optional identifier specifying the compression and encryption method of the file	None
file_size	Optional[int]	Optional expected size of the file in bytes	None

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by RequestFileTransfer.interpret_response

block_generator(data, block_size) staticmethod

Generate blocks of data to be used in a transfer sequence.

This method generates blocks of data from the provided data with the specified block size. It yields tuples containing sequence numbers and corresponding data blocks.

:yields: A tuple containing sequence number and data block

Parameters:

Name	Type	Description	Default
data	bytes	The data to be divided into blocks	required
block_size	int	The size of each data block	required

change_session(session)

Change the diagnostic session to the specified session.

client_threadsafe(block=True, timeout=DEFAULT_CLIENT_LOCK_TIMEOUT_SEC)

Context manager for accessing the underlying udsoncan.client.Client object safely.

Acquires a lock, yields the client object, and releases the lock when done.

Example Usage:

with codec.client_threadsafe() as client:
    client.read_data_by_identifier(...)

:yields: The underlying udsoncan.client.Client object

client_threadsafe_async() async

In a round about way, wraps the threading.Lock with an async-friendly condition. Async Context manager for accessing the underlying udsoncan.client.Client object safely. Acquires a lock, yields the client object, and releases the lock when done.

:yields: The underlying udsoncan.client.Client object

delete_file(path_and_name)

Requests the server to delete an existing file.

Parameters:

Name	Type	Description	Default
path_and_name	str	Path and name of the file to delete	required

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by RequestFileTransfer.interpret_response

ecu_reset(reset_type=ECUReset.ResetType.hardReset, response_required=True)

Resets the ECU.

Parameters:

Name	Type	Description	Default
reset_type	ResetType	The type of reset to perform	hardReset

freeze_current_state(did, return_response=False)

Freezes the current state of an input/output specific parameter.

Parameters:

Name	Type	Description	Default
did	int	The data identifier for IO control	required

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by InputOutputControlByIdentifier.interpret_response

get_did_value(values, did) staticmethod

Retrieve the value corresponding to a data identifier from a response.

This method extracts the value associated with the provided data identifier from a response object. The data identifier can be provided as a UdsSignal object, a string, or an integer.

Parameters:

Name	Type	Description	Default
values	Any	The response.service_data.values dictionary from a response.	required
did	Union[UdsSignal, str, int]	The data identifier whose value needs to be retrieved.	required

Returns:

Type	Description
Any	The value associated with the provided data identifier.

get_routine_results(routine_id, data=None, return_response=False)

Get results from a UDS routine with the specified routine identifier.

Parameters:

Name	Type	Description	Default
routine_id	int	The identifier of the routine to get results from	required
data	bytes	Optional data to send with the routine control request	None
return_response	bool	Whether to return the response from the routine control request	False

Returns:

Type	Description
Any	The response from the routine control request

io_control(did, control_param, control_state=None, control_mask=None, return_response=False)

Performs an InputOutputControlByIdentifier service request.

Parameters:

Name	Type	Description	Default
did	int	The data identifier for IO control	required
control_param	int	The control parameter (e.g., returnControlToECU, resetToDefault, freezeCurrentState, shortTermAdjustment)	required
control_state	Any	Optional control state data	None
control_mask	Any	Optional control mask data	None

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by InputOutputControlByIdentifier.interpret_response

normalize_identifier(did)

Normalize a Data Identifier (DID) or list of DIDs into a list of integers.

This method takes a Data Identifier (DID) or a list of DIDs and returns a list of corresponding integer identifiers. If a single DID is provided, it is converted into a list with one element.

Parameters:

Name	Type	Description	Default
did	Union[UdsDidIdentifier, list[UdsDidIdentifier]]	The Data Identifier (DID) or list of DIDs to normalize.	required

Returns:

Type	Description
list[int]	A list of integer identifiers corresponding to the normalized DID(s).

perform_download_sequence(data, address=0, address_format=32, block_size=None)

Perform a request download and transfer data sequence.

This method initiates a request download and transfers the provided data in blocks over the CAN bus. It handles the sequence of actions required for the download process, including changing the diagnostic session to programming session, requesting download, transferring data in blocks, and requesting transfer exit.

Parameters:

Name	Type	Description	Default
data	bytes	The data to be downloaded	required
address	int	The starting address for the download operation. Defaults to 0x0	0
address_format	int	The format of the memory address. Defaults to 32-bit	32
block_size	Optional[int]	The size of each block of data to transfer. If not provided, uses server-provided size	None

Returns:

Type	Description
None	None

perform_upload_sequence(size, address=0, address_format=32, block_size=None)

Perform a request upload and transfer data sequence.

This method initiates a request upload and receives the data in blocks from the ECU. It handles the sequence of actions required for the upload process, including changing the diagnostic session to programming session, requesting upload, receiving data blocks, and requesting transfer exit.

Parameters:

Name	Type	Description	Default
size	int	The size of data to upload in bytes	required
address	int	The starting address for the upload operation. Defaults to 0x0	0
address_format	int	The format of the memory address. Defaults to 32-bit	32
block_size	Optional[int]	The size of each block of data to transfer. If not provided, uses server-provided size	None

Returns:

Type	Description
bytes	The uploaded data as bytes

read_data_by_identifier(did, return_response=False)

Perform a read data by identifier transaction and return the corresponding data.

This method performs a read data by identifier transaction for the specified Data Identifier (DID) or list of DIDs. It returns a dictionary containing the retrieved data, where the keys represent the identifiers and the values represent the corresponding data.

Parameters:

Name	Type	Description	Default
did	Union[UdsDidIdentifier, list[UdsDidIdentifier]]	The Data Identifier (DID) or list of DIDs to read data from.	required

Returns:

Type	Description
dict[int, Any]	A dictionary containing the retrieved data.

read_dir(path_and_name)

Requests to read a directory.

Parameters:

Name	Type	Description	Default
path_and_name	str	Path of the directory to read	required

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by RequestFileTransfer.interpret_response

read_file(path_and_name, data_format_identifier=None)

Requests to read data from an existing file.

Parameters:

Name	Type	Description	Default
path_and_name	str	Path and name of the file to read	required
data_format_identifier	Optional[int]	Optional identifier specifying the compression and encryption method of the file	None

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by RequestFileTransfer.interpret_response

replace_file(path_and_name, data_format_identifier=None, file_size=None)

Requests the server to replace an existing file.

Parameters:

Name	Type	Description	Default
path_and_name	str	Path and name of the file to replace	required
data_format_identifier	Optional[int]	Optional identifier specifying the compression and encryption method of the file	None
file_size	Optional[int]	Optional expected size of the file in bytes	None

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by RequestFileTransfer.interpret_response

reset_to_default(did, return_response=False)

Resets an input/output specific parameter to default value.

Parameters:

Name	Type	Description	Default
did	int	The data identifier for IO control	required

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by InputOutputControlByIdentifier.interpret_response

return_control_to_ecu(did, return_response=False)

Returns control of an input/output specific parameter to the ECU.

Parameters:

Name	Type	Description	Default
did	int	The data identifier for IO control	required

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by InputOutputControlByIdentifier.interpret_response

routine_control(routine_id, control_type, data=None, return_response=False)

Internal method to handle routine control requests.

Parameters:

Name	Type	Description	Default
routine_id	int	The identifier of the routine	required
control_type	int	The type of routine control (start/stop/results)	required
data	bytes	Optional data to send with the routine control request	None
return_response	bool	Whether to return the response from the routine control request	False

Returns:

Type	Description
Any	The response from the routine control request

send_tp(response_required=False)

Sends a Tester Present (TP) message over the diagnostic session.

Parameters:

Name	Type	Description	Default
response_required	bool	Whether a response is required after sending the TP message. Defaults to False.	False

short_term_adjustment(did, control_state, control_mask=None, return_response=False)

Performs short term adjustment of an input/output specific parameter.

Parameters:

Name	Type	Description	Default
did	int	The data identifier for IO control	required
control_state	Any	The control state to set	required
control_mask	Any	Optional control mask	None

Returns:

Type	Description
Optional[InterpretedResponse]	The server response parsed by InputOutputControlByIdentifier.interpret_response

start_did_periodic(did_periodic)

Starts a periodic task for the specified Data Identifier (DID).

This method adds a periodic task for the specified Data Identifier (DID) to the _did_periodics dictionary. If a periodic task already exists for the DID, it is stopped before creating a new one.

Parameters:

Name	Type	Description	Default
did_periodic	Union[UdsDidPeriodic, dict]	The Data Identifier (DID) periodic task definition.	required

Returns:

Type	Description
None	None

start_periodics()

Starts all periodic tasks configured in the UDS codec.

This method starts all configured periodic tasks, including the Tester Present (TP) periodic task and any Data Identifier (DID) periodic tasks specified in the configuration.

Returns:

Type	Description
None	None

start_routine(routine_id, data=None, return_response=False)

Start a UDS routine with the specified routine identifier.

Parameters:

Name	Type	Description	Default
routine_id	int	The identifier of the routine to start	required
data	bytes	Optional data to send with the routine control request	None
return_response	bool	Whether to return the response from the routine control request	False

Returns:

Type	Description
Any	The response from the routine control request

start_tp(period=DEFAULT_TP_PERIOD_SEC)

Starts the Tester Present (TP) periodic task.

This method starts the Tester Present (TP) periodic task with the specified period. If a TP periodic task is already active, it is stopped before starting a new one.

Parameters:

Name	Type	Description	Default
period	float	The period in seconds at which the TP task should execute. Defaults to DEFAULT_TP_PERIOD_SEC if not provided.	DEFAULT_TP_PERIOD_SEC

Returns:

Type	Description
None	None

stop_did_periodic(did)

Stops all periodics that match the given Data Identifier(s).

This method cancels the periodic task associated with the specified Data Identifier (DID). If the DID is not found in the _did_periodics dictionary, no action is taken.

Parameters:

Name	Type	Description	Default
did	Union[UdsDidIdentifier, list[UdsDidIdentifier]]	The Data Identifier (DID) or list of DIDs to stop periodic tasks for.	required

Returns:

Type	Description
None	None

stop_periodics()

Stops all periodic tasks including Tester Present (TP) and Data Identifier (DID) periodics.

This method cancels all active periodic tasks, including the TP periodic task and any DID periodic tasks registered in the _did_periodics dictionary.

Returns:

Type	Description
None	None

stop_routine(routine_id, data=None, return_response=False)

Stop a UDS routine with the specified routine identifier.

Parameters:

Name	Type	Description	Default
routine_id	int	The identifier of the routine to stop	required
data	bytes	Optional data to send with the routine control request	None
return_response	bool	Whether to return the response from the routine control request	False

Returns:

Type	Description
Any	The response from the routine control request

stop_tp()

Stops the Tester Present (TP) periodic task.

This method cancels the Tester Present (TP) periodic task if it is currently active.

Returns:

Type	Description
None	None

write_data_by_identifier(did, value)

Perform a write to a data identifier.

This method performs a write operation to the specified Data Identifier (DID) or list of DIDs with the provided value.

Parameters:

Name	Type	Description	Default
did	UdsDidIdentifier	The Data Identifier (DID).	required
value		The value to write to the specified DID(s).	required

Returns:

Type	Description
Any	The result of the write operation.

UdsConfig

Bases: TypedDict

Configuration class for UdsCodec

UdsDidDefinition(identifier, did_codec, name=None) dataclass

Data class defining a Data Identifier register

UdsSignal(identifier, did_codec, codec, name=None, **kwargs)

Bases: ReadWriteSignal

Represents a signal in a message on a CAN bus.

Attributes:

Initialize the UdsSignal.

did_codec property

returns the data identifier codec

identifier property

returns the identifier value

zeloscloud.codecs.visa

Concrete codec/signal implementation classes for visa comms

VisaCodec(name, link, config=None, **kwargs)

Bases: CodecBase

Codec for interacting with a cmd/query like interface

The codec uses a configuration to set up the signals representing commands or queries on a link.

Initialize the VisaCodec with a name, link, and a configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of the codec.	required
link	LinkBase	Interface to the physical bus, (ex: visa link)	required
config	Dict[str, Union[str, Dict[str, Any]]]	command/query signal configurations	None

zeloscloud.aio.codecs.grpc

GrpcCodec(name, link, config, **kwargs)

Bases: CodecBase

Async gRPC codec implementation that handles protocol-level communication.

Initialize the gRPC codec.

Parameters:

Name	Type	Description	Default
name	str	Name of the codec	required
link	LinkBase	GrpcLink instance	required
config	GrpcCodecConfig	Configuration containing stub_class	required
kwargs	Any	Additional arguments	{}

Raises:

Type	Description
ValueError	If required configuration is missing

bidirectional_stream(method_name, request_iterator, timeout=None, metadata=None, message_name=None, request_name=None, response_name=None) async

Make a stream-stream gRPC call.

Parameters:

Name	Type	Description	Default
method_name	str	Name of the gRPC method to call	required
request_iterator	Iterator[Message]	Iterator yielding request messages	required
timeout	Optional[float]	Optional timeout in seconds	None
metadata	Optional[list]	Optional metadata list	None

Returns:

Type	Description
AsyncGenerator[Message, None]	AsyncGenerator yielding response messages

Raises:

Type	Description
grpc.RpcError	On gRPC errors

call(method_name, request, timeout=None, metadata=None, message_name=None, request_name=None, response_name=None) async

Make a unary-unary gRPC call.

Parameters:

Name	Type	Description	Default
method_name	str	Name of the gRPC method to call	required
request	Union[Message, ProtoMessage]	Request message	required
timeout	Optional[float]	Optional timeout in seconds	None
metadata	Optional[list]	Optional metadata list	None

Returns:

Type	Description
Message	Response message

Raises:

Type	Description
grpc.RpcError	On gRPC errors

client_stream(method_name, request_iterator, timeout=None, metadata=None, message_name=None, request_name=None, response_name=None) async

Make a stream-unary gRPC call.

Parameters:

Name	Type	Description	Default
method_name	str	Name of the gRPC method to call	required
request_iterator	Iterator[Union[Message, ProtoMessage]]	Iterator yielding request messages	required
timeout	Optional[float]	Optional timeout in seconds	None
metadata	Optional[list]	Optional metadata list	None

Returns:

Type	Description
Message	Response message

Raises:

Type	Description
grpc.RpcError	On gRPC errors

stream(method_name, request, timeout=None, metadata=None, message_name=None, request_name=None, response_name=None) async

Make a unary-stream gRPC call.

Parameters:

Name	Type	Description	Default
method_name	str	Name of the gRPC method to call	required
request	Union[Message, ProtoMessage]	Request message	required
timeout	Optional[float]	Optional timeout in seconds	None
metadata	Optional[list]	Optional metadata list	None

Returns:

Type	Description
AsyncGenerator[Message, None]	AsyncGenerator yielding response messages

Raises:

Type	Description
grpc.RpcError	On gRPC errors

Links

zeloscloud.links

Links are the way to connect and communicate with hardware.

zeloscloud.links.serial_link

This module defines the SerialLink class, an implementation of a link for sending and receiving data over a serial connection. Under the hood, the SerialLink class handles serial communication using the pyserial library.

Example Usage:

To use the SerialLink, you need to specify the serial port and baud rate. Optionally, you can specify a timeout and other parameters supported by pyserial.

Example:

from zeloscloud.links import SerialLink

# Initialize the serial link and open the link
with SerialLink(name="MySerialLink", port="/dev/ttyUSB0", baud=9600) as link:
    # Send some data
    link.send(b'Hello, world!')

    # Receive data
    data = link.recv()
    print(f"Received data: {data}")

See Also:

• pyserial documentation: https://pyserial.readthedocs.io/en/latest/pyserial.html

SerialLink(name, port, baud, timeout_s=1, bytes_to_strip=[b'\r', b'\n'], **kwargs)

Bases: LinkBase

Represents a serial link for sending and receiving data on a serial connection.

    Constructs all the necessary attributes for the SerialLink object.

    :param name: The name of the serial link.
    :param port: The port of the serial connection.
    :param baud: The baud rate for the serial connection.
    :param timeout_s: The timeout in seconds for the serial connection (default: 1).
    :param bytes_to_strip: A list of bytes to automatically remove from received data
                        (default: [b'

', b' ']).

recv(num_bytes=None, timeout_s=None)

Receives data from the serial connection. By default, if num_bytes is none, it will read until a newline or a timeout occurs.

Parameters:

Name	Type	Description	Default
num_bytes	Optional[int]	The number of bytes to read, defaults to None.	None
timeout_s	Optional[int]	The timeout for reading, defaults to None.	None

Returns:

Type	Description
bytes	The received data.

Raises:

Type	Description
IOError	If the connection is not open.

send(data)

Sends data on the serial connection.

Parameters:

Name	Type	Description	Default
data	bytes	The data to be sent.	required

Raises:

Type	Description
IOError	If the connection is not open.

zeloscloud.links.can_link

Defines an implementation class for a link that sends/receives data over CAN

CanLink(config, loop=None, name=None, **kwargs)

Bases: LinkBase

Link for sending and receiving on a CAN connection.

Initialize a CAN link.

Parameters:

Name	Type	Description	Default
config	BusConfig	Configuration for the CAN bus.	required
loop		Event loop, optional.	None
name	Optional[str]	Name of the link, optional.	None
kwargs		Additional keyword arguments.	{}

add_listener(listener)

Adds a listener to the CAN bus

Parameters:

Name	Type	Description	Default
listener	MessageRecipient	The listener to add	required

recv(**kwargs)

Receives a message from a CAN bus.

Parameters:

Name	Type	Description	Default
kwargs		Additional keyword arguments.	{}

Returns:

Type	Description
Message	The received CAN message.

Raises:

Type	Description
IOError	If the connection is not open.

send(message, **kwargs)

Sends a message on a CAN bus.

Parameters:

Name	Type	Description	Default
message	Message	The CAN message to be sent.	required
kwargs		Additional keyword arguments.	{}

Raises:

Type	Description
IOError	If the connection is not open.
NotImplementedError	If the message type is unknown.

send_periodic(message, **kwargs)

Sends a message periodically on a CAN bus.

Parameters:

Name	Type	Description	Default
message	Message	The CAN message to be sent periodically.	required
kwargs		Additional keyword arguments.	{}

Returns:

Type	Description
CyclicSendTaskABC	A task that manages the periodic sending of the message.

Raises:

Type	Description
IOError	If the connection is not open.
NotImplementedError	If the message type is unknown.

zeloscloud.links.uds_link

Defines an implementation class for a link that sends/receives data over CAN

UdsLink(name, connection, **kwargs)

Bases: LinkBase

Link for send/recv'ing UDS A thin wrapper around udsoncan's BaseConnection

Parameters:

Name	Type	Description	Default
name		the name of the UDS link	required
connection	BaseConnection	a concrete instance of a BaseConnection	required

Initialize a can link.

connection property

Exposes the underlying BaseConnection

can_connection(can_link, isotp_addr, isotp_overrides=None) staticmethod

'Factory' function to assist in building a python-can based connection and corresponding UdsLink

Parameters:

Name	Type	Description	Default
isotp_addr		args to pass to isotp. Addressing docs: https://can-isotp.readthedocs.io/en/latest/isotp/addressing.html	required
isotp_overrides	Optional[dict]	optional arg to override default isotp parameters	None

recv(**kwargs)

Wraps BaseConnection.wait_until and satisfies the LinkBase.recv abstractmethod

send(data, **kwargs)

Sends Raw UDS traffic on a connection

zeloscloud.links.visa_link

Defines an implementation class for a link that sends/receives data over visa

VisaLink(name, identifier, id_query=DEFAULT_ID_QUERY, resource_config=None, library='@py', **kwargs)

Bases: LinkBase

Link for send/recv'ing on a visa connection

:name str: the name of the visa link

Initialize a visa link.

query(query, **kwargs)

Sends query on a visa connection.

recv(**kwargs)

Receives message from a visa connection.

send(message, **kwargs)

Sends message on a visa connection.

VisaLinkException

Bases: Exception

Visa Link Exception

zeloscloud.links.ble

Provides link interfaces for BLE (Bluetooth Low Energy) communication.

BleClientLinkBase(name, link_config=None, *args, **kwargs)

Bases: LinkBase

Abstract base class for BLE link objects.

Initialize the BLE link base class.

Parameters:

Name	Type	Description	Default
name	str	The name of the link.	required
link_config	Optional[Dict]	Configuration options for the BLE link.	None
args		Positional arguments passed to the function.	()
kwargs		Keyword arguments passed to the function.	{}

connect() abstractmethod

Connect to a BLE device.

disconnect() abstractmethod

Disconnect from the BLE device.

discover_devices() abstractmethod

Discover available BLE devices.

Returns:

Type	Description
List[Dict[str, str]]	A list of discovered BLE devices, each represented as a dictionary with 'address' and 'name' keys.

discover_services() abstractmethod

Discover services provided by the connected BLE device.

Returns:

Type	Description
List	A list of services.

is_connected() abstractmethod

Checks if the link was successfully connected or not.

Returns:

Type	Description
bool	True if the client successfully connected, else False.

read_characteristic(characteristic_uuid) abstractmethod

Read a characteristic value.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to read.	required

Returns:

Type	Description
bytes	The value of the characteristic.

subscribe_to_notifications(characteristic_uuid, callback) abstractmethod

Subscribe to notifications from a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic.	required
callback	Callable[[str, bytes], None]	The function to call when a notification is received.	required

unsubscribe_from_notifications(characteristic_uuid) abstractmethod

Unsubscribe from notifications from a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic.	required

write_characteristic(characteristic_uuid, data) abstractmethod

Write a value to a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to write.	required
data	bytes	The value to write.	required

BleClientLinkError

Bases: Exception

Base Exception for a BleClientLink

BleakBleLink(name, link_config)

Bases: BleClientLinkBase

Concrete implementation of BleClientLinkBase using bleak.

For more on bleak: https://bleak.readthedocs.io/en/latest/index.html

Initialize the BleakBleLink.

Parameters:

Name	Type	Description	Default
name	str	The name of the ble link.	required
link_config	Dict[str, Any]	Configuration options for the BLE link. - address (str): The address of the BLE device to connect to. - timeout (Optional[float]): Connection timeout in seconds. - adapter (Optional[str]): The Bluetooth adapter to use. - mtu (Optional[int]): The maximum transmission unit size. - disconnected_callback (Optional[Callable[[BleakClient], None]]): A callback function to handle disconnections. - scan_filter (Optional[Dict[str, Any]]): A filter to apply when scanning for devices.	required

client: Optional[BleakClient] property

Get the Bleak client handler.

is_connected: bool property

Checks the connection status between the client and the server.

Returns:

Type	Description
bool	True if connected, else False.

async_connect() async

Asynchronous method to connect to a BLE device.

async_disconnect() async

Asynchronous method to disconnect from the BLE device.

async_discover_devices() async

Asynchronous method to discover available BLE devices.

Returns:

Type	Description
List[Tuple[str, str]]	A list of discovered BLE devices.

async_discover_services() async

Asynchronous method to discover services provided by the connected BLE device.

Returns:

Type	Description
List	A list of services.

async_pair(**kwargs) async

Asynchronous method to pair with the specified GATT server.

This method is not available on macOS. Instead of manually initiating pairing, the user will be prompted to pair the device the first time that a characteristic that requires authentication is read or written. This method may have backend-specific additional keyword arguments.

Returns:

Type	Description
bool	True if paired and connected, else False

async_read_characteristic(characteristic_uuid) async

Asynchronous method to read a characteristic value.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to read.	required

Returns:

Type	Description
bytes	The value of the characteristic.

async_subscribe_to_notifications(characteristic_uuid, callback) async

Asynchronous method to subscribe to notifications from a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic.	required
callback	Callable[[str, bytes], None]	The function to call when a notification is received.	required

async_unpair() async

Asynchronous method to unpair from the specified GATT server and disconnect the device.

This method is only available on Windows and Linux and will raise an exception on other platforms.

Returns:

Type	Description
bool	True if unpaired with device.

async_unsubscribe_from_notifications(characteristic_uuid, callback=None) async

Asynchronous method to unsubscribe from notifications from a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic.	required
callback	Optional[Callable[[str, bytes], None]]	The specific callback to remove. If None, remove all callbacks.	None

async_write_characteristic(characteristic_uuid, data) async

Asynchronous method to write a value to a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to write.	required
data	bytes	The value to write.	required

connect()

Connect to a BLE device.

disconnect()

Disconnect from the BLE device.

discover_devices()

Discover available BLE devices.

Returns:

Type	Description
List[Tuple[str, str]]	A list of discovered BLE devices.

discover_services()

Discover services provided by the connected BLE device.

Returns:

Type	Description
List	A list of services.

pair(**kwargs)

Pair with the specified GATT server.

Returns:

Type	Description
bool	True if paired and connected, else False

read_characteristic(characteristic_uuid)

Read a characteristic value.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to read.	required

Returns:

Type	Description
bytes	The value of the characteristic.

recv(characteristic_uuid)

Recv a characteristic value.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to recv.	required

Returns:

Type	Description
bytes	The value of the characteristic.

send(characteristic_uuid, data)

Send data to a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to write.	required
data	bytes	The value to send.	required

subscribe_to_notifications(characteristic_uuid, callback)

Subscribe to notifications from a characteristic.

Example callback:

def callback(sender, data: bytearray): print(f"{sender}: {data}")

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic.	required
callback	Callable[[str, bytes], None]	The function to call when a notification is received.	required

unpair()

Unpair from the specified GATT server and disconnect the device.

Returns:

Type	Description
None	True if unpaired with device.

unsubscribe_from_notifications(characteristic_uuid, callback=None)

Unsubscribe from notifications from a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic.	required
callback	Optional[Callable[[str, bytes], None]]	The specific callback to remove. If None, remove all callbacks.	None

write_characteristic(characteristic_uuid, data)

Write a value to a characteristic.

Parameters:

Name	Type	Description	Default
characteristic_uuid	str	The UUID of the characteristic to write.	required
data	bytes	The value to write.	required

zeloscloud.aio.links

zeloscloud.aio.links.grpc_link

Defines an implementation class for an async link that sends/receives data over gRPC.

Raises:

Type	Description
ValueError	If required configuration fields are missing

GrpcChannelConfig

Bases: TypedDict

TypedDict for gRPC channel configuration with type hints and defaults. Only target is required, all other fields are optional.

Attributes:

Name	Type	Description
target	str	The server address (required)
use_secure	bool	Whether to use a secure channel
credentials	Optional[ChannelCredentials]	SSL credentials for secure connections
connect_timeout	float	Timeout for initial connection in seconds
options	List[Tuple[str, Union[str, int, float]]]	List of additional channel options as tuples
metadata	Optional[List[Tuple[str, str]]]	List of metadata key-value pairs

GrpcLink(name, config, **kwargs)

Bases: LinkBase

Async gRPC link implementation for handling raw channel communication.

Provides low-level gRPC channel operations with: - Automatic channel management - Connection state handling - Metadata support - Channel options configuration

Example config: { "target": "localhost:50051", "use_secure": False, "credentials": grpc.ssl_channel_credentials(), "options": [ ("grpc.keepalive_time_ms", 10000), ("grpc.keepalive_timeout_ms", 5000), ], "metadata": [ ("key", "value"), ] }

Initialize the gRPC link.

Parameters:

Name	Type	Description	Default
name	str	Name of the link	required
config	dict	Link configuration	required
kwargs	Any	Additional keyword arguments	{}

Raises:

Type	Description
ValueError	If required configuration is missing

channel: grpc.aio.Channel property

Get the underlying gRPC channel.

Returns:

Type	Description
Channel	The gRPC channel

Raises:

Type	Description
RuntimeError	If channel is not initialized

metadata: Optional[List[Tuple[str, str]]] property

Get the metadata for gRPC calls.

Returns:

Type	Description
Optional[List[Tuple[str, str]]]	List of metadata key-value pairs or None

check_connectivity() async

Get current channel connectivity state.

Returns:

Type	Description
ChannelConnectivity	Current connectivity state

recv(*args, **kwargs) async

Receive is not implemented for gRPC links. gRPC communication should be handled through the channel directly.

Parameters:

Name	Type	Description	Default
args	Any	Positional arguments	()
kwargs	Any	Keyword arguments	{}

Raises:

Type	Description
NotImplementedError	Always raised as method is not implemented

send(*args, **kwargs) async

Send is not implemented for gRPC links. gRPC communication should be handled through the channel directly.

Parameters:

Name	Type	Description	Default
args	Any	Positional arguments	()
kwargs	Any	Keyword arguments	{}

Raises:

Type	Description
NotImplementedError	Always raised as method is not implemented

wait_for_state_change(timeout=None) async

Wait for the channel state to change.

Parameters:

Name	Type	Description	Default
timeout	Optional[float]	Maximum time to wait in seconds	None

Returns:

Type	Description
bool	True if state changed, False if timeout

create_channel_config(config)

Create a properly typed channel configuration by merging user config with defaults.

Parameters:

Name	Type	Description	Default
config	dict	User-provided configuration dictionary	required

Returns:

Type	Description
GrpcChannelConfig	Merged configuration with proper typing

Raises:

Type	Description
ValueError	If required fields are missing

Plugins

zeloscloud.pytest.checker

Checker plugin for pytest.

Use via:

pytest_plugins = ["zeloscloud.pytest.checker"]

Checker(title=None, fail_fast=True)

Bases: ContextBase

A collection of checker assertions.

Initialize a Checklist with a name.

Parameters:

Name	Type	Description	Default
title	Optional[str]	Optional, adds additional title text to the checker board	None
fail_fast	bool	Optional, Flag to determine if a check should fail right away or delay until the end of all checks. (Default: True)	True

that(lhs, op='is true', rhs=None, for_duration_s=None, within_duration_s=None, interval_s=0.1, blocking=True, args=(), kwargs={})

Add and evaluate a checker assertion.

Parameters:

Name	Type	Description	Default
lhs	Union[SignalBase, Any]	Left-hand side operand.	required
op	Union[str, DescriptiveOp]	Operator for the check.	'is true'
rhs	Union[SignalBase, Any]	Right-hand side operand.	None
within_duration_s	float	The condition should be true within this duration.	None
for_duration_s	float	The condition should be true for this duration.	None
interval_s	float	Interval between condition checks.	0.1
blocking	bool	Whether the polling should be blocking.	True
args	tuple	Additional arguments for the operator.	()
kwargs	dict	Additional keyword arguments for the operator.	{}

zeloscloud.pytest.checker.ops

This module provides a framework for defining and using descriptive operations, primarily utilized in assertion checks. The core of this framework is the DescriptiveOp class, which wraps callable operations with descriptive text.

Example Usage:

from zeloscloud.pytest.checker import ops

# Registering custom operations
def custom_op(x, y): return x * 2 == y
ops.register_op(ops.DescriptiveOp(custom_op, "double equals"))

# Using a registered operation
eq_op = ops.get_op("equals")
assert eq_op(3, 3)

# Listing all available operations
ops.list_ops()
# prints -> ['equals', 'double equals']

DescriptiveOp(op, description=None)

Represents a descriptive callable operation. This is used to display more descriptive text in the check board operator output.

Parameters:

Name	Type	Description	Default
op	Callable[..., bool]	A callable object that implements the operation logic.	required
description	str	A string describing the operation. Defaults to name of function.	None

__call__(*args, **kwargs)

Executes the operation with the provided arguments.

Parameters:

Name	Type	Description	Default
args		Arguments to pass to the operation's function.	()
kwargs		Additional keyword arguments to pass to the operation's function.	{}

Returns:

Type	Description
bool	The result of the operation as a boolean.

Raises:

Type	Description
OpsError	If there is an error in executing the operation.

OpsError

Bases: Exception

Base exception class for operator errors.

get_op(description)

Retrieves an operation based on its description.

Parameters:

Name	Type	Description	Default
description	str	The description of the operation.	required

Returns:

Type	Description
DescriptiveOp	Instance of DescriptiveOp corresponding to the description.

Raises:

Type	Description
OpsError	If no operation matches the description.

list_ops()

Returns a list of all registered operation descriptions.

Returns:

Type	Description
list	List of operation descriptions.

pytest_approx(lhs, rhs, rel=None, abs=None, nan_ok=False)

Wraps the pytest approx function

register_op(op, overwrite=False)

Registers a new operation using its description as the key. Optionally allows overwriting an existing operation.

Parameters:

Name	Type	Description	Default
op	DescriptiveOp	Instance of DescriptiveOp to be registered.	required
overwrite	bool	Boolean flag to allow overwriting existing operations (default is False).	False

Raises:

Type	Description
ValueError	If an operation with the same description is already registered and overwrite is False.

zeloscloud.pytest.report

Report plugin for pytest.

Use via:

pytest_plugins = ["zeloscloud.pytest.report"]

pytest_zelos_configure(config)

configure zelos report plugin

zeloscloud.pytest.trace

Zelos Cloud trace plugin

Use via:

pytest_plugins = ["zeloscloud.pytest.trace"]

pytest_addhooks(pluginmanager)

This example assumes the hooks are grouped in the 'hooks' module.

pytest_addoption(parser)

Parses flags that can enable/disable specific event handlers.

Parameters:

Name	Type	Description	Default
parser	Parser	A pytest Parser object to add the command-line options.	required

trace_class(request)

Initialize and manage trace handlers for each test class.

:yield: Manages the lifecycle of trace handlers without returning a value.

Parameters:

Name	Type	Description	Default
request		The pytest request object.	required

trace_function(request)

Initialize and manage trace handlers for each test function.

:yield: Manages the lifecycle of trace handlers without returning a value.

Parameters:

Name	Type	Description	Default
request		The pytest request object.	required

trace_module(request)

Initialize and manage trace handlers for each module.

:yield: Manages the lifecycle of trace handlers without returning a value.

Parameters:

Name	Type	Description	Default
request		The pytest request object.	required

trace_session(request)

Initialize and manage trace handlers for the entire test session.

:yield: Manages the lifecycle of trace handlers without returning a value.

Parameters:

Name	Type	Description	Default
request		The pytest request object.	required

zeloscloud.pytest.zr

Zelos Cloud task plugin

Use via:

pytest_plugins = ["zeloscloud.pytest.zr"]

zeloscloud.pytest.zr.zirconium

Zirconium class

Provides a synchronous interface for interacting with a Zirconium board via gRPC

To generate the Python classes from the proto files, run (from api/py/src): python3 -m grpc_tools.protoc --proto_path ../../../components/agent/proto/ --python_out=zeloscloud/pytest/zr/proto/ --grpc_python_out=zeloscloud/pytest/zr/proto/ --mypy_out=zeloscloud/pytest/zr/proto/ --mypy_grpc_out=zeloscloud/pytest/zr/proto/ hello_world.proto zirconium.proto

ZirconiumCodec(name, grpc, config, pericon, carbon, task=None, **kwargs)

Bases: CodecBase

Codec for interacting with a Zirconium board via gRPC Provides synchronous interface by wrapping async operations in a Task

Initialize the ZirconiumCodec with gRPC support

dev_mode property

Get the current dev mode configuration

gpio property

Get the current state of all GPIOs

logging property

Get the current logging configuration

__getattr__(name)

Delegate attribute access to the gRPC stub

get_all_current_gpio()

Get all last known GPIO states without making a new request

Returns:

Type	Description
Dict[str, bool]	Dictionary of GPIO names to their current states

get_all_gpio(timeout_s=ZR_DEFAULT_TIMEOUT)

Get all GPIO states

get_all_gpio_directions(timeout_s=ZR_DEFAULT_TIMEOUT)

Get all GPIO directions

get_current_gpio_state(name)

Get the last known state of a GPIO without making a new request

Parameters:

Name	Type	Description	Default
name	str	GPIO name	required

Returns:

Type	Description
Optional[bool]	Current state or None if unknown

get_gpio_direction(name, timeout_s=ZR_DEFAULT_TIMEOUT)

Get a GPIO's current direction

get_gpio_value(name, timeout_s=ZR_DEFAULT_TIMEOUT)

Get a GPIO's current value

set_gpio_direction(name, direction, timeout_s=ZR_DEFAULT_TIMEOUT)

Configure a GPIO's direction

set_gpio_value(name, value, timeout_s=ZR_DEFAULT_TIMEOUT)

Set a GPIO's value (for outputs)

start_monitor(name, callback)

Monitor a GPIO with a callback

Parameters:

Name	Type	Description	Default
name	str	GPIO name to monitor	required
callback	Callable[[bool], None]	Function to call with GPIO state updates	required

stop_monitor(name)

Stop monitoring a GPIO

ZirconiumConfig

Bases: TypedDict

Configuration for the ZirconiumCodec

ZirconiumDebugConfig

Bases: TypedDict

Debug configuration for the ZirconiumCodec

ZirconiumGpioSignal(name, value, direction, codec, fixed=False)

Bases: ReadWriteSignal

A GPIO signal that can be read and written to

set(value)

Sets the value of the GPIO and updates the codec

set_direction(direction)

Sets the direction of the GPIO and updates the codec

update(value)

Updates only the value

ZirconiumLoggingConfig

Bases: TypedDict

Logging configuration for the ZirconiumCodec

ZirconiumTraceConfig

Bases: TypedDict

Trace configuration for the ZirconiumCodec

Config

zeloscloud.config

zeloscloud.config.station

Helpers for parsing configuration from a file.

ZelosNamespace(objects)

Bases: ContextBase

construct_overrides(args)

Parse a list of strings in the format dotted.key.path=literal to a nested dictionary. Since common shells (bash / zsh) strip quotes from arguments, fall back to assuming users mean to treat arguments as a string.

Examples:

>>> construct_overrides(['can.interface=can0'])
{'can': {'interface': 'can0'}}

>>> construct_overrides(['serial.baud=9600'])
{'serial': {'baud': 9600}}

>>> construct_overrides(['can.interface="can0"', 'can_codec.dbc="./example.dbc"'])
{'can': {'interface': 'can0'}, 'can_codec': {'dbc': './example.dbc'}}

load_config(config, overrides=None)

Taking a config ordered dictionary and an optional dictionary of overrides, initialize objects from the config file and return a ZelosNamespace object that contains all of the un-opened objects.

load_file(file, overrides=None)

Read a configuration from a file path, optionally specifying overrides.

load_string(config, overrides=None)

Read a configuration from a string, optionally specifying overrides.