katgpucbf.main module

Utilities for writing the main programs of engines.

class katgpucbf.main.SubParser

Bases: object

Parser for a single command-line argument that has suboptions.

The sub-options use key=value syntax, separated by commas. This is not as complete as a full argparse parser, but supports some basics:

• required arguments

• defaults

• types

It does not currently support actions (such as store_true). Specifying a keyword more than once results in an error.

It is callable, so that an instance can be passed directly as the type argument of argparse.ArgumentParser.add_argument().

add_argument(name: str, *, type: ~collections.abc.Callable[[str], ~typing.Any] = <class 'str'>, required: bool = False, default: ~typing.Any = None) → None

Add an argument.

katgpucbf.main.add_common_arguments(parser: ArgumentParser, *, katcp: bool = True, prometheus: bool = True, aiomonitor: bool = True, version: bool = True) → None

Add command-line arguments to the parser.

katgpucbf.main.add_gc_stats() → None

Add Prometheus metrics for garbage collection timing.

It is only safe to call this once.

katgpucbf.main.add_recv_arguments(parser: ArgumentParser, *, multi: bool = False) → None

Add arguments for receiving interface (supporting ibverbs).

If multi is true, the arguments take comma-separated lists.

katgpucbf.main.add_send_arguments(parser: ArgumentParser, *, prefix: str = 'send-', multi: bool = False, ibverbs: bool = True, affinity: bool = True, rate_factor: bool = True) → None

Add arguments for sending interface (optionally supporting ibverbs).

Parameters:

• parser – Parser to which arguments are added.

• prefix – Prefix to use on argument names.

• multi – If true, multiple interfaces are supported.

• ibverbs – If true, ibverbs-related options will be added.

• affinity – If true, a separate thread is used for sending, and the core affinity may be set.

• rate_factor – If true, provide the --send-rate-factor command-line argument.

katgpucbf.main.add_signal_handlers(server: DeviceServer) → None

Arrange for clean shutdown on SIGINT (Ctrl-C) or SIGTERM.

katgpucbf.main.add_time_converter_arguments(parser: ArgumentParser, sync_time_required: bool = True) → None

Add arguments necessary for constructing a TimeConverter.

katgpucbf.main.comma_split(base_type: Callable[[str], T], count: int | None = None, allow_single=False) → Callable[[str], list[T]]

Return a function to split a comma-delimited str into a list of type T.

This function is used to parse lists of parameters which come from the command-line as comma-separated strings, but are obviously more useful as a list.

Parameters:

• base_type – The base type of thing you expect in the list, e.g. int, float.

• count – How many of them you expect to be in the list. None means the list could be any length.

• allow_single – If true (defaults to false), allow a single value to be used when count is greater than 1. In this case, it will be repeated count times.

katgpucbf.main.engine_main(args: Namespace, start_engine: Callable[[Namespace, TaskGroup, AsyncExitStack, MutableMapping], Awaitable[DeviceServer]]) → None

Run an engine.

This takes care of:

• running an event loop;

• setting up logging;

• running a web server for Prometheus scraping if requested on the command line;

• running aiomonitor

• adding Prometheus statistics for the garbage collector (GC);

• freezing the GC after starting the engine and unfreezing it on shutdown;

Parameters:

• args – The command-line arguments.

• start_engine –

  The function that sets up and starts the engine. It takes the following parameters:

  • The command-line arguments (args)

  • A task group that can be used to schedule on-going work that will be waited on before shutting down.

  • An asynchronous exit stack that can be used to enter contexts or schedule cleanup work.

  • A dictionary of variables to expose to aiomonitor.

katgpucbf.main.parse_dither(value: str) → DitherType

Parse a string into a dither type.

katgpucbf.main.parse_enum(name: str, value: str, cls: type[E]) → E

Parse a command-line argument into an enum type.

katgpucbf.main.parse_source(value: str) → list[tuple[str, int]] | str

Parse a string into a list of IP endpoints or a filename.

katgpucbf.main.parse_source_ipv4(value: str, default_port=7148) → list[tuple[str, int]]

Parse a string into a list of IPv4 endpoints.