chore: implement venting emitter dto

src/libecalc/domain/infrastructure/emitters/venting_emitter.py

@@ -0,0 +1,167 @@
+from typing import Optional
+
+import numpy as np
+
+from libecalc.application.energy.component_energy_context import ComponentEnergyContext
+from libecalc.application.energy.emitter import Emitter
+from libecalc.application.energy.energy_component import EnergyComponent
+from libecalc.application.energy.energy_model import EnergyModel
+from libecalc.common.component_type import ComponentType
+from libecalc.common.string.string_utils import generate_id
+from libecalc.common.temporal_model import TemporalModel
+from libecalc.common.time_utils import Period
+from libecalc.common.units import Unit
+from libecalc.common.utils.rates import Rates, RateType, TimeSeriesFloat, TimeSeriesStreamDayRate
+from libecalc.common.variables import ExpressionEvaluator
+from libecalc.core.result.emission import EmissionResult
+from libecalc.dto.types import ConsumerUserDefinedCategoryType
+from libecalc.dto.utils.validators import convert_expression
+from libecalc.expression import Expression
+from libecalc.presentation.yaml.yaml_types.emitters.yaml_venting_emitter import (
+    YamlVentingEmission,
+    YamlVentingType,
+    YamlVentingVolume,
+)
+
+
+class VentingEmitter(Emitter, EnergyComponent):
+    def __init__(
+        self,
+        name: str,
+        expression_evaluator: ExpressionEvaluator,
+        component_type: ComponentType,
+        user_defined_category: dict[Period, ConsumerUserDefinedCategoryType],
+        emitter_type: YamlVentingType,
+    ):
+        self.name = name
+        self.expression_evaluator = expression_evaluator
+        self.component_type = component_type
+        self.user_defined_category = user_defined_category
+        self.emitter_type = emitter_type
+        self._regularity_evaluated = None
+
+    @property
+    def id(self) -> str:
+        return generate_id(self.name)
+
+    @property
+    def regularity_evaluated(self):
+        return self.expression_evaluator.evaluate(expression=TemporalModel(self._regularity)).tolist()
+
+    def evaluate_emissions(
+        self,
+        energy_context: ComponentEnergyContext,
+        energy_model: EnergyModel,
+    ) -> Optional[dict[str, EmissionResult]]:
+        self._regularity = energy_model.get_regularity(self.id)
+        venting_emitter_results = {}
+        emission_rates = self.get_emissions()
+
+        for emission_name, emission_rate in emission_rates.items():
+            emission_result = EmissionResult(
+                name=emission_name,
+                periods=self.expression_evaluator.get_periods(),
+                rate=emission_rate,
+            )
+            venting_emitter_results[emission_name] = emission_result
+        return venting_emitter_results
+
+    def get_emissions(self) -> dict[str, TimeSeriesStreamDayRate]:
+        raise NotImplementedError("Subclasses should implement this method")
+
+    def _evaluate_emission_rate(self, emission):
+        emission_rate = self.expression_evaluator.evaluate(Expression.setup_from_expression(value=emission.rate.value))
+        if emission.rate.type == RateType.CALENDAR_DAY:
+            emission_rate = Rates.to_stream_day(
+                calendar_day_rates=np.asarray(emission_rate), regularity=self.regularity_evaluated
+            ).tolist()
+        unit = emission.rate.unit.to_unit()
+        emission_rate = unit.to(Unit.TONS_PER_DAY)(emission_rate)
+        return emission_rate
+
+    def _create_time_series(self, emission_rate):
+        return TimeSeriesStreamDayRate(
+            periods=self.expression_evaluator.get_periods(),
+            values=emission_rate,
+            unit=Unit.TONS_PER_DAY,
+        )
+
+    def is_fuel_consumer(self) -> bool:
+        return False
+
+    def is_electricity_consumer(self) -> bool:
+        return False
+
+    def get_component_process_type(self) -> ComponentType:
+        return self.component_type
+
+    def get_name(self) -> str:
+        return self.name
+
+    def is_provider(self) -> bool:
+        return False
+
+    def is_container(self) -> bool:
+        return False
+
+
+class DirectVentingEmitter(VentingEmitter):
+    def __init__(self, emissions: list[YamlVentingEmission], **kwargs):
+        super().__init__(**kwargs)
+        self.emissions = emissions
+        self.emitter_type = YamlVentingType.DIRECT_EMISSION
+
+    def get_emissions(self) -> dict[str, TimeSeriesStreamDayRate]:
+        emissions = {}
+        for emission in self.emissions:
+            emission_rate = self._evaluate_emission_rate(emission)
+            emissions[emission.name] = self._create_time_series(emission_rate)
+        return emissions
+
+
+class OilVentingEmitter(VentingEmitter):
+    def __init__(self, volume: YamlVentingVolume, **kwargs):
+        super().__init__(**kwargs)
+        self.volume = volume
+        self.emitter_type = YamlVentingType.OIL_VOLUME
+
+    def get_emissions(self) -> dict[str, TimeSeriesStreamDayRate]:
+        oil_rates = self.expression_evaluator.evaluate(
+            expression=Expression.setup_from_expression(value=self.volume.rate.value)
+        )
+        if self.volume.rate.type == RateType.CALENDAR_DAY:
+            oil_rates = Rates.to_stream_day(
+                calendar_day_rates=np.asarray(oil_rates), regularity=self.regularity_evaluated
+            ).tolist()
+        emissions = {}
+        for emission in self.volume.emissions:
+            factors = self.expression_evaluator.evaluate(
+                Expression.setup_from_expression(value=emission.emission_factor)
+            )
+            unit = self.volume.rate.unit.to_unit()
+            oil_rates = unit.to(Unit.STANDARD_CUBIC_METER_PER_DAY)(oil_rates)
+            emission_rate = [oil_rate * factor for oil_rate, factor in zip(oil_rates, factors)]
+            emission_rate = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_rate)
+            emissions[emission.name] = self._create_time_series(emission_rate)
+        return emissions
+
+    def get_oil_rates(
+        self,
+        regularity: TimeSeriesFloat,
+    ) -> TimeSeriesStreamDayRate:
+        oil_rates = self.expression_evaluator.evaluate(expression=convert_expression(self.volume.rate.value))
+
+        if self.volume.rate.type == RateType.CALENDAR_DAY:
+            oil_rates = Rates.to_stream_day(
+                calendar_day_rates=np.asarray(oil_rates),
+                regularity=regularity.values,
+            ).tolist()
+
+        unit = self.volume.rate.unit.to_unit()
+        oil_rates = unit.to(Unit.STANDARD_CUBIC_METER_PER_DAY)(oil_rates)
+
+        return TimeSeriesStreamDayRate(
+            periods=self.expression_evaluator.get_periods(),
+            values=oil_rates,
+            unit=Unit.STANDARD_CUBIC_METER_PER_DAY,
+        )

