Public API

gallia.command

class gallia.command.AsyncScript(config: AsyncScriptConfig)

Bases: ABC

AsyncScript is the baseclass for all gallia commands. This class can be used in standalone scripts via the gallia command line interface facility.

This class needs to be subclassed and all the abstract methods need to be implemented. The artifacts_dir is generated based on the COMMAND, GROUP, SUBGROUP properties (falls back to the class name if all three are not set).

The main entry_point is entry_point().

CAUGHT_EXCEPTIONS: list[type[Exception]] = []

A list of exception types for which tracebacks are suppressed at the top level. For these exceptions a log message with level critical is logged.

CONFIG_TYPE

alias of AsyncScriptConfig

EPILOG: str | None = None

The string which is shown at the bottom of –help.

SHORT_HELP: str | None = None

The string which is shown on the cli with –help.

class gallia.command.UDSScanner(config: UDSScannerConfig)

Bases: AsyncScript, ABC

UDSScanner is a baseclass, particularly for scanning tasks related to the UDS protocol. It has the following properties:

• It loads transports via TargetURIs; available via self.transport.

• main() is the relevant entry_point for the scanner and must be implemented.

• Controlling PowerSupplies via the opennetzteil API is supported.

• setup() can be overwritten (do not forget to call super().setup()) for preparation tasks, such as establishing a network connection or starting background tasks.

• teardown() can be overwritten (do not forget to call super().teardown()) for cleanup tasks, such as terminating a network connection or background tasks.

• self.ecu contains a OEM specific UDS client object.

• A background tasks sends TesterPresent regularly to avoid timeouts.

CAUGHT_EXCEPTIONS: list[type[Exception]] = [<class 'ConnectionError'>, <class 'gallia.services.uds.core.exception.UDSException'>]

A list of exception types for which tracebacks are suppressed at the top level. For these exceptions a log message with level critical is logged.

result: Any = None

Depending on the Subclass of UDSScanner this property can contain results of various types. By default, it is None.

gallia.log

class gallia.log.Logger(name, level=0)

Bases: Logger

class gallia.log.Loglevel(*values)

Bases: IntEnum

A wrapper around the constants exposed by python’s logging module. Since gallia adds two additional loglevel’s (NOTICE and TRACE), this class provides a type safe way to access the loglevels. The level NOTICE was added to conform better to RFC3164. Subsequently, TRACE was added to have a facility for optional debug messages. Loglevel describes python specific values for loglevels which are required to integrate with the python ecosystem. For generic priority values, see PenlogPriority.

class gallia.log.PenlogPriority(*values)

Bases: IntEnum

PenlogPriority holds the values which are written to json log records. These values conform to RFC3164 with the addition of TRACE. Since Python uses different int values for the loglevels, there are two enums in gallia describing loglevels. PenlogPriority describes generic priority values which are included in json log records.

classmethod from_level(value: int) → PenlogPriority

Converts an int value (e.g. from python’s logging module) to an instance of this class.

classmethod from_str(string: str) → PenlogPriority

Converts a string to an instance of PenlogPriority. string can be a numeric value (0 to 8 inclusive) or a string with a case insensitive name of the level (e.g. debug).

to_level() → Loglevel

Converts an instance of PenlogPriority to Loglevel.

class gallia.log.PenlogRecord(module: 'str', host: 'str', data: 'str', datetime: 'datetime.datetime', priority: 'PenlogPriority', tags: 'list[str] | None' = None, colors: 'bool' = False, line: 'str | None' = None, stacktrace: 'str | None' = None, _python_level_no: 'int | None' = None, _python_level_name: 'str | None' = None, _python_func_name: 'str | None' = None)

Bases: object

gallia.log.remove_zst_log_handler(logger_name: str, handler: _ZstdFileHandler) → None

This function removes the handler from the specified logger and closes it

gallia.log.setup_logging(level: Loglevel = Loglevel.DEBUG, logger_name: str = 'gallia', volatile_info: bool = False, colors: bool = True, syslog_format: bool = False) → None

Enable and configure gallia’s logging system. If this fuction is not called as early as possible, the logging system is in an undefined state und might not behave as expected. Always use this function to initialize gallia’s logging. For instance, setup_logging() initializes a QueueHandler to avoid blocking calls during logging.

Parameters:

• level – The loglevel to enable for the console handler.

• file_level – The loglevel to enable for the file handler.

• path – The path to the logfile containing json records.

• color_mode – The color mode to use for the console.

gallia.plugins

gallia.services.uds

class gallia.services.uds.ECU(transport: BaseTransport, timeout: float, max_retry: int = 0, power_supply: PowerSupply | None = None)

Bases: UDSClient

ECU is a high level interface wrapping a UDSClient class. It provides semantically correct higher level interfaces such as read_session() or ping(). Vendor specific implementations can be derived from this class. For the arguments of the constructor, please check uds.uds.UDS.

