refactor: implement energy chain with consumer, provider and energy system abstractions

src/libecalc/domain/energy/infrastructure_contracts.py

@@ -0,0 +1,34 @@
+"""Thin protocols that infrastructure models satisfy."""
+
+from typing import Protocol
+
+
+class TurbineEnergyUsage(Protocol):
+    values: list[float]
+
+
+class TurbineEnergyResult(Protocol):
+    energy_usage: TurbineEnergyUsage
+
+
+class TurbineResult(Protocol):
+    exceeds_maximum_load: list[bool]
+
+    def get_energy_result(self) -> TurbineEnergyResult: ...
+
+
+class TurbineDriver(Protocol):
+    """Anything that converts mechanical load [MW] to fuel, with load limits."""
+
+    def evaluate(self, load) -> TurbineResult: ...
+
+    @property
+    def max_power(self) -> float: ...
+
+
+class FuelConverter(Protocol):
+    """Anything that converts electrical power [MW] to fuel [Sm³/day]."""
+
+    def evaluate_fuel_usage(self, power_mw: float) -> float: ...
+
+    def evaluate_power_capacity_margin(self, power_mw: float) -> float: ...

src/libecalc/domain/energy/network/energy_node.py

@@ -0,0 +1,35 @@
+import uuid
+from abc import ABC, abstractmethod
+from typing import NewType
+
+from libecalc.domain.energy.network.energy_stream import EnergyStream
+
+EnergyNodeId = NewType("EnergyNodeId", uuid.UUID)
+
+
+def create_energy_node_id() -> EnergyNodeId:
+    return EnergyNodeId(uuid.uuid4())
+
+
+class EnergyNode(ABC):
+    """Base for anything in the energy network."""
+
+    @abstractmethod
+    def get_id(self) -> EnergyNodeId: ...
+
+
+class Consumer(EnergyNode, ABC):
+    """Knows its own demand — leaf node, entry point of the network."""
+
+    @abstractmethod
+    def get_demand(self) -> EnergyStream: ...
+
+
+class Provider(EnergyNode, ABC):
+    """Receives demand, transforms or fulfills it."""
+
+    @abstractmethod
+    def supply(self, demand: EnergyStream) -> EnergyStream: ...
+
+    @abstractmethod
+    def get_capacity_margin(self, demand: EnergyStream) -> float: ...

src/libecalc/domain/energy/network/energy_stream.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
+from libecalc.common.units import Unit
+
+
+class EnergyType(Enum):
+    MECHANICAL = "MECHANICAL"
+    ELECTRICAL = "ELECTRICAL"
+    FUEL = "FUEL"
+
+
+@dataclass(frozen=True)
+class EnergyStream:
+    """What flows between energy nodes."""
+
+    energy_type: EnergyType
+    value: float
+    unit: Unit
+
+    @staticmethod
+    def mechanical(power_mw: float) -> EnergyStream:
+        return EnergyStream(energy_type=EnergyType.MECHANICAL, value=power_mw, unit=Unit.MEGA_WATT)
+
+    @staticmethod
+    def electrical(power_mw: float) -> EnergyStream:
+        return EnergyStream(energy_type=EnergyType.ELECTRICAL, value=power_mw, unit=Unit.MEGA_WATT)
+
+    @staticmethod
+    def fuel(rate_sm3_per_day: float) -> EnergyStream:
+        return EnergyStream(energy_type=EnergyType.FUEL, value=rate_sm3_per_day, unit=Unit.STANDARD_CUBIC_METER_PER_DAY)

src/libecalc/domain/energy/network/energy_system.py

