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