优化仿真求解性能并修复流量闭合问题(初版)

This commit is contained in:
ljz committed 2026-08-16 17:46:05 +08:00
1 parent 57b459bc72
commit 5332a788f3
55 files changed
+8973 -549

No files matched your search

+45 -19
View File
@@ -3,6 +3,8 @@ from __future__ import annotations
from dataclasses import dataclass
from math import isfinite
from app.simulation.core.base import Component
from app.simulation.core.ports import PortState
from app.simulation.performance import profile_phase
from app.simulation.systems.network import Endpoint, SimulationNetwork
@@ -19,35 +21,62 @@ class PneumaticVolumeDiagnostics:
}
@dataclass(frozen=True)
class _PneumaticVolumeConnectionBinding:
connected_endpoint: Endpoint
connected_port: PortState
class PneumaticVolumeResolver:
"""Propagate AMESim pneumatic external-volume connector variables."""
def __init__(self, network: SimulationNetwork) -> None:
self.network = network
self._pneumatic_ports = tuple(
component.get_port(definition.name)
for component in network.components.values()
for definition in component.active_port_definitions
if definition.kind == "physical" and definition.domain == "pneumatic"
)
self._output_components = tuple(
component
for component in network.components.values()
if type(component).pneumatic_volume_outputs
is not Component.pneumatic_volume_outputs
)
self._connected_endpoint = self._build_connection_map()
self.last_diagnostics: PneumaticVolumeDiagnostics | None = None
def _build_connection_map(self) -> dict[Endpoint, Endpoint]:
result: dict[Endpoint, Endpoint] = {}
def _build_connection_map(
self,
) -> dict[Endpoint, _PneumaticVolumeConnectionBinding]:
result: dict[Endpoint, _PneumaticVolumeConnectionBinding] = {}
for connection in self.network.connections:
if connection.kind != "physical" or connection.domain != "pneumatic":
continue
first, second = connection.endpoints
result[first] = second
result[second] = first
result[first] = _PneumaticVolumeConnectionBinding(
connected_endpoint=second,
connected_port=self.network.components[second.component].get_port(
second.port
),
)
result[second] = _PneumaticVolumeConnectionBinding(
connected_endpoint=first,
connected_port=self.network.components[first.component].get_port(
first.port
),
)
return result
@profile_phase("simulation.pneumatic_volume", minimum_mode="audit")
def solve(self) -> PneumaticVolumeDiagnostics:
for component in self.network.components.values():
for definition in component.active_port_definitions:
if definition.kind == "physical" and definition.domain == "pneumatic":
port = component.get_port(definition.name)
port.volume = 0.0
port.volume_flow = 0.0
for port in self._pneumatic_ports:
port.volume = 0.0
port.volume_flow = 0.0
outputs: dict[Endpoint, tuple[float, float]] = {}
for component in self.network.components.values():
for component in self._output_components:
for port_name, raw_values in component.pneumatic_volume_outputs().items():
port = component.get_port(port_name)
definition = port.definition
@@ -73,18 +102,15 @@ class PneumaticVolumeResolver:
propagated = 0
for endpoint, values in outputs.items():
connected = self._connected_endpoint.get(endpoint)
if connected is None:
binding = self._connected_endpoint.get(endpoint)
if binding is None:
continue
if connected in outputs:
if binding.connected_endpoint in outputs:
raise ValueError(
"A pneumatic connection cannot contain two external-volume "
f"sources: {endpoint} and {connected}."
f"sources: {endpoint} and {binding.connected_endpoint}."
)
connected_port = self.network.components[connected.component].get_port(
connected.port
)
connected_port.volume, connected_port.volume_flow = values
binding.connected_port.volume, binding.connected_port.volume_flow = values
propagated += 1
diagnostics = PneumaticVolumeDiagnostics(