Helper Modules

Those modules exist to ease the development of most common use cases.

andi.autosar

Helper module for extract PDU messages from ethernet messages and transform it on receive message event of other ethernet messages to handle PDU messages. See documentation for more details PDU messages tutorial.

class andi.autosar.SocketAdaptor

Extract autosar PDUs from messages

get_pdus(message)

Parameters: message -- The ethernet message received from start capture method.
Returns: List of pdu messages from the captured ethernet message.

Return Type:: List[MessagePDU]

transform(handler)

Decorator function to transform ethernet messages to PDU messages.

Parameters: handler -- Get the list of PDU messages from ethernet message.

andi.ptp

Helper module for synchronizing ECUs clocks with PC clock

andi.ptp.cm_set_timestamp(channel, timestamp=None)

Parameters

channel (ChannelEthernet) -- The channel to send UDP message on.
timestamp (float) -- Timestamp to send in the UDP message. Defaults to current time.

andi.ptp.send_pdelay_request(channel, sequence_id, timestamp)

Parameters

channel (ChannelEthernet) -- The channel to send peer delay request message on.
sequence_id (int) -- Sequence ID of the peer delay request message.
timestamp (float) -- Timestamp to send in the peer delay request message.

Returns

Timestamp when the peer delay request message was sent.

Return type

timestamp (float)

andi.ptp.send_pdelay_response(channel, request)

Parameters

channel (ChannelEthernet) -- The channel to send peer delay response message on.
request (int) -- PTP message requesting of the peer delay response message.

Returns

Timestamp when the peer delay response message was sent.

Return type

timestamp (float)

andi.ptp.send_pdelay_response_followup(channel, request, timestamp)

Parameters

channel (ChannelEthernet) -- The channel to send peer delay response follow-up message on.
request (int) -- PTP message requesting of the peer delay response message.
timestamp (float) -- Timestamp when the peer delay response was sent.

Returns

Timestamp when the peer delay response follow-up message was sent.

Return type

timestamp (float)

andi.ptp.sync(channel, timestamp=None, sequence_id=0)

Parameters

channel (ChannelEthernet) -- The channel to send PTP messages on.
timestamp (float) -- Timestamp to send in the PTP messages. Defaults to current time.
sequence_id (int) -- Sequence ID to send in the PTP messages. Defaults 0.

andi.tecmp

Helper module for extract TECMP status messages from ethernet messages and return the parsed status. See documentation for more details TECMP messages tutorial.

andi.tecmp.get_bus_status(message)

Get bus status from an ethernet message.

Parameters: message (MessageEthernet) -- The received ethernet message.
Returns: BusStatus object if message is a valid tecmp bus status message otherwise 'none'.

Return Type:: BusStatus

andi.tecmp.get_cm_status(message)

Get capture module status from an ethernet message.

Parameters: message (MessageEthernet) -- The received ethernet message.
Returns: CmStatus object if message is a valid tecmp capture module status message otherwise 'none'.

Return Type:: CmStatus

andi.tecmp.get_control_data(message)

Get control message event from an ethernet message.

Parameters

message (MessageEthernet) -- The received ethernet message.
Returns --

:param ControlMessageData object if message is a valid tecmp control message otherwise 'none'.:

Return Type:: ControlMessageData

andi.e2e

Helper module to handle E2E messages. See documentation for more details E2E tutorial.

andi.e2e.get_actual_alive(message, e2e=None): Deprecated, use get_actual_counter() instead.

andi.e2e.get_actual_counter(message, e2e=None)

Gets counter value from a message based on e2e information.

Parameters

message (IMessageBase) -- The concerned message.
e2e (IE2EInformation) -- The e2e information.

Returns

UInt64 actual alive value if coherent message and e2e otherwise 'None'.

Return Type:: UInt64

andi.e2e.get_actual_crc(message, e2e=None)

Gets CRC value from a message based on e2e information.

Parameters

message (IMessageBase) -- The concerned message.
e2e (IE2EInformation) -- The e2e information.

Returns

UInt64 actual crc value if coherent message and e2e otherwise 'None'.

Return Type:: UInt64

andi.e2e.get_expected_crc(message, e2e=None)

Calculates CRC value based on a message and e2e information.

Parameters

message (IMessageBase) -- The concerned message.
e2e (IE2EInformation) -- The e2e information.

Returns

UInt64 expected crc value if coherent message and e2e otherwise 'None'.

Return Type:: UInt64

andi.e2e.get_next_counter(message, e2e=None)

Calculates next counter value based on the previous message and e2e information.

Parameters

message (IMessageBase) -- The previous message.
e2e (IE2EInformation) -- The e2e information.

Returns

UInt64 next alive value if coherent message and e2e otherwise 'None'.

Return Type:: UInt64

andi.e2e.increment_counter(message, e2e=None)

Updates the counter in the message's payload.

