model

Models

config

class libdots.model.config.ServiceConfig(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | None = None, _cli_ignore_unknown_args: bool | None = None, _cli_kebab_case: bool | None = None, _secrets_dir: PathType | None = None, *, log_level: Literal['debug', 'info', 'warning', 'warn', 'error', 'fatal', 'critical', 'DEBUG', 'INFO', 'WARNING', 'WARN', 'ERROR', 'FATAL', 'CRITICAL'] = 'info', simulation_id: str, model_id: str, mqtt_host: str = 'localhost', mqtt_port: int = 1883, mqtt_qos: int = 0, mqtt_username: str = '', mqtt_password: SecretStr = SecretStr(''), influxdb_host: str = '', influxdb_port: int = 8086, influxdb_user: str = '', influxdb_password: SecretStr = SecretStr(''), influxdb_name: str = '')

Bases: BaseSettings

The configuration of the calculation service. This uses pydantic_settings with dotenv support. Configuration values are (case-insensitive) read from:

• environment variables

• .env

• .env.docker

If you’re using pyright’s strict typing in your library, you need to add an ignore statement when instantiating it. pydantic-settings reads the missing parameters from environment variables so the missing parameters can be safely ignored.

config = ServiceConfig() # pyright:ignore[reportCallIssue]

influxdb_host: str

influxdb_name: str

influxdb_password: SecretStr

influxdb_port: int

influxdb_user: str

log_level: Literal['debug', 'info', 'warning', 'warn', 'error', 'fatal', 'critical', 'DEBUG', 'INFO', 'WARNING', 'WARN', 'ERROR', 'FATAL', 'CRITICAL']

model_id: str

mqtt_host: str

mqtt_password: SecretStr

mqtt_port: int

mqtt_qos: int

mqtt_username: str

simulation_id: str

esdl_parser

class libdots.model.esdl_parser.ESDLParser(receives_service_names_list: list[str])

Bases: object

add_calc_services_from_all_objects(calculation_services: list[CalculationServiceDescription], connected_input_esdl_objects: dict[str, list[str]], energy_system: EnergySystem)

add_calc_services_from_non_connected_objects(calculation_services: list[CalculationServiceDescription], connected_input_esdl_objects: dict[str, list[str]], energy_system: Item | EnergySystem | GenericProfile | DataSource)

add_calc_services_from_output_ports(calculation_services: list[CalculationServiceDescription], connected_input_esdl_objects: dict[str, list[str]], model_esdl_asset: EnergyAsset)

add_calc_services_from_ports(calculation_services: list[CalculationServiceDescription], connected_input_esdl_objects: dict[str, list[str]], model_esdl_asset: EnergyAsset)

add_esdl_object(connected_input_esdl_objects: dict[str, list[str]], esdl_obj: Item | EnergySystem | GenericProfile | DataSource, calculation_services: list[CalculationServiceDescription])

get_connected_input_esdl_objects(esdl_id: str, calculation_services: list[CalculationServiceDescription], energy_system: EnergySystem) → dict[str, list[str]]

get_connected_output_esdl_objects(esdl_id: str, calculation_services: list[CalculationServiceDescription], energy_system: EnergySystem) → dict[str, list[str]]

get_energy_system(esdl_base64string: str) → EnergySystem

get_model_esdl_object(esdl_id: str, energy_system: EnergySystem) → Item | EnergySystem | GenericProfile | DataSource

influxdb_connector

class libdots.model.influxdb_connector.InfluxDBConnector(influx_host: str, influx_port: str, influx_user: str, influx_password: str, influx_database_name: str)

Bases: object

A connector writes data to an InfluxDB database.

add_measurement(points: list[dict[str, Any]], esdl_id: str, timestamp: str, fields: dict[str, Any])

property client: InfluxDBClient

close()

connect() → InfluxDBClient

create_database()

init_profile_output_data(simulation_id: str, model_id: str, esdl_type: str, start_date: datetime, time_step_seconds: int, nr_of_time_steps: int, esdl_ids: list[str], output_names: list[str], esdl_objects: dict[str, Item | EnergySystem | GenericProfile | DataSource])

query(query: str)

set_summary_data_point(esdl_id: str, output_name: str, value: float)

set_time_step_data_point(esdl_id: str, output_name: str, time_step_nr: int, value: float)

