Architecture

This document describes how the KPI Calculator is structured, how data flows through it, and where to look when making changes.

Role in the OMOTES Toolkit

The KPI calculator can be used standalone (via its Python API or command line), but within the OMOTES design toolkit it typically runs as part of a larger pipeline. The ESDL MapEditor serves as the interface for drawing energy systems, triggering computations, and displaying results. The MapEditor already supports KPI visualization: dashboards, color-coded maps based on KPI values, and tabular result views.

The typical flow is:

  1. A user designs an energy system in the MapEditor

  2. The MapEditor sends the ESDL to the OMOTES orchestrator

  3. A worker (simulator or optimizer) processes the ESDL and calls the KPI calculator

  4. The KPI calculator returns results, which are written back into the ESDL as DistributionKPI elements

  5. The MapEditor renders the KPIs from the returned ESDL

This means that decisions about output format — asset-level vs. system-level KPIs, category breakdowns, export structure — should be guided by what the MapEditor can consume and display. The richer the ESDL KPI output, the more the MapEditor can visualize.

Overview

The package has four layers:

┌──────────────────────────────────────────────────────────┐
│  Public API                                              │
│    api.py → calculate_kpis()                             │
├──────────────────────────────────────────────────────────┤
│  KPI Manager (Orchestrator)                              │
│    kpi_manager.py → KpiManager                           │
├──────────────────────────────────────────────────────────┤
│  Calculators                                             │
│    financial_calculator    energy_calculator    emission_calc  │
├──────────────────────────────────────────────────────────┤
│  Adapters                                                │
│    esdl_adapter    time_series_manager    database_loader │
├──────────────────────────────────────────────────────────┤
│  Common Model                                            │
│    Asset    TimeSeries    EnergySystem                    │
└──────────────────────────────────────────────────────────┘

Adapters parse external data (ESDL files, time series) into the common model. Calculators consume the common model and produce KPI results. The KPI Manager coordinates loading and calculation. The API is a thin wrapper that validates inputs and delegates to the manager.

Data Flow

When calculate_kpis("model.esdl") is called:

  1. api.py validates the file path and creates a KpiManager

  2. KpiManager.load_from_esdl() calls the ESDL adapter

  3. The ESDL adapter:

    1. Parses the ESDL file using PyESDL (EnergySystemHandler.load_file())

    2. Stores the parsed PyESDL object on EnergySystem.esdl_energy_system for later export

    3. Iterates over all assets in es.eAllContents(), extracting physical properties and costs

    4. Delegates time series loading to TimeSeriesManager

    5. Returns a populated EnergySystem containing Asset objects

  4. KpiManager.calculate_all_kpis() runs the three calculators against the EnergySystem

  5. Results are returned as a KpiResults TypedDict

Common Model

The common model (adapters/common_model.py) is the contract between adapters and calculators. Everything goes through these types.

Asset

@dataclass
class Asset:
    id: str
    name: str
    asset_type: AssetType  # Producer, Consumer, Storage, Transport, etc.

    # Physical properties
    power: float = 0.0      # Watts
    length: float = 0.0     # meters (pipes)
    volume: float = 0.0     # m³ (storage)
    cop: float = 0.0        # Coefficient of performance

    # Cost properties (6 types, each with value + unit)
    investment_cost: float = 0.0
    investment_cost_unit: str = "EUR"
    installation_cost: float = 0.0
    installation_cost_unit: str = "EUR"
    fixed_operational_cost: float = 0.0
    fixed_operational_cost_unit: str = "EUR/yr"
    variable_operational_cost: float = 0.0
    variable_operational_cost_unit: str = "EUR/MWh"
    fixed_maintenance_cost: float = 0.0
    fixed_maintenance_cost_unit: str = "EUR/yr"
    variable_maintenance_cost: float = 0.0
    variable_maintenance_cost_unit: str = "EUR/MWh"

    # Lifecycle
    technical_lifetime: float = 40.0   # years
    discount_rate: float = 5.0         # percent
    emission_factor: float = 0.0       # kg CO2/GJ

    # Aggregation
    aggregation_count: int = 1

    # Time series data
    time_series: dict[str, TimeSeries] = field(default_factory=dict)