async check_and_set_session(expected_session: int, retries: int = 3) → bool

check_and_set_session() reads the current session and (re)tries to set the session to the expected session if they do not match.

Returns True if the current session matches the expected session, or if read_session is not supported by the ECU or in the current session.

async clear_dtc(config: UDSRequestConfig | None = None) → NegativeResponse | ClearDiagnosticInformationResponse

Clear all dtc records on the ecu.

async leave_session(session: int, config: UDSRequestConfig | None = None, sleep: float | None = None) → bool

leave_session() is a hook which can be called explicitly by a scanner when a session is to be disabled. Use this hook if resetting the ECU is required, e.g. when disabling the programming session.

async power_cycle(sleep: float = 5) → bool

Perform a power cycle and wait for the ECU to recover.

Returns False on Error, and True if power cycle was successful or there is no power supply.

async read_dtc(config: UDSRequestConfig | None = None) → NegativeResponse | ReportDTCByStatusMaskResponse

Read all dtc records from the ecu.

async read_session(config: UDSRequestConfig | None = None) → int

Read out current session.

Returns:

The current session as int.

async read_vin(config: UDSRequestConfig | None = None) → NegativeResponse | ReadDataByIdentifierResponse

Read the VIN of the vehicle

async refresh_state(reset_state: bool = False) → None

Refresh the attributes of the ECU states, if possible. By, default, old values are only overwritten in case the corresponding information can be requested from the ECU and could be retrieved from a positive response from the ECU.

Parameters:

reset_state – If True, the ECU state is reset before updating it.

async set_session_post(session: int, config: UDSRequestConfig | None = None) → bool

set_session_post() is called after the diagnostic session control pdu was written on the wire. Implement this if there are special cleanup routines or sleeping until a certain moment is required.

Parameters:

• uds – The UDSClient class where this hook is embedded. The caller typically calls this function with self as the first argument.

• session – The desired session identifier.

Returns:

True on success, False on error.

async set_session_pre(session: int, config: UDSRequestConfig | None = None) → bool

set_session_pre() is called before the diagnostic session control pdu is written on the wire. Implement this if there are special preconditions for a particular session, such as disabling error logging.

Parameters:

• uds – The UDSClient class where this hook is embedded. The caller typically calls this function with self as the first argument.

• session – The desired session identifier.

Returns:

True on success, False on error.

async transmit_data(data: bytes, block_length: int, max_block_length: int = 4095, config: UDSRequestConfig | None = None) → None

transmit_data splits the data to be sent in several blocks of size block_length, transfers all of them and concludes the transmission with RequestTransferExit

async wait_for_ecu(timeout: float = 10) → bool

Wait for ECU to be alive again (e.g. after reset) for at most timeout.

If present, this method utilizes TesterPresentSender background job, otherwise it sends a TesterPresent request every 1s.

This function automatically handles necessary reconnects!

class gallia.services.uds.NegativeResponse(request_service_id: int, response_code: UDSErrorCodes)

Bases: NegativeResponseBase

class gallia.services.uds.PositiveResponse

Bases: UDSResponse, ABC

class gallia.services.uds.SubFunctionRequest(suppress_response: bool)

Bases: UDSRequest, ABC

RESPONSE_TYPE

alias of SubFunctionResponse

class gallia.services.uds.SubFunctionResponse

Bases: PositiveResponse, ABC

class gallia.services.uds.UDSErrorCodes(*values)

Bases: IntEnum

class gallia.services.uds.UDSIsoServices(*values)

Bases: IntEnum

class gallia.services.uds.UDSRequest

Bases: ABC

class gallia.services.uds.UDSRequestConfig(timeout: float | None = None, max_retry: int | None = None, skip_hooks: bool = False, tags: list[str] | None = None)

Bases: object

class gallia.services.uds.UDSResponse

Bases: ABC

gallia.transports

All available transports are documented in Transports.

class gallia.transports.BaseTransport(target: TargetURI)

Bases: ABC

BaseTransport is the base class providing the required interface for all transports used by gallia.

A transport usually is some kind of network protocol which carries an application level protocol. A good example is DoIP carrying UDS requests which acts as a minimal middleware on top of TCP.

This class is to be used as a subclass with all abstractmethods implemented and the SCHEME property filled.

A few methods provide a tags argument. The debug logs of these calls include these tags in the tags property of the relevant gallia.log.PenlogRecord.

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = ''

The scheme for the implemented protocol, e.g. “doip”.

classmethod check_scheme(target: TargetURI) → None

Checks if the provided URI has the correct scheme.

abstractmethod async close() → None

Terminates the connection and clean up all allocated resources.

abstractmethod async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

abstractmethod async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

abstractmethod async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async reconnect(timeout: float | None = None) → None

Closes the connection to the target and attempts to reconnect every 100 ms until at max timeout. If timeout is None, only attempt to connect once. This method is safe for concurrent use.

async request(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → bytes

Chains a write() call with a read() call. The call is protected by a mutex and is thus safe for concurrent use.

async request_unsafe(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → bytes

Chains a write() call with a read() call. The call is not protected by a mutex. Only use this method when you know what you are doing.

abstractmethod async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.

class gallia.transports.DoIPTransport(target: TargetURI)

Bases: BaseTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'doip'

The scheme for the implemented protocol, e.g. “doip”.

async close() → None

Terminates the connection and clean up all allocated resources.

async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async reconnect(timeout: float | None = None) → None

Closes the connection to the target and attempts to reconnect every 100 ms until at max timeout. If timeout is None, only attempt to connect once. This method is safe for concurrent use.

async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.

class gallia.transports.DummyTransport(target: TargetURI)

Bases: BaseTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'dummy'

The scheme for the implemented protocol, e.g. “doip”.

async close() → None

Terminates the connection and clean up all allocated resources.

async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.

class gallia.transports.HSFZTransport(target: TargetURI)

Bases: BaseTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'hsfz'

The scheme for the implemented protocol, e.g. “doip”.

async close() → None

Terminates the connection and clean up all allocated resources.

async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.

class gallia.transports.ISOTPTransport(target: TargetURI)

Bases: BaseTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'isotp'

The scheme for the implemented protocol, e.g. “doip”.

async close() → None

Terminates the connection and clean up all allocated resources.

async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.

class gallia.transports.RawCANTransport(target: TargetURI)

Bases: BaseTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'can-raw'

The scheme for the implemented protocol, e.g. “doip”.

async close() → None

Terminates the connection and clean up all allocated resources.

async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

async get_idle_traffic(sniff_time: float) → list[int]

Listen to traffic on the bus and return list of IDs which are seen in the specified period of time. The output of this function can be used as input to set_filter.

async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.

class gallia.transports.TCPLinesTransport(target: TargetURI)

Bases: LinesTransportMixin, TCPTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'tcp-lines'

The scheme for the implemented protocol, e.g. “doip”.

class gallia.transports.TCPTransport(target: TargetURI)

Bases: BaseTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'tcp'

The scheme for the implemented protocol, e.g. “doip”.

async close() → None

Terminates the connection and clean up all allocated resources.

async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.

class gallia.transports.TargetURI(raw: str)

Bases: object

TargetURI represents a target to which gallia can connect. The target string must conform to a URI is specified by RFC3986.

Basically, this is a wrapper around Python’s urlparse() and parse_qs() methods. TargetURI provides frequently used properties for a more userfriendly usage. Instances are meant to be passed to BaseTransport.connect() of transport implementations.

classmethod from_parts(scheme: str, host: str, port: int | None, args: dict[str, Any]) → Self

Constructs a instance of TargetURI with the given arguments. The args dict is used for the query string.

property hostname: str | None

The hostname (without port)

property location: str

A URI string which only consists of the relevant scheme, the host and the port.

property netloc: str

The hostname and the portnumber, separated by a colon.

property path: str

The path property of the url.

property port: int | None

The port number

property qs_flat: dict[str, str]

A dict which contains the query string’s key/value pairs. In case a key appears multiple times, this variant only contains the first found key/value pair. In contrast to qs, this variant avoids lists and might be easier to use for some cases.

property scheme: TransportScheme

The URI scheme

class gallia.transports.TransportScheme(*values)

Bases: StrEnum

class gallia.transports.UnixLinesTransport(target: TargetURI)

Bases: LinesTransportMixin, UnixTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'unix-lines'

The scheme for the implemented protocol, e.g. “doip”.

class gallia.transports.UnixTransport(target: TargetURI)

Bases: BaseTransport

BUFSIZE: int = 131072

The buffersize of the transport. Might be used in read() calls. Defaults to io.DEFAULT_BUFFER_SIZE.

SCHEME: str = 'unix'

The scheme for the implemented protocol, e.g. “doip”.

async close() → None

Terminates the connection and clean up all allocated resources.

async connect(timeout: float | None = None) → None

Classmethod to connect the transport to a relevant target. The target argument is a URI, such as doip://192.0.2.2:13400?src_addr=0xf4&dst_addr=0x1d” An instance of the relevant transport class is returned.

async dumpcap_argument_list() → list[str] | None

Return the arguments required for Dumpcap to capture traffic of this transport or None if not supported.

async read(timeout: float | None = None, tags: list[str] | None = None) → bytes

Reads one message and returns its raw byte representation. An example for one message is ‘one line, terminated by newline’ for a TCP transport yielding lines.

async write(data: bytes, timeout: float | None = None, tags: list[str] | None = None) → int

Writes one message and return the number of written bytes.