Parameters

message (IMessageBase) -- The concerned message.
e2e (IE2EInformation) -- The e2e information.

Returns

bool if the counter is updated or not.

Return Type:: bool

andi.e2e.protect(message, e2e=None)

Updates the CRC, Length and DataId in the message's payload.

Parameters

message (IMessageBase) -- The concerned message.
e2e (IE2EInformation) -- The e2e information.

Returns

bool if the fields are updated or not.

Return Type:: bool

andi.e2e.update_crc(message, e2e=None)

Updates the CRC in the message's payload.

Parameters

message (IMessageBase) -- The concerned message.
e2e (IE2EInformation) -- The e2e information.

Returns

bool if the CRC is updated or not.

Return Type:: bool

andi.isotp

Helper module to handle ISO-TP Socket implementation. See documentation for more details ISO-TP tutorial.

andi.ssh

Helper module to handle SSH (Secure Shell) implementation. See documentation for more details Using SSH Client.

andi.uds

Helper module to handle UDS messages. See documentation for more details UDS.

class andi.uds.DoIpConnection(message, protocol_version=2, activation_response_timeout=1000)

open()

Set up the connection object.

Returns: None

specific_send(payload)

The implementation of the send method.

Parameters: payload (bytes) -- Data to send
Returns: None

class andi.uds.HsfzConnection(message)

Sends and receives data through a HSFZ message.

specific_send(payload)

The implementation of the send method.

Parameters: payload (bytes) -- Data to send
Returns: None

class andi.uds.IsoTpConnection(message)

Sends and receives data through a ISO-TP message.

specific_send(payload)

The implementation of the send method.

Parameters: payload (bytes) -- Data to send
Returns: None

class andi.uds.MessageConnection(message)

close()

Close the connection object

Returns: None

empty_rxqueue()

Empty all unread data in the reception buffer.

Returns: None

open()

Set up the connection object.

Returns: None

specific_wait_frame(timeout=2)

The implementation of the wait_frame method.

Parameters: timeout (int) -- The maximum amount of time to wait before giving up
Returns: Received data
Return type: bytes or None

class andi.uds.UdsClient(conn, config={'data_identifiers': {}, 'dtc_snapshot_did_size': 2, 'exception_on_invalid_response': True, 'exception_on_negative_response': True, 'exception_on_unexpected_response': True, 'ignore_all_zero_dtc': True, 'input_output': {}, 'p2_star_timeout': 5, 'p2_timeout': 1, 'request_timeout': 5, 'security_algo': None, 'security_algo_params': None, 'server_address_format': None, 'server_memorysize_format': None, 'standard_version': 2020, 'tolerate_zero_padding': True, 'use_server_timing': True}, request_timeout=None)

Returns UDS client with specific target address target_addr.

Parameters: target_addr -- Target address (e.g. ECUName=0x10)
Returns: The UDS client
Return type: UdsClient

andi.someip

Helper module to handle SOME/IP messages.

andi.someip.reassemble(timeout: float = 1, forward_discarded: bool = True) → Callable[[SomeIPCallback], SomeIPCallback]

This is a Python decorator factory you can add to on_message_received callbacks It changes the callback so that it receives reassembled SOME/IP-TP messages

See https://docs.python.org/3/glossary.html#term-decorator for more inforamtion about Python decorators.

Parameters

timeout (float) -- Maximum time, after which incomplete TP transmissions are discarded
forward_discarded (bool) -- Should discarded TP messages be forwarded to the callback or not

Returns

a Python decorator

andi.someip.segmentize(someip: MessageSomeIP, max_segment_length: int) → Iterable[MessageSomeIP]

Segmentize the given SOME/IP message into a list of TP segment messages. If the payload length <= the given max segment length the original message is returned as a one-element list.

Parameters

someip (MessageSomeIP) -- the original message
max_segment_length (int) -- a segment's maximum payload length

Returns

segmented messages as a list

andi.converters

Helper module to convert PCAP or PCAPNG files with various options. See documentation for more details andi.converters.

andi.converters.archivize(infile: Union[str, PathLike], formats: List[str], *, tecmp_mapping: Optional[Dict[int, str]] = None) → None

Parameters

infile (AnyPath) -- The path to the input network packet capture file.
formats (List[str]) -- A list of file formats to convert the input file to.
tecmp_mapping (Optional[Dict[int, str]]) -- A dictionary mapping technique IDs to their names, optional.

Returns

None

andi.converters.convert(infile: Union[str, PathLike], outfile: Union[str, PathLike], *, tecmp_mapping: Optional[Dict[int, str]] = None) → None

Parameters

infile (AnyPath) -- The path to the input network packet capture file.
outfile (AnyPath) -- The path to the output file, including optional compression.
tecmp_mapping (Optional[Dict[int, str]]) -- A dictionary mapping technique IDs to their names, optional.

Returns

None