src/libecalc/domain/infrastructure/energy_components/installation/installation.py

@@ -5,6 +5,7 @@
 from libecalc.common.string.string_utils import generate_id
 from libecalc.common.time_utils import Period
 from libecalc.common.variables import ExpressionEvaluator
+from libecalc.domain.infrastructure.emitters.venting_emitter import VentingEmitter
 from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system_dto import ConsumerSystem
 from libecalc.domain.infrastructure.energy_components.fuel_consumer.fuel_consumer import FuelConsumer
 from libecalc.domain.infrastructure.energy_components.generator_set.generator_set_dto import GeneratorSet
@@ -15,9 +16,6 @@
     validate_temporal_model,
 )
 from libecalc.expression import Expression
-from libecalc.presentation.yaml.yaml_types.emitters.yaml_venting_emitter import (
-    YamlVentingEmitter,
-)
 
 
 class Installation(EnergyComponent):
@@ -28,7 +26,7 @@ def __init__(
         hydrocarbon_export: dict[Period, Expression],
         fuel_consumers: list[Union[GeneratorSet, FuelConsumer, ConsumerSystem]],
         expression_evaluator: ExpressionEvaluator,
-        venting_emitters: Optional[list[YamlVentingEmitter]] = None,
+        venting_emitters: Optional[list[VentingEmitter]] = None,
         user_defined_category: Optional[InstallationUserDefinedCategoryType] = None,
     ):
         self.name = name

src/libecalc/presentation/exporter/infrastructure.py

@@ -12,6 +12,7 @@
 from libecalc.common.utils.rates import TimeSeries, TimeSeriesFloat, TimeSeriesRate, TimeSeriesStreamDayRate
 from libecalc.core.result import GeneratorSetResult
 from libecalc.domain.infrastructure import FuelConsumer, GeneratorSet
