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, pv_prefix: str, runinfo_topic: str = 'runnervmp4gaq_runInfo', 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

flush_timeout_s: int

The timeout for a flush after producing run info messages

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

runinfo_topic: str

Run info topic to push run starts/stops to

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.