@@ -0,0 +1,144 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+from libecalc.domain.energy.network.energy_node import (
+    Consumer,
+    EnergyNode,
+    EnergyNodeId,
+    Provider,
+    create_energy_node_id,
+)
+from libecalc.domain.energy.network.energy_stream import EnergyStream, EnergyType
+
+
+@dataclass(frozen=True)
+class EnergyChainResult:
+    """Complete result from running an energy chain."""
+
+    demand: EnergyStream  # mechanical demand from consumer
+    fuel: EnergyStream  # fuel consumption from supply chain
+    capacity_margin_mw: float  # tightest margin across all providers
+
+
+class EnergySystem(Provider, ABC):
+    """A composite provider that contains other energy nodes."""
+
+    @abstractmethod
+    def get_children(self) -> list[EnergyNode]: ...
+
+
+@dataclass
+class EnergyChain:
+    """A consumer connected to its supply chain."""
+
+    consumer: Consumer
+    supply_chain: EnergySystem
+
+    def run(self) -> EnergyChainResult:
+        """Run the full energy chain: demand → supply → fuel + margin."""
+        demand = self.consumer.get_demand()
+        fuel = self.supply_chain.supply(demand)
+        capacity_margin = self.supply_chain.get_capacity_margin(demand)
+        return EnergyChainResult(
+            demand=demand,
+            fuel=fuel,
+            capacity_margin_mw=capacity_margin,
+        )
+
+    def get_all_nodes(self) -> list[EnergyNode]:
+        """Full graph including demand side."""
+        return [self.consumer, *self.supply_chain.get_children()]
+
+
+class PowerBus(EnergySystem):
+    """
+    Distributes electrical demand across multiple sources by priority.
+
+    Sources are tried in order. If a source cannot fully meet the remaining
+    demand (indicated by get_capacity_margin), it delivers what it can and the
+    remainder spills to the next source.
+    """
+
+    def __init__(self, sources: list[Provider]):
+        self._id = create_energy_node_id()
+        self._sources = sources
+
+    def get_id(self) -> EnergyNodeId:
+        return self._id
+
+    def get_children(self) -> list[EnergyNode]:
+        return list(self._sources)
+
+    def supply(self, demand: EnergyStream) -> EnergyStream:
+        if demand.energy_type != EnergyType.ELECTRICAL:
+            raise ValueError(f"Power bus expects electrical demand, got {demand.energy_type}")
+
+        remaining = demand.value
+        total_fuel = 0.0
+
+        for source in self._sources:
+            if remaining <= 0:
+                break
+
+            margin = source.get_capacity_margin(EnergyStream.electrical(remaining))
+            if margin < 0:
+                # Source can only partially cover remaining demand.
+                deliverable = max(0.0, remaining + margin)
+                result = source.supply(EnergyStream.electrical(deliverable))
+                total_fuel += result.value
+                remaining -= deliverable
+                continue
+
+            # Source can cover all remaining demand.
+            result = source.supply(EnergyStream.electrical(remaining))
+            total_fuel += result.value
+            remaining = 0.0
+
+        return EnergyStream.fuel(total_fuel)
+
+    def get_capacity_margin(self, demand: EnergyStream) -> float:
+        """Total spare capacity across all sources."""
+        total_margin = 0.0
+        for source in self._sources:
+            total_margin += max(0.0, source.get_capacity_margin(demand))
+        return total_margin - demand.value
+
+
+class SerialEnergySystem(EnergySystem):
+    """
+    Chain of providers where each node transforms demand and passes it to the next.
+
+    Demand enters the first node. If the node outputs a non-fuel stream (e.g. electrical),
+    that becomes the demand for the next node. When a node outputs fuel, the chain terminates.
+    """
+
+    def __init__(self, nodes: list[Provider]):
+        self._id = create_energy_node_id()
+        self._nodes = nodes
+
+    def get_id(self) -> EnergyNodeId:
+        return self._id
+
+    def get_children(self) -> list[EnergyNode]:
+        return list(self._nodes)
+
+    def supply(self, demand: EnergyStream) -> EnergyStream:
+        """Propagate demand through each node until fuel is reached."""
+        current = demand
+        for node in self._nodes:
+            current = node.supply(current)
+            if current.energy_type == EnergyType.FUEL:
+                return current
+        return current
+
+    def get_capacity_margin(self, demand: EnergyStream) -> float:
+        """Tightest margin across all nodes in the chain."""
+        min_margin = float("inf")
+        current = demand
+        for node in self._nodes:
+            margin = node.get_capacity_margin(current)
+            min_margin = min(min_margin, margin)
+            current = node.supply(current)
+            if current.energy_type == EnergyType.FUEL:
+                break
+        return min_margin

src/libecalc/domain/energy/network/nodes/consumers.py

@@ -0,0 +1,47 @@
+from libecalc.domain.energy.network.energy_node import Consumer, EnergyNodeId, create_energy_node_id
+from libecalc.domain.energy.network.energy_stream import EnergyStream
+
+
+class RotatingEquipment(Consumer):
+    """A single piece of rotating equipment as an energy node."""
+
+    def __init__(self, demand: float):
+        self._id = create_energy_node_id()
+        self._demand = demand
+
+    def get_id(self) -> EnergyNodeId:
+        return self._id
+
+    def get_demand(self) -> EnergyStream:
+        return EnergyStream.mechanical(power_mw=self._demand)
+
+
+class Shaft(Consumer):
+    """
+    Aggregates mechanical demand from rotating equipment on a common shaft,
+    adjusted for mechanical losses (bearings, gearbox, friction).
+    """
+
+    def __init__(self, rotating_equipment: list[RotatingEquipment], mechanical_efficiency: float = 1.0):
+        self._id = create_energy_node_id()
+        self._rotating_equipment = rotating_equipment
+        self._mechanical_efficiency = mechanical_efficiency
+
+    def get_id(self) -> EnergyNodeId:
+        return self._id
+
+    def get_children(self) -> list[RotatingEquipment]:
+        return self._rotating_equipment
+
+    def get_demand(self) -> EnergyStream:
+        demands = [eq.get_demand() for eq in self._rotating_equipment]
+        if not demands:
+            return EnergyStream.mechanical(0.0)
+
+        expected_unit = demands[0].unit
+        for d in demands[1:]:
+            if d.unit != expected_unit:
+                raise ValueError(f"Inconsistent units in rotating equipment demands: {d.unit} vs {expected_unit}")
+
+        total = sum(d.value for d in demands)
+        return EnergyStream.mechanical(total / self._mechanical_efficiency)