Source code for kpicalculator.calculators.energy_calculator

# src/kpicalculator/calculators/energy_calculator.py
import logging

from ..adapters.common_model import Asset, AssetType, EnergySystem
from ..common.constants import (
    CONSUMPTION_FIELDS,
    DEMAND_FIELDS,
    PRODUCTION_FIELDS,
    SECONDS_PER_YEAR,
)

logger = logging.getLogger(__name__)




class EnergyCalculator:
    """Calculator for energy-related KPIs."""

    def __init__(self, energy_system: EnergySystem):
        """Initialize the energy calculator.

        Args:
            energy_system: Energy system to calculate KPIs for
        """
        self.energy_system = energy_system



    def get_total_energy_consumption_per_year(self) -> float:
        """Calculate total energy consumption per year.

        Returns:
            Total energy consumption in joules per year
        """
        total_consumption = 0.0

        for asset in self.energy_system.assets:
            if asset.asset_type == AssetType.CONSUMER:
                total_consumption += self._calculate_asset_energy_consumption(asset)

        return total_consumption




    def get_total_energy_demand_per_year(self) -> float:
        """Calculate total energy demand per year.

        Returns:
            Total energy demand in joules per year
        """
        total_demand = 0.0

        for asset in self.energy_system.assets:
            if asset.asset_type == AssetType.CONSUMER:
                total_demand += self._calculate_asset_energy_demand(asset)

        return total_demand




    def get_total_energy_production_per_year(self) -> float:
        """Calculate total energy production per year.

        Returns:
            Total energy production in joules per year
        """
        total_production = 0.0

        for asset in self.energy_system.assets:
            if asset.asset_type in [AssetType.PRODUCER, AssetType.GEOTHERMAL]:
                total_production += self._calculate_asset_energy_production(asset)

        return total_production




    def calculate_system_efficiency(self) -> float:
        """Calculate overall system efficiency.

        Returns:
            System efficiency as a ratio (0-1)
        """
        production = self.get_total_energy_production_per_year()
        consumption = self.get_total_energy_consumption_per_year()

        if production <= 0:
            return 0.0

        return consumption / production


    def _calculate_asset_energy_consumption(self, asset: Asset) -> float:
        """Calculate energy consumption for a specific asset.

        Args:
            asset: Asset to calculate energy consumption for

        Returns:
            Energy consumption in joules per year
        """
        if not asset.time_series:
            logger.debug(
                "No time series data for asset '%s'. Consumption returned as 0.0.",
                asset.name,
            )
            return 0.0

        # Look for consumption time series
        ts_name = None
        for name in CONSUMPTION_FIELDS:
            if name in asset.time_series:
                ts_name = name
                break

        if not ts_name:
            logger.debug(
                "No consumption field found in time series for asset '%s'. Returning 0.0.",
                asset.name,
            )
            return 0.0

        ts = asset.time_series[ts_name]

        # Calculate annual energy
        duration = ts.time_step * len(ts.values)
        if duration <= 0:
            logger.warning(
                "Non-positive duration in consumption time series for asset '%s'. Returning 0.0.",
                asset.name,
            )
            return 0.0
        time_factor = SECONDS_PER_YEAR / duration
        energy_sum = sum(ts.values) * ts.time_step

        return energy_sum * time_factor

    def _calculate_asset_energy_demand(self, asset: Asset) -> float:
        """Calculate energy demand for a specific asset.

        Args:
            asset: Asset to calculate energy demand for

        Returns:
            Energy demand in joules per year
        """
        if not asset.time_series:
            logger.debug(
                "No time series data for asset '%s'. Demand returned as 0.0.",
                asset.name,
            )
            return 0.0

        # Look for demand time series
        ts_name = None
        for name in DEMAND_FIELDS:
            if name in asset.time_series:
                ts_name = name
                break

        if not ts_name:
            return self._calculate_asset_energy_consumption(asset)  # Fall back to consumption

        ts = asset.time_series[ts_name]

        # Calculate annual energy
        duration = ts.time_step * len(ts.values)
        if duration <= 0:
            logger.warning(
                "Non-positive duration in demand time series for asset '%s'. Returning 0.0.",
                asset.name,
            )
            return 0.0
        time_factor = SECONDS_PER_YEAR / duration
        energy_sum = sum(ts.values) * ts.time_step

        return energy_sum * time_factor



    def get_asset_energy_production_per_year(self, asset: Asset) -> float:
        """Calculate annual energy production for a single producing asset.

        Args:
            asset: Producing asset (PRODUCER or GEOTHERMAL type).

        Returns:
            Energy production in joules per year, or 0.0 if no time series data.
        """
        return self._calculate_asset_energy_production(asset)


    def _calculate_asset_energy_production(self, asset: Asset) -> float:
        """Calculate energy production for a specific asset.

        Args:
            asset: Asset to calculate energy production for

        Returns:
            Energy production in joules per year
        """
        if not asset.time_series:
            logger.debug(
                "No time series data for asset '%s'. Production returned as 0.0.",
                asset.name,
            )
            return 0.0

        # Look for production time series
        ts_name = None
        for name in PRODUCTION_FIELDS:
            if name in asset.time_series:
                ts_name = name
                break

        if not ts_name:
            logger.debug(
                "No production field found in time series for asset '%s'. Returning 0.0.",
                asset.name,
            )
            return 0.0

        ts = asset.time_series[ts_name]

        # Calculate annual energy
        duration = ts.time_step * len(ts.values)
        if duration <= 0:
            logger.warning(
                "Non-positive duration in production time series for asset '%s'. Returning 0.0.",
                asset.name,
            )
            return 0.0
        time_factor = SECONDS_PER_YEAR / duration
        energy_sum = sum(ts.values) * ts.time_step

        return energy_sum * time_factor