Cost units are stored as-is from ESDL. The cost calculator handles the conversion (see Cost Unit Conversion).

TimeSeries

@dataclass
class TimeSeries:
    time_step: float        # seconds between data points
    values: list[float]     # measured values

EnergySystem

@dataclass
class EnergySystem:
    name: str
    assets: list[Asset]
    unit_conversion: dict[str, float] = field(default_factory=dict)
    source_metadata: dict[str, str] = field(default_factory=dict)
    esdl_energy_system: esdl.EnergySystem | None = None

This is the top-level container passed to all calculators. unit_conversion holds cost unit conversion factors (automatically populated from COST_UNIT_FACTORS in constants.py). source_metadata records how the system was loaded (e.g., {"esdl_file": "model.esdl"} or {"esdl_source": "string"}). esdl_energy_system holds the original PyESDL object set by the adapter for both file-loaded and string-loaded systems, enabling ESDL export without re-reading from disk.

ESDL Adapter

The ESDL adapter (adapters/esdl_adapter.py) is the most complex component in the package. It does two things: extract assets with costs, and load time series.

Asset and Cost Extraction

The adapter iterates over es.eAllContents() and converts each ESDL asset to the common model Asset. For each asset, it:

  1. Maps the ESDL type to an AssetType enum value

  2. Extracts physical properties (power, length, volume)

  3. Extracts cost data from the ESDL costInformation element

The cost extraction maps six ESDL fields to internal fields:

ESDL costInformation field       →  Asset field
─────────────────────────────────────────────────
investmentCosts                  →  investment_cost
installationCosts                →  installation_cost
fixedOperationalCosts            →  fixed_operational_cost
variableOperationalCosts         →  variable_operational_cost
fixedMaintenanceCosts            →  fixed_maintenance_cost
variableMaintenanceCosts         →  variable_maintenance_cost

Each cost element in ESDL has a profileQuantityAndUnit that specifies the unit. The adapter stores both the raw value and the unit string on the Asset.

Time Series Loading

Time series loading is handled by TimeSeriesManager (adapters/time_series_manager.py). The priority list is built dynamically by EsdlAdapter._process_energy_system() based on which sources are enabled:

source_priority = ["dataframes"]
if use_database_profiles:
    source_priority.append("database")
if time_series_file:
    source_priority.append("xml")
source_priority.append("empty")

The full priority order when all sources are enabled is:

  1. pandas DataFrames — passed directly via the timeseries_dataframes parameter

  2. InfluxDB profiles — loaded from InfluxDBProfile references found in the ESDL

  3. XML files — loaded from the time_series parameter (xml_time_series_adapter.py, used in tests only — not a supported production input)

  4. No data — calculators return zero for energy values (no rated-capacity fallback is implemented)

Each source is attempted in order. If one fails, it logs a warning and tries the next. In the OMOTES pipeline, the simulator-worker provides time series via DataFrames, bypassing InfluxDB entirely.

InfluxDB disabled in the default API: The public API (KpiManager.load_from_esdl()) passes use_database_profiles=False to the adapter. This means InfluxDB is not added to the priority list at all — it is excluded, not tried-and-skipped. The effective priority for the default API is: DataFrames → XML → empty. This is intentional: the default entry point is designed to work without a database connection. The adapter layer itself fully supports InfluxDB; re-enabling it requires passing use_database_profiles=True when constructing the adapter directly.

DataFrame Composite Key Mapping

TimeSeriesManager._load_from_dataframes() iterates over every column in each DataFrame and stores the data under a composite key in the format asset_id|column_name. This matches the format produced by the InfluxDB and XML loaders, so EsdlAdapter._build_asset() can resolve the time series without any special-casing for the DataFrame path.

