对齐AMESim管阻孔口与储气耦合

app/simulation/solvers/pneumatic_storage.py

@@ -0,0 +1,256 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Iterable, Sequence
+
+from app.simulation.components.amesim.flow.pipes import (
+    AmesimPnl0001,
+    AmesimPnl0002,
+    AmesimPnl0003,
+)
+from app.simulation.solvers.mechanical import MechanicalConstraintGroup
+from app.simulation.systems.network import Endpoint, SimulationNetwork
+
+if TYPE_CHECKING:
+    from app.simulation.core.base import DynamicComponent
+    from app.simulation.solvers.mechanical import MechanicalStateReducer
+
+
+@dataclass(frozen=True)
+class PneumaticStoragePartition:
+    """One fixed-volume ``[mass, internal energy]`` pressure-state partition."""
+
+    component: DynamicComponent
+    state_offset: int
+    volume: float
+
+    @property
+    def key(self) -> tuple[str, int]:
+        return self.component.name, self.state_offset
+
+
+@dataclass(frozen=True)
+class IdealPneumaticStorageGroup:
+    partitions: tuple[PneumaticStoragePartition, ...]
+
+    @property
+    def names(self) -> tuple[str, ...]:
+        return tuple(partition.component.name for partition in self.partitions)
+
+
+def pneumatic_storage_partition(
+    network: SimulationNetwork,
+    endpoint: Endpoint,
+) -> PneumaticStoragePartition | None:
+    """Map an AMESim pressure-state port to its fixed gas-volume state slice.
+
+    PNL0001 exposes its C side at ``port_2``. PNL0003 exposes one compliance
+    at each end. PNL0002 has resistances at both external ports, so its center
+    compliance is intentionally not returned here.
+    """
+
+    component = network.components[endpoint.component]
+    if isinstance(component, AmesimPnl0003):
+        if endpoint.port == "port_1":
+            return PneumaticStoragePartition(
+                component=component,
+                state_offset=0,
+                volume=component.compliance_volume,
+            )
+        if endpoint.port == "port_2":
+            return PneumaticStoragePartition(
+                component=component,
+                state_offset=2,
+                volume=component.compliance_volume,
+            )
+        return None
+    if isinstance(component, AmesimPnl0001) and not isinstance(
+        component, AmesimPnl0002
+    ):
+        if endpoint.port == "port_2":
+            return PneumaticStoragePartition(
+                component=component,
+                state_offset=0,
+                volume=component.volume,
+            )
+    return None
+
+
+def ideal_storage_group_is_reducible(
+    network: SimulationNetwork,
+    storage_endpoints: Iterable[Endpoint],
+) -> bool:
+    partitions = [
+        pneumatic_storage_partition(network, endpoint)
+        for endpoint in storage_endpoints
+    ]
+    if not partitions or any(partition is None for partition in partitions):
+        return False
+    unique = {partition.key: partition for partition in partitions if partition}
+    if len(unique) < 2:
+        return False
+    media = {id(partition.component.medium) for partition in unique.values()}
+    return len(media) == 1 and all(partition.volume > 0.0 for partition in unique.values())
+
+
+def _pressure_storage_endpoint_groups(
+    network: SimulationNetwork,
+) -> tuple[tuple[Endpoint, ...], ...]:
+    pneumatic_endpoints = {
+        Endpoint(component.name, definition.name)
+        for component in network.components.values()
+        for definition in component.port_definitions
+        if definition.kind == "physical" and definition.domain == "pneumatic"
+    }
+    parent = {endpoint: endpoint for endpoint in pneumatic_endpoints}
+
+    def find(endpoint: Endpoint) -> Endpoint:
+        root = endpoint
+        while parent[root] != root:
+            root = parent[root]
+        while parent[endpoint] != endpoint:
+            next_endpoint = parent[endpoint]
+            parent[endpoint] = root
+            endpoint = next_endpoint
+        return root
+
+    def union(first: Endpoint, second: Endpoint) -> None:
+        first_root = find(first)
+        second_root = find(second)
+        if first_root != second_root:
+            parent[second_root] = first_root
+
+    for connection in network.connections:
+        first, second = connection.endpoints
+        if first in pneumatic_endpoints and second in pneumatic_endpoints:
+            union(first, second)
+
+    storage_endpoints: set[Endpoint] = set()
+    for component in network.components.values():
+        for equation in component.pressure_flow_equation_residuals():
+            pressure_endpoints = [
+                Endpoint(component.name, variable.rsplit(".", 2)[1])
+                for variable in equation.variables
+                if variable.startswith(f"{component.name}.") and variable.endswith(".p")
+            ]
+            if equation.relation == "equal":
+                for endpoint in pressure_endpoints[1:]:
+                    union(pressure_endpoints[0], endpoint)
+            elif equation.relation == "state":
+                storage_endpoints.update(pressure_endpoints)
+
+    by_root: dict[Endpoint, list[Endpoint]] = {}
+    for endpoint in storage_endpoints:
+        by_root.setdefault(find(endpoint), []).append(endpoint)
+    return tuple(tuple(endpoints) for endpoints in by_root.values())
+
+
+class IdealPneumaticStorageReducer:
+    """Project supported ideal C-C connections onto one thermodynamic state.
+
+    AMESim permits compatible pipe compliances to share an ideal pneumatic
+    junction. The public ODE solver keeps the original result states but
+    projects their mass and energy densities together and distributes the
+    group's total derivative by physical volume. This removes the redundant
+    pressure constraint without adding a fictitious resistance.
+    """
+
+    def __init__(
+        self,
+        network: SimulationNetwork,
+        mechanical_state_reducer: MechanicalStateReducer,
+    ) -> None:
+        self.network = network
+        self.mechanical_state_reducer = mechanical_state_reducer
+        self.groups = self._build_groups()
+        self._component_offsets = self._build_component_offsets()
+
+    def _build_groups(self) -> tuple[IdealPneumaticStorageGroup, ...]:
+        groups: list[IdealPneumaticStorageGroup] = []
+        for endpoints in _pressure_storage_endpoint_groups(self.network):
+            partitions = [
+                pneumatic_storage_partition(self.network, endpoint)
+                for endpoint in endpoints
+            ]
+            unique = {
+                partition.key: partition
+                for partition in partitions
+                if partition is not None
+            }
+            if len(unique) > 1 and len(unique) == len(
+                {endpoint.component for endpoint in endpoints}
+            ):
+                group = IdealPneumaticStorageGroup(tuple(unique.values()))
+                if ideal_storage_group_is_reducible(self.network, endpoints):
+                    groups.append(group)
+        return tuple(groups)
+
+    def _build_component_offsets(self) -> dict[str, int]:
+        offsets: dict[str, int] = {}
+        cursor = 0
+        for entry in self.mechanical_state_reducer.state_entries:
+            if isinstance(entry, MechanicalConstraintGroup):
+                cursor += 2
+            else:
+                offsets[entry.name] = cursor
+                cursor += entry.state_size
+        return offsets
+
+    def _global_offset(self, partition: PneumaticStoragePartition) -> int:
+        return self._component_offsets[partition.component.name] + partition.state_offset
+
+    def synchronize_state_vector(
+        self,
+        values: Sequence[float],
+        *,
+        validate: bool = False,
+    ) -> list[float]:
+        projected = [float(value) for value in values]
+        for group in self.groups:
+            volumes = [partition.volume for partition in group.partitions]
+            offsets = [self._global_offset(partition) for partition in group.partitions]
+            mass_densities = [
+                projected[offset] / volume
+                for offset, volume in zip(offsets, volumes)
+            ]
+            energy_densities = [
+                projected[offset + 1] / volume
+                for offset, volume in zip(offsets, volumes)
+            ]
+            if validate:
+                mass_scale = max([abs(value) for value in mass_densities] + [1.0])
+                energy_scale = max([abs(value) for value in energy_densities] + [1.0])
+                if (
+                    max(mass_densities) - min(mass_densities) > 1.0e-9 * mass_scale
+                    or max(energy_densities) - min(energy_densities)
+                    > 1.0e-9 * energy_scale
+                ):
+                    raise ValueError(
+                        "Ideally coupled AMESim pipe compliances require consistent "
+                        "initial pressure and temperature: " + ", ".join(group.names)
+                    )
+            total_volume = sum(volumes)
+            mass_density = sum(projected[offset] for offset in offsets) / total_volume
+            energy_density = (
+                sum(projected[offset + 1] for offset in offsets) / total_volume
+            )
+            for offset, volume in zip(offsets, volumes):
+                projected[offset] = mass_density * volume
+                projected[offset + 1] = energy_density * volume
+        return projected
+
+    def coupled_derivatives(self, values: Sequence[float]) -> list[float]:
+        derivatives = [float(value) for value in values]
+        for group in self.groups:
+            volumes = [partition.volume for partition in group.partitions]
+            offsets = [self._global_offset(partition) for partition in group.partitions]
+            total_volume = sum(volumes)
+            total_mass_derivative = sum(derivatives[offset] for offset in offsets)
+            total_energy_derivative = sum(
+                derivatives[offset + 1] for offset in offsets
+            )
+            for offset, volume in zip(offsets, volumes):
+                fraction = volume / total_volume
+                derivatives[offset] = total_mass_derivative * fraction
+                derivatives[offset + 1] = total_energy_derivative * fraction
+        return derivatives