Public API
gallia.command
- class gallia.command.AsyncScript(parser: ArgumentParser, config: Config = {})[source]
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 = {})[source]
Bases:
FlockMixin
,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 = {})[source]
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 = {})[source]
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 = {})[source]
Bases:
Scanner
- GROUP: str | None = 'discover'
The group name when used in the gallia CLI.
- class gallia.command.UDSScanner(parser: ArgumentParser, config: Config = {})[source]
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)[source]
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.Loglevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
IntEnum
A wrapper around the constants exposed by python’s
logging
module. Since gallia adds two additional loglevel’s (NOTICE
andTRACE
), this class provides a type safe way to access the loglevels. The levelNOTICE
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, seePenlogPriority
.
- class gallia.log.PenlogPriority(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]
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 [source]
Converts an int value (e.g. from python’s logging module) to an instance of this class.
- classmethod from_str(string: str) PenlogPriority [source]
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
).
- 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)[source]
Bases:
object
- gallia.log.resolve_color_mode(mode: ~gallia.log.ColorMode, stream: ~typing.TextIO = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>) bool [source]
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 [source]
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
- gallia.plugins.add_cli_group(parent: Parsers, group: str, help_: str, metavar: str, description: str | None = None, epilog: str | None = None) None [source]
Adds a group to the gallia CLI interface. The arguments correspond to the arguments of
argparse.ArgumentParser.add_argument()
. Theparent
argument must contain the relevant entry point to the cli parse tree. The parse tree is passed to the entry_pointgallia_cli_init
.
- gallia.plugins.load_cli_init_plugin_eps() list[EntryPoint] [source]
Loads the
gallia_cli_init
entry_point.
- gallia.plugins.load_cli_init_plugins() list[Callable[[Parsers], None]] [source]
Loads the
gallia_cli_init
entry_point and imports the functions behind it.
- gallia.plugins.load_command_plugin_eps() list[EntryPoint] [source]
Loads the
gallia_commands
entry_point.
- gallia.plugins.load_command_plugins() list[type[BaseCommand]] [source]
Loads the
gallia_commands
entry_point and imports the command classes with some sanity checks.
- gallia.plugins.load_ecu(vendor: str) type[ECU] [source]
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 stringdefault
selects a generic ECU.
- gallia.plugins.load_ecu_plugin_eps() list[EntryPoint] [source]
Loads the
gallia_uds_ecus
entry_point.
- gallia.plugins.load_ecu_plugins() list[type[ECU]] [source]
Loads the
gallia_uds_ecus
entry_point and imports the ecu classes with some sanity checks.
- gallia.plugins.load_transport(target: TargetURI) type[BaseTransport] [source]
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] [source]
Loads the
gallia_transports
entry_point.
- gallia.plugins.load_transport_plugins() list[type[BaseTransport]] [source]
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 = 0, power_supply: PowerSupply | None = None)[source]
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 [source]
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 [source]
Clear all dtc records on the ecu.
- async leave_session(level: int, config: UDSRequestConfig | None = None, sleep: int | None = None) bool [source]
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 [source]
Send an UDS TesterPresent message.
- Returns:
UDS response.
- async read_dtc(config: UDSRequestConfig | None = None) NegativeResponse | ReportDTCByStatusMaskResponse [source]
Read all dtc records from the ecu.
- async read_session(config: UDSRequestConfig | None = None) int [source]
Read out current session.
- Returns:
The current session as int.
- async read_vin(config: UDSRequestConfig | None = None) NegativeResponse | ReadDataByIdentifierResponse [source]
Read the VIN of the vehicle
- async refresh_state(reset_state: bool = False) None [source]
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 [source]
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 [source]
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 [source]
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
- class gallia.services.uds.NegativeResponse(request_service_id: int, response_code: UDSErrorCodes)[source]
Bases:
NegativeResponseBase
- class gallia.services.uds.PositiveResponse[source]
Bases:
UDSResponse
,ABC
- class gallia.services.uds.SubFunctionRequest(suppress_response: bool)[source]
Bases:
UDSRequest
,ABC
- RESPONSE_TYPE
alias of
SubFunctionResponse
- class gallia.services.uds.SubFunctionResponse[source]
Bases:
PositiveResponse
,ABC
- class gallia.services.uds.UDSErrorCodes(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
IntEnum
- class gallia.services.uds.UDSIsoServices(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
IntEnum
gallia.transports
All available transports are documented in Transports.
- class gallia.transports.BaseTransport(target: TargetURI)[source]
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 thetags
property of the relevantgallia.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 [source]
Checks if the provided URI has the correct scheme.
- abstract async close() None [source]
Terminates the connection and clean up all allocated ressources.
- abstract async classmethod connect(target: str | TargetURI, timeout: float | None = None) Self [source]
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 [source]
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 [source]
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. 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 [source]
Chains a
write()
call with aread()
call. The call is protected by a mutex and is thus safe for concurrent use.
- class gallia.transports.DoIPTransport(target: TargetURI, port: int, config: DoIPConfig, conn: DoIPConnection)[source]
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 classmethod connect(target: str | TargetURI, timeout: float | None = None) Self [source]
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 [source]
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 [source]
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. A new instance of this class is returned rendering the old one obsolete. This method is safe for concurrent use.
- class gallia.transports.HSFZTransport(target: TargetURI, port: int, config: HSFZConfig, conn: HSFZConnection)[source]
Bases:
BaseTransport
- BUFSIZE: int = 8192
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 classmethod connect(target: str | TargetURI, timeout: float | None = None) HSFZTransport [source]
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.
- class gallia.transports.ISOTPTransport(target: TargetURI, config: ISOTPConfig, sock: socket)[source]
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 classmethod connect(target: str | TargetURI, timeout: float | None = None) Self [source]
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.
- class gallia.transports.RawCANTransport(target: TargetURI, config: RawCANConfig, sock: socket)[source]
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 classmethod connect(target: str | TargetURI, timeout: float | None = None) Self [source]
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] [source]
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.
- class gallia.transports.TCPLinesTransport(target: TargetURI, reader: StreamReader, writer: StreamWriter)[source]
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)[source]
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 classmethod connect(target: str | TargetURI, timeout: float | None = None) Self [source]
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.
- class gallia.transports.TargetURI(raw: str)[source]
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()
andparse_qs()
methods. TargetURI provides frequently used properties for a more userfriendly usage. Instances are meant to be passed toBaseTransport.connect()
of transport implementations.- classmethod from_parts(scheme: str, host: str, port: int | None, args: dict[str, Any]) Self [source]
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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
StrEnum
- class gallia.transports.UnixLinesTransport(target: TargetURI, reader: StreamReader, writer: StreamWriter)[source]
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)[source]
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 classmethod connect(target: str | TargetURI, timeout: float | None = None) Self [source]
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.
opennetzteil.netzteil
- class opennetzteil.netzteil.BaseNetzteil(target: TargetURI, timeout: float | None)[source]
Bases:
ABC
- PRODUCT_ID = ''
The product_id is used to choose the relevant implementation.
- async classmethod connect(target: TargetURI, timeout: float | None) BaseNetzteil [source]
Connects to
target
and checks for connectivity usingprobe()
.
- abstract async get_ocp(channel: int) bool [source]
Gets the state (on/off) of the overcurrent protection.
- abstract async get_output(channel: int) bool [source]
Returns the state (on/off) of the supplied channel.
- abstract async get_ovp(channel: int) bool [source]
Gets the state (on/off) of the overvoltage protection.
- async probe() None [source]
Checks for connectivity. The default implementation reads out the version number and model string by calling
get_ident()
.
- abstract async set_ocp(channel: int, enabled: bool) None [source]
Sets the state (on/off) of the overcurrent protection.
- abstract async set_output(channel: int, enabled: bool) None [source]
Sets the state (on/off) of the supplied channel.