chore: use pipeline sections instead of pipelines for solvers (#1668)

The solvers got a new signature previously, accepting PipelineSections instead of
Pipelines. A Pipeline consists of one or more pipeline sections. Each pipeline
section is considered for now to be able to be handled separately, but in sequence,
in according to the order in the (parent) pipeline.

Both multipressuresolver and pipelinesectionsolver are able to handle pipeline sections,
and we therefore supply a complete set of data/params with a pipeline section for 
resolving, and no need to know about the complete pipeline.

In order to easily be able to set up the sections both when parsing yaml and when
we actually get the timeseries data to solve it, we create the sections when parsing,
and reuse the same sections when solving.

Refs: equinor/ecalc-internal#2008

Author: tj098895

src/libecalc/ecalc_model/process_simulation.py

@@ -14,11 +14,17 @@
 )
 from libecalc.ecalc_model.time_series_stream import TimeSeriesStream
 from libecalc.presentation.yaml.domain.time_series_expression import TimeSeriesExpression
-from libecalc.process.process_pipeline.process_pipeline import ProcessPipelineId
+from libecalc.process.process_pipeline.process_pipeline import (
+    ProcessPipelineId,
+    ProcessPipelineSectionId,
+    ProcessUnitConnectionId,
+)
 from libecalc.process.process_pipeline.process_unit import ProcessUnitId
 from libecalc.process.process_solver.anti_surge.anti_surge_strategy import AntiSurgeType
 from libecalc.process.process_solver.configuration_handler import ConfigurationHandler
-from libecalc.process.process_solver.pressure_control.pressure_control_strategy import PressureControlType
+from libecalc.process.process_solver.pressure_control.pressure_control_strategy import (
+    PressureControlType,
+)
 from libecalc.process.stream_distribution.common_stream_distribution import Overflow
 
 
@@ -52,60 +58,87 @@ class AntiSurgeConfig:
 @value_object
 class Constraint:
     outlet_pressure: TimeSeriesExpression
-    pressure_control: PressureControlConfig
-    anti_surge: AntiSurgeConfig
     target_process_unit_id: ProcessUnitId
+    target_process_connection_id: ProcessUnitConnectionId  # Currently the outlet of the process_unit above
 
 
 ProcessProblemId = NewType("ProcessProblemId", UUID)
+ProcessProblemSectionId = NewType("ProcessProblemSectionId", UUID)
 
 
-@value_object
-class ProcessProblemSection:
-    """A process section assembled for solver execution."""
+class ProcessProblemSection(Entity[ProcessProblemSectionId]):
+    def __init__(
+        self,
+        process_pipeline_section_id: ProcessPipelineSectionId,
+        constraint: Constraint,
+        pressure_control: PressureControlConfig,
+        anti_surge: AntiSurgeConfig,
+        process_problem_section_id: ProcessProblemSectionId | None = None,
+    ):
+        self._process_pipeline_section_id = process_pipeline_section_id
+        self._constraint = constraint
+        self._pressure_control = pressure_control
+        self._anti_surge = anti_surge
+        self._id: Final[ProcessProblemSectionId] = process_problem_section_id or ProcessProblemSection._create_id()
+
+    def get_id(self) -> ProcessProblemSectionId:
+        return self._id
 
-    process_unit_ids: list[ProcessUnitId]
-    configuration_handlers: Sequence[ConfigurationHandler]
-    constraint: Constraint
+    def get_process_pipeline_section_id(self) -> ProcessPipelineSectionId:
+        return self._process_pipeline_section_id
 
+    def get_constraint(self) -> Constraint:
+        return self._constraint
 
-class ProcessProblem(Entity[ProcessProblemId]):  # 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
+    def get_pressure_control(self) -> PressureControlConfig:
+        return self._pressure_control
 
+    def get_anti_surge(self) -> AntiSurgeConfig:
+        return self._anti_surge
+
+    @classmethod
+    def _create_id(cls: type[Self]) -> ProcessProblemSectionId:
+        return ProcessProblemSectionId(ecalc_id_generator())
+
+
+class ProcessProblem(Entity[ProcessProblemId]):
     def __init__(
         self,
-        process_problem_sections: Sequence[ProcessProblemSection],
-        configuration_handlers: Sequence[ConfigurationHandler],
+        process_problem_sections: Sequence[
+            ProcessProblemSection
+        ],  # Currently we consider that ProblemSections needs to be solved in sequence, therefore we have instance here and not ID
+        configuration_handlers: Sequence[
+            ConfigurationHandler
+        ],  # inter or intra section config handlers, stored here for now
         process_pipeline_id: ProcessPipelineId,
         process_problem_id: ProcessProblemId | None = None,
     ):