+from libecalc.domain.infrastructure.emitters.venting_emitter import DirectVentingEmitter, OilVentingEmitter
 from libecalc.dto.utils.validators import convert_expression
 from libecalc.presentation.exporter.domain.exportable import (
     Attribute,
@@ -22,8 +23,6 @@
     ExportableType,
 )
 from libecalc.presentation.yaml.yaml_types.emitters.yaml_venting_emitter import (
-    YamlDirectTypeEmitter,
-    YamlOilTypeEmitter,
     YamlVentingType,
 )
 
@@ -135,7 +134,7 @@ def get_maximum_electricity_production(self, unit: Unit) -> AttributeSet:
     def get_storage_volumes(self, unit: Unit) -> AttributeSet:
         attributes = []
         for venting_emitter in self._installation_dto.venting_emitters:
-            if venting_emitter.type != YamlVentingType.OIL_VOLUME:
+            if venting_emitter.emitter_type != YamlVentingType.OIL_VOLUME:
                 continue
 
             oil_rates = venting_emitter.get_oil_rates(
@@ -362,7 +361,7 @@ def get_emissions(self, unit: Unit) -> AttributeSet:
                     )
 
         for venting_emitter in self._installation_dto.venting_emitters:
-            assert isinstance(venting_emitter, YamlOilTypeEmitter | YamlDirectTypeEmitter)
+            assert isinstance(venting_emitter, DirectVentingEmitter | OilVentingEmitter)
 
             emissions = self._installation_graph.get_emissions(venting_emitter.id)
             for emission in emissions.values():

src/libecalc/presentation/yaml/mappers/component_mapper.py

@@ -13,6 +13,11 @@
     Asset,
     Installation,
 )
+from libecalc.domain.infrastructure.emitters.venting_emitter import (
+    DirectVentingEmitter,
+    OilVentingEmitter,
+    VentingEmitter,
+)
 from libecalc.domain.infrastructure.energy_components.common import Consumer
 from libecalc.domain.infrastructure.energy_components.electricity_consumer.electricity_consumer import (
     ElectricityConsumer,
@@ -38,6 +43,11 @@
 )
 from libecalc.presentation.yaml.yaml_types.components.yaml_generator_set import YamlGeneratorSet
 from libecalc.presentation.yaml.yaml_types.components.yaml_installation import YamlInstallation
+from libecalc.presentation.yaml.yaml_types.emitters.yaml_venting_emitter import (
+    YamlDirectTypeEmitter,
+    YamlOilTypeEmitter,
+    YamlVentingEmitter,
+)
 
 energy_usage_model_to_component_type_map = {
     ConsumerType.PUMP: ComponentType.PUMP,
@@ -55,6 +65,29 @@
 ]
 
 
+def map_yaml_to_emitter(yaml_emitter: YamlVentingEmitter, expression_evaluator: ExpressionEvaluator) -> VentingEmitter:
+    if isinstance(yaml_emitter, YamlOilTypeEmitter):
+        return OilVentingEmitter(
+            name=yaml_emitter.name,
+            expression_evaluator=expression_evaluator,
+            component_type=yaml_emitter.component_type,
+            user_defined_category=yaml_emitter.user_defined_category,
+            emitter_type=yaml_emitter.type,
+            volume=yaml_emitter.volume,
+        )
+    elif isinstance(yaml_emitter, YamlDirectTypeEmitter):
+        return DirectVentingEmitter(
+            name=yaml_emitter.name,
+            expression_evaluator=expression_evaluator,
+            component_type=yaml_emitter.component_type,
+            user_defined_category=yaml_emitter.user_defined_category,
+            emitter_type=yaml_emitter.type,
+            emissions=yaml_emitter.emissions,
+        )
+    else:
+        raise ValueError(f"Unsupported YAML emitter type: {type(yaml_emitter)}")
+
+
 def _get_component_type(energy_usage_models: dict[Period, ConsumerFunction]) -> ComponentType:
     energy_usage_model_types = {energy_usage_model.typ for energy_usage_model in energy_usage_models.values()}
 
@@ -283,7 +316,7 @@ def from_yaml_to_dto(self, data: YamlInstallation, expression_evaluator: Express
         )
 
         venting_emitters = [
-            venting_emitter.set_expression_evaluator(expression_evaluator) or venting_emitter
+            map_yaml_to_emitter(venting_emitter, expression_evaluator)
             for venting_emitter in data.venting_emitters or []
         ]