The column name is used as the field name and must match one of the names recognised by the KPI calculators. The full set is defined in KNOWN_TIME_SERIES_FIELDS (common/constants.py), derived from the field name tuples imported by the calculators:

CONSUMPTION_FIELDS        ThermalConsumption, Consumption, Energy,
                          heat_power_primary
DEMAND_FIELDS             ThermalDemand, Demand
PRODUCTION_FIELDS         ThermalProduction, Production, heat_supplied, Energy,
                          heat_power_secondary
ELECTRICAL_CONSUMPTION    ElectricalConsumption
CONVERSION_FIELDS         ElectricalConsumption, electricity_consumption,
                          ThermalProduction

heat_power_primary and heat_power_secondary are the field names emitted by omotes_simulator_core for heat pump and heat exchanger assets (primary side = consumer of heat, secondary side = producer of heat). electricity_consumption is the lowercase field name used by simulator-core for HeatPump and AirToWaterHeatPump assets; it maps to CONVERSION_FIELDS alongside the ESDL-native ElectricalConsumption.

Columns with unrecognised names are stored in the time series dict but will not be picked up by any calculator — a warning is logged at load time so callers can catch the mismatch early. Non-numeric columns are rejected with an error recorded in ValidationResult and never stored.

Cost Unit Conversion

The financial calculator (calculators/financial_calculator.py) converts cost values from ESDL units to EUR using the asset’s physical properties and built-in conversion factors defined in COST_UNIT_FACTORS (common/constants.py). The factors are looked up by _get_unit_factor(unit) from energy_system.unit_conversion, which is automatically populated by the adapter.

Investment and installation costs — allowed units:

Unit            Conversion
──────────────────────────────────────────────────
EUR             value (no factor needed)
EUR/kW          value × power_W × 0.001
EUR/MW          value × power_W × 1e-6
EUR/m           value × length_m
EUR/km          value × length_m × 0.001
EUR/m3          value × volume_m3 (no factor needed)

Fixed operational and fixed maintenance costs — allowed units:

Unit            Conversion
──────────────────────────────────────────────────
EUR, EUR/yr     value (used directly as annual cost)
EUR/MW          value × power_W × 1e-6
% OF CAPEX      value × (investment + installation) × 0.01

Variable operational and variable maintenance costs — allowed units:

Unit            Conversion
──────────────────────────────────────────────────
EUR, EUR/yr     value (used directly as annual cost)
EUR/kWh         value × annual_energy_J × 2.78e-7
EUR/MWh         value × annual_energy_J × 2.78e-10

Variable costs use the first time series on the asset (regardless of key name). For geothermal assets with COP > 0, the energy is divided by COP before applying the cost rate.

To add a new unit: Add the unit string to the allowed_units list in the relevant cost method, add a conversion branch, add the factor to COST_UNIT_FACTORS in constants.py, and add a test case.

Calculators

All three calculators take an EnergySystem and return part of the KpiResults dict. They are independent of each other.

Error Handling

All calculators use Python’s logging module with the logger name set to their module path (e.g. kpicalculator.calculators.financial_calculator). Two log levels are used:

  • DEBUG — expected missing-data conditions: no time series on an asset, no matching field name in the time series. These are normal when assets lack simulation results.

  • WARNING — unexpected or potentially lossy conditions: unsupported cost units (the cost is zeroed), non-positive duration time series (would cause ZeroDivisionError or incorrect annualization).

In all cases the calculator returns 0.0 for the affected asset rather than raising an exception. Input validation (e.g. technical_lifetime <= 0) is handled upstream by BaseAdapter._validate_energy_system() and is not duplicated in the calculators.

Financial Calculator

