chore: move expression evaluator to components

src/libecalc/application/energy/emitter.py

@@ -3,13 +3,12 @@
 
 from libecalc.application.energy.component_energy_context import ComponentEnergyContext
 from libecalc.application.energy.energy_model import EnergyModel
-from libecalc.common.variables import ExpressionEvaluator
 from libecalc.core.result.emission import EmissionResult
 
 
 class Emitter(abc.ABC):
     """
-    Something that emits something.
+    Something that emits something..
     """
 
     @property
@@ -21,5 +20,4 @@ def evaluate_emissions(
         self,
         energy_context: ComponentEnergyContext,
         energy_model: EnergyModel,
-        expression_evaluator: ExpressionEvaluator,
     ) -> Optional[dict[str, EmissionResult]]: ...

src/libecalc/application/energy_calculator.py

@@ -70,11 +70,7 @@ def evaluate_energy_usage(self) -> dict[str, EcalcModelResult]:
         for energy_component in energy_components:
             if hasattr(energy_component, "evaluate_energy_usage"):
                 context = self._get_context(energy_component.id)
-                self._consumer_results.update(
-                    energy_component.evaluate_energy_usage(
-                        context=context, expression_evaluator=self._expression_evaluator
-                    )
-                )
+                self._consumer_results.update(energy_component.evaluate_energy_usage(context=context))
 
         self._consumer_results = Numbers.format_results_to_precision(self._consumer_results, precision=6)
         return self._consumer_results
@@ -91,7 +87,6 @@ def evaluate_emissions(self) -> dict[str, dict[str, EmissionResult]]:
                 emission_result = energy_component.evaluate_emissions(
                     energy_context=self._get_context(energy_component.id),
                     energy_model=self._energy_model,
-                    expression_evaluator=self._expression_evaluator,
                 )
 
                 if emission_result is not None:

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/consumer_system/consumer_system_dto.py

@@ -60,6 +60,7 @@ def __init__(
         component_conditions: SystemComponentConditions,
         stream_conditions_priorities: Priorities[SystemStreamConditions],
         consumers: Union[list[CompressorComponent], list[PumpComponent]],
+        expression_evaluator: ExpressionEvaluator,
         fuel: Optional[dict[Period, FuelType]] = None,
         component_type: Literal[ComponentType.CONSUMER_SYSTEM_V2] = ComponentType.CONSUMER_SYSTEM_V2,
     ):
