from __future__ import annotations from collections.abc import Iterable, Mapping from dataclasses import dataclass from math import isclose, isfinite from types import MappingProxyType from app.simulation.core.metadata import ParameterDefinition from app.simulation.core.medium import GasMedium AMESIM_BUILTIN_AIR_GAS_INDEX = 0 AMESIM_DEFAULT_GAS_INDEX = AMESIM_BUILTIN_AIR_GAS_INDEX AMESIM_MIN_GAS_INDEX = AMESIM_BUILTIN_AIR_GAS_INDEX AMESIM_MIN_DEFINED_GAS_INDEX = 1 AMESIM_MAX_GAS_INDEX = 99 AMESIM_GAS_INDEX_PARAMETER = ParameterDefinition( "gi", float(AMESIM_DEFAULT_GAS_INDEX), label="介质物性模型（gi）", quantity="dimensionless", unit="", minimum=float(AMESIM_MIN_GAS_INDEX), maximum=float(AMESIM_MAX_GAS_INDEX), editor="amesimGasReference", description=( "选择本元件使用的气体介质定义索引；0 表示内置空气，" "1–99 引用画布中的介质定义组件。" ), ) AMESIM_GAS_DEFINITION_INDEX_PARAMETER = ParameterDefinition( "gi", float(AMESIM_MIN_DEFINED_GAS_INDEX), label="介质定义索引（gi）", quantity="dimensionless", unit="", minimum=float(AMESIM_MIN_DEFINED_GAS_INDEX), maximum=float(AMESIM_MAX_GAS_INDEX), description=( "介质定义在当前模型中的唯一索引；由画布自动分配，" "0 保留给内置空气。" ), ) def normalize_amesim_gas_index(value: float | int) -> int: """Validate an AMESim gas reference. Index 0 is reserved for the built-in ideal-gas air profile. Positive indices refer to medium-definition components placed in the project. """ if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError("AMESim gas type index gi must be a number.") numeric = float(value) if not isfinite(numeric): raise ValueError("AMESim gas type index gi must be finite.") rounded = round(numeric) if not isclose(numeric, rounded, rel_tol=0.0, abs_tol=1.0e-12): raise ValueError("AMESim gas type index gi must be an integer value.") index = int(rounded) if not AMESIM_MIN_GAS_INDEX <= index <= AMESIM_MAX_GAS_INDEX: raise ValueError( "AMESim gas type index gi must be between " f"{AMESIM_MIN_GAS_INDEX} and {AMESIM_MAX_GAS_INDEX}." ) return index def normalize_amesim_defined_gas_index(value: float | int) -> int: """Validate a positive index owned by a project medium definition.""" index = normalize_amesim_gas_index(value) if index < AMESIM_MIN_DEFINED_GAS_INDEX: raise ValueError( "AMESim medium definition index gi must be between " f"{AMESIM_MIN_DEFINED_GAS_INDEX} and {AMESIM_MAX_GAS_INDEX}; " "gi=0 is reserved for built-in ideal-gas air." ) return index @dataclass(frozen=True) class AmesimGasDefinition: """One AMESim PNGD-style gas-definition slot. ``fluid_type`` and ``eos_type`` are intentionally optional today. They reserve the metadata needed to map a future PNGD00 helium definition while the executable behavior is supplied by ``medium``. """ gi: int label: str medium: GasMedium fluid_type: int | None = None eos_type: int | None = None def __post_init__(self) -> None: normalized = normalize_amesim_gas_index(self.gi) object.__setattr__(self, "gi", normalized) if not self.label.strip(): raise ValueError("AMESim gas definition label must not be empty.") class AmesimGasRegistry: """Resolve AMESim component ``gi`` references to thermodynamic media.""" def __init__( self, definitions: Iterable[AmesimGasDefinition] = (), *, default_gi: int = AMESIM_DEFAULT_GAS_INDEX, ) -> None: from app.simulation.components.amesim.media.mediums import ( AmesimIdealAirMedium, ) self.default_gi = normalize_amesim_gas_index(default_gi) self._definitions: dict[int, AmesimGasDefinition] = { AMESIM_BUILTIN_AIR_GAS_INDEX: AmesimGasDefinition( gi=AMESIM_BUILTIN_AIR_GAS_INDEX, label="空气（理想气体，内置默认）", medium=AmesimIdealAirMedium(), ) } for definition in definitions: self.register(definition) @property def definitions(self) -> Mapping[int, AmesimGasDefinition]: return MappingProxyType(self._definitions) def register(self, definition: AmesimGasDefinition) -> None: if not isinstance(definition, AmesimGasDefinition): raise TypeError("AMESim gas registry entries must use AmesimGasDefinition.") if definition.gi == AMESIM_BUILTIN_AIR_GAS_INDEX: raise ValueError( "AMESim gas type index gi=0 is reserved for built-in " "ideal-gas air and cannot be replaced." ) if definition.gi in self._definitions: raise ValueError( f"AMESim gas type index gi={definition.gi} is already defined." ) self._definitions[definition.gi] = definition def copy(self) -> AmesimGasRegistry: """Return an independent registry for one project compilation.""" copied = AmesimGasRegistry( ( definition for index, definition in self._definitions.items() if index != AMESIM_BUILTIN_AIR_GAS_INDEX ), default_gi=self.default_gi, ) copied._definitions[AMESIM_BUILTIN_AIR_GAS_INDEX] = self._definitions[ AMESIM_BUILTIN_AIR_GAS_INDEX ] return copied def resolve( self, gi: float | int, *, component_name: str | None = None, ) -> GasMedium: index = normalize_amesim_gas_index(gi) try: return self._definitions[index].medium except KeyError as exc: owner = f" for component '{component_name}'" if component_name else "" available = ", ".join(str(index) for index in sorted(self._definitions)) available_message = available or "none" raise ValueError( f"AMESim gas type index gi={index}{owner} is not defined. " "Register a PNGD-style gas definition before using this index. " f"Available indices: {available_message}." ) from exc @property def default_medium(self) -> GasMedium: return self.resolve(self.default_gi) def resolve_network_media( self, component_gas_indices: Mapping[str, float | int | None], pneumatic_connections: Iterable[tuple[str, str]], ) -> dict[str, GasMedium]: """Assign one medium to every connected pneumatic circuit. Components without ``gi`` inherit the explicit index used by their circuit. Conflicting indices inside one circuit are rejected instead of silently mixing different gases. """ parents = {component_id: component_id for component_id in component_gas_indices} def find(component_id: str) -> str: parent = parents[component_id] while parent != parents[parent]: parent = parents[parent] while component_id != parent: next_component = parents[component_id] parents[component_id] = parent component_id = next_component return parent def union(left: str, right: str) -> None: left_root = find(left) right_root = find(right) if left_root != right_root: parents[right_root] = left_root for source, target in pneumatic_connections: if source in parents and target in parents: union(source, target) members_by_root: dict[str, list[str]] = {} for component_id in component_gas_indices: members_by_root.setdefault(find(component_id), []).append(component_id) media: dict[str, GasMedium] = {} for members in members_by_root.values(): indexed_components: dict[int, list[str]] = {} for component_id in members: raw_index = component_gas_indices[component_id] if raw_index is None: continue index = normalize_amesim_gas_index(raw_index) indexed_components.setdefault(index, []).append(component_id) if len(indexed_components) > 1: details = ", ".join( f"gi={index} ({', '.join(sorted(component_ids))})" for index, component_ids in sorted(indexed_components.items()) ) raise ValueError( "Connected pneumatic circuit contains conflicting AMESim " f"gas definitions: {details}." ) index = ( next(iter(indexed_components)) if indexed_components else self.default_gi ) indexed_members = indexed_components.get(index, members) medium = self.resolve( index, component_name=", ".join(sorted(indexed_members)), ) for component_id in members: media[component_id] = medium return media def default_amesim_gas_registry() -> AmesimGasRegistry: """Create a registry containing only built-in gi=0 ideal-gas air.""" return AmesimGasRegistry()