refactor: separate optimization from system (#245)

* feat: multiple streams in system

Multiple streams can be used to handle crossover streams individually in
a train. Currently no additional streams can be given, as the system
does not support providing several rates/streams per consumer.

* refactor: separate optimization from system

src/libecalc/common/priority_optimizer.py

@@ -0,0 +1,95 @@
+import operator
+import typing
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime
+from functools import reduce
+from typing import Dict, Generic, List, TypeVar
+
+import numpy as np
+from libecalc.common.units import Unit
+from libecalc.common.utils.rates import TimeSeriesBoolean, TimeSeriesInt
+
+TResult = TypeVar("TResult")
+TPriority = TypeVar("TPriority")
+
+
+@dataclass
+class PriorityOptimizerResult(Generic[TResult]):
+    priorities_used: TimeSeriesInt
+    priority_results: Dict[datetime, Dict[int, Dict[str, TResult]]]
+
+
+@dataclass
+class EvaluatorResult(Generic[TResult]):
+    id: str
+    result: TResult
+    is_valid: TimeSeriesBoolean
+
+
+class PriorityOptimizer(Generic[TResult, TPriority]):
+    def optimize(
+        self,
+        timesteps: List[datetime],
+        priorities: List[TPriority],
+        evaluator: typing.Callable[[datetime, TPriority], List[EvaluatorResult[TResult]]],
+    ) -> PriorityOptimizerResult:
+        """
+        Given a list of priorities, evaluate each priority using the evaluator. If the result of an evaluation is valid
+        the priority is selected, if invalid try the next priority.
+
+        We process each timestep separately.
+
+        Args:
+            timesteps: The timesteps that we want to figure out which priority to use for.
+            priorities: List of priorities, index is used to identify the priority in the results.
+            evaluator: The evaluator function gives a list of results back, each result with its own unique id.
+
+        Returns:
+            PriorityOptimizerResult: result containing priorities used and a map of the calculated results. The keys of
+                the results map are the timestep used, the priority index and the id of the result.
+
+        """
+        """
+
+        """
+        is_valid = TimeSeriesBoolean(timesteps=timesteps, values=[False] * len(timesteps), unit=Unit.NONE)
+        priorities_used = TimeSeriesInt(timesteps=timesteps, values=[0] * len(timesteps), unit=Unit.NONE)
+        priority_results: Dict[datetime, Dict[int, Dict[str, TResult]]] = defaultdict(dict)
+
+        for timestep_index, timestep in enumerate(timesteps):
+            priority_results[timestep] = defaultdict(dict)
+            for priority_index, priority_value in enumerate(priorities):
+                evaluator_results = evaluator(timestep, priority_value)
+                for evaluator_result in evaluator_results:
+                    priority_results[timestep][priority_index][evaluator_result.id] = evaluator_result.result
+
+                # Check if consumers are valid for this operational setting, should be valid for all consumers
+                all_evaluator_results_valid = reduce(
+                    operator.mul, [evaluator_result.is_valid for evaluator_result in evaluator_results]
+                )
+                all_evaluator_results_valid_indices = np.nonzero(all_evaluator_results_valid.values)[0]
+                all_evaluator_results_valid_indices_period_shifted = [
+                    axis_indices + timestep_index for axis_indices in all_evaluator_results_valid_indices
+                ]
+
+                # Remove already valid indices, so we don't overwrite priority used with the latest valid
+                new_valid_indices = [
+                    i for i in all_evaluator_results_valid_indices_period_shifted if not is_valid.values[i]
+                ]
+
+                # Register the valid timesteps as valid and keep track of the operational setting used
+                is_valid[new_valid_indices] = True
+                priorities_used[new_valid_indices] = priority_index
+
+                if all(is_valid.values):
+                    # quit as soon as all time-steps are valid. This means that we do not need to test all settings.
+                    break
+                elif priority_index + 1 == len(priorities):
+                    # If we are at the last operational_setting and not all indices are valid
+                    invalid_indices = [i for i, x in enumerate(is_valid.values) if not x]
+                    priorities_used[invalid_indices] = [priority_index for _ in invalid_indices]
+        return PriorityOptimizerResult(
+            priorities_used=priorities_used,
+            priority_results=dict(priority_results),
+        )

src/libecalc/core/consumers/consumer_system.py

@@ -1,17 +1,15 @@
 import itertools
 import operator
 from abc import abstractmethod
-from collections import defaultdict
 from datetime import datetime
 from functools import reduce
 from typing import Dict, List, Optional, Protocol, Tuple, TypeVar, Union
 
 import networkx as nx
 import numpy as np
+from libecalc.common.priority_optimizer import EvaluatorResult, PriorityOptimizer
 from libecalc.common.stream import Stream
-from libecalc.common.units import Unit
 from libecalc.common.utils.rates import (
-    TimeSeriesBoolean,
     TimeSeriesInt,
 )
 from libecalc.core.consumers.base import BaseConsumer
@@ -132,7 +130,7 @@ def _get_operational_settings_adjusted_for_crossover(
     def evaluate(
         self,
         variables_map: VariablesMap,
-        operational_settings: List[Union[SystemOperationalSettings]],
+        operational_setting_priorities: List[SystemOperationalSettings],
     ) -> EcalcModelResult:
         """
         Evaluating a consumer system that may be composed of both consumers and other consumer systems. It will default
@@ -142,69 +140,48 @@ def evaluate(
             - We use 1-indexed operational settings output. We should consider changing this to default 0-index, and
                 only convert when presenting results to the end-user.
         """
-        is_valid = TimeSeriesBoolean(
-            timesteps=variables_map.time_vector, values=[False] * len(variables_map.time_vector), unit=Unit.NONE
-        )
-        operational_settings_used = TimeSeriesInt(
-            timesteps=variables_map.time_vector, values=[0] * len(variables_map.time_vector), unit=Unit.NONE
-        )
-        operational_settings_results: Dict[datetime, Dict[int, Dict[str, EcalcModelResult]]] = defaultdict(dict)
 
-        for timestep_index, timestep in enumerate(variables_map.time_vector):
-            variables_map_for_timestep = variables_map.get_subset_for_timestep(timestep)
-            operational_settings_results[timestep] = defaultdict(dict)
-
-            for operational_setting_index, operational_setting in enumerate(operational_settings):
-                operational_setting_for_timestep = operational_setting.for_timestep(timestep)
-                adjusted_operational_settings = self._get_operational_settings_adjusted_for_crossover(
-                    operational_setting=operational_setting_for_timestep,
-                    variables_map=variables_map_for_timestep,
-                )
+        optimizer = PriorityOptimizer()
 
-                for consumer, adjusted_operational_setting in zip(self._consumers, adjusted_operational_settings):
-                    consumer_result = consumer.evaluate(adjusted_operational_setting)
-                    operational_settings_results[timestep][operational_setting_index][consumer.id] = consumer_result
-
-                consumer_results = operational_settings_results[timestep][operational_setting_index].values()
-
-                # Check if consumers are valid for this operational setting, should be valid for all consumers
-                all_consumer_results_valid = reduce(
-                    operator.mul, [consumer_result.component_result.is_valid for consumer_result in consumer_results]
+        def evaluator(timestep: datetime, operational_setting: SystemOperationalSettings) -> List[EvaluatorResult]:
+            operational_setting_for_timestep = operational_setting.for_timestep(timestep)
+            adjusted_operational_settings = self._get_operational_settings_adjusted_for_crossover(
+                operational_setting=operational_setting_for_timestep,
+                variables_map=variables_map.get_subset_for_timestep(timestep),
+            )
+            consumer_results_for_priority = [
+                consumer.evaluate(adjusted_operational_setting)
+                for consumer, adjusted_operational_setting in zip(self._consumers, adjusted_operational_settings)
+            ]
+            return [
+                EvaluatorResult(
+                    id=consumer_result_for_priority.component_result.id,
+                    result=consumer_result_for_priority,
+                    is_valid=consumer_result_for_priority.component_result.is_valid,
                 )
-                all_consumer_results_valid_indices = np.nonzero(all_consumer_results_valid.values)[0]
-                all_consumer_results_valid_indices_period_shifted = [
-                    axis_indices + timestep_index for axis_indices in all_consumer_results_valid_indices
-                ]
+                for consumer_result_for_priority in consumer_results_for_priority
+            ]
 
-                # Remove already valid indices, so we don't overwrite operational setting used with the latest valid
-                new_valid_indices = [
-                    i for i in all_consumer_results_valid_indices_period_shifted if not is_valid.values[i]
-                ]
-
-                # Register the valid timesteps as valid and keep track of the operational setting used
-                is_valid[new_valid_indices] = True
-                operational_settings_used[new_valid_indices] = operational_setting_index
-
-                if all(is_valid.values):
-                    # quit as soon as all time-steps are valid. This means that we do not need to test all settings.
-                    break
-                elif operational_setting_index + 1 == len(operational_settings):
-                    # If we are at the last operational_setting and not all indices are valid
-                    invalid_indices = [i for i, x in enumerate(is_valid.values) if not x]
-                    operational_settings_used[invalid_indices] = [operational_setting_index for _ in invalid_indices]
-
-        # Avoid mistakes after this point by not using defaultdict
-        operational_settings_results = dict(operational_settings_results)
+        optimizer_result = optimizer.optimize(
+            timesteps=variables_map.time_vector, priorities=operational_setting_priorities, evaluator=evaluator
+        )
 
-        consumer_results = self.collect_consumer_results(operational_settings_used, operational_settings_results)
+        consumer_results = self.collect_consumer_results(
+            operational_settings_used=optimizer_result.priorities_used,
+            operational_settings_results=optimizer_result.priority_results,
+        )
 
+        operational_settings_used = optimizer_result.priorities_used.copy()
         # Change to 1-index operational_settings_used
         for index, value in enumerate(operational_settings_used.values):
             operational_settings_used[index] = value + 1
 
         consumer_system_result = ConsumerSystemResult(
             id=self.id,
-            is_valid=is_valid,
+            is_valid=reduce(
+                operator.mul,
+                [consumer_result.is_valid for consumer_result in consumer_results],
+            ),
             timesteps=sorted({*itertools.chain(*(consumer_result.timesteps for consumer_result in consumer_results))}),
             energy_usage=reduce(
                 operator.add,

src/libecalc/core/ecalc.py

@@ -61,7 +61,7 @@ def evaluate_energy_usage(self, variables_map: dto.VariablesMap) -> Dict[str, Ec
                     variables_map=variables_map,
                 )
                 system_result = pump_system.evaluate(
-                    variables_map=variables_map, operational_settings=evaluated_operational_settings
+                    variables_map=variables_map, operational_setting_priorities=evaluated_operational_settings
                 )
                 consumer_results[component_dto.id] = system_result
                 for consumer_result in system_result.sub_components:
@@ -80,7 +80,7 @@ def evaluate_energy_usage(self, variables_map: dto.VariablesMap) -> Dict[str, Ec
                     variables_map=variables_map,
                 )
                 system_result = compressor_system.evaluate(
-                    variables_map=variables_map, operational_settings=evaluated_operational_settings
+                    variables_map=variables_map, operational_setting_priorities=evaluated_operational_settings
                 )
                 consumer_results[component_dto.id] = system_result
                 for consumer_result in system_result.sub_components:

src/libecalc/fixtures/cases/consumer_system_v2_multiple_streams/data/genset.csv

@@ -0,0 +1,4 @@
+POWER,FUEL
+0,0
+0.1,0.1
+1000000,1000000