oedisi.types package¶

Submodules¶

oedisi.types.data_types module¶

Power system data types for OEDISI measurements and control.

class oedisi.types.data_types.AdmittanceMatrix(*, admittance_matrix: list[list[tuple[float, float]]], ids: list[str], units: str = 'S')¶

Bases: BaseModel

Dense representation of network admittance matrix.

admittance_matrix: list[list[Complex]]¶: Dense matrix for admittance

ids: list[str]¶: Row and column bus IDs

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for admittance

class oedisi.types.data_types.AdmittanceSparse(*, from_equipment: list[str], to_equipment: list[str], equipment_type: list[str] | None = None, admittance_list: list[tuple[float, float]], units: str = 'S')¶

Bases: Incidence

Sparse representation of network admittance matrix.

admittance_list: list[Complex]¶: Sparse admittance values with incidence connections

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for admittance

class oedisi.types.data_types.BusArray(*, values: list[float], ids: list[str], units: str, accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: MeasurementArray

Measurements for or at power system buses (primarily voltages).

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.CapacitorStates(*, values: list[int], ids: list[str], time: datetime | None = None)¶

Bases: StateArray

Capacitor state data for power system equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.Command(*, obj_name: str, obj_property: str, val: str)¶

Bases: BaseModel

JSON Configuration for external object commands for OpenDSS.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

obj_name: str¶: Name of OpenDSS object to change

obj_property: str¶: Property of OpenDSS object to change

val: str¶: New value. All OpenDSS transfers must be str

class oedisi.types.data_types.CostArray(*, values: list[list[float]], ids: list[str], units: str = '$', time: datetime | None = None)¶

Bases: BaseModel

Base class for cost-related data arrays.

Extended by classes:: “RealCostFunctions”, “ReactiveCostFunctions”, “RealWholesalePrices”, “ReactiveWholesalePrices”, “OperationalCosts”.

ids: list[str]¶: List of ids which values applies to

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

time: datetime.datetime | None¶: Time of measurement

units: str¶: Cost unit of float

values: list[list[float]]¶: List of cost arrays

class oedisi.types.data_types.CurrentsAngle(*, values: list[float], ids: list[str], units: str = 'radians', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Current angle measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for angle

class oedisi.types.data_types.CurrentsImaginary(*, values: list[float], ids: list[str], units: str = 'A', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Imaginary component of current measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for current

class oedisi.types.data_types.CurrentsMagnitude(*, values: list[float], ids: list[str], units: str = 'A', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Current magnitude measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for current

class oedisi.types.data_types.CurrentsReal(*, values: list[float], ids: list[str], units: str = 'A', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Real component of current measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for current

class oedisi.types.data_types.EquipmentArray(*, values: list[float], ids: list[str], units: str, accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: MeasurementArray

Measurements at equipment nodes (currents, impedances, environmental).

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.EquipmentNodeArray(*, values: list[float], ids: list[str], units: str, accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None, equipment_ids: list[str])¶

Bases: MeasurementArray

Power measurements at equipment nodes (primary key: ids + equipment_ids).

Primary key is ids + equipment_ids where ids correspond to node ids (e.g., “113.1”, “113.2”, “113.3”) and equipment_ids correspond to equipment identifiers (e.g., PVSystem.113).

equipment_ids: list[str]¶: Unique ID for values such as ‘PVSystem.113’

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.ImpedanceAngle(*, values: list[float], ids: list[str], units: str = 'radians', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Impedance angle measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for angle

class oedisi.types.data_types.ImpedanceImaginary(*, values: list[float], ids: list[str], units: str = 'Ohm', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Imaginary component of impedance measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for impedance

class oedisi.types.data_types.ImpedanceMagnitude(*, values: list[float], ids: list[str], units: str = 'Ohm', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Impedance magnitude measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for impedance

class oedisi.types.data_types.ImpedanceReal(*, values: list[float], ids: list[str], units: str = 'Ohm', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Real component of impedance measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for impedance

class oedisi.types.data_types.Incidence(*, from_equipment: list[str], to_equipment: list[str], equipment_type: list[str] | None = None)¶

Bases: BaseModel

Incidence relationships between equipment in the power system.

Each list should have the same length. 3W transformers are transformed into 2 separate edges.

equipment_type: list[str] | None¶: For connection i, equipment_type[i] is the target bus.

from_equipment: list[str]¶: For connection i, from_equpiment[i] is the source bus.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_equipment: list[str]¶: For connection i, to_equipment[i] is the source bus.

class oedisi.types.data_types.IncidenceList(*, from_equipment: list[str], to_equipment: list[str], equipment_type: list[str] | None = None, ids: list[str])¶

Bases: Incidence

Incidence relationships with associated identifiers.

ids: list[str]¶: String ID for each connection such as branch ID

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.Injection(*, current_real: ~oedisi.types.data_types.CurrentsReal = <factory>, current_imaginary: ~oedisi.types.data_types.CurrentsImaginary = <factory>, power_real: ~oedisi.types.data_types.PowersReal = <factory>, power_imaginary: ~oedisi.types.data_types.PowersImaginary = <factory>, impedance_real: ~oedisi.types.data_types.ImpedanceReal = <factory>, impedance_imaginary: ~oedisi.types.data_types.ImpedanceImaginary = <factory>)¶

Bases: BaseModel

Current and power injections at network nodes.

current_imaginary: CurrentsImaginary¶

current_real: CurrentsReal¶

impedance_imaginary: ImpedanceImaginary¶

impedance_real: ImpedanceReal¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

power_imaginary: PowersImaginary¶

power_real: PowersReal¶

class oedisi.types.data_types.InverterControl(*, pvsystem_list: list[str] | None = None, vvcontrol: VVControl | None = None, vwcontrol: VWControl | None = None, mode: InverterControlMode = InverterControlMode.voltvar)¶

Bases: BaseModel

InverterControl with volt-var control and/or volt-watt control.

See https://dss-extensions.org/dss-format/InvControl.html.

classmethod check_mode(values)¶: Make sure that mode reflects vvcontrol and vwcontrol data.

mode: InverterControlMode¶: Inverter control mode

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pvsystem_list: list[str] | None¶: List of pvsystems to apply controls to

vvcontrol: VVControl | None¶: Volt-var control settings

vwcontrol: VWControl | None¶: Volt-watt control settings

class oedisi.types.data_types.InverterControlMode(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)¶

Bases: Enum

Inverter control mode.

voltvar = 'VOLTVAR'¶

voltvar_voltwatt = 'VV_VW'¶

voltwatt = 'VOLTWATT'¶

class oedisi.types.data_types.MeasurementArray(*, values: list[float], ids: list[str], units: str, accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: BaseModel

Base class for measurement data arrays.

Extended by classes:: “BusArray”, “EquipmentArray”, “EquipmentNodeArray”.

accuracy: list[float] | None¶: Estimated or known std error at each location

bad_data_threshold: list[float] | None¶: Threshold after which value should be considered junk

ids: list[str]¶: List of ids which values applies to

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

time: datetime.datetime | None¶: Time of original measurement

units: str¶: Unit of each float

values: list[float]¶: List of values

class oedisi.types.data_types.OperationalCosts(*, values: list[list[float]], ids: list[str], units: str = '$', time: datetime | None = None)¶

Bases: CostArray

Operational cost data for equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.PowersAngle(*, values: list[float], ids: list[str], units: str = 'radians', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None, equipment_ids: list[str])¶

Bases: EquipmentNodeArray

Power angle measurements at equipment nodes.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for angle

class oedisi.types.data_types.PowersImaginary(*, values: list[float], ids: list[str], units: str = 'kVAR', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None, equipment_ids: list[str])¶

Bases: EquipmentNodeArray

Reactive power measurements at equipment nodes.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for power

class oedisi.types.data_types.PowersMagnitude(*, values: list[float], ids: list[str], units: str = 'kVA', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None, equipment_ids: list[str])¶

Bases: EquipmentNodeArray

Power magnitude (apparent power) measurements at equipment nodes.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for power

class oedisi.types.data_types.PowersReal(*, values: list[float], ids: list[str], units: str = 'kW', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None, equipment_ids: list[str])¶

Bases: EquipmentNodeArray

Real power measurements at equipment nodes.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for power

class oedisi.types.data_types.ReactiveCostFunctions(*, values: list[list[float]], ids: list[str], units: str = '$', time: datetime | None = None)¶

Bases: CostArray

Reactive power cost functions for equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.ReactivePowerSetting(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)¶

Bases: Enum

Reactive power setting, almost always VARAVAL_WATTS.

See https://dss-extensions.org/dss-format/InvControl.html.

VARAVAL_WATTS = 'VARAVAL_WATTS'¶: Base absorbed reactive power is equal to available

VARMAX_VARS = 'VARMAX_VARS'¶: Base absorbed reactive power equal to kvar maximum

VARMAX_WATTS = 'VARMAX_WATTS'¶: Base absorbed reactive power equal to power magnitude maximum

class oedisi.types.data_types.ReactiveWholesalePrices(*, values: list[list[float]], ids: list[str], units: str = '$', time: datetime | None = None)¶

Bases: CostArray

Reactive power wholesale price data for equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.RealCostFunctions(*, values: list[list[float]], ids: list[str], units: str = '$', time: datetime | None = None)¶

Bases: CostArray

Real power cost functions for equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.RealWholesalePrices(*, values: list[list[float]], ids: list[str], units: str = '$', time: datetime | None = None)¶

Bases: CostArray

Real power wholesale price data for equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.RegulatorStates(*, values: list[int], ids: list[str], time: datetime | None = None)¶

Bases: StateArray

Voltage regulator state data for power system equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.SolarIrradiances(*, values: list[float], ids: list[str], units: str = 'kW/m^2', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Solar irradiance measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for power per area

class oedisi.types.data_types.StateArray(*, values: list[int], ids: list[str], time: datetime | None = None)¶

Bases: BaseModel

Base class for power system equipment state arrays.

Extended by classes:: “SwitchStates”, “CapacitorStates”, “RegulatorStates”.

ids: list[str]¶: List of ids which values applies to

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

time: datetime.datetime | None¶: Time of original measurement

values: list[int]¶: List of values

class oedisi.types.data_types.StatesOfCharge(*, values: list[float], ids: list[str], units: str = 'percent', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

State of charge measurements for energy storage equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Dimensionless unit

class oedisi.types.data_types.SwitchStates(*, values: list[int], ids: list[str], time: datetime | None = None)¶

Bases: StateArray

Switch state data for power system equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.types.data_types.Temperatures(*, values: list[float], ids: list[str], units: str = 'C', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Temperature measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for temperature

class oedisi.types.data_types.Topology(**data: Any)¶

Bases: BaseModel

Power system network topology with admittance and injection data.

admittance: AdmittanceSparse | AdmittanceMatrix¶: Admittance matrix either as AdmittanceSparse or AdmittanceMatrix

base_voltage_angles: VoltagesAngle | None¶: Base voltage angles at each bus

base_voltage_magnitudes: VoltagesMagnitude | None¶: Base voltage mangitudes at each bus

incidences: IncidenceList | None¶: Connectivity of all equipment

injections: Injection¶: Base injections for all equipment

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

slack_bus: list[str]¶: Slack bus (usually 3 buses for each phase)

class oedisi.types.data_types.VVControl(*, deltaq_factor: float = -1.0, varchangetolerance: float = 0.025, voltagechangetolerance: float = 0.0001, vv_refreactivepower: ReactivePowerSetting = ReactivePowerSetting.VARAVAL_WATTS, voltage: list[float], reactive_response: list[float])¶

Bases: BaseModel

OpenDSS setting for volt-var control.

deltaq_factor: float¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reactive_response: list[float]¶

varchangetolerance: float¶

voltage: list[float]¶

voltagechangetolerance: float¶

vv_refreactivepower: ReactivePowerSetting¶

class oedisi.types.data_types.VWControl(*, deltap_factor: float = -1.0, voltage: list[float], power_response: list[float])¶

Bases: BaseModel

OpenDSS setting for volt-watt control.

See https://dss-extensions.org/dss-format/InvControl.html.

deltap_factor: float¶

DeltaP_factor used to limit change for control iterations.

By default uses OpenDSS to figure it out.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

power_response: list[float]¶: Power response coordinates (p.u. of VAs)

voltage: list[float]¶: Voltage response coordinates (V)

class oedisi.types.data_types.VoltagesAngle(*, values: list[float], ids: list[str], units: str = 'radians', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: BusArray

Voltage angle measurements at buses.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for angle

class oedisi.types.data_types.VoltagesImaginary(*, values: list[float], ids: list[str], units: str = 'V', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: BusArray

Imaginary component of voltage measurements at buses.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for voltage

class oedisi.types.data_types.VoltagesMagnitude(*, values: list[float], ids: list[str], units: str = 'V', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: BusArray

Voltage magnitude measurements at buses.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for voltage

class oedisi.types.data_types.VoltagesReal(*, values: list[float], ids: list[str], units: str = 'V', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: BusArray

Real component of voltage measurements at buses.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for voltage

class oedisi.types.data_types.WindSpeeds(*, values: list[float], ids: list[str], units: str = 'm/s', accuracy: list[float] | None = None, bad_data_threshold: list[float] | None = None, time: datetime | None = None)¶

Bases: EquipmentArray

Wind speed measurements at equipment.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

units: str¶: Unit for speed

oedisi.types.generate_schema module¶

Generate JSON schemas for all OEDISI data types.

Module contents¶

Bases: BaseModel

HELICS broker connection parameters.

Parameters:

host (str | None) – Broker hostname or IP address.
port (int | None) – Broker port number.
key (str | None) – Broker key for authentication.
auto (bool | None) – Whether to automatically configure broker connection.
initstring (str | None) – Additional initialization string for broker connection.

auto: bool | None¶

classmethod from_rest_config(rest: BrokerConfig) → HELICSBrokerConfig¶: Convert REST API BrokerConfig to HELICS native broker config.

host: str | None¶

initstring: str | None¶

key: str | None¶

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

port: int | None¶

class oedisi.types.HELICSFederateConfig(*, name: str, coreType: str | None = None, coreName: str | None = None, coreInitString: str | None = None, broker: HELICSBrokerConfig | None = None)¶

Bases: BaseModel

Full HELICS federate configuration.

This is what federates receive in static_inputs.json, containing various HELICS configuration at the top-level (name, core_type, broker, etc). Subtype this in your applications for custom configuration.

Parameters:

name (str) – Federate name (derived from Component.name).
core_type (str | None) – HELICS core type (e.g., “zmq”, “tcp”, “inproc”).
core_name (str | None) – Core name for this federate (derived per-component).
core_init (str | None) – Core initialization string.
broker (oedisi.types.helics_config.HELICSBrokerConfig | None) – Broker connection configuration.

Examples

>>> config = HELICSFederateConfig(
...     name="state_estimator",
...     core_type="zmq",
...     broker=HELICSBrokerConfig(port=23404)
... )
>>> config.to_json()
'{"name": "state_estimator", "coreType": "zmq", ...}'

apply_to_federate_info(info) → None¶

Apply configuration to a helics.HelicsFederateInfo object.

Parameters:: info – A helics.HelicsFederateInfo object to configure.

Notes

This method requires the helics Python package to be installed. It modifies the info object in place.

broker: HELICSBrokerConfig | None¶

core_init: str | None¶

core_name: str | None¶

core_type: str | None¶

classmethod from_multicontainer(broker_config: BrokerConfig, params: dict, core_type: str = 'zmq', **kwargs) → HELICSFederateConfig¶

Create federate config for multicontainer deployments.

This is a convenience method for components running in Docker/Kubernetes that receive BrokerConfig from the broker service’s /run endpoint. Supports subclasses of HELICSFederateConfig with additional parameters.

Parameters:

broker_config – REST API broker configuration from /run endpoint.
params – Component parameters from static_inputs.json. If provided, must contain ‘name’ key. All other keys are passed to the config.
core_type – HELICS core type, defaults to “zmq”.
**kwargs – Additional config options (overrides params dict if both provided).

Returns:

Complete federate configuration ready to apply. If called on a subclass, returns an instance of that subclass.

Return type:

HELICSFederateConfig

Examples

>>> # Basic usage with params dict
>>> with open("static_inputs.json") as f:
...     params = json.load(f)
>>> config = HELICSFederateConfig.from_multicontainer(
...     broker_config=broker_config,
...     params=params
... )
>>> fedinfo = h.helicsCreateFederateInfo()
>>> config.apply_to_federate_info(fedinfo)

>>> # With custom subclass
>>> class MyComponentConfig(HELICSFederateConfig):
...     my_param: str
>>> config = MyComponentConfig.from_multicontainer(
...     broker_config=broker_config,
...     params={"name": "comp1", "my_param": "value"}
... )

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

to_dict() → dict¶: Convert to dictionary using HELICS-style camelCase keys.

to_json() → str¶: Serialize to JSON string using HELICS-style camelCase keys.

class oedisi.types.SharedFederateConfig(*, coreType: str | None = None, coreInitString: str | None = None, broker: HELICSBrokerConfig | None = None)¶

Bases: BaseModel

Shared federate settings at the WiringDiagram level.

This contains settings that are shared across all federates in a simulation. Does NOT include name/core_name (those are per-component). Users are expected to write this as part of the wiring_diagram.json, and invidual federates get passed config.to_federate_config().

Parameters:

core_type (str | None) – HELICS core type (e.g., “zmq”, “tcp”, “inproc”).
core_init (str | None) – Core initialization string.
broker (oedisi.types.helics_config.HELICSBrokerConfig | None) – Broker connection configuration.

Examples

>>> shared = SharedFederateConfig(
...     core_type="zmq",
...     broker=HELICSBrokerConfig(port=23404)
... )
>>> config = shared.to_federate_config("my_federate")
>>> config.name
'my_federate'

broker: HELICSBrokerConfig | None¶

core_init: str | None¶

core_type: str | None¶

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_federate_config(name: str, core_name: str | None = None) → HELICSFederateConfig¶

Create a full HELICSFederateConfig for a specific component.

During building, a HELICSFederateConfig is created for each component, combined with other static configuration from the wiring diagram.

Parameters:

name – Federate name (typically Component.name).
core_name – Optional core name for this federate.

Returns:

Complete federate configuration with per-component values set.

Return type:

HELICSFederateConfig