chore: mapping from yaml to temporary domain/DTO representation (#1490)

Mainly parsing and mapping from YAML to a semi-domain representation for further processing.

Author: tj098895

.gitignore

@@ -148,3 +148,5 @@ docs/.docusaurus/
 src/sandbox/
 
 .ruff_cache
+
+*.iml

libecalc.iml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/tests" isTestSource="true" />
+      <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
+      <excludeFolder url="file://$MODULE_DIR$/docs" />
+    </content>
+    <orderEntry type="jdk" jdkName="Python 3.12 (libecalc)" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PyDocumentationSettings">
+    <option name="format" value="GOOGLE" />
+    <option name="myDocStringFormat" value="Google" />
+  </component>
+  <component name="TestRunnerService">
+    <option name="PROJECT_TEST_RUNNER" value="py.test" />
+  </component>
+</module>

src/libecalc/domain/process/process_simulation.py

@@ -1,7 +1,12 @@
-from dataclasses import dataclass
-from typing import Literal
+import uuid
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from typing import Literal, NewType
+from uuid import UUID
 
-from libecalc.domain.process.process_system.process_system import ProcessSystem, ProcessSystemId
+from libecalc.domain.process.process_system.process_system import ProcessSystem
+from libecalc.domain.process.process_system.process_unit import ProcessUnit
+from libecalc.domain.process.process_system.stream_propagator import StreamPropagator
 from libecalc.domain.process.stream_distribution.common_stream_distribution import Overflow
 from libecalc.domain.process.value_objects.fluid_stream.time_series_stream import TimeSeriesStream
 from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
@@ -14,7 +19,7 @@ class CommonStreamSettings:
 
 
 @dataclass
-class CommonStreamDistributionConfig:
+class CommonStreamDistributionConfig:  # TODO: Rather a strategy that splits a stream in a method according to a policy?
     inlet_stream: TimeSeriesStream
     settings: list[CommonStreamSettings]
 
@@ -25,27 +30,98 @@ class IndividualStreamDistributionConfig:
 
 
 @dataclass
-class Constraint:
-    process_system_id: ProcessSystemId
+class Constraint:  # should this instead be more flexible wrt. matching one or more stream conditions?
     outlet_pressure: TimeSeriesExpression
 
 
 @dataclass
-class PressureControlConfig:
-    process_system_id: ProcessSystemId
+class PressureControlConfig:  # Spec
     type: Literal["UPSTREAM_CHOKE", "DOWNSTREAM_CHOKE", "COMMON_ASV", "INDIVIDUAL_ASV_RATE", "INDIVIDUAL_ASV_PRESSURE"]
 
 
 @dataclass
 class AntiSurgeConfig:
-    process_system_id: ProcessSystemId
     type: Literal["INDIVIDUAL_ASV", "COMMON_ASV"]
 
 
+ProcessProblemId = NewType("ProcessProblemId", UUID)
+
+
+def create_process_problem_id() -> ProcessProblemId:
+    return ProcessProblemId(uuid.uuid4())
+
+
+ProcessPipelineId = NewType("ProcessPipelineId", UUID)
+
+
+def create_process_pipeline_id() -> ProcessPipelineId:
+    return ProcessPipelineId(uuid.uuid4())
+
+
 @dataclass
-class ProcessSimulation:
-    process_systems: list[ProcessSystem]
-    pressure_control_strategies: list[PressureControlConfig]
-    anti_surge_strategies: list[AntiSurgeConfig]
-    constraints: list[Constraint]
-    stream_distribution: IndividualStreamDistributionConfig | CommonStreamDistributionConfig
+class ProcessPipeline:  # or simulator?
+    """
+    A part of a process topology that is calculated independently
+    container propagators - ie systems or units ...
+
+    the static physical stuff that we know a priori
+    TODO: subpipelines?
+    """
+
+    id: ProcessPipelineId
+    stream_propagators: Sequence[StreamPropagator]
+
+    def get_process_units(self) -> list[ProcessUnit]:
+        process_units = []
+        for stream_propagator in self.stream_propagators:
+            match stream_propagator:
+                case ProcessSystem():
+                    process_units.extend(stream_propagator.get_process_units())
+                case ProcessUnit():
+                    process_units.append(stream_propagator)
+
+        return process_units
+
+
+@dataclass
+class ProcessProblem:  # TODO: Rename to subproblem?
+    # can a problem exist wo. a simulation? yes, e.g. get max rate ...
+    # given a physical pipeline (a contained problem, such as a compressor train), the user needs to define strategies to find a solution for the sub problem
+    # TODO: might have subproblems, or dependencies, but we may want to add those as problems that depend on each other and needs to be evaluated in a given order
+    pressure_control_strategy: PressureControlConfig
+    anti_surge_strategy: AntiSurgeConfig
+    constraint: Constraint  # ie target pressure - and intermediate pressures ...
+    process_pipeline_id: UUID  # embedded ref here now for convenience, but not a part of this aggr, so FK/ID later
+    id: ProcessProblemId = field(default_factory=create_process_problem_id)
+
+    def get_id(self) -> ProcessProblemId:
+        return self.id
+
+    def get_constraint(self) -> Constraint:
+        return self.constraint
+
+    def get_pressure_control_strategy(self) -> PressureControlConfig:
+        return self.pressure_control_strategy
+
+    def get_anti_surge_strategy(self) -> AntiSurgeConfig:
+        return self.anti_surge_strategy
+
+
+@dataclass
+class ProcessSimulation:  # process_model?
+    """
+    TODO: one or more subproblems, where we first need to find the stream distribution before looking at each subproblem separately
+    quit and notify as soon as we notice we are not able to find a solution, or always finish?
+    """
+
+    id: UUID
+    # we wait with stream distr ...
+    # might be input param instead ...
+    # stream_distribution: (
+    #    IndividualStreamDistributionConfig | CommonStreamDistributionConfig
+    # )  # the inlet stream is only indirectly a part of sim, through the strategy. It could be separate, where the strategy is just a policy how to distr it
+    inlet_streams: list[TimeSeriesStream]
+    process_problems: list[ProcessProblem]  # todo: subproblem? TODO: a part of aggr or not?
+
+    # def get_stream_distribution_config(self) -> IndividualStreamDistributionConfig | CommonStreamDistributionConfig:
+    #    return self.stream_distribution

src/libecalc/domain/process/process_system/serial_process_system.py

@@ -2,6 +2,7 @@
 
 from libecalc.domain.process.process_system.process_system import ProcessSystem, ProcessSystemId
 from libecalc.domain.process.process_system.process_unit import ProcessUnit
+from libecalc.domain.process.process_system.stream_propagator import StreamPropagator
 from libecalc.domain.process.value_objects.fluid_stream import FluidStream
 
 
@@ -26,6 +27,9 @@ def get_process_units(self) -> Sequence[ProcessUnit | ProcessSystem]:
                 process_units.append(propagator)
         return process_units
 
+    def get_propagators(self) -> Sequence[StreamPropagator]:
+        return self._propagators
+
     def propagate_stream(self, inlet_stream: FluidStream) -> FluidStream:
         current_inlet = inlet_stream
         for process_unit in self._propagators:

src/libecalc/domain/process/value_objects/fluid_stream/time_series_stream.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
 
 from libecalc.domain.process.value_objects.fluid_stream import FluidModel
-from libecalc.domain.time_series_flow_rate import TimeSeriesFlowRate
+from libecalc.presentation.yaml.domain.expression_time_series_flow_rate import ExpressionTimeSeriesFlowRate
 from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
 
 
@@ -10,4 +10,4 @@ class TimeSeriesStream:
     fluid_model: FluidModel
     pressure_bara: TimeSeriesExpression
     temperature_kelvin: TimeSeriesExpression
-    standard_rate_m3_per_day: TimeSeriesFlowRate
+    standard_rate_m3_per_day: ExpressionTimeSeriesFlowRate

src/libecalc/domain/shared/__init__.py

@@ -50,6 +50,9 @@ def _validate(self):
             if rate < 0:
                 raise InvalidFlowRateException(rate, str(self._time_series_expression.get_expression()))
 
+    def get_original_expression(self) -> TimeSeriesExpression:
+        return self._time_series_expression
+
     def _get_stream_day_values(self) -> list[float]:
         """
         Returns the stream day flow rate values.

src/libecalc/presentation/yaml/domain/expression_time_series_flow_rate.py

@@ -26,6 +26,7 @@ def __init__(
 
         """
 
+        self._original_expression = expression
         self._expression = convert_expression(expression)
         self.expression_evaluator = expression_evaluator
         self._condition = convert_expression(condition) if condition is not None else None
@@ -36,6 +37,9 @@ def get_expression(self) -> Expression:
         """
         return self._expression
 
+    def get_original_expression(self) -> ExpressionType:
+        return self._original_expression
+
     def get_evaluated_expression(self) -> list[float]:
         """
         Evaluates expression and returns the result as a NumPy array.