refactor: pump v2 support for single speed

src/libecalc/application/energy_calculator.py

@@ -1,15 +1,16 @@
 from collections import defaultdict
 from datetime import datetime
 from functools import reduce
-from typing import Dict
+from typing import Dict, List, Optional
 
 import numpy as np
 
 import libecalc.dto.components
 from libecalc import dto
 from libecalc.common.list.list_utils import elementwise_sum
 from libecalc.common.priorities import PriorityID
-from libecalc.common.priority_optimizer import PriorityOptimizer
+from libecalc.common.priority_optimizer import EvaluatorResult, PriorityOptimizer
+from libecalc.common.temporal_equipment import TemporalEquipment
 from libecalc.common.units import Unit
 from libecalc.common.utils.rates import TimeSeriesInt, TimeSeriesString
 from libecalc.core.consumers.consumer_system import ConsumerSystem
@@ -19,6 +20,7 @@
 from libecalc.core.models.fuel import FuelModel
 from libecalc.core.result import ComponentResult, EcalcModelResult
 from libecalc.core.result.emission import EmissionResult
+from libecalc.domain.stream_conditions import StreamConditions
 from libecalc.dto.component_graph import ComponentGraph
 from libecalc.dto.types import ConsumptionType
 from libecalc.presentation.yaml.yaml_types.emitters.yaml_venting_emitter import (
@@ -37,14 +39,32 @@ def __init__(
     ):
         self._graph = graph
 
-    def evaluate_energy_usage(self, variables_map: dto.VariablesMap) -> Dict[str, EcalcModelResult]:
+    def evaluate_energy_usage(
+        self,
+        variables_map: dto.VariablesMap,
+        stream_conditions: Optional[Dict[str, Dict[datetime, List[StreamConditions]]]] = None,
+    ) -> Dict[str, EcalcModelResult]:
+        """
+
+        Args:
+            variables_map:
+            stream_conditions: V2 only. Stream conditions for components supporting and requiring that. component_name -> timestep -> stream_conditions
+
+        Returns:
+
+        """
         component_ids = list(reversed(self._graph.sorted_node_ids))
         component_dtos = [self._graph.get_node(component_id) for component_id in component_ids]
 
         consumer_results: Dict[str, EcalcModelResult] = {}
 
         for component_dto in component_dtos:
-            if isinstance(component_dto, (dto.ElectricityConsumer, dto.FuelConsumer)):
+            if isinstance(component_dto, (dto.Asset, dto.Installation)):
+                # Asset and installation are just containers/aggregators, do not evaluate itself
+                pass
+            elif isinstance(component_dto, TemporalEquipment):
+                consumer_results[component_dto.id] = component_dto.evaluate(stream_conditions.get(component_dto.name))
+            elif isinstance(component_dto, (dto.ElectricityConsumer, dto.FuelConsumer)):
                 consumer = Consumer(consumer_dto=component_dto)
                 consumer_results[component_dto.id] = consumer.evaluate(variables_map=variables_map)
             elif isinstance(component_dto, dto.GeneratorSet):
@@ -93,7 +113,7 @@ def evaluate_energy_usage(self, variables_map: dto.VariablesMap) -> Dict[str, Ec
                         component_conditions=component_dto.component_conditions,
                     )
 
-                    def evaluator(priority: PriorityID):
+                    def evaluator(priority: PriorityID) -> List[EvaluatorResult]:
                         stream_conditions_for_priority = evaluated_stream_conditions[priority]
                         stream_conditions_for_timestep = {
                             component_id: [
@@ -140,6 +160,10 @@ def evaluator(priority: PriorityID):
                         sub_components=[],
                         models=[],
                     )
+            else:
+                print(
+                    f"Unknown component not evaluated: {type(component_dto)}"
+                )  # Added to more easily see when something that should be evaluated is skipped
 
         return consumer_results
 
@@ -157,7 +181,9 @@ def evaluate_emissions(
         """
         emission_results: Dict[str, Dict[str, EmissionResult]] = {}
         for consumer_dto in self._graph.nodes.values():
-            if isinstance(consumer_dto, (dto.FuelConsumer, dto.GeneratorSet)):
+            if isinstance(consumer_dto, (dto.FuelConsumer, dto.GeneratorSet)) or (
+                isinstance(consumer_dto, TemporalEquipment) and consumer_dto.fuel
+            ):  # Only for fuel driven, el driven are exempted for emissions of course ...
                 fuel_model = FuelModel(consumer_dto.fuel)
                 energy_usage = consumer_results[consumer_dto.id].component_result.energy_usage
                 emission_results[consumer_dto.id] = fuel_model.evaluate_emissions(
@@ -187,4 +213,9 @@ def evaluate_emissions(
                     )
                 }
                 emission_results[consumer_dto.id] = emission_result
+            else:
+                print(
+                    f"Ignoring collecting emissions for {type(consumer_dto)}"
+                )  # Added to more easily see when something that should be evaluated for emissions is skipped
+
         return emission_results

src/libecalc/application/graph_result.py

@@ -162,6 +162,7 @@ def _evaluate_installations(self, variables_map: dto.VariablesMap) -> List[libec
                 [
                     emission_dto_results[fuel_consumer_id]
                     for fuel_consumer_id in self.graph.get_successors(installation.id)
+                    if fuel_consumer_id in emission_dto_results
                 ]
             )
 
@@ -828,7 +829,11 @@ def get_asset_result(self) -> libecalc.dto.result.EcalcModelResult:
                             )
                         ]
                     )
-            elif consumer_node_info.component_type in [ComponentType.PUMP, ComponentType.PUMP_SYSTEM]:
+            elif consumer_node_info.component_type in [
+                ComponentType.PUMP,
+                ComponentType.PUMP_SYSTEM,
+                ComponentType.PUMP_V2,
+            ]:
                 component = self.graph.get_node(consumer_id)
                 for model in consumer_result.models:
                     models.extend(
@@ -970,12 +975,12 @@ def get_asset_result(self) -> libecalc.dto.result.EcalcModelResult:
                     id=consumer_result.component_result.id,
                     is_valid=consumer_result.component_result.is_valid,
                 )
-            elif consumer_node_info.component_type == ComponentType.PUMP:
+            elif consumer_node_info.component_type in [ComponentType.PUMP, ComponentType.PUMP_V2]:
                 obj = dto.result.results.PumpResult(
                     name=consumer_node_info.name,
                     parent=self.graph.get_predecessor(consumer_id),
                     component_level=consumer_node_info.component_level,
-                    componentType=consumer_node_info.component_type,
+                    componentType=ComponentType.PUMP,  # TODO: Since v1 and v2 currently has same structure, we specify as v1 to avoid having to update Web
                     emissions=self._parse_emissions(self.emission_results[consumer_id], regularity)
                     if consumer_id in self.emission_results
                     else [],
@@ -1211,7 +1216,8 @@ def get_asset_result(self) -> libecalc.dto.result.EcalcModelResult:
                     },
                 )
 
-            sub_components.append(obj)
+            if obj:
+                sub_components.append(obj)
 
         for installation in asset.installations:
             regularity = regularities[installation.id]  # Already evaluated regularities

src/libecalc/common/string/string_utils.py

@@ -23,6 +23,9 @@ def generate_id(*args: str) -> str:
     the id, i.e. it should not be used to get the name of a consumer, even if the name might be used to create the id.
 
     If there are many strings they are joined together.
+
+    TODO: Deprecate. This was needed when names on components etc were NON-UNIQUE, and we were not able to enforce uniqueness. Now we have been able to enforce uniqueness for names, and can therefore remove this method.
+    We may however want to generate "internal" IDs again, if it turns out that it is difficult to deal with names due to encoding etc, such as when using in URLs etc
     """
     return "-".join(args)
 

src/libecalc/common/temporal_equipment.py

@@ -0,0 +1,63 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Dict, Generic, List, Optional, Protocol, TypeVar
+
+from libecalc.common.temporal_model import TemporalModel
+
+# from libecalc.core.result import EcalcModelResult # TODO: Cannot include due to circular import ...yet
+from libecalc.domain.stream_conditions import StreamConditions
+from libecalc.dto.base import ComponentType, ConsumerUserDefinedCategoryType
+
+
+class Evaluationable(Protocol):
+    def evaluate(self, streams: List[StreamConditions]) -> Any:
+        ...
+
+
+EquipmentType = TypeVar("EquipmentType", bound=Evaluationable)
+
+
+@dataclass
+class TemporalEquipment(Generic[EquipmentType]):
+    """
+    V2 only.
+
+    The temporal layer of equipment, includes metadata and the equipment domain model itself for each timestep, basically
+    a parsed and flattened version of the yaml
+
+    Except from data, the properties are just needed for metadata, aggregation etc for LTP and similar
+
+    This class should be the same for all equipment represented in e.g. yaml, basically a wrapper around the core domain layer
+    """
+
+    id: Optional[str]
+    component_type: Optional[ComponentType]
+    name: Optional[str]
+    user_defined_category: Optional[Dict[datetime, ConsumerUserDefinedCategoryType]]
+    fuel: Optional[Dict[datetime, Any]]  # cannot specify FuelType due to circular import ..
+    data: TemporalModel[EquipmentType]
+
+    def evaluate(self, stream_conditions: Dict[datetime, List[StreamConditions]]) -> Any:
+        """
+        Evaluate the temporal domain models.
+
+        TODO: Here we might want to use cache somehow, either for this single run wrt same results for same timesteps,
+        but also across runs etc
+
+        Args:
+            stream_conditions:
+
+        Returns:
+
+        """
+        if not stream_conditions:
+            raise ValueError(f"Missing stream conditions for {self.name}")
+
+        result = None
+        for timestep, model in self.data.items():
+            if result is None:  # TODO: Use map reduce
+                result = model.evaluate(stream_conditions.get(timestep.start))
+            else:
+                result.extend(model.evaluate(stream_conditions.get(timestep.start)))
+
+        return result

src/libecalc/common/temporal_model.py

@@ -16,8 +16,13 @@ class Model(Generic[ModelType]):
 
 
 class TemporalModel(Generic[ModelType]):
+    """
+    Very generic data structure to support temporal models of any type
+    """
+
     def __init__(self, data: Dict[datetime, ModelType]):
         self._data = data
+
         start_times = list(data.keys())
         end_times = [*start_times[1:], datetime.max]
         self.models = [

src/libecalc/common/time_utils.py

@@ -116,6 +116,10 @@ def define_time_model_for_period(
 ) -> Optional[Dict[datetime, Any]]:
     """Process time model based on the target period.
 
+    TODO: We should probably in general get a model for a timestep instead of a period. A period may have more than one
+    model, unless we are 100% sure that the period is actually 2 successive time steps. Also, a period is only needed
+    when we need to calculate volumes, which would only be relevant wrt final aggregations and calculations ...
+
     Steps:
         - Add a default start date if the model is not already a time model
         - Filter definitions outside given time period

src/libecalc/core/consumers/pump/component.py

@@ -23,7 +23,7 @@ class Pump(BaseConsumerWithoutOperationalSettings):
     def __init__(self, id: str, pump_model: PumpModel):
         self.id = id
         self._pump_model = pump_model
-        self._operational_settings: Optional[PumpOperationalSettings] = None
+        self._operational_settings: Optional[PumpOperationalSettings] = None  # TODO: Always set, also for single pumps?
 
     def get_max_rate(self, inlet_stream: StreamConditions, target_pressure: Pressure) -> List[float]:
         """

src/libecalc/core/models/chart/single_speed_chart.py

@@ -1,4 +1,7 @@
+from typing import List
+
 import numpy as np
+from typing_extensions import Self
 
 from libecalc.core.models.chart.base import ChartCurve
 from libecalc.dto.types import ChartAreaFlag
@@ -14,6 +17,27 @@ class SingleSpeedChart(ChartCurve):
     Note: For a single speed chart the speed is optional, but it is good practice to include it.
     """
 
+    @classmethod
+    def create(
+        cls,
+        speed_rpm: float,
+        rate_actual_m3_hour: List[float],
+        polytropic_head_joule_per_kg: List[float],
+        efficiency_fraction: List[float],
+    ) -> Self:
+        """
+        V2 only.
+
+        This class method to construct single speed chart is being used for V2 only to avoid using old init constructor
+        used for v1 (since we must be backwards compatible, and the v1 constructor uses a single DTO for initialization)
+        """
+        instance = super().__new__(cls)
+        instance.speed_rpm = speed_rpm
+        instance.rate_actual_m3_hour = rate_actual_m3_hour
+        instance.polytropic_head_joule_per_kg = polytropic_head_joule_per_kg
+        instance.efficiency_fraction = efficiency_fraction
+        return instance
+
     def get_chart_area_flag(self, rate: float) -> ChartAreaFlag:
         """Set chart area flag based on rate [Am3/h]."""
         if rate < self.minimum_rate:

src/libecalc/core/result/base.py

@@ -33,7 +33,16 @@ def extend(self, other: Self) -> Self:
                 # In case of nested models such as compressor with turbine
                 values.extend(other_values)
             elif isinstance(values, list):
-                if isinstance(other_values, list):
+                # Temporary v2 only
+                # We are not able to currently import PumpModelResult due to circular import, therefore we use duck typing...
+                if (
+                    len(values) == 1
+                    and len(other_values) == 1
+                    and hasattr(values[0], "extend")
+                    and hasattr(other_values[0], "extend")
+                ):
+                    values[0].extend(other_values[0])
+                elif isinstance(other_values, list):
                     self.__setattr__(attribute, values + other_values)
                 else:
                     self.__setattr__(attribute, values + [other_values])

src/libecalc/domain/stream_conditions.py

@@ -106,7 +106,7 @@ def copy(self, update: Dict = None):
     @classmethod
     def mix_all(cls, streams: List[StreamConditions]) -> StreamConditions:
         if len(streams) == 0:
-            raise ValueError("No streams to mix")
+            raise ValueError("No streams to mix.")
         if len(streams) == 1:
             return streams[0].copy()
 

src/libecalc/dto/components.py

@@ -1,7 +1,7 @@
 from abc import ABC
 from collections import defaultdict
 from datetime import datetime
-from typing import Dict, List, Literal, Optional, TypeVar, Union
+from typing import Any, Dict, List, Literal, Optional, TypeVar, Union
 
 try:
     from pydantic.v1 import Field, root_validator
@@ -25,6 +25,8 @@
     TimeSeriesFloat,
     TimeSeriesStreamDayRate,
 )
+
+# from libecalc.core.consumers.pump import Pump # TODO: Cannot import due to circular deps ..
 from libecalc.dto.base import (
     Component,
     ComponentType,
@@ -293,7 +295,10 @@ class GeneratorSet(BaseEquipment):
     component_type: Literal[ComponentType.GENERATOR_SET] = ComponentType.GENERATOR_SET
     fuel: Dict[datetime, FuelType]
     generator_set_model: Dict[datetime, GeneratorSetSampled]
-    consumers: List[Union[ElectricityConsumer, ConsumerSystem]] = Field(default_factory=list)
+    consumers: List[Union[ElectricityConsumer, ConsumerSystem, Any]] = Field(
+        default_factory=list
+    )  # Any here is Pump, that cannot be explicitly specified due to circular import ...temporalequipment?
+    # Any must be set in order for pydantic to accept/validate the consumer
     _validate_genset_temporal_models = validator("generator_set_model", "fuel", allow_reuse=True)(
         validate_temporal_model
     )
@@ -382,6 +387,30 @@ def id(self):
     def installation_ids(self) -> List[str]:
         return [installation.id for installation in self.installations]
 
+    def get_component_by_id(self, id: str) -> Optional[Component]:
+        """
+        Get a component by id, if it exists, otherwise None
+        Args:
+            id:
+
+        Returns:
+
+        """
+        for installation in self.installations:
+            if installation.id == id:
+                return installation
+            for fuel_consumer in installation.fuel_consumers:
+                if fuel_consumer.id == id:
+                    return fuel_consumer
+                if isinstance(fuel_consumer, dto.GeneratorSet):
+                    for electricity_consumer in fuel_consumer.consumers:
+                        if electricity_consumer.id == id:
+                            return electricity_consumer
+            for venting_emitter in installation.venting_emitters:
+                if venting_emitter.id == id:
+                    return venting_emitter
+        return None
+
     def get_component_ids_for_installation_id(self, installation_id: str) -> List[str]:
         installation = self.get_installation(installation_id)
         component_ids = []