@@ -71,6 +72,7 @@ def __init__(
         self.component_conditions = component_conditions
         self.stream_conditions_priorities = stream_conditions_priorities
         self.consumers = consumers
+        self.expression_evaluator = expression_evaluator
         self.fuel = self.validate_fuel_exist(name=self.name, fuel=fuel, consumes=consumes)
         self.component_type = component_type
 
@@ -118,18 +120,14 @@ def get_component_process_type(self) -> ComponentType:
     def get_name(self) -> str:
         return self.name
 
-    def evaluate_energy_usage(
-        self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext
-    ) -> dict[str, EcalcModelResult]:
+    def evaluate_energy_usage(self, context: ComponentEnergyContext) -> dict[str, EcalcModelResult]:
         consumer_results = {}
-        evaluated_stream_conditions = self.evaluate_stream_conditions(
-            expression_evaluator=expression_evaluator,
-        )
+        evaluated_stream_conditions = self.evaluate_stream_conditions()
         optimizer = PriorityOptimizer()
 
         results_per_period: dict[str, dict[Period, ComponentResult]] = defaultdict(dict)
         priorities_used = []
-        for period in expression_evaluator.get_periods():
+        for period in self.expression_evaluator.get_periods():
             consumers_for_period = [
                 create_consumer(
                     consumer=consumer,
@@ -161,7 +159,7 @@ def evaluator(priority: PriorityID):
                 results_per_period[consumer_result.id][period] = consumer_result
 
         priorities_used = TimeSeriesString(
-            periods=expression_evaluator.get_periods(),
+            periods=self.expression_evaluator.get_periods(),
             values=priorities_used,
             unit=Unit.NONE,
         )
@@ -201,7 +199,6 @@ def evaluate_emissions(
         self,
         energy_context: ComponentEnergyContext,
         energy_model: EnergyModel,
-        expression_evaluator: ExpressionEvaluator,
     ) -> Optional[dict[str, EmissionResult]]:
         if self.is_fuel_consumer():
             assert self.fuel is not None
@@ -211,8 +208,8 @@ def evaluate_emissions(
             assert fuel_usage is not None
 
             return fuel_model.evaluate_emissions(
-                expression_evaluator=expression_evaluator,
                 fuel_rate=fuel_usage.values,
+                expression_evaluator=self.expression_evaluator,
             )
 
     def get_graph(self) -> ComponentGraph:
@@ -223,9 +220,7 @@ def get_graph(self) -> ComponentGraph:
             graph.add_edge(self.id, consumer.id)
         return graph
 
-    def evaluate_stream_conditions(
-        self, expression_evaluator: ExpressionEvaluator
-    ) -> Priorities[dict[ConsumerID, list[TimeSeriesStreamConditions]]]:
+    def evaluate_stream_conditions(self) -> Priorities[dict[ConsumerID, list[TimeSeriesStreamConditions]]]:
         parsed_priorities: Priorities[dict[ConsumerID, list[TimeSeriesStreamConditions]]] = defaultdict(dict)
         for priority_name, priority in self.stream_conditions_priorities.items():
             for consumer_name, streams_conditions in priority.items():
@@ -234,9 +229,9 @@ def evaluate_stream_conditions(
                         id=generate_id(consumer_name, stream_name),
                         name="-".join([consumer_name, stream_name]),
                         rate=TimeSeriesStreamDayRate(
-                            periods=expression_evaluator.get_periods(),
+                            periods=self.expression_evaluator.get_periods(),
                             values=list(
-                                expression_evaluator.evaluate(
+                                self.expression_evaluator.evaluate(
                                     Expression.setup_from_expression(stream_conditions["rate"].value)
                                 )
                             ),
@@ -245,9 +240,9 @@ def evaluate_stream_conditions(
                         if stream_conditions and "rate" in stream_conditions and stream_conditions["rate"] is not None
                         else None,
                         pressure=TimeSeriesFloat(
-                            periods=expression_evaluator.get_periods(),
+                            periods=self.expression_evaluator.get_periods(),
                             values=list(
-                                expression_evaluator.evaluate(
+                                self.expression_evaluator.evaluate(
                                     expression=Expression.setup_from_expression(stream_conditions["pressure"].value)
                                 )
                             ),
@@ -258,9 +253,9 @@ def evaluate_stream_conditions(
                         and stream_conditions["pressure"] is not None
                         else None,
                         fluid_density=TimeSeriesFloat(
-                            periods=expression_evaluator.get_periods(),
+                            periods=self.expression_evaluator.get_periods(),
                             values=list(
-                                expression_evaluator.evaluate(
+                                self.expression_evaluator.evaluate(
                                     expression=Expression.setup_from_expression(
                                         stream_conditions["fluid_density"].value
                                     )

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

@@ -38,6 +38,7 @@ def __init__(
             ComponentType.COMPRESSOR_SYSTEM,
         ],
         energy_usage_model: dict[Period, ElectricEnergyUsageModel],
+        expression_evaluator: ExpressionEvaluator,
         consumes: Literal[ConsumptionType.ELECTRICITY] = ConsumptionType.ELECTRICITY,
     ):
         self.name = name
@@ -47,6 +48,7 @@ def __init__(
         self.energy_usage_model = self.check_energy_usage_model(energy_usage_model)
         self._validate_el_consumer_temporal_model(self.energy_usage_model)
         self._check_model_energy_usage(self.energy_usage_model)
+        self.expression_evaluator = expression_evaluator
         self.consumes = consumes
         self.component_type = component_type
 
@@ -78,9 +80,10 @@ def get_component_process_type(self) -> ComponentType:
     def get_name(self) -> str:
         return self.name
 
-    def evaluate_energy_usage(
-        self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext
-    ) -> dict[str, EcalcModelResult]:
+    def set_consumer_results(self, consumer_results: dict[str, EcalcModelResult]):
+        self.consumer_results = consumer_results
+
+    def evaluate_energy_usage(self, context: ComponentEnergyContext) -> dict[str, EcalcModelResult]:
         consumer_results: dict[str, EcalcModelResult] = {}
         consumer = ConsumerEnergyComponent(
             id=self.id,
@@ -95,7 +98,7 @@ def evaluate_energy_usage(
                 }
             ),
         )
-        consumer_results[self.id] = consumer.evaluate(expression_evaluator=expression_evaluator)
+        consumer_results[self.id] = consumer.evaluate(expression_evaluator=self.expression_evaluator)
 
         return consumer_results