calculators/financial_calculator.py — the largest calculator. The public class is FinancialCalculator.

  • CAPEX: Sum of (investment + installation) per asset, converted from ESDL units, grouped by asset category (Production, Transport, Storage, Conversion, Consumption, All)

  • OPEX: Sum of (fixed_operational + variable_operational + fixed_maintenance + variable_maintenance) per asset

  • NPV: Standard discounted cash flow: CAPEX + Sum(OPEX_t / (1 + rate/100)^t) over system lifetime (default 30 years, 5% discount rate). Includes asset replacement costs when technical lifetime < system lifetime.

  • LCOE: NPV / discounted_energy_delivered, computed inside FinancialCalculator using pre-calculated energy values.

  • Per-asset financial breakdown: get_asset_financial_breakdown(annual_energy_mwh_by_asset) returns per-asset CAPEX, OPEX, NPV, and LCOE as AssetFinancialResult entries, stored in KpiResults["asset_financials"].

  • Category aggregation: aggregate_by_category(asset_results) groups per-asset results into the category-level totals (Production, Transport, etc.) stored under KpiResults["financials"].

Each of the six _calculate_*_cost() methods validates the unit string against an allowed_units list. If the unit is not supported, a warning is logged and the cost is returned as 0.0. Variable cost methods (operational and maintenance) also guard against zero-duration time series.

Lifetime Handling

Two different lifetimes are used:

  • System lifetime (default 30 years): The analysis period for NPV and LCOE. Passed as a parameter to calculate_kpis().

  • Technical lifetime (default 40 years per asset): How long each asset lasts. Extracted from ESDL costInformation.technicalLifetime. When an asset’s technical lifetime is shorter than the system lifetime, the cost calculator includes replacement costs.

Energy Calculator

calculators/energy_calculator.py — ~170 lines.

  • Consumption/Production/Demand: Sums time series values × time step, annualized

  • Efficiency: consumption / production

Without time series, energy values are returned as zero (no rated-capacity fallback is implemented). Missing data is logged at DEBUG level; zero-duration time series are logged at WARNING level.

Emission Calculator

calculators/emission_calculator.py — ~135 lines.

  • Total emissions: Sum of (emission_factor × energy) per asset, converted to tonnes

  • Emissions per MWh: total_emissions / consumption_in_MWh

Emission factors come from ESDL carrier definitions (kg CO2/GJ). Only EnergyCarrier assets carry an .emission attribute; HeatCommodity and other carrier types do not. When a non-EnergyCarrier carrier is encountered, the factor is set to 0.0 and a DEBUG message is logged — emissions for that asset are not included in the total. The same logging conventions apply: missing data at DEBUG, zero duration at WARNING.

ESDL Export

The exporter (reporting/esdl_kpi_exporter.py) writes KPI results back into an ESDL structure using DistributionKPI elements. Currently supports system-level export only.

The exporter does not perform calculations — it receives pre-calculated results and formats them into ESDL-compliant XML elements.

String-Loaded ESDL Support

The adapter stores the parsed PyESDL object on the EnergySystem.esdl_energy_system field for both file-loaded and string-loaded systems. The exporter checks this field first and reuses the stored object, avoiding the need to re-load from disk.

ESDL object resolution in the exporter:

  1. Check energy_system.esdl_energy_system — use it if present (covers both load paths)

  2. Otherwise, fall back to handler.load_file() using source_metadata["esdl_file"] or the source_esdl_file parameter

  3. If neither is available, raise ValueError

This enables workflows where ESDL never touches disk: load from string → calculate KPIs → export to ESDL object → serialize to string or pass to another service. Critical for simulator-worker integration and MapEditor REST service.

Project Layout