write(msgs: list[dict[str, Iterable[Any]]])

write_output()

service

class libdots.model.service.BaseService(config: ServiceConfig)

Bases: ABC

Abstract Base Class for the actual Service object. This object needs to be overriden and its service_calc_class property implemented.

This can be done like this:

from typing import override

from libdots.model.config import ServiceConfig
from libdots.model.service import BaseService
from libdots.model.service_calc import CalculationFunction
from libdots.model.service_calc import ServiceCalc
from .service_cal import MyServiceCalc

class MyService(BaseService):
    @property
    @override
    def service_calc_class(self):
        return MyServiceCalc

config = ServiceConfig() # pyright:ignore[reportCallIssue]
service = MyService(config)
service.start()

abstract property service_calc_class: type[ServiceCalc[CalculationFunction[Any, Any]]]

Return the class (type not instance) to be instantiated in this Service.

start()

service_calc

class libdots.model.service_calc.CalculationFunction(*args, **kwargs)

Bases: Protocol[InputDataInterfaceT, OutputDataInterfaceT]

Protocol describing the calculation functions.

class libdots.model.service_calc.OutputDataInterfaceT

With python 3.14 https://peps.python.org/pep-0728/

class AllInputDataInterface(TypedDict, extra_items=Sequence[IODataInterface]):

new_step: NewStep

alias of TypeVar(‘OutputDataInterfaceT’, bound=tuple[Mapping[str, IODataInterface], …], covariant=True)

class libdots.model.service_calc.ServiceCalc(simulation_id: str, model_id: str, influxdb_host: str, influxdb_port: int, influxdb_user: str, influxdb_password: str, influxdb_name: str)

Bases: ABC, Generic[CalculationFunctionT]

Abstract Base Class for the main calculation service. The calculation service has 4 stages:

• __init__: Initialize the class and its parameters

• setup: After the first mqtt message is received with the ESDL file and model parameters, parses the energy system and runs similar setup code.

• time step: For each time step, the corresponding calculation functions are called once their required input data was received

• teardown: Store all data in influxdb

There can be multiple calculation functions per service. Each function can have its own input and output data. This makes it possible to have multiple stages of calculation in a single service.

abstract base_setup() → None

Setup code to run before we start looping over all individual esdl objects, but after the esdl file was parsed. This can for instance be used to setup data from static profiles using URIProfile’s in the ESDL file.

calc_function(calc_name: str, input_data_dict: Mapping[Literal['new_step'] | str, NewStep | Sequence[IODataInterface]])

Gets called by mqtt client when all input data has been received.

property calculation_function_input_types: dict[str, list[type[IODataInterface]]]

Returns a dictionary of calculation function names and a list of IODataInterface classes in its input_data. It uses typing introspection for this, and its used to tell the data inventory what data to wait for per calculation function.

abstract property calculation_functions: Mapping[str, CalculationFunctionT]

Determines what message types we expect as input data per calculation function before it can run. This data comes from the calculation services that we expect input data from. Once each of the message types is received for the current timestep, the corresponding function is started.

Should return a dictionary mapping calculation function names to actual methods. Example

{
    "pre_battery": self.pre_battery,
    "post_battery": self.post_battery
}

This defines there are two calculation functions with the names pre_battery and post_battery and references the functions that run the logic for each timestep once all required data has been received.

esdl_energy_system: EnergySystem

esdl_ids: list[str]

esdl_parser: ESDLParser

get_profile_uri_by_id(id: str) → str

Get a ESDL ProfileURI by the esdl id

nr_of_time_steps: int

abstract process_esdl_object(esdl_id: str, esdl_object: Item | EnergySystem | GenericProfile | DataSource)

Runs during the model setup phase. This code runs for each esdl object handled by this calculation service.

abstract property receives_service_names: list[str]

Used during the model __init__. Should contain the names of the calculation services this service is expecting input data from.

abstract property service_name: str

The name of the service.

setup(model_parameters: ModelParametersDescription)

setup_influxdb_output()

Setup the output to store in influxdb. Runs at the send of setup. This is optional, and determines the fields to store in influxdb for each step.

simulation_name: str

simulation_start_date: datetime

time_step_seconds: int

write_to_influxdb()

Write collected data to influxdb