257 lines
9.9 KiB
Python
257 lines
9.9 KiB
Python
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
|