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, 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)

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.

board_ip: IPv4Address

IP address of the streaming control board

board_xml: Path

Path to the streaming control board’s firmware-generated XML file

events_topic: str

The topic to use when putting ev44 placeholders in the run start nexus structure

flush_timeout_s: int

The timeout for a flush after producing run info messages

instrument_name: str

Name of the instrument

kafka_producer: dict[str, str]

Kafka producer configuration

local_ip: IPv4Address

Local IP to set the control board IP register to

model_config = {}

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

poll_interval_s: float

How often to poll the streaming control board

pv_prefix: str

PV prefix of all PVs in this IOC

pv_update_interval_s: float

How often to update the PVs

read_port: int

The read port to use for reading from the streaming control board

property register_map: dict[str, int]

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

runinfo_topic: str

Run info topic to push run starts/stops to

sample_env_topic: str

The topic to use when putting f144 placeholders in the run start nexus structure

state_file: Path

Path to the file to save/restore state to

write_port: int

The write port to use for writing to the streaming control board

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

Validate and load a config file at the specified path.