-        self.process_problem_sections = process_problem_sections
-        self.configuration_handlers = configuration_handlers
-        self.process_pipeline_id = process_pipeline_id
+        self._process_problem_sections = process_problem_sections
+        self._configuration_handlers = configuration_handlers
+        self._process_pipeline_id = process_pipeline_id
         self._id: Final[ProcessProblemId] = process_problem_id or ProcessProblem._create_id()
 
     def get_id(self) -> ProcessProblemId:
         return self._id
 
+    def get_process_pipeline_id(self) -> ProcessPipelineId:
+        return self._process_pipeline_id
+
     @classmethod
     def _create_id(cls: type[Self]) -> ProcessProblemId:
         return ProcessProblemId(ecalc_id_generator())
 
-    def get_constraints(self) -> list[Constraint]:
-        return [section.constraint for section in self.process_problem_sections]
+    def get_process_problem_sections(self) -> Sequence[ProcessProblemSection]:
+        return self._process_problem_sections
+
+    def get_configuration_handlers(self) -> Sequence[ConfigurationHandler]:
+        return self._configuration_handlers
 
 
 ProcessSimulationId = NewType("ProcessSimulationId", UUID)
 
 
 class ProcessSimulation(Entity[ProcessSimulationId]):  # 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?
-    """
-
     def __init__(
         self,
         name: str,

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

@@ -1,7 +1,7 @@
 from typing import assert_never, get_args
 
 from libecalc.common.errors.ecalc_validation_error import EcalcValidationException
-from libecalc.common.errors.exceptions import InvalidResourceException
+from libecalc.common.errors.exceptions import InvalidResourceException, ProgrammingError
 from libecalc.common.time_utils import Period
 from libecalc.common.units import Unit
 from libecalc.common.variables import ExpressionEvaluator
@@ -63,6 +63,7 @@
 from libecalc.process.process_pipeline.process_pipeline import (
     ProcessPipeline,
     ProcessPipelineId,
+    ProcessPipelineSection,
 )
 from libecalc.process.process_pipeline.process_unit import ProcessUnit, ProcessUnitId
 from libecalc.process.process_solver.feasibility_solver import FeasibilitySolver
@@ -336,9 +337,12 @@ def map_process_simulation(
                 assert isinstance(compressor, Compressor)
                 shaft.connect(compressor)
 
+            # Shaft is currently (potentially) an inter-section configuration handler
             problem_configuration_handlers.append(shaft)
 
-            process_units: list[ProcessUnit] = []
+            # Since we, in addition to intra-section connection, have inter-section connections, we keep them separate
+            # from section, and keep track of them at pipeline level
+            process_pipeline_sections: list[ProcessPipelineSection] = []
             process_problem_sections: list[ProcessProblemSection] = []
             pipeline_constraints = yaml_process_simulation.constraints.get(item.name)
 
@@ -354,30 +358,57 @@ def map_process_simulation(
                 mapped_sections=mapped_sections, fluid_service=self._fluid_service
             )
 
-            for mapped_section, assembled_section in zip(mapped_sections, assembled_sections, strict=True):
-                constraint = Constraint(
-                    outlet_pressure=TimeSeriesExpression(
-                        expression=mapped_section.constraint.outlet_pressure,
-                        expression_evaluator=self._expression_evaluator,
-                    ),
-                    pressure_control=PressureControlConfig(type=mapped_section.constraint.pressure_control),
-                    anti_surge=AntiSurgeConfig(mapped_section.constraint.anti_surge),
-                    target_process_unit_id=mapped_section.target_process_unit_id,
-                )
+            # Set up pipeline and pipeline sections
+            for nr, assembled_section in enumerate(assembled_sections):
+                process_section_process_units = list(assembled_section.process_units)
+                if nr == 0:  # first section
+                    process_section_process_units.insert(0, Inlet())
+                if nr == len(mapped_sections) - 1:  # last section
+                    process_section_process_units.append(Outlet())
+
+                process_pipeline_sections.append(ProcessPipelineSection(process_units=process_section_process_units))
+
+            # TODO: We should move this class to this module/layer
+            # in particular because it creates necessary connections, which means that they will get new IDs
+            process_pipeline = ProcessPipeline(
+                name=item.name,
+                process_pipeline_sections=process_pipeline_sections,
+            )
+
+            # Set up problem and problem sections
+            for pipeline_section, mapped_section, assembled_section in zip(
+                process_pipeline_sections, mapped_sections, assembled_sections, strict=True
+            ):
+                try:
+                    constraint = Constraint(
+                        outlet_pressure=TimeSeriesExpression(
+                            expression=mapped_section.constraint.outlet_pressure,
+                            expression_evaluator=self._expression_evaluator,
+                        ),
+                        target_process_unit_id=mapped_section.target_process_unit_id,
+                        target_process_connection_id=next(
+                            process_unit_connection.get_id()
+                            for process_unit_connection in process_pipeline.get_process_unit_connections()
+                            if process_unit_connection.get_from_process_unit_id()
+                            == mapped_section.target_process_unit_id
+                        ),
+                    )
+                except StopIteration:
+                    raise ProgrammingError(
+                        f"Not able to set constraint on OUTLET of process unit '{mapped_section.target_process_unit_id}' because the Connection ID was not found."
+                    ) from None
+
                 process_problem_sections.append(
                     ProcessProblemSection(
-                        process_unit_ids=[u.get_id() for u in assembled_section.process_units],
-                        configuration_handlers=assembled_section.configuration_handlers,
+                        process_pipeline_section_id=pipeline_section.get_id(),
+                        # configuration_handlers=assembled_section.configuration_handlers,  # TODO: We currently store config handlers at problem level. They may be inter or intra section ...
                         constraint=constraint,
+                        pressure_control=PressureControlConfig(type=mapped_section.constraint.pressure_control),
+                        anti_surge=AntiSurgeConfig(mapped_section.constraint.anti_surge),
                     )
                 )
-                process_units.extend(assembled_section.process_units)
-
-            # A pipeline must have a start and an end - always add process units for that (ie owner of inlet and outlet streams)
-            process_units.append(Outlet())
-            process_units.insert(0, Inlet())
-
-            process_pipeline = ProcessPipeline(name=item.name, stream_propagators=process_units)
+                # Choke and recirculation configuration handlers are currently intra-section config handlers
+                problem_configuration_handlers.extend(assembled_section.configuration_handlers)
 
             predefined_configurations[process_pipeline.get_id()] = problem_time_series_configurations
 

src/libecalc/process/process_pipeline/process_pipeline.py

@@ -4,17 +4,83 @@
 
 from libecalc.common.ddd.entity import Entity
 from libecalc.common.utils.ecalc_uuid import ecalc_id_generator
-from libecalc.process.process_pipeline.process_unit import ProcessUnit
+from libecalc.process.process_pipeline.process_unit import ProcessUnit, ProcessUnitId
 
 ProcessPipelineId = NewType("ProcessPipelineId", UUID)
+ProcessPipelineSectionId = NewType("ProcessPipelineSectionId", UUID)
+ProcessUnitConnectionId = NewType("ProcessUnitConnectionId", UUID)
+
+
+class ProcessUnitConnection(Entity[ProcessUnitConnectionId]):
+    def __init__(
+        self,
+        from_process_unit_id: ProcessUnitId,
+        to_process_unit_id: ProcessUnitId,
+        process_unit_connection_id: ProcessUnitConnectionId | None = None,
+    ):
+        self._from_process_unit_id = from_process_unit_id
+        self._to_process_unit_id = to_process_unit_id
+        self._id: Final[ProcessUnitConnectionId] = process_unit_connection_id or ProcessUnitConnection._create_id()
+
+    def get_id(self) -> ProcessUnitConnectionId:
+        return self._id
+
+    def get_from_process_unit_id(self) -> ProcessUnitId:
+        return self._from_process_unit_id
+
+    def get_to_process_unit_id(self) -> ProcessUnitId:
+        return self._to_process_unit_id
+
+    @classmethod
+    def _create_id(cls: type[Self]) -> ProcessUnitConnectionId:
+        return ProcessUnitConnectionId(ecalc_id_generator())
+
+    def __str__(self):
+        return f"ProcessUnitConnection(process_unit_connection_id={self._id}, from_process_unit_id={self._from_process_unit_id}, to_process_unit_id={self._to_process_unit_id})"
+
+
+class ProcessPipelineSection(Entity[ProcessPipelineSectionId]):
+    def __init__(
+        self,
+        process_units: Sequence[ProcessUnit],  # TODO: Reassure they are in order
+        process_pipeline_section_id: ProcessPipelineSectionId | None = None,
+    ):
+        self._process_units = process_units
+        self._id: Final[ProcessPipelineSectionId] = process_pipeline_section_id or ProcessPipelineSection._create_id()
+
+    def get_id(self) -> ProcessPipelineSectionId:
+        return self._id
+
+    def get_process_units(self) -> list[ProcessUnit]:
+        return list(self._process_units)
+
+    @classmethod
+    def _create_id(cls: type[Self]) -> ProcessPipelineSectionId:
+        return ProcessPipelineSectionId(ecalc_id_generator())
+
+    def __str__(self):
+        return f"ProcessPipelineSection(process_pipeline_section_id={self._id}, process_units={self._process_units})"
 
 
 class ProcessPipeline(Entity[ProcessPipelineId]):
+    """
+    TODO: We define this class in process, but we do not use it here. We use it in the ephemeral mapping layer,
+    when storing in db And some testing. We should move it.
+    In particular because it creates the necessary connections, which means that they will get new IDs
+    """
+
     def __init__(
-        self, name: str, stream_propagators: Sequence[ProcessUnit], process_pipeline_id: ProcessPipelineId | None = None
+        self,
+        name: str,
+        process_pipeline_sections: Sequence[ProcessPipelineSection],
+        process_pipeline_id: ProcessPipelineId | None = None,
     ):
         self._name = name
-        self._stream_propagators = stream_propagators
+        self._process_pipeline_sections = process_pipeline_sections
+        self._process_unit_connections = ProcessPipeline._create_process_unit_connections(
+            process_pipeline_sections=process_pipeline_sections
+        )
+        self._process_pipeline_id = process_pipeline_id
         self._id: Final[ProcessPipelineId] = process_pipeline_id or ProcessPipeline._create_id()
 
     def get_id(self) -> ProcessPipelineId:
@@ -23,12 +89,63 @@ def get_id(self) -> ProcessPipelineId:
     def get_name(self) -> str:
         return self._name
 
-    def get_process_units(self) -> list[ProcessUnit]:
-        return list(self._stream_propagators)
+    def get_process_pipeline_sections(self) -> Sequence[ProcessPipelineSection]:
+        return self._process_pipeline_sections
+
+    def get_process_unit_connections(self) -> Sequence[ProcessUnitConnection]:
+        return self._process_unit_connections
+
+    def get_process_units(self) -> Sequence[ProcessUnit]:
+        return [
+            process_unit
+            for process_section in self.get_process_pipeline_sections()
+            for process_unit in process_section.get_process_units()
+        ]
 
     @classmethod
     def _create_id(cls: type[Self]) -> ProcessPipelineId:
         return ProcessPipelineId(ecalc_id_generator())
 
+    @staticmethod
+    def _create_process_unit_connections(
+        process_pipeline_sections: Sequence[ProcessPipelineSection],
+    ) -> Sequence[ProcessUnitConnection]:
+        """
+        Connections kept at this level for now. Could potentially be handled at section level, which makes more sense,
+        if we define the owner of a connection to be the process unit with the OUTLET (the last process unit will then not
+        own any connections, as it doesn't have an outlet ...). In this class we can therefore just gather/build connections
+        from the process unit sections, either represented as connections or just "outlet"s, which is what we need to store
+        stream info on. inlet is just the result from previous process unit or section or sth else. The parameter basically.
+
+        Currently we keep it here though, because inlet and outlet are "equivalent", and we therefore have
+        intra and inter connections between process units, ie. across sections. So, we need to make a decision on who,
+        is the owner - the section or the pipeline. Since it is just an identifier based on the surrogate of from and to,
+        with a unique id, with no extra information, we can generate it on the fly.
+
+        Private, because this should be internal information handled by pipeline, and only exposed for read
+
+        Args:
+            process_pipeline_sections:
+
+        Returns:
+
+        """
+        process_unit_connections: list[ProcessUnitConnection] = []
+
+        previous_process_unit: ProcessUnit | None = None
+        for process_section in process_pipeline_sections:  # Ordered, in sequence!
+            for process_unit in process_section.get_process_units():  # Ordered, in sequence!
+                if previous_process_unit is not None:
+                    process_unit_connections.append(
+                        ProcessUnitConnection(
+                            from_process_unit_id=previous_process_unit.get_id(),
+                            to_process_unit_id=process_unit.get_id(),
+                        )
+                    )
+
+                previous_process_unit = process_unit
+
+        return process_unit_connections
+
     def __str__(self):
-        return f"ProcessPipeline(process_pipeline_id={self._id}, name={self._name}, stream_propagators={self._stream_propagators})"
+        return f"ProcessPipeline(process_pipeline_id={self._id}, name={self._name}, process_pipeline_sections={self._process_pipeline_sections})"