kafka_dae_control.config

Utilities for reading Control IOC configuration from TOML.

Members

ControlConfig	Create a new model by parsing and validating input data from keyword arguments.
load_config	Validate and load a config file at the specified path.

class kafka_dae_control.config.ControlConfig(*, board_ip: IPv4Address, board_xml: Path, pv_prefix: str, instrument_name: str, sample_env_topic: str, events_topic: str, runinfo_topic: str, vetoes_topic: str, local_ip: IPv4Address, poll_interval_s: float = 1.0, kafka_producer: dict[str, str], state_file: Path = WindowsPath('state.json'), pv_update_interval_s: float = 0.1, read_port: int = 10000, write_port: int = 10002, flush_timeout_s: int = 1, veto_names: Annotated[list[str], MinLen(min_length=32), MaxLen(max_length=32)] = ['veto_0', 'veto_1', 'veto_2', 'veto_3', 'veto_4', 'veto_5', 'veto_6', 'veto_7', 'veto_8', 'veto_9', 'veto_10', 'veto_11', 'veto_12', 'veto_13', 'veto_14', 'veto_15', 'veto_16', 'veto_17', 'veto_18', 'veto_19', 'veto_20', 'veto_21', 'veto_22', 'veto_23', 'veto_24', 'veto_25', 'veto_26', 'veto_27', 'veto_28', 'veto_29', 'veto_30', 'veto_31'], resend_ip_after_connection_failures: Annotated[int, Gt(gt=0)] = 10)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property register_map: dict[str, int]

Parse the register XML file to get a mapping of register names to addresses.

kafka_dae_control.config.load_config(config_path: str) → ControlConfig

Validate and load a config file at the specified path.