src/kpicalculator/
├── api.py                          # Public API: calculate_kpis()
├── kpi_manager.py                  # Orchestrator + result TypedDicts
├── exceptions.py                   # Custom exception hierarchy
├── adapters/
│   ├── base_adapter.py             # Abstract base class + ValidationResult
│   ├── common_model.py             # Asset, TimeSeries, EnergySystem
│   ├── esdl_adapter.py             # ESDL parsing + cost extraction
│   ├── time_series_manager.py      # Multi-source time series loading
│   ├── database_time_series_loader.py  # InfluxDB integration
│   └── xml_time_series_adapter.py  # XML time series (testing only)
├── calculators/
│   ├── financial_calculator.py          # CAPEX, OPEX, NPV, LCOE
│   ├── energy_calculator.py        # Consumption, production, efficiency
│   └── emission_calculator.py      # CO2 emissions
├── reporting/
│   ├── base_exporter.py            # Export interface
│   └── esdl_kpi_exporter.py        # Write KPIs back to ESDL
├── security/
│   ├── credential_manager.py       # Database credential handling
│   └── input_validator.py          # Path and input validation
└── common/
    ├── constants.py                # Defaults and conversion factors
    ├── types.py                    # Pydantic models
    └── logging_utils.py            # Structured logging

Test Layout

Tests live in unit_test/ alongside the source. Test data (ESDL files, XML time series) is in unit_test/data/.

unit_test/
├── data/
│   ├── Unit_test_ESDL.esdl       # Main test ESDL fixture
│   └── power_timeseries.xml      # Matching XML time series
├── test_api.py                   # Public API integration tests
├── test_kpi_calculator.py        # End-to-end KPI calculation + DataFrame mapping + ESDL export
├── test_examples.py              # README code examples (regression tests)
├── test_esdl_adapter.py          # ESDL adapter branch coverage
├── test_esdl_cost_extraction.py  # Cost extraction + unit conversion
├── test_esdl_kpi_exporter.py     # Unit: EsdlKpiExporter in isolation (mocked)
├── test_emission_calculator.py   # Emission calculator edge cases
├── test_database_time_series_loader.py  # InfluxDB loader (mocked connections)
├── test_database_connectivity.py # Database connection handling
├── test_credential_manager.py    # Credential loading and environment variables
├── test_input_validator.py       # Path, host, and input validation
├── test_pydantic_models.py       # Pydantic model validation (property-based)
└── test_logging_utils.py         # Structured logging

The test files fall into three categories:

  • Integration tests (test_kpi_calculator.py, test_examples.py, test_api.py) — load real ESDL fixtures, run the full pipeline, and assert specific output values.

  • Unit tests (all other test_*.py files) — test individual classes in isolation, using mocks for external dependencies (database connections, file I/O).

  • Security tests (test_credential_manager.py, test_input_validator.py) — validate threat detection, path traversal prevention, and credential management.

Run the full suite with coverage:

uv run pytest unit_test/

Adding a New Adapter

To add a new data source (e.g., MESIDO optimization results):

  1. Create adapters/mesido_adapter.py

  2. Parse the source data and construct Asset objects with the properties your source provides (costs, time series, physical properties)

  3. Return a populated EnergySystem

  4. Add a load_from_mesido() method to KpiManager

The calculators don’t need to change — they only depend on the common model.

BaseAdapter design principle: the base class enforces only the EnergySystem return type. Each adapter owns its own loading signature. KpiManager calls the specific adapter it knows about directly, not through the base class interface. The SimulatorAdapter (adapters/simulator_adapter.py) is the reference implementation for this pattern: it accepts a (pd.DataFrame, esdl_string) pair, resolves port IDs to asset IDs, and delegates cost extraction to EsdlAdapter.load_from_esdl_object().

BaseAdapter also provides _parse_esdl_string(esdl_string) — a shared static helper that wraps EnergySystemHandler.load_from_string() and raises a uniform ValidationError on failure. Both EsdlAdapter and SimulatorAdapter use it, keeping ESDL parsing logic in one place.

The # type: ignore[override] annotations on load_data and validate_source in concrete adapters are intentional — they narrow the inherited Any signature to specific types without breaking Liskov substitutability in practice, because callers always use the concrete type directly.