Public API

gallia.command

class gallia.command.AsyncScript(parser: ArgumentParser, config: Config = {})

Bases: BaseCommand, ABC

AsyncScript is a base class for a asynchronous gallia command. To implement an async script, create a subclass and implement the .main() method.

GROUP: str | None = 'script'

The group name when used in the gallia CLI.

class gallia.command.BaseCommand(parser: ArgumentParser, config: Config = {})

Bases: ABC

BaseCommand 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().

CATCHED_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.

COMMAND: str | None = None

The command name when used in the gallia CLI.

EPILOG: str | None = None

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

GROUP: str | None = None

The group name when used in the gallia CLI.

HAS_ARTIFACTS_DIR: bool = False

Enable a artifacts_dir. Setting this property to True enables the creation of a logfile.

SHORT_HELP: str | None = None

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

SUBGROUP: str | None = None

The subgroup name when used in the gallia CLI.

class gallia.command.Scanner(parser: ArgumentParser, config: Config = {})

Bases: AsyncScript, ABC

Scanner is a base class for all scanning related commands. A scanner has the following properties:

• It is async.

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

• 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.

• pcap logfiles can be recorded via a Dumpcap background task.

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

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

CATCHED_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.

GROUP: str | None = 'scan'

The group name when used in the gallia CLI.

HAS_ARTIFACTS_DIR: bool = True

Enable a artifacts_dir. Setting this property to True enables the creation of a logfile.

class gallia.command.Script(parser: ArgumentParser, config: Config = {})

Bases: BaseCommand, ABC

Script is a base class for a synchronous gallia command. To implement a script, create a subclass and implement the .main() method.

GROUP: str | None = 'script'

The group name when used in the gallia CLI.

class gallia.command.UDSDiscoveryScanner(parser: ArgumentParser, config: Config = {})

Bases: Scanner

GROUP: str | None = 'discover'

The group name when used in the gallia CLI.

class gallia.command.UDSScanner(parser: ArgumentParser, config: Config = {})

Bases: Scanner

UDSScanner is a baseclass, particularly for scanning tasks related to the UDS protocol. The differences to Scanner are:

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

• A background tasks sends TesterPresent regularly to avoid timeouts.

GROUP: str | None = 'scan'

The group name when used in the gallia CLI.

SUBGROUP: str | None = 'uds'

The subgroup name when used in the gallia CLI.

gallia.log

class gallia.log.ColorMode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

ColorMode is used as an argument to set_color_mode().

ALWAYS = 'always'

Colors are always turned on.

AUTO = 'auto'

Colors are turned off if the target stream (e.g. stderr) is not a tty.

NEVER = 'never'

No colors are used. In other words, no ANSI escape codes are included.

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

Bases: Logger

class gallia.log.Loglevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

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, colored: '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.resolve_color_mode(mode: ~gallia.log.ColorMode, stream: ~typing.TextIO = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>) → bool

Sets the color mode of the console log handler.

Parameters:

• mode – The available options are described in ColorMode.

• stream – Used as a reference for ColorMode.AUTO.

gallia.log.setup_logging(level: Loglevel | None = None, color_mode: ColorMode = ColorMode.AUTO, no_volatile_info: bool = False, logger_name: str = 'gallia') → 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. If this argument is None, the env variable GALLIA_LOGLEVEL (see Environment Variables) is read.

• 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

class gallia.plugins.Parsers

Bases: TypedDict

gallia.plugins.add_cli_group(parent: Parsers, group: str, help_: str, metavar: str, description: str | None = None, epilog: str | None = None) → None

Adds a group to the gallia CLI interface. The arguments correspond to the arguments of argparse.ArgumentParser.add_argument(). The parent argument must contain the relevant entry point to the cli parse tree. The parse tree is passed to the entry_point gallia_cli_init.

gallia.plugins.load_cli_init_plugin_eps() → list[EntryPoint]

Loads the gallia_cli_init entry_point.

gallia.plugins.load_cli_init_plugins() → list[Callable[[Parsers], None]]

Loads the gallia_cli_init entry_point and imports the functions behind it.

gallia.plugins.load_command_plugin_eps() → list[EntryPoint]

Loads the gallia_commands entry_point.

gallia.plugins.load_command_plugins() → list[type[BaseCommand]]

Loads the gallia_commands entry_point and imports the command classes with some sanity checks.

gallia.plugins.load_ecu(vendor: str) → type[ECU]

Selects an ecu class depending on a vendor string. The lookup is performed in builtin ecus and all classes behind the gallia_uds_ecus entry_point. The vendor string default selects a generic ECU.

gallia.plugins.load_ecu_plugin_eps() → list[EntryPoint]

Loads the gallia_uds_ecus entry_point.

gallia.plugins.load_ecu_plugins() → list[type[ECU]]

Loads the gallia_uds_ecus entry_point and imports the ecu classes with some sanity checks.

gallia.plugins.load_transport(target: TargetURI) → type[BaseTransport]

Selects a transport class depending on a TargetURI. The lookup is performed in builtin transports and all classes behind the gallia_transports entry_point.

gallia.plugins.load_transport_plugin_eps() → list[EntryPoint]

Loads the gallia_transports entry_point.

gallia.plugins.load_transport_plugins() → list[type[BaseTransport]]

Loads the gallia_transports entry_point and imports the transport classes with some sanity checks.

gallia.services.uds

class gallia.services.uds.ECU(transport: BaseTransport, timeout: float, max_retry: int = 1, 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(level: int, config: UDSRequestConfig | None = None, sleep: int | 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 ping(config: UDSRequestConfig | None = None) → NegativeResponse | TesterPresentResponse

Send an UDS TesterPresent message.

Returns:

UDS response.

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(level: 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(level: 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 | None = None) → bool

Wait for ecu to be alive again (e.g. after reset). Sends a ping every 0.5s and waits at most timeout

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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

class gallia.services.uds.UDSIsoServices(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

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 = 8192

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.

abstract async close() → None

Terminates the connection and clean up all allocated ressources.

abstract async classmethod connect(target: str | TargetURI, timeout: float | None = None) → Self

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.

abstract 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) → Self

Closes the connection to the target and reconnects. A new instance of this class is returned rendering the old one obsolete. 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.

abstract 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, port: int, config: DoIPConfig, conn: DoIPConnection)

Bases: BaseTransport

BUFSIZE: int = 8192

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 ressources.

async classmethod connect(target: str | TargetURI, timeout: float | None = None) → DoIPTransport

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 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, config: ISOTPConfig, sock: socket)

Bases: BaseTransport

BUFSIZE: int = 8192

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 ressources.

async classmethod connect(target: str | TargetURI, timeout: float | None = None) → ISOTPTransport

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 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, config: RawCANConfig, sock: socket)

Bases: BaseTransport

BUFSIZE: int = 8192

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 ressources.

async classmethod connect(target: str | TargetURI, timeout: float | None = None) → RawCANTransport

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 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, reader: StreamReader, writer: StreamWriter)

Bases: LinesTransportMixin, TCPTransport

BUFSIZE: int = 8192

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, reader: StreamReader, writer: StreamWriter)

Bases: BaseTransport

BUFSIZE: int = 8192

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 ressources.

async classmethod connect(target: str | TargetURI, timeout: float | None = None) → TCPTransport

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 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]) → TargetURI

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: str

The URI scheme

class gallia.transports.UnixLinesTransport(target: TargetURI, reader: StreamReader, writer: StreamWriter)

Bases: LinesTransportMixin, UnixTransport

BUFSIZE: int = 8192

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, reader: StreamReader, writer: StreamWriter)

Bases: BaseTransport

BUFSIZE: int = 8192

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 ressources.

async classmethod connect(target: str | TargetURI, timeout: float | None = None) → UnixTransport

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 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.

opennetzteil.netzteil

class opennetzteil.netzteil.BaseNetzteil(target: TargetURI, timeout: float | None)

Bases: ABC

PRODUCT_ID = ''

The product_id is used to choose the relevant implementation.

async classmethod connect(target: TargetURI, timeout: float | None) → BaseNetzteil

Connects to target and checks for connectivity using probe().

abstract async get_channels() → int

Returns the number of available channels.

abstract async get_current(channel: int) → float

Returns the max. current value.

abstract async get_ident() → str

Reads the version number and model string.

abstract async get_master() → bool

Reads the status of the master switch.

abstract async get_ocp(channel: int) → bool

Gets the state (on/off) of the overcurrent protection.

abstract async get_output(channel: int) → bool

Returns the state (on/off) of the supplied channel.

abstract async get_ovp(channel: int) → bool

Gets the state (on/off) of the overvoltage protection.

abstract async get_voltage(channel: int) → float

Returns the max. voltage value.

async probe() → None

Checks for connectivity. The default implementation reads out the version number and model string by calling get_ident().

abstract async set_beep(enabled: bool) → None

Sets the state (on/off) of the alarm.

abstract async set_current(channel: int, value: float) → None

Sets the max. current value.

abstract async set_master(enabled: bool) → None

Sets the status of the master switch.

abstract async set_ocp(channel: int, enabled: bool) → None

Sets the state (on/off) of the overcurrent protection.

abstract async set_output(channel: int, enabled: bool) → None

Sets the state (on/off) of the supplied channel.

abstract async set_ovp(channel: int, enabled: bool) → None

Sets the state (on/off) of the overvoltage protection.

abstract async set_voltage(channel: int, value: float) → None

Sets the max. voltage value.