feat: integrate AMESim media models and editor UI

This commit is contained in:
ljz committed 2026-07-31 23:36:58 +08:00
1 parent 3e4f466114
commit e7177ab03e
181 files changed
+7042 -25270

No files matched your search

@@ -1,5 +1,10 @@
from __future__ import annotations
from app.simulation.examples.test_mql.system import (
TestMqlRunConfig,
TestMqlSimulationResult,
TestMqlSystem,
)
from app.simulation.examples.test_mql.run import (
PreparedTestMqlRun,
TestMqlRunResult,
@@ -10,7 +15,10 @@ from app.simulation.examples.test_mql.run import (
__all__ = [
"PreparedTestMqlRun",
"TestMqlRunConfig",
"TestMqlRunResult",
"TestMqlSimulationResult",
"TestMqlSystem",
"prepare_test_mql_run",
"run_prepared_test_mql",
"run_test_mql",
@@ -0,0 +1,77 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from app.simulation.reporting.amesim_results import AmesimResults, load_test_mql_amesim_results
from app.simulation.reporting.test_mql_comparison import TestMqlComparisonResult
from app.simulation.reporting.test_mql_observations import (
TestMqlObservationCatalog,
build_test_mql_observation_catalog,
)
from app.simulation.reporting.test_mql_output_schema import (
TestMqlOutputSchema,
build_test_mql_output_schema,
)
from app.simulation.reporting.test_mql_output_validation import (
TestMqlValidatedOutput,
compare_validated_test_mql_output,
validate_test_mql_output,
)
@dataclass(frozen=True)
class TestMqlBaselineRun:
amesim_results: AmesimResults
observation_catalog: TestMqlObservationCatalog
output_schema: TestMqlOutputSchema
output: TestMqlValidatedOutput
comparison: TestMqlComparisonResult
@property
def sample_count(self) -> int:
return len(self.output.times)
@property
def signal_count(self) -> int:
return len(self.output.data_paths)
def run_test_mql_baseline_passthrough(
archive_path: Path,
*,
data_paths: tuple[str, ...] | list[str] | None = None,
) -> TestMqlBaselineRun:
amesim_results = load_test_mql_amesim_results(archive_path)
observation_catalog = build_test_mql_observation_catalog(amesim_results)
output_schema = build_test_mql_output_schema(
amesim_results,
observation_catalog=observation_catalog,
)
selected_paths = tuple(data_paths) if data_paths is not None else output_schema.data_paths()
baseline_series = observation_catalog.baseline_series_by_data_path(
amesim_results,
selected_paths,
)
output = validate_test_mql_output(
times=amesim_results.times,
series_by_data_path=baseline_series,
schema=output_schema,
data_paths=selected_paths,
require_all_schema_paths=data_paths is None,
)
comparison = compare_validated_test_mql_output(
times=output.times,
series_by_data_path=output.series_by_data_path,
schema=output_schema,
amesim_results=amesim_results,
data_paths=output.data_paths,
require_all_schema_paths=data_paths is None,
)
return TestMqlBaselineRun(
amesim_results=amesim_results,
observation_catalog=observation_catalog,
output_schema=output_schema,
output=output,
comparison=comparison,
)
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,468 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from app.simulation.examples.test_mql.primitives.pneumatic import m3_to_cm3
from app.simulation.reporting.amesim_results import AmesimResults, load_test_mql_amesim_results
from app.simulation.reporting.test_mql_comparison import TestMqlComparisonResult
from app.simulation.reporting.test_mql_observations import (
TestMqlObservationCatalog,
build_test_mql_observation_catalog,
)
from app.simulation.reporting.test_mql_output_schema import (
TestMqlOutputSchema,
build_test_mql_output_schema,
)
from app.simulation.reporting.test_mql_output_validation import (
TestMqlValidatedOutput,
compare_validated_test_mql_output,
validate_test_mql_output,
)
from app.simulation.reporting.test_mql_variables import build_test_mql_variable_catalog
from app.simulation.examples.test_mql.mechanical import (
TestMqlMechanicalAssembly,
build_test_mql_mechanical_assembly,
)
from app.simulation.examples.test_mql.pneumatic import (
TestMqlPneumaticAssembly,
build_test_mql_pneumatic_assembly,
)
@dataclass(frozen=True)
class TestMqlComputedPistonGeometryRun:
amesim_results: AmesimResults
observation_catalog: TestMqlObservationCatalog
output_schema: TestMqlOutputSchema
mechanical_assembly: TestMqlMechanicalAssembly
output: TestMqlValidatedOutput
comparison: TestMqlComparisonResult
@property
def sample_count(self) -> int:
return len(self.output.times)
@property
def signal_count(self) -> int:
return len(self.output.data_paths)
@dataclass(frozen=True)
class TestMqlComputedGeometryRun:
amesim_results: AmesimResults
observation_catalog: TestMqlObservationCatalog
output_schema: TestMqlOutputSchema
mechanical_assembly: TestMqlMechanicalAssembly
pneumatic_assembly: TestMqlPneumaticAssembly
output: TestMqlValidatedOutput
comparison: TestMqlComparisonResult
@property
def sample_count(self) -> int:
return len(self.output.times)
@property
def signal_count(self) -> int:
return len(self.output.data_paths)
@dataclass(frozen=True)
class TestMqlComputedLineRelationsRun:
amesim_results: AmesimResults
observation_catalog: TestMqlObservationCatalog
output_schema: TestMqlOutputSchema
output: TestMqlValidatedOutput
comparison: TestMqlComparisonResult
@property
def sample_count(self) -> int:
return len(self.output.times)
@property
def signal_count(self) -> int:
return len(self.output.data_paths)
@dataclass(frozen=True)
class TestMqlComputedPneumaticRelationsRun:
amesim_results: AmesimResults
observation_catalog: TestMqlObservationCatalog
output_schema: TestMqlOutputSchema
output: TestMqlValidatedOutput
comparison: TestMqlComparisonResult
@property
def sample_count(self) -> int:
return len(self.output.times)
@property
def signal_count(self) -> int:
return len(self.output.data_paths)
@dataclass(frozen=True)
class TestMqlComputedMechanicalRelationsRun:
amesim_results: AmesimResults
observation_catalog: TestMqlObservationCatalog
output_schema: TestMqlOutputSchema
mechanical_assembly: TestMqlMechanicalAssembly
output: TestMqlValidatedOutput
comparison: TestMqlComparisonResult
@property
def sample_count(self) -> int:
return len(self.output.times)
@property
def signal_count(self) -> int:
return len(self.output.data_paths)
def run_test_mql_computed_piston_geometry(
archive_path: Path,
) -> TestMqlComputedPistonGeometryRun:
amesim_results = load_test_mql_amesim_results(archive_path)
observation_catalog = build_test_mql_observation_catalog(amesim_results)
output_schema = build_test_mql_output_schema(
amesim_results,
observation_catalog=observation_catalog,
)
variable_catalog = build_test_mql_variable_catalog(amesim_results)
mechanical_assembly = build_test_mql_mechanical_assembly(
amesim_results=amesim_results,
variable_catalog=variable_catalog,
)
output_series = _compute_piston_geometry_series(amesim_results, mechanical_assembly)
output_data_paths = tuple(output_series)
output = validate_test_mql_output(
times=amesim_results.times,
series_by_data_path=output_series,
schema=output_schema,
data_paths=output_data_paths,
)
comparison = compare_validated_test_mql_output(
times=output.times,
series_by_data_path=output.series_by_data_path,
schema=output_schema,
amesim_results=amesim_results,
data_paths=output.data_paths,
)
return TestMqlComputedPistonGeometryRun(
amesim_results=amesim_results,
observation_catalog=observation_catalog,
output_schema=output_schema,
mechanical_assembly=mechanical_assembly,
output=output,
comparison=comparison,
)
def run_test_mql_computed_geometry(
archive_path: Path,
) -> TestMqlComputedGeometryRun:
amesim_results = load_test_mql_amesim_results(archive_path)
observation_catalog = build_test_mql_observation_catalog(amesim_results)
output_schema = build_test_mql_output_schema(
amesim_results,
observation_catalog=observation_catalog,
)
variable_catalog = build_test_mql_variable_catalog(amesim_results)
mechanical_assembly = build_test_mql_mechanical_assembly(
amesim_results=amesim_results,
variable_catalog=variable_catalog,
)
pneumatic_assembly = build_test_mql_pneumatic_assembly()
output_series = {
**_compute_piston_geometry_series(amesim_results, mechanical_assembly),
**_compute_variable_chamber_volume_series(
amesim_results,
mechanical_assembly,
pneumatic_assembly,
),
}
output_data_paths = tuple(output_series)
output = validate_test_mql_output(
times=amesim_results.times,
series_by_data_path=output_series,
schema=output_schema,
data_paths=output_data_paths,
)
comparison = compare_validated_test_mql_output(
times=output.times,
series_by_data_path=output.series_by_data_path,
schema=output_schema,
amesim_results=amesim_results,
data_paths=output.data_paths,
)
return TestMqlComputedGeometryRun(
amesim_results=amesim_results,
observation_catalog=observation_catalog,
output_schema=output_schema,
mechanical_assembly=mechanical_assembly,
pneumatic_assembly=pneumatic_assembly,
output=output,
comparison=comparison,
)
def run_test_mql_computed_line_relations(
archive_path: Path,
) -> TestMqlComputedLineRelationsRun:
amesim_results = load_test_mql_amesim_results(archive_path)
observation_catalog = build_test_mql_observation_catalog(amesim_results)
output_schema = build_test_mql_output_schema(
amesim_results,
observation_catalog=observation_catalog,
)
output_series = _compute_line_reversed_series(amesim_results, observation_catalog)
output_data_paths = tuple(output_series)
output = validate_test_mql_output(
times=amesim_results.times,
series_by_data_path=output_series,
schema=output_schema,
data_paths=output_data_paths,
)
comparison = compare_validated_test_mql_output(
times=output.times,
series_by_data_path=output.series_by_data_path,
schema=output_schema,
amesim_results=amesim_results,
data_paths=output.data_paths,
)
return TestMqlComputedLineRelationsRun(
amesim_results=amesim_results,
observation_catalog=observation_catalog,
output_schema=output_schema,
output=output,
comparison=comparison,
)
def run_test_mql_computed_pneumatic_relations(
archive_path: Path,
) -> TestMqlComputedPneumaticRelationsRun:
amesim_results = load_test_mql_amesim_results(archive_path)
observation_catalog = build_test_mql_observation_catalog(amesim_results)
output_schema = build_test_mql_output_schema(
amesim_results,
observation_catalog=observation_catalog,
)
output_series = {
**_compute_chamber_duplicate_series(amesim_results, observation_catalog),
**_compute_orifice_reversed_series(amesim_results, observation_catalog),
}
output_data_paths = tuple(output_series)
output = validate_test_mql_output(
times=amesim_results.times,
series_by_data_path=output_series,
schema=output_schema,
data_paths=output_data_paths,
)
comparison = compare_validated_test_mql_output(
times=output.times,
series_by_data_path=output.series_by_data_path,
schema=output_schema,
amesim_results=amesim_results,
data_paths=output.data_paths,
)
return TestMqlComputedPneumaticRelationsRun(
amesim_results=amesim_results,
observation_catalog=observation_catalog,
output_schema=output_schema,
output=output,
comparison=comparison,
)
def run_test_mql_computed_mechanical_relations(
archive_path: Path,
) -> TestMqlComputedMechanicalRelationsRun:
amesim_results = load_test_mql_amesim_results(archive_path)
observation_catalog = build_test_mql_observation_catalog(amesim_results)
output_schema = build_test_mql_output_schema(
amesim_results,
observation_catalog=observation_catalog,
)
variable_catalog = build_test_mql_variable_catalog(amesim_results)
mechanical_assembly = build_test_mql_mechanical_assembly(
amesim_results=amesim_results,
variable_catalog=variable_catalog,
)
output_series = {
**_compute_mass_duplicate_series(amesim_results, mechanical_assembly),
**_compute_inactive_mass_force_series(amesim_results, mechanical_assembly),
**_compute_zero_force_source_series(amesim_results, mechanical_assembly),
}
output_data_paths = tuple(output_series)
output = validate_test_mql_output(
times=amesim_results.times,
series_by_data_path=output_series,
schema=output_schema,
data_paths=output_data_paths,
)
comparison = compare_validated_test_mql_output(
times=output.times,
series_by_data_path=output.series_by_data_path,
schema=output_schema,
amesim_results=amesim_results,
data_paths=output.data_paths,
)
return TestMqlComputedMechanicalRelationsRun(
amesim_results=amesim_results,
observation_catalog=observation_catalog,
output_schema=output_schema,
mechanical_assembly=mechanical_assembly,
output=output,
comparison=comparison,
)
def _compute_piston_geometry_series(
amesim_results: AmesimResults,
mechanical_assembly: TestMqlMechanicalAssembly,
) -> dict[str, tuple[float, ...]]:
series_by_data_path: dict[str, tuple[float, ...]] = {}
for alias in sorted(mechanical_assembly.pistons):
piston = mechanical_assembly.pistons[alias]
geometry = piston.geometry()
x4 = amesim_results.series(f"x4@{alias}")
x5 = amesim_results.series(f"x5@{alias}")
v4 = amesim_results.series(f"v4@{alias}")
v5 = amesim_results.series(f"v5@{alias}")
series_by_data_path[f"length@{alias}"] = tuple(
geometry.chamber_length_mm(port4, port5)
for port4, port5 in zip(x4, x5)
)
series_by_data_path[f"vol1@{alias}"] = tuple(
geometry.chamber_volume_cm3(port4, port5)
for port4, port5 in zip(x4, x5)
)
series_by_data_path[f"vvol1@{alias}"] = tuple(
geometry.chamber_volume_rate_l_min(port4, port5)
for port4, port5 in zip(v4, v5)
)
return series_by_data_path
def _compute_variable_chamber_volume_series(
amesim_results: AmesimResults,
mechanical_assembly: TestMqlMechanicalAssembly,
pneumatic_assembly: TestMqlPneumaticAssembly,
) -> dict[str, tuple[float, ...]]:
series_by_data_path: dict[str, tuple[float, ...]] = {}
for chamber_alias in sorted(pneumatic_assembly.variable_chambers):
chamber = pneumatic_assembly.variable_chambers[chamber_alias]
piston_alias = _piston_alias_for_variable_chamber(chamber_alias)
piston = mechanical_assembly.pistons[piston_alias]
geometry = piston.geometry()
x4 = amesim_results.series(f"x4@{piston_alias}")
x5 = amesim_results.series(f"x5@{piston_alias}")
dead_volume_cm3 = m3_to_cm3(chamber.dead_volume)
series_by_data_path[f"vol@{chamber_alias}"] = tuple(
dead_volume_cm3 + geometry.chamber_volume_cm3(port4, port5)
for port4, port5 in zip(x4, x5)
)
return series_by_data_path
def _piston_alias_for_variable_chamber(chamber_alias: str) -> str:
if not chamber_alias.startswith("pn_c1"):
raise ValueError(f"Unexpected PNCH012 alias: {chamber_alias}")
return chamber_alias.replace("pn_c1", "pn_brp2", 1)
def _compute_mass_duplicate_series(
amesim_results: AmesimResults,
mechanical_assembly: TestMqlMechanicalAssembly,
) -> dict[str, tuple[float, ...]]:
series_by_data_path: dict[str, tuple[float, ...]] = {}
for alias in sorted(mechanical_assembly.masses):
for signal_name in ("x1", "v1", "acc1"):
source_path = f"{signal_name}@{alias}"
duplicate_path = f"{signal_name}dup@{alias}"
series_by_data_path[duplicate_path] = tuple(
-value for value in amesim_results.series(source_path)
)
return series_by_data_path
def _compute_inactive_mass_force_series(
amesim_results: AmesimResults,
mechanical_assembly: TestMqlMechanicalAssembly,
) -> dict[str, tuple[float, ...]]:
series_by_data_path: dict[str, tuple[float, ...]] = {}
for alias in sorted(mechanical_assembly.masses):
mass = mechanical_assembly.masses[alias].endstop()
x1 = amesim_results.series(f"x1@{alias}")
v1 = amesim_results.series(f"v1@{alias}")
series_by_data_path[f"Fmin@{alias}"] = tuple(
mass.lower_static_force_magnitude(displacement)
for displacement in x1
)
series_by_data_path[f"Fvisc@{alias}"] = tuple(
mass.viscous_friction_force(velocity)
for velocity in v1
)
series_by_data_path[f"Ffric@{alias}"] = tuple(0.0 for _ in x1)
return series_by_data_path
def _compute_zero_force_source_series(
amesim_results: AmesimResults,
mechanical_assembly: TestMqlMechanicalAssembly,
) -> dict[str, tuple[float, ...]]:
return {
f"fzero@{alias}": tuple(0.0 for _ in amesim_results.times)
for alias in sorted(mechanical_assembly.zero_force_sources)
}
def _compute_chamber_duplicate_series(
amesim_results: AmesimResults,
observation_catalog: TestMqlObservationCatalog,
) -> dict[str, tuple[float, ...]]:
series_by_data_path: dict[str, tuple[float, ...]] = {}
for binding in observation_catalog.chambers.bindings:
pressure_series = amesim_results.series(binding.pressure_path)
temperature_series = amesim_results.series(binding.temperature_path)
for duplicate_path in binding.pressure_duplicate_paths:
series_by_data_path[duplicate_path] = tuple(pressure_series)
for duplicate_path in binding.temperature_duplicate_paths:
series_by_data_path[duplicate_path] = tuple(temperature_series)
return series_by_data_path
def _compute_orifice_reversed_series(
amesim_results: AmesimResults,
observation_catalog: TestMqlObservationCatalog,
) -> dict[str, tuple[float, ...]]:
series_by_data_path: dict[str, tuple[float, ...]] = {}
for binding in observation_catalog.orifices.bindings:
series_by_data_path[binding.reversed_mass_flow_path] = tuple(
-value for value in amesim_results.series(binding.primary_mass_flow_path)
)
series_by_data_path[binding.reversed_enthalpy_flow_path] = tuple(
-value for value in amesim_results.series(binding.primary_enthalpy_flow_path)
)
return series_by_data_path
def _compute_line_reversed_series(
amesim_results: AmesimResults,
observation_catalog: TestMqlObservationCatalog,
) -> dict[str, tuple[float, ...]]:
series_by_data_path: dict[str, tuple[float, ...]] = {}
for binding in observation_catalog.lines.by_submodel("PNL00R"):
if len(binding.mass_flow_paths) != 2 or len(binding.enthalpy_flow_paths) != 2:
raise ValueError(f"Expected two PNL00R flow paths for {binding.alias}.")
primary_mass_path, reversed_mass_path = binding.mass_flow_paths
primary_enthalpy_path, reversed_enthalpy_path = binding.enthalpy_flow_paths
series_by_data_path[reversed_mass_path] = tuple(
-value for value in amesim_results.series(primary_mass_path)
)
series_by_data_path[reversed_enthalpy_path] = tuple(
-value for value in amesim_results.series(primary_enthalpy_path)
)
return series_by_data_path
+151
View File
@@ -0,0 +1,151 @@
from __future__ import annotations
import ast
import operator
from dataclasses import dataclass
from math import isfinite
from typing import Any
from app.simulation.core.peng_robinson import HELIUM_PR, PengRobinsonFluid
from app.simulation.examples.test_mql.system import COMPONENT_SPECS, GLOBAL_PARAMETERS
_BINARY_OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
}
_UNARY_OPERATORS = {
ast.UAdd: operator.pos,
ast.USub: operator.neg,
}
class TestMqlExpressionError(ValueError):
"""Raised when an AMESim parameter expression cannot be resolved safely."""
@dataclass(frozen=True)
class TestMqlResolvedParameter:
name: str
title: str
raw_value: str
units: str
value: float | None
@property
def is_numeric(self) -> bool:
return self.value is not None
@dataclass(frozen=True)
class TestMqlResolvedComponent:
alias: str
component_name: str
submodel: str
label: str
parameters: dict[str, TestMqlResolvedParameter]
def parameter_value(self, name: str) -> float:
parameter = self.parameters[name]
if parameter.value is None:
raise KeyError(f"Parameter {name!r} on {self.alias!r} is not numeric")
return parameter.value
@dataclass(frozen=True)
class TestMqlConfig:
raw_global_parameters: dict[str, str]
global_parameters: dict[str, float]
fluid: PengRobinsonFluid
components: tuple[TestMqlResolvedComponent, ...]
@classmethod
def from_amesim_specs(cls) -> "TestMqlConfig":
raw_globals = dict(GLOBAL_PARAMETERS)
numeric_globals = {
name: value
for name, raw in raw_globals.items()
if (value := resolve_numeric_expression(raw, {})) is not None
}
components = tuple(
_resolve_component(spec, numeric_globals)
for spec in COMPONENT_SPECS
)
return cls(
raw_global_parameters=raw_globals,
global_parameters=numeric_globals,
fluid=HELIUM_PR,
components=components,
)
def component(self, alias: str) -> TestMqlResolvedComponent:
for component in self.components:
if component.alias == alias:
return component
raise KeyError(alias)
def components_by_submodel(self, submodel: str) -> tuple[TestMqlResolvedComponent, ...]:
return tuple(component for component in self.components if component.submodel == submodel)
def _resolve_component(
spec: dict[str, Any],
variables: dict[str, float],
) -> TestMqlResolvedComponent:
parameters = {}
for parameter in spec.get("parameters", []):
name = str(parameter["name"])
raw_value = str(parameter["value"])
parameters[name] = TestMqlResolvedParameter(
name=name,
title=str(parameter["title"]),
raw_value=raw_value,
units=str(parameter["units"]),
value=resolve_numeric_expression(raw_value, variables),
)
return TestMqlResolvedComponent(
alias=str(spec["alias"]),
component_name=str(spec["component_name"]),
submodel=str(spec["submodel"]),
label=str(spec["label"]),
parameters=parameters,
)
def resolve_numeric_expression(
expression: str,
variables: dict[str, float],
) -> float | None:
expression = expression.strip()
if not expression:
return None
normalized = expression.replace("^", "**")
try:
parsed = ast.parse(normalized, mode="eval")
value = float(_eval_node(parsed.body, variables))
except (SyntaxError, TestMqlExpressionError, ValueError, TypeError, ZeroDivisionError):
return None
return value if isfinite(value) else None
def _eval_node(node: ast.AST, variables: dict[str, float]) -> float:
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return float(node.value)
if isinstance(node, ast.Name):
if node.id not in variables:
raise TestMqlExpressionError(f"Unknown variable: {node.id}")
return float(variables[node.id])
if isinstance(node, ast.BinOp):
operator_type = type(node.op)
if operator_type not in _BINARY_OPERATORS:
raise TestMqlExpressionError(f"Unsupported binary operator: {operator_type}")
return float(_BINARY_OPERATORS[operator_type](_eval_node(node.left, variables), _eval_node(node.right, variables)))
if isinstance(node, ast.UnaryOp):
operator_type = type(node.op)
if operator_type not in _UNARY_OPERATORS:
raise TestMqlExpressionError(f"Unsupported unary operator: {operator_type}")
return float(_UNARY_OPERATORS[operator_type](_eval_node(node.operand, variables)))
raise TestMqlExpressionError(f"Unsupported expression node: {type(node)}")
@@ -0,0 +1,451 @@
from __future__ import annotations
import re
import tarfile
from dataclasses import dataclass
from pathlib import Path
from app.simulation.examples.test_mql.system import CONNECTION_SPECS, GLOBAL_PARAMETERS
from app.simulation.examples.test_mql.config import resolve_numeric_expression
AMESIM_REFERENCE_PRESSURE_PA = 101_300.0
@dataclass(frozen=True)
class TestMqlPnl0001Spec:
alias: str
source_component: str
source_port: str
target_component: str
target_port: str
diameter_mm: float
length_m: float
relative_roughness: float
polytropic_constant: float
heat_transfer_coefficient: float
external_temperature_k: float
gas_type_index: int
mode: int
initial_temperature_k: float
initial_gauge_pressure_pa: float
@property
def initial_absolute_pressure_pa(self) -> float:
return self.initial_gauge_pressure_pa + AMESIM_REFERENCE_PRESSURE_PA
@dataclass(frozen=True)
class TestMqlPnl0002Spec:
alias: str
source_component: str
source_port: str
target_component: str
target_port: str
diameter_mm: float
length_m: float
relative_roughness: float
polytropic_constant: float
heat_transfer_coefficient: float
external_temperature_k: float
gas_type_index: int
mode: int
initial_center_temperature_k: float
initial_center_gauge_pressure_pa: float
@property
def initial_center_absolute_pressure_pa(self) -> float:
return self.initial_center_gauge_pressure_pa + AMESIM_REFERENCE_PRESSURE_PA
@dataclass(frozen=True)
class TestMqlPnl0003Spec:
alias: str
source_component: str
source_port: str
target_component: str
target_port: str
diameter_mm: float
length_m: float
relative_roughness: float
polytropic_constant: float
heat_transfer_coefficient: float
external_temperature_k: float
gas_type_index: int
mode: int
initial_temperature_1_k: float
initial_gauge_pressure_1_pa: float
initial_temperature_2_k: float
initial_gauge_pressure_2_pa: float
@property
def initial_absolute_pressure_1_pa(self) -> float:
return self.initial_gauge_pressure_1_pa + AMESIM_REFERENCE_PRESSURE_PA
@property
def initial_absolute_pressure_2_pa(self) -> float:
return self.initial_gauge_pressure_2_pa + AMESIM_REFERENCE_PRESSURE_PA
@dataclass(frozen=True)
class TestMqlPnl00rSpec:
alias: str
source_component: str
source_port: str
target_component: str
target_port: str
diameter_mm: float
length_m: float
relative_roughness: float
gas_type_index: int
def load_test_mql_pnl0001_specs(
archive_path: str | Path,
*,
cir_member: str = "test_mql_.cir",
) -> tuple[TestMqlPnl0001Spec, ...]:
"""Load resolved PNL0001 geometry and initial states from the AMESim source."""
with tarfile.open(archive_path) as archive:
cir_file = archive.extractfile(cir_member)
if cir_file is None:
raise ValueError(f"Missing AMESim circuit member: {cir_member}")
cir_text = cir_file.read().decode("latin1")
numeric_globals = {
name: value
for name, expression in GLOBAL_PARAMETERS.items()
if (value := resolve_numeric_expression(expression, {})) is not None
}
connections = {
str(connection["alias"]): connection
for connection in CONNECTION_SPECS
if connection["submodel"] == "PNL0001"
}
specs = []
for block in re.findall(r"<LINE>.*?</LINE>", cir_text, flags=re.DOTALL):
if _optional_text(block, "SUB_NAME") != "PNL0001":
continue
alias = _required_text(block, "ALIAS")
connection = connections.get(alias)
if connection is None:
raise ValueError(f"PNL0001 line {alias!r} is absent from CONNECTION_SPECS")
real_parameters = _parameter_expressions(block, "RPARAM")
integer_parameters = _parameter_expressions(block, "IPARAM")
state_values = _evar_values(block)
specs.append(
TestMqlPnl0001Spec(
alias=alias,
source_component=str(connection["source_component"]),
source_port=str(connection["source_port"]),
target_component=str(connection["target_component"]),
target_port=str(connection["target_port"]),
diameter_mm=_required_numeric(
alias, "diam", real_parameters, numeric_globals
),
length_m=_required_numeric(alias, "le", real_parameters, numeric_globals),
relative_roughness=_required_numeric(
alias, "rr", real_parameters, numeric_globals
),
polytropic_constant=_required_numeric(
alias, "k", real_parameters, numeric_globals
),
heat_transfer_coefficient=_required_numeric(
alias, "kth", real_parameters, numeric_globals
),
external_temperature_k=_required_numeric(
alias, "extemp", real_parameters, numeric_globals
),
gas_type_index=int(
_required_numeric(alias, "gi", integer_parameters, numeric_globals)
),
mode=int(
_required_numeric(alias, "mode", integer_parameters, numeric_globals)
),
initial_temperature_k=_required_numeric(
alias, "t2", state_values, numeric_globals
),
initial_gauge_pressure_pa=_required_numeric(
alias, "p2", state_values, numeric_globals
),
)
)
if set(connections) != {spec.alias for spec in specs}:
missing = sorted(set(connections) - {spec.alias for spec in specs})
raise ValueError(f"Missing PNL0001 parameter blocks: {missing}")
return tuple(specs)
def load_test_mql_pnl0002_specs(
archive_path: str | Path,
*,
cir_member: str = "test_mql_.cir",
) -> tuple[TestMqlPnl0002Spec, ...]:
"""Load resolved PNL0002 geometry and center compliance initial state."""
with tarfile.open(archive_path) as archive:
cir_file = archive.extractfile(cir_member)
if cir_file is None:
raise ValueError(f"Missing AMESim circuit member: {cir_member}")
cir_text = cir_file.read().decode("latin1")
numeric_globals = {
name: value
for name, expression in GLOBAL_PARAMETERS.items()
if (value := resolve_numeric_expression(expression, {})) is not None
}
connections = {
str(connection["alias"]): connection
for connection in CONNECTION_SPECS
if connection["submodel"] == "PNL0002"
}
specs = []
for block in re.findall(r"<LINE>.*?</LINE>", cir_text, flags=re.DOTALL):
if _optional_text(block, "SUB_NAME") != "PNL0002":
continue
alias = _required_text(block, "ALIAS")
connection = connections.get(alias)
if connection is None:
raise ValueError(f"PNL0002 line {alias!r} is absent from CONNECTION_SPECS")
real_parameters = _parameter_expressions(block, "RPARAM")
integer_parameters = _parameter_expressions(block, "IPARAM")
state_values = _ivar_values(block)
specs.append(
TestMqlPnl0002Spec(
alias=alias,
source_component=str(connection["source_component"]),
source_port=str(connection["source_port"]),
target_component=str(connection["target_component"]),
target_port=str(connection["target_port"]),
diameter_mm=_required_numeric(
alias, "diam", real_parameters, numeric_globals
),
length_m=_required_numeric(alias, "le", real_parameters, numeric_globals),
relative_roughness=_required_numeric(
alias, "rr", real_parameters, numeric_globals
),
polytropic_constant=_required_numeric(
alias, "k", real_parameters, numeric_globals
),
heat_transfer_coefficient=_required_numeric(
alias, "kth", real_parameters, numeric_globals
),
external_temperature_k=_required_numeric(
alias, "extemp", real_parameters, numeric_globals
),
gas_type_index=int(
_required_numeric(alias, "gi", integer_parameters, numeric_globals)
),
mode=int(
_required_numeric(alias, "mode", integer_parameters, numeric_globals)
),
initial_center_temperature_k=_required_numeric(
alias, "tctr", state_values, numeric_globals
),
initial_center_gauge_pressure_pa=_required_numeric(
alias, "pctr", state_values, numeric_globals
),
)
)
if set(connections) != {spec.alias for spec in specs}:
missing = sorted(set(connections) - {spec.alias for spec in specs})
raise ValueError(f"Missing PNL0002 parameter blocks: {missing}")
return tuple(specs)
def load_test_mql_pnl0003_specs(
archive_path: str | Path,
*,
cir_member: str = "test_mql_.cir",
) -> tuple[TestMqlPnl0003Spec, ...]:
"""Load resolved PNL0003 geometry and both compliance initial states."""
with tarfile.open(archive_path) as archive:
cir_file = archive.extractfile(cir_member)
if cir_file is None:
raise ValueError(f"Missing AMESim circuit member: {cir_member}")
cir_text = cir_file.read().decode("latin1")
numeric_globals = {
name: value
for name, expression in GLOBAL_PARAMETERS.items()
if (value := resolve_numeric_expression(expression, {})) is not None
}
connections = {
str(connection["alias"]): connection
for connection in CONNECTION_SPECS
if connection["submodel"] == "PNL0003"
}
specs = []
for block in re.findall(r"<LINE>.*?</LINE>", cir_text, flags=re.DOTALL):
if _optional_text(block, "SUB_NAME") != "PNL0003":
continue
alias = _required_text(block, "ALIAS")
connection = connections.get(alias)
if connection is None:
raise ValueError(f"PNL0003 line {alias!r} is absent from CONNECTION_SPECS")
real_parameters = _parameter_expressions(block, "RPARAM")
integer_parameters = _parameter_expressions(block, "IPARAM")
state_values = _evar_values(block)
specs.append(
TestMqlPnl0003Spec(
alias=alias,
source_component=str(connection["source_component"]),
source_port=str(connection["source_port"]),
target_component=str(connection["target_component"]),
target_port=str(connection["target_port"]),
diameter_mm=_required_numeric(
alias, "diam", real_parameters, numeric_globals
),
length_m=_required_numeric(alias, "le", real_parameters, numeric_globals),
relative_roughness=_required_numeric(
alias, "rr", real_parameters, numeric_globals
),
polytropic_constant=_required_numeric(
alias, "k", real_parameters, numeric_globals
),
heat_transfer_coefficient=_required_numeric(
alias, "kth", real_parameters, numeric_globals
),
external_temperature_k=_required_numeric(
alias, "extemp", real_parameters, numeric_globals
),
gas_type_index=int(
_required_numeric(alias, "gi", integer_parameters, numeric_globals)
),
mode=int(
_required_numeric(alias, "mode", integer_parameters, numeric_globals)
),
initial_temperature_1_k=_required_numeric(
alias, "t1", state_values, numeric_globals
),
initial_gauge_pressure_1_pa=_required_numeric(
alias, "p1", state_values, numeric_globals
),
initial_temperature_2_k=_required_numeric(
alias, "t2", state_values, numeric_globals
),
initial_gauge_pressure_2_pa=_required_numeric(
alias, "p2", state_values, numeric_globals
),
)
)
if set(connections) != {spec.alias for spec in specs}:
missing = sorted(set(connections) - {spec.alias for spec in specs})
raise ValueError(f"Missing PNL0003 parameter blocks: {missing}")
return tuple(specs)
def load_test_mql_pnl00r_specs(
archive_path: str | Path,
*,
cir_member: str = "test_mql_.cir",
) -> tuple[TestMqlPnl00rSpec, ...]:
"""Load resolved PNL00R geometry from the AMESim source."""
with tarfile.open(archive_path) as archive:
cir_file = archive.extractfile(cir_member)
if cir_file is None:
raise ValueError(f"Missing AMESim circuit member: {cir_member}")
cir_text = cir_file.read().decode("latin1")
numeric_globals = {
name: value
for name, expression in GLOBAL_PARAMETERS.items()
if (value := resolve_numeric_expression(expression, {})) is not None
}
connections = {
str(connection["alias"]): connection
for connection in CONNECTION_SPECS
if connection["submodel"] == "PNL00R"
}
specs = []
for block in re.findall(r"<LINE>.*?</LINE>", cir_text, flags=re.DOTALL):
if _optional_text(block, "SUB_NAME") != "PNL00R":
continue
alias = _required_text(block, "ALIAS")
connection = connections.get(alias)
if connection is None:
raise ValueError(f"PNL00R line {alias!r} is absent from CONNECTION_SPECS")
real_parameters = _parameter_expressions(block, "RPARAM")
integer_parameters = _parameter_expressions(block, "IPARAM")
specs.append(
TestMqlPnl00rSpec(
alias=alias,
source_component=str(connection["source_component"]),
source_port=str(connection["source_port"]),
target_component=str(connection["target_component"]),
target_port=str(connection["target_port"]),
diameter_mm=_required_numeric(
alias, "diam", real_parameters, numeric_globals
),
length_m=_required_numeric(alias, "le", real_parameters, numeric_globals),
relative_roughness=_required_numeric(
alias, "rr", real_parameters, numeric_globals
),
gas_type_index=int(
_required_numeric(alias, "gi", integer_parameters, numeric_globals)
),
)
)
if set(connections) != {spec.alias for spec in specs}:
missing = sorted(set(connections) - {spec.alias for spec in specs})
raise ValueError(f"Missing PNL00R parameter blocks: {missing}")
return tuple(specs)
def _parameter_expressions(block: str, tag_name: str) -> dict[str, str]:
parameters = {}
for parameter_block in re.findall(
rf"<{tag_name}>.*?</{tag_name}>",
block,
flags=re.DOTALL,
):
parameters[_required_text(parameter_block, "VARNAME")] = _required_text(
parameter_block,
"VALUE",
)
return parameters
def _ivar_values(block: str) -> dict[str, str]:
values = {}
for variable_block in re.findall(r"<IVAR>.*?</IVAR>", block, flags=re.DOTALL):
value = _optional_text(variable_block, "VALUE")
if value:
values[_required_text(variable_block, "VARNAME")] = value
return values
def _evar_values(block: str) -> dict[str, str]:
values = {}
for variable_block in re.findall(r"<EVAR>.*?</EVAR>", block, flags=re.DOTALL):
value = _optional_text(variable_block, "VALUE")
if value:
values[_required_text(variable_block, "VARNAME")] = value
return values
def _required_numeric(
alias: str,
name: str,
expressions: dict[str, str],
variables: dict[str, float],
) -> float:
if name not in expressions:
raise ValueError(f"Missing {name!r} on line {alias!r}")
value = resolve_numeric_expression(expressions[name], variables)
if value is None:
raise ValueError(
f"Cannot resolve {name!r}={expressions[name]!r} on line {alias!r}"
)
return value
def _required_text(block: str, tag_name: str) -> str:
value = _optional_text(block, tag_name)
if value is None:
raise ValueError(f"Missing AMESim circuit element: {tag_name}")
return value
def _optional_text(block: str, tag_name: str) -> str | None:
match = re.search(rf"<{tag_name}>(.*?)</{tag_name}>", block, flags=re.DOTALL)
return match.group(1).strip() if match is not None else None
+102
View File
@@ -0,0 +1,102 @@
from __future__ import annotations
import re
from collections import Counter
from dataclasses import dataclass
from app.simulation.reporting.amesim_results import AmesimResults
from app.simulation.reporting.test_mql_variables import (
TestMqlVariableCatalog,
build_test_mql_variable_catalog,
)
from app.simulation.examples.test_mql.system import CONNECTION_SPECS
TEST_MQL_PNEUMATIC_LINE_SUBMODELS = ("PNL0001", "PNL0002", "PNL0003", "PNL00R")
_LINE_PATTERN_RE = re.compile(r"\(([^()]+)\)\s*$")
@dataclass(frozen=True)
class TestMqlLineConnection:
index: int
alias: str
submodel: str
pattern: str
source_component: str
source_port: str
target_component: str
target_port: str
label: str
data_paths: tuple[str, ...]
signal_names: tuple[str, ...]
@property
def has_compliance(self) -> bool:
return "C" in self.pattern
@property
def has_resistance(self) -> bool:
return "R" in self.pattern
@dataclass(frozen=True)
class TestMqlLineAssembly:
lines: tuple[TestMqlLineConnection, ...]
@property
def line_count(self) -> int:
return len(self.lines)
def by_alias(self, alias: str) -> TestMqlLineConnection:
for line in self.lines:
if line.alias == alias:
return line
raise KeyError(alias)
def by_submodel(self, submodel: str) -> tuple[TestMqlLineConnection, ...]:
return tuple(line for line in self.lines if line.submodel == submodel)
def counts_by_submodel(self) -> dict[str, int]:
return dict(Counter(line.submodel for line in self.lines))
def aliases(self) -> tuple[str, ...]:
return tuple(line.alias for line in self.lines)
def build_test_mql_line_assembly(
amesim_results: AmesimResults,
variable_catalog: TestMqlVariableCatalog | None = None,
) -> TestMqlLineAssembly:
variable_catalog = variable_catalog or build_test_mql_variable_catalog(amesim_results)
lines = []
for spec in CONNECTION_SPECS:
submodel = str(spec["submodel"])
if submodel not in TEST_MQL_PNEUMATIC_LINE_SUBMODELS:
continue
data_paths = variable_catalog.data_paths_for_owner(str(spec["alias"]))
signal_names = tuple(path.rsplit("@", 1)[0] for path in data_paths)
lines.append(
TestMqlLineConnection(
index=int(spec["index"]),
alias=str(spec["alias"]),
submodel=submodel,
pattern=_line_pattern(str(spec["label"]), submodel),
source_component=str(spec["source_component"]),
source_port=str(spec["source_port"]),
target_component=str(spec["target_component"]),
target_port=str(spec["target_port"]),
label=str(spec["label"]),
data_paths=data_paths,
signal_names=signal_names,
)
)
return TestMqlLineAssembly(lines=tuple(lines))
def _line_pattern(label: str, submodel: str) -> str:
match = _LINE_PATTERN_RE.search(label)
if match is not None:
return match.group(1)
if submodel == "PNL00R":
return "R"
return submodel
@@ -0,0 +1,544 @@
from __future__ import annotations
from dataclasses import dataclass
from app.simulation.examples.test_mql.primitives.mechanical import (
AmesimElasticEndstop,
AmesimMassFrictionEndstops,
AmesimPistonGeometry,
circular_area,
mm_to_m,
)
from app.simulation.reporting.amesim_results import AmesimResults
from app.simulation.reporting.test_mql_variables import (
TestMqlVariableCatalog,
build_test_mql_variable_catalog,
)
from app.simulation.examples.test_mql.config import TestMqlConfig, TestMqlResolvedComponent
MM_TO_M = 1.0e-3
N_PER_MM_TO_N_PER_M = 1.0e3
N_PER_MM_PER_S_TO_N_PER_M_PER_S = 1.0e3
@dataclass(frozen=True)
class TestMqlPistonSpec:
alias: str
piston_diameter_m: float
rod_diameter_m: float
zero_displacement_m: float
piston_area_m2: float
rod_area_m2: float
annulus_area_m2: float
data_paths: tuple[str, ...]
def geometry(self) -> AmesimPistonGeometry:
return AmesimPistonGeometry(
piston_diameter_m=self.piston_diameter_m,
rod_diameter_m=self.rod_diameter_m,
zero_length_m=self.zero_displacement_m,
)
@dataclass(frozen=True)
class TestMqlMassEndstopSpec:
alias: str
mass_kg: float
xmin_m: float
xmax_m: float
min_stiffness_n_per_m: float
max_stiffness_n_per_m: float
min_damping_n_per_m_per_s: float
max_damping_n_per_m_per_s: float
min_penetration_m: float
max_penetration_m: float
stiction_force_n: float
coulomb_friction_n: float
viscous_friction_n_per_m_per_s: float
windage_n_per_m2_per_s2: float
stick_velocity_threshold_m_s: float
reset_velocity_threshold_m_s: float
rest_coeff: float
stribeck_constant_m_s: float
use_friction: bool
stop_type: int
initial_velocity_m_s: float
initial_displacement_m: float
data_paths: tuple[str, ...]
def endstop(self) -> AmesimMassFrictionEndstops:
return AmesimMassFrictionEndstops(
mass_kg=self.mass_kg,
lower_limit_m=self.xmin_m,
upper_limit_m=self.xmax_m,
lower_stiffness_n_per_m=self.min_stiffness_n_per_m,
upper_stiffness_n_per_m=self.max_stiffness_n_per_m,
lower_damping_n_per_m_per_s=self.min_damping_n_per_m_per_s,
upper_damping_n_per_m_per_s=self.max_damping_n_per_m_per_s,
viscous_friction_n_per_m_per_s=self.viscous_friction_n_per_m_per_s,
coulomb_friction_n=self.coulomb_friction_n,
stiction_force_n=self.stiction_force_n,
windage_n_per_m2_per_s2=self.windage_n_per_m2_per_s2,
)
@dataclass(frozen=True)
class TestMqlElasticEndstopSpec:
alias: str
gap_m: float
contact_stiffness_n_per_m: float
contact_damping_n_per_m_per_s: float
spring_diameter_m: float
wire_diameter_m: float
data_paths: tuple[str, ...]
def endstop(self) -> AmesimElasticEndstop:
return AmesimElasticEndstop(
contact_stiffness_n_per_m=self.contact_stiffness_n_per_m,
contact_damping_n_per_m_per_s=self.contact_damping_n_per_m_per_s,
gap0_m=self.gap_m,
)
@dataclass(frozen=True)
class TestMqlMechanicalNodeSpec:
alias: str
port_count: int
sum_mode: int
data_paths: tuple[str, ...]
@dataclass(frozen=True)
class TestMqlPiecewiseLinearSignalSpec:
alias: str
t_start_s: float
starts: tuple[float, ...]
ends: tuple[float, ...]
durations_s: tuple[float, ...]
stage_count: int
is_cyclic: bool
data_paths: tuple[str, ...]
def output_at(self, time_s: float) -> float:
if self.stage_count <= 0:
return 0.0
elapsed = max(time_s - self.t_start_s, 0.0)
active_durations = self.durations_s[: self.stage_count]
total_duration = sum(active_durations)
if self.is_cyclic and total_duration > 0.0:
elapsed = elapsed % total_duration
stage_start_time = 0.0
for index, duration in enumerate(active_durations):
stage_end_time = stage_start_time + duration
if elapsed < stage_end_time or index == self.stage_count - 1:
if duration <= 0.0:
return self.ends[index]
fraction = (elapsed - stage_start_time) / duration
return self.starts[index] + fraction * (self.ends[index] - self.starts[index])
stage_start_time = stage_end_time
return self.ends[self.stage_count - 1]
@dataclass(frozen=True)
class TestMqlForceConnectorSpec:
alias: str
signal_alias: str
target_mass_alias: str
data_paths: tuple[str, ...]
def force_at(
self,
time_s: float,
signals: dict[str, TestMqlPiecewiseLinearSignalSpec],
) -> float:
return signals[self.signal_alias].output_at(time_s)
@dataclass(frozen=True)
class TestMqlMechanicalAssembly:
pistons: dict[str, TestMqlPistonSpec]
masses: dict[str, TestMqlMassEndstopSpec]
elastic_endstops: dict[str, TestMqlElasticEndstopSpec]
mechanical_nodes: dict[str, TestMqlMechanicalNodeSpec]
piecewise_signals: dict[str, TestMqlPiecewiseLinearSignalSpec]
force_connectors: dict[str, TestMqlForceConnectorSpec]
zero_force_sources: tuple[str, ...]
@property
def component_count(self) -> int:
return (
len(self.pistons)
+ len(self.masses)
+ len(self.elastic_endstops)
+ len(self.mechanical_nodes)
+ len(self.piecewise_signals)
+ len(self.force_connectors)
+ len(self.zero_force_sources)
)
@property
def aliases(self) -> tuple[str, ...]:
return tuple(
[
*self.pistons,
*self.masses,
*self.elastic_endstops,
*self.mechanical_nodes,
*self.piecewise_signals,
*self.force_connectors,
*self.zero_force_sources,
]
)
@dataclass(frozen=True)
class TestMqlMechanicalMassState:
alias: str
velocity_m_s: float
displacement_m: float
def as_vector(self) -> list[float]:
return [self.velocity_m_s, self.displacement_m]
@dataclass(frozen=True)
class TestMqlMechanicalNodeKinematics:
alias: str
velocities_m_s: dict[int, float]
displacements_m: dict[int, float]
@dataclass(frozen=True)
class TestMqlPistonKinematics:
alias: str
port_2_velocity_m_s: float
port_2_displacement_m: float
port_3_velocity_m_s: float
port_3_displacement_m: float
@dataclass(frozen=True)
class TestMqlMechanicalMassSnapshot:
states: tuple[TestMqlMechanicalMassState, ...]
node_kinematics_by_alias: dict[str, TestMqlMechanicalNodeKinematics]
piston_kinematics_by_alias: dict[str, TestMqlPistonKinematics]
@property
def state_count(self) -> int:
return 2 * len(self.states)
class TestMqlMechanicalMassClosure:
def __init__(self, assembly: TestMqlMechanicalAssembly) -> None:
self.assembly = assembly
self.mass_aliases = tuple(assembly.masses)
def initial_state_vector(self) -> list[float]:
state: list[float] = []
for alias in self.mass_aliases:
spec = self.assembly.masses[alias]
state.extend([spec.initial_velocity_m_s, spec.initial_displacement_m])
return state
def snapshot(self, state_vector: list[float] | None = None) -> TestMqlMechanicalMassSnapshot:
values = self.initial_state_vector() if state_vector is None else list(state_vector)
if len(values) != 2 * len(self.mass_aliases):
raise ValueError("mechanical mass state vector requires two values per mass")
states = tuple(
TestMqlMechanicalMassState(
alias=alias,
velocity_m_s=values[2 * index],
displacement_m=values[2 * index + 1],
)
for index, alias in enumerate(self.mass_aliases)
)
node_kinematics = self._node_kinematics_by_alias(states)
return TestMqlMechanicalMassSnapshot(
states=states,
node_kinematics_by_alias=node_kinematics,
piston_kinematics_by_alias=self._piston_kinematics_by_alias(
states,
node_kinematics,
),
)
def _node_kinematics_by_alias(
self,
states: tuple[TestMqlMechanicalMassState, ...],
) -> dict[str, TestMqlMechanicalNodeKinematics]:
state_by_alias = {state.alias: state for state in states}
front = state_by_alias["mass_friction_endstops_18"]
rear = state_by_alias["mass_friction_endstops_19"]
return {
"dynamic_mechanical_node_alternative_2": TestMqlMechanicalNodeKinematics(
alias="dynamic_mechanical_node_alternative_2",
velocities_m_s={port: -front.velocity_m_s for port in range(1, 9)},
displacements_m={port: -front.displacement_m for port in range(1, 9)},
),
"dynamic_mechanical_node_alternative_3": TestMqlMechanicalNodeKinematics(
alias="dynamic_mechanical_node_alternative_3",
velocities_m_s={port: rear.velocity_m_s for port in range(1, 9)},
displacements_m={port: rear.displacement_m for port in range(1, 9)},
),
}
def _piston_kinematics_by_alias(
self,
states: tuple[TestMqlMechanicalMassState, ...],
node_kinematics_by_alias: dict[str, TestMqlMechanicalNodeKinematics],
) -> dict[str, TestMqlPistonKinematics]:
state_by_alias = {state.alias: state for state in states}
rear_node = node_kinematics_by_alias["dynamic_mechanical_node_alternative_3"]
piston_bindings = (
("pn_brp2_8", "mass_friction_endstops_10", 8),
("pn_brp2_9", "mass_friction_endstops_11", 7),
("pn_brp2_10", "mass_friction_endstops_12", 6),
("pn_brp2_11", "mass_friction_endstops_13", 5),
("pn_brp2_12", "mass_friction_endstops_14", 4),
("pn_brp2_13", "mass_friction_endstops_15", 3),
("pn_brp2_14", "mass_friction_endstops_16", 2),
("pn_brp2_15", "mass_friction_endstops_17", 1),
)
return {
piston_alias: TestMqlPistonKinematics(
alias=piston_alias,
port_2_velocity_m_s=state_by_alias[mass_alias].velocity_m_s,
port_2_displacement_m=state_by_alias[mass_alias].displacement_m,
port_3_velocity_m_s=rear_node.velocities_m_s[rear_node_port],
port_3_displacement_m=rear_node.displacements_m[rear_node_port],
)
for piston_alias, mass_alias, rear_node_port in piston_bindings
}
def rhs(
self,
state_vector: list[float],
*,
force_by_mass_alias: dict[str, float] | None = None,
constrained_mass_aliases: set[str] | None = None,
) -> list[float]:
snapshot = self.snapshot(state_vector)
force_by_mass_alias = force_by_mass_alias or {}
constrained_mass_aliases = constrained_mass_aliases or set()
derivatives: list[float] = []
for state in snapshot.states:
spec = self.assembly.masses[state.alias]
mass = spec.endstop()
applied_force = force_by_mass_alias.get(state.alias, 0.0)
acceleration, velocity = mass.derivatives(
velocity_m_s=state.velocity_m_s,
displacement_m=state.displacement_m,
port_1_force_n=applied_force,
)
if state.alias in constrained_mass_aliases and _limit_constraint_holds(
spec,
state,
applied_force,
):
acceleration = 0.0
velocity = 0.0
derivatives.extend([acceleration, velocity])
return derivatives
def build_test_mql_mechanical_assembly(
config: TestMqlConfig | None = None,
amesim_results: AmesimResults | None = None,
variable_catalog: TestMqlVariableCatalog | None = None,
) -> TestMqlMechanicalAssembly:
config = config or TestMqlConfig.from_amesim_specs()
if variable_catalog is None and amesim_results is not None:
variable_catalog = build_test_mql_variable_catalog(amesim_results)
pistons = {
component.alias: _build_piston(component, variable_catalog)
for component in config.components_by_submodel("PNRP17")
}
masses = {
component.alias: _build_mass(component, variable_catalog, amesim_results)
for component in config.components_by_submodel("MECMAS21")
}
elastic_endstops = {
component.alias: _build_elastic_endstop(component, variable_catalog)
for component in config.components_by_submodel("LSTP00A")
}
mechanical_nodes = {
component.alias: _build_mechanical_node(component, variable_catalog)
for component in config.components_by_submodel("LMECHN1")
}
piecewise_signals = {
component.alias: _build_piecewise_signal(component, variable_catalog)
for component in config.components_by_submodel("UD00")
}
force_connectors = {
component.alias: _build_force_connector(component, variable_catalog)
for component in config.components_by_submodel("FORC")
}
zero_force_sources = tuple(component.alias for component in config.components_by_submodel("F000"))
return TestMqlMechanicalAssembly(
pistons=pistons,
masses=masses,
elastic_endstops=elastic_endstops,
mechanical_nodes=mechanical_nodes,
piecewise_signals=piecewise_signals,
force_connectors=force_connectors,
zero_force_sources=zero_force_sources,
)
def _build_piston(
component: TestMqlResolvedComponent,
variable_catalog: TestMqlVariableCatalog | None,
) -> TestMqlPistonSpec:
geometry = AmesimPistonGeometry(
piston_diameter_m=mm_to_m(component.parameter_value("dp")),
rod_diameter_m=mm_to_m(component.parameter_value("dr")),
zero_length_m=mm_to_m(component.parameter_value("x0")),
)
return TestMqlPistonSpec(
alias=component.alias,
piston_diameter_m=geometry.piston_diameter_m,
rod_diameter_m=geometry.rod_diameter_m,
zero_displacement_m=geometry.zero_length_m,
piston_area_m2=geometry.piston_area_m2,
rod_area_m2=geometry.rod_area_m2,
annulus_area_m2=geometry.annulus_area_m2,
data_paths=_data_paths(variable_catalog, component.alias),
)
def _build_mass(
component: TestMqlResolvedComponent,
variable_catalog: TestMqlVariableCatalog | None,
amesim_results: AmesimResults | None,
) -> TestMqlMassEndstopSpec:
return TestMqlMassEndstopSpec(
alias=component.alias,
mass_kg=component.parameter_value("mass"),
xmin_m=component.parameter_value("xmin"),
xmax_m=component.parameter_value("xmax"),
min_stiffness_n_per_m=n_per_mm_to_n_per_m(component.parameter_value("Kbmin")),
max_stiffness_n_per_m=n_per_mm_to_n_per_m(component.parameter_value("Kbmax")),
min_damping_n_per_m_per_s=n_per_mm_per_s_to_n_per_m_per_s(component.parameter_value("Dbmin")),
max_damping_n_per_m_per_s=n_per_mm_per_s_to_n_per_m_per_s(component.parameter_value("Dbmax")),
min_penetration_m=mm_to_m(component.parameter_value("Pdmin")),
max_penetration_m=mm_to_m(component.parameter_value("Pdmax")),
stiction_force_n=component.parameter_value("fstick"),
coulomb_friction_n=component.parameter_value("fcoul"),
viscous_friction_n_per_m_per_s=component.parameter_value("rvisc"),
windage_n_per_m2_per_s2=component.parameter_value("wind"),
stick_velocity_threshold_m_s=component.parameter_value("dvel"),
reset_velocity_threshold_m_s=component.parameter_value("restdvel"),
rest_coeff=component.parameter_value("restcoeff"),
stribeck_constant_m_s=component.parameter_value("astrib"),
use_friction=bool(int(component.parameter_value("useFriction"))),
stop_type=int(component.parameter_value("stoptype")),
initial_velocity_m_s=_initial_value(amesim_results, f"v1@{component.alias}"),
initial_displacement_m=_initial_value(amesim_results, f"x1@{component.alias}"),
data_paths=_data_paths(variable_catalog, component.alias),
)
def _build_elastic_endstop(
component: TestMqlResolvedComponent,
variable_catalog: TestMqlVariableCatalog | None,
) -> TestMqlElasticEndstopSpec:
return TestMqlElasticEndstopSpec(
alias=component.alias,
gap_m=mm_to_m(component.parameter_value("gap0")),
contact_stiffness_n_per_m=component.parameter_value("kcont"),
contact_damping_n_per_m_per_s=component.parameter_value("rcont"),
spring_diameter_m=mm_to_m(component.parameter_value("sdiam")),
wire_diameter_m=mm_to_m(component.parameter_value("wdiam")),
data_paths=_data_paths(variable_catalog, component.alias),
)
def _build_mechanical_node(
component: TestMqlResolvedComponent,
variable_catalog: TestMqlVariableCatalog | None,
) -> TestMqlMechanicalNodeSpec:
return TestMqlMechanicalNodeSpec(
alias=component.alias,
port_count=int(component.parameter_value("v1")),
sum_mode=int(component.parameter_value("sum")),
data_paths=_data_paths(variable_catalog, component.alias),
)
def _limit_constraint_holds(
spec: TestMqlMassEndstopSpec,
state: TestMqlMechanicalMassState,
applied_force_n: float,
) -> bool:
if abs(state.velocity_m_s) > spec.stick_velocity_threshold_m_s:
return False
at_lower_limit = state.displacement_m <= spec.xmin_m + spec.min_penetration_m
at_upper_limit = state.displacement_m >= spec.xmax_m - spec.max_penetration_m
return (at_lower_limit and applied_force_n <= 0.0) or (
at_upper_limit and applied_force_n >= 0.0
)
def _build_piecewise_signal(
component: TestMqlResolvedComponent,
variable_catalog: TestMqlVariableCatalog | None,
) -> TestMqlPiecewiseLinearSignalSpec:
starts = tuple(component.parameter_value(f"start{index}") for index in range(1, 9))
ends = tuple(component.parameter_value(f"end{index}") for index in range(1, 9))
durations = tuple(component.parameter_value(f"t{index}") for index in range(1, 9))
return TestMqlPiecewiseLinearSignalSpec(
alias=component.alias,
t_start_s=component.parameter_value("tstart"),
starts=starts,
ends=ends,
durations_s=durations,
stage_count=int(component.parameter_value("nstages")),
is_cyclic=bool(int(component.parameter_value("iscyclic"))),
data_paths=_data_paths(variable_catalog, component.alias),
)
def _build_force_connector(
component: TestMqlResolvedComponent,
variable_catalog: TestMqlVariableCatalog | None,
) -> TestMqlForceConnectorSpec:
signal_alias_by_force_connector = {
"forcecon_1": "piecewiselinear",
"forcecon_2": "piecewiselinear_1",
}
target_mass_by_force_connector = {
"forcecon_1": "mass_friction_endstops_19",
"forcecon_2": "mass_friction_endstops_18",
}
return TestMqlForceConnectorSpec(
alias=component.alias,
signal_alias=signal_alias_by_force_connector[component.alias],
target_mass_alias=target_mass_by_force_connector[component.alias],
data_paths=_data_paths(variable_catalog, component.alias),
)
def n_per_mm_to_n_per_m(value: float) -> float:
return value * N_PER_MM_TO_N_PER_M
def n_per_mm_per_s_to_n_per_m_per_s(value: float) -> float:
return value * N_PER_MM_PER_S_TO_N_PER_M_PER_S
def _initial_value(amesim_results: AmesimResults | None, data_path: str) -> float:
if amesim_results is None:
return 0.0
return float(amesim_results.series(data_path)[0])
def _data_paths(
variable_catalog: TestMqlVariableCatalog | None,
alias: str,
) -> tuple[str, ...]:
if variable_catalog is None:
return ()
return variable_catalog.data_paths_for_owner(alias)
+215
View File
@@ -0,0 +1,215 @@
from __future__ import annotations
from dataclasses import dataclass
from app.simulation.examples.test_mql.system import COMPONENT_SPECS
@dataclass(frozen=True)
class TestMqlPneumaticNode3Balance:
temperature_k: float
pressure_pa: float
port_1_enthalpy_flow_w: float
port_1_mass_flow_g_s: float
port_1_volume_derivative_l_min: float
port_1_volume_cm3: float
port_2_enthalpy_flow_w: float
port_2_mass_flow_g_s: float
port_2_volume_derivative_l_min: float
port_2_volume_cm3: float
port_3_enthalpy_flow_w: float
port_3_mass_flow_g_s: float
port_3_volume_derivative_l_min: float
port_3_volume_cm3: float
@dataclass(frozen=True)
class TestMqlPneumaticNode3:
"""Exact algebraic contract of AMESim ``PN3NODE2``.
Pressure and temperature are fixed by port 2 and duplicated to ports 1 and
3. Flow and volume signals at port 2 are the sums of ports 1 and 3, matching
the ``EXPRESS2`` equations stored in ``test_mql_.cir``.
"""
alias: str
def balance(
self,
*,
port_2_temperature_k: float,
port_2_pressure_pa: float,
port_1_enthalpy_flow_w: float,
port_1_mass_flow_g_s: float,
port_3_enthalpy_flow_w: float,
port_3_mass_flow_g_s: float,
port_1_volume_derivative_l_min: float = 0.0,
port_1_volume_cm3: float = 0.0,
port_3_volume_derivative_l_min: float = 0.0,
port_3_volume_cm3: float = 0.0,
) -> TestMqlPneumaticNode3Balance:
if port_2_temperature_k <= 0.0:
raise ValueError("port_2_temperature_k must be positive")
if port_2_pressure_pa <= 0.0:
raise ValueError("port_2_pressure_pa must be positive")
return TestMqlPneumaticNode3Balance(
temperature_k=port_2_temperature_k,
pressure_pa=port_2_pressure_pa,
port_1_enthalpy_flow_w=port_1_enthalpy_flow_w,
port_1_mass_flow_g_s=port_1_mass_flow_g_s,
port_1_volume_derivative_l_min=port_1_volume_derivative_l_min,
port_1_volume_cm3=port_1_volume_cm3,
port_2_enthalpy_flow_w=(
port_1_enthalpy_flow_w + port_3_enthalpy_flow_w
),
port_2_mass_flow_g_s=port_1_mass_flow_g_s + port_3_mass_flow_g_s,
port_2_volume_derivative_l_min=(
port_1_volume_derivative_l_min + port_3_volume_derivative_l_min
),
port_2_volume_cm3=port_1_volume_cm3 + port_3_volume_cm3,
port_3_enthalpy_flow_w=port_3_enthalpy_flow_w,
port_3_mass_flow_g_s=port_3_mass_flow_g_s,
port_3_volume_derivative_l_min=port_3_volume_derivative_l_min,
port_3_volume_cm3=port_3_volume_cm3,
)
@dataclass(frozen=True)
class TestMqlPneumaticNode4Balance:
temperature_k: float
pressure_pa: float
port_1_enthalpy_flow_w: float
port_1_mass_flow_g_s: float
port_1_volume_derivative_l_min: float
port_1_volume_cm3: float
port_2_enthalpy_flow_w: float
port_2_mass_flow_g_s: float
port_2_volume_derivative_l_min: float
port_2_volume_cm3: float
port_3_enthalpy_flow_w: float
port_3_mass_flow_g_s: float
port_3_volume_derivative_l_min: float
port_3_volume_cm3: float
port_4_enthalpy_flow_w: float
port_4_mass_flow_g_s: float
port_4_volume_derivative_l_min: float
port_4_volume_cm3: float
@dataclass(frozen=True)
class TestMqlPneumaticNode4:
"""Exact algebraic contract of AMESim ``P4NODE2``.
Pressure and temperature are fixed by port 2 and duplicated to ports 1, 3,
and 4. Flow and volume signals at port 2 are the sums of ports 1, 3, and
4, matching the saved AMESim variables for ``pnnode4_*`` instances.
"""
alias: str
def balance(
self,
*,
port_2_temperature_k: float,
port_2_pressure_pa: float,
port_1_enthalpy_flow_w: float,
port_1_mass_flow_g_s: float,
port_3_enthalpy_flow_w: float,
port_3_mass_flow_g_s: float,
port_4_enthalpy_flow_w: float,
port_4_mass_flow_g_s: float,
port_1_volume_derivative_l_min: float = 0.0,
port_1_volume_cm3: float = 0.0,
port_3_volume_derivative_l_min: float = 0.0,
port_3_volume_cm3: float = 0.0,
port_4_volume_derivative_l_min: float = 0.0,
port_4_volume_cm3: float = 0.0,
) -> TestMqlPneumaticNode4Balance:
if port_2_temperature_k <= 0.0:
raise ValueError("port_2_temperature_k must be positive")
if port_2_pressure_pa <= 0.0:
raise ValueError("port_2_pressure_pa must be positive")
return TestMqlPneumaticNode4Balance(
temperature_k=port_2_temperature_k,
pressure_pa=port_2_pressure_pa,
port_1_enthalpy_flow_w=port_1_enthalpy_flow_w,
port_1_mass_flow_g_s=port_1_mass_flow_g_s,
port_1_volume_derivative_l_min=port_1_volume_derivative_l_min,
port_1_volume_cm3=port_1_volume_cm3,
port_2_enthalpy_flow_w=(
port_1_enthalpy_flow_w
+ port_3_enthalpy_flow_w
+ port_4_enthalpy_flow_w
),
port_2_mass_flow_g_s=(
port_1_mass_flow_g_s
+ port_3_mass_flow_g_s
+ port_4_mass_flow_g_s
),
port_2_volume_derivative_l_min=(
port_1_volume_derivative_l_min
+ port_3_volume_derivative_l_min
+ port_4_volume_derivative_l_min
),
port_2_volume_cm3=(
port_1_volume_cm3 + port_3_volume_cm3 + port_4_volume_cm3
),
port_3_enthalpy_flow_w=port_3_enthalpy_flow_w,
port_3_mass_flow_g_s=port_3_mass_flow_g_s,
port_3_volume_derivative_l_min=port_3_volume_derivative_l_min,
port_3_volume_cm3=port_3_volume_cm3,
port_4_enthalpy_flow_w=port_4_enthalpy_flow_w,
port_4_mass_flow_g_s=port_4_mass_flow_g_s,
port_4_volume_derivative_l_min=port_4_volume_derivative_l_min,
port_4_volume_cm3=port_4_volume_cm3,
)
@dataclass(frozen=True)
class TestMqlP4NodePortConnection:
line_alias: str
local_node_alias: str
local_port: str
remote_node_alias: str
remote_port: str
@dataclass(frozen=True)
class TestMqlP4NodePrimaryConnection:
line_alias: str
node_alias: str
node_port: str
chamber_alias: str
chamber_port: str
@dataclass(frozen=True)
class TestMqlP4NodeOrificeConnection:
orifice_alias: str
node_alias: str
node_port: str
direct_line_alias: str
@dataclass(frozen=True)
class TestMqlP4NodeNeighborhood:
node_alias: str
primary: TestMqlP4NodePrimaryConnection
port_1: TestMqlP4NodePortConnection
port_3: TestMqlP4NodePortConnection
port_4: TestMqlP4NodeOrificeConnection
def build_test_mql_node3_assembly() -> dict[str, TestMqlPneumaticNode3]:
return {
str(spec["alias"]): TestMqlPneumaticNode3(alias=str(spec["alias"]))
for spec in COMPONENT_SPECS
if spec["submodel"] == "PN3NODE2"
}
def build_test_mql_node4_assembly() -> dict[str, TestMqlPneumaticNode4]:
return {
str(spec["alias"]): TestMqlPneumaticNode4(alias=str(spec["alias"]))
for spec in COMPONENT_SPECS
if spec["submodel"] == "P4NODE2"
}
@@ -0,0 +1,273 @@
from __future__ import annotations
from dataclasses import dataclass
from app.simulation.examples.test_mql.primitives.pneumatic import (
HELIUM_PNEUMATIC_GAS,
AmesimPneumaticGas,
AmesimPneumaticOrifice,
AmesimPneumaticVolume,
AmesimVariablePneumaticVolume,
)
from app.simulation.examples.test_mql.config import TestMqlConfig, TestMqlResolvedComponent
AMESIM_REFERENCE_PRESSURE_PA = 101_300.0
BAR_TO_PA = 1.0e5
DEFAULT_TEST_MQL_TEMPERATURE_K = 293.15
DEFAULT_VARIABLE_CHAMBER_PRESSURE_BAR = 1.0
# Matched to PNVO001 event-window mass flow near the 0.04 s opening event.
TEST_MQL_PNVO001_FLOW_COEFFICIENT_MULTIPLIER = 0.99805
@dataclass(frozen=True)
class TestMqlStepSignalSpec:
alias: str
initial_output: float
final_output: float
step_time_s: float
transition_duration_s: float
transition_type: int
def output_at(self, time_s: float) -> float:
if self.transition_type != 1:
raise ValueError(
f"unsupported STEP0 transition type {self.transition_type} on {self.alias}"
)
return self.initial_output if time_s < self.step_time_s else self.final_output
@dataclass(frozen=True)
class TestMqlVariableOrificeControl:
orifice_alias: str
step: TestMqlStepSignalSpec
def opening_at(self, time_s: float) -> float:
return self.step.output_at(time_s)
@dataclass(frozen=True)
class TestMqlPneumaticAssembly:
fixed_chambers: dict[str, AmesimPneumaticVolume]
variable_chambers: dict[str, AmesimVariablePneumaticVolume]
fixed_orifices: dict[str, AmesimPneumaticOrifice]
variable_orifices: dict[str, AmesimPneumaticOrifice]
variable_orifice_controls: dict[str, TestMqlVariableOrificeControl]
fixed_initial_absolute_pressure_pa: float
variable_initial_absolute_pressure_pa: float
@property
def initial_pressure_pa(self) -> float:
return pressure_to_amesim_gauge_pa(self.fixed_initial_absolute_pressure_pa)
@property
def fixed_initial_gauge_pressure_pa(self) -> float:
return pressure_to_amesim_gauge_pa(self.fixed_initial_absolute_pressure_pa)
@property
def variable_initial_gauge_pressure_pa(self) -> float:
return pressure_to_amesim_gauge_pa(self.variable_initial_absolute_pressure_pa)
@property
def chamber_count(self) -> int:
return len(self.fixed_chambers) + len(self.variable_chambers)
@property
def orifice_count(self) -> int:
return len(self.fixed_orifices) + len(self.variable_orifices)
@property
def component_count(self) -> int:
return self.chamber_count + self.orifice_count
@property
def variable_orifice_control_count(self) -> int:
return len(self.variable_orifice_controls)
def set_variable_orifice_openings(self, time_s: float) -> None:
for alias, control in self.variable_orifice_controls.items():
self.variable_orifices[alias].opening = control.opening_at(time_s)
@property
def aliases(self) -> tuple[str, ...]:
return tuple(
[
*self.fixed_chambers,
*self.variable_chambers,
*self.fixed_orifices,
*self.variable_orifices,
]
)
def build_test_mql_pneumatic_assembly(
config: TestMqlConfig | None = None,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
) -> TestMqlPneumaticAssembly:
config = config or TestMqlConfig.from_amesim_specs()
fixed_initial_absolute_pressure_pa = absolute_pressure_from_amesim_bar_parameter(
config.global_parameters["P0"]
)
variable_initial_absolute_pressure_pa = absolute_pressure_from_amesim_bar_parameter(
DEFAULT_VARIABLE_CHAMBER_PRESSURE_BAR
)
fixed_chambers = {
component.alias: _build_chamber(
component,
volume_parameter="cvol",
gas=gas,
initial_pressure_pa=fixed_initial_absolute_pressure_pa,
)
for component in config.components_by_submodel("PNCH023")
}
variable_chambers = {
component.alias: _build_chamber(
component,
volume_parameter="cvol0",
gas=gas,
initial_pressure_pa=variable_initial_absolute_pressure_pa,
)
for component in config.components_by_submodel("PNCH012")
}
fixed_orifices = {
component.alias: _build_orifice(
component,
area_parameter="area",
gas=gas,
opening=1.0,
)
for component in config.components_by_submodel("PNOR001")
}
variable_orifice_controls = _build_variable_orifice_controls(config)
variable_orifices = {
component.alias: _build_orifice(
component,
area_parameter="area0",
gas=gas,
opening=variable_orifice_controls[component.alias].opening_at(0.0),
)
for component in config.components_by_submodel("PNVO001")
}
return TestMqlPneumaticAssembly(
fixed_chambers=fixed_chambers,
variable_chambers=variable_chambers,
fixed_orifices=fixed_orifices,
variable_orifices=variable_orifices,
variable_orifice_controls=variable_orifice_controls,
fixed_initial_absolute_pressure_pa=fixed_initial_absolute_pressure_pa,
variable_initial_absolute_pressure_pa=variable_initial_absolute_pressure_pa,
)
def _build_variable_orifice_controls(
config: TestMqlConfig,
) -> dict[str, TestMqlVariableOrificeControl]:
from app.simulation.examples.test_mql.system import CONNECTION_SPECS
components_by_alias = {component.alias: component for component in config.components}
variable_orifice_aliases = {
component.alias for component in config.components_by_submodel("PNVO001")
}
controls: dict[str, TestMqlVariableOrificeControl] = {}
for connection in CONNECTION_SPECS:
if connection["submodel"] != "DIRECT":
continue
source_alias = str(connection["source_component"])
target_alias = str(connection["target_component"])
if target_alias in variable_orifice_aliases:
orifice_alias = target_alias
step_alias = source_alias
elif source_alias in variable_orifice_aliases:
orifice_alias = source_alias
step_alias = target_alias
else:
continue
step_component = components_by_alias.get(step_alias)
if step_component is None or step_component.submodel != "STEP0":
continue
controls[orifice_alias] = TestMqlVariableOrificeControl(
orifice_alias=orifice_alias,
step=TestMqlStepSignalSpec(
alias=step_alias,
initial_output=step_component.parameter_value("out0"),
final_output=step_component.parameter_value("out1"),
step_time_s=step_component.parameter_value("t0"),
transition_duration_s=step_component.parameter_value("td"),
transition_type=int(step_component.parameter_value("transitionType")),
),
)
missing = variable_orifice_aliases - controls.keys()
if missing:
raise ValueError(
"missing STEP0 controls for PNVO001 components: "
+ ", ".join(sorted(missing))
)
return controls
def absolute_pressure_from_amesim_bar_parameter(pressure_bar: float) -> float:
return pressure_bar * BAR_TO_PA
def pressure_to_amesim_gauge_pa(absolute_pressure_pa: float) -> float:
return absolute_pressure_pa - AMESIM_REFERENCE_PRESSURE_PA
def pressure_from_amesim_bar_parameter(pressure_bar: float) -> float:
return pressure_to_amesim_gauge_pa(absolute_pressure_from_amesim_bar_parameter(pressure_bar))
def _build_chamber(
component: TestMqlResolvedComponent,
*,
volume_parameter: str,
gas: AmesimPneumaticGas,
initial_pressure_pa: float,
) -> AmesimPneumaticVolume:
if volume_parameter == "cvol0":
return AmesimVariablePneumaticVolume.from_liters(
name=component.alias,
dead_volume_liters=component.parameter_value(volume_parameter),
gas=gas,
p0=initial_pressure_pa,
T0=_component_temperature(component),
heat_transfer_coefficient=component.parameter_value("kth"),
heat_transfer_area=component.parameter_value("sth"),
external_temperature_k=_component_temperature(component),
)
return AmesimPneumaticVolume.from_liters(
name=component.alias,
volume_liters=component.parameter_value(volume_parameter),
gas=gas,
p0=initial_pressure_pa,
T0=_component_temperature(component),
heat_transfer_coefficient=component.parameter_value("kth"),
heat_transfer_area=component.parameter_value("sth"),
external_temperature_k=_component_temperature(component),
)
def _build_orifice(
component: TestMqlResolvedComponent,
*,
area_parameter: str,
gas: AmesimPneumaticGas,
opening: float,
) -> AmesimPneumaticOrifice:
flow_coefficient = component.parameter_value("cq")
if component.submodel == "PNVO001":
flow_coefficient *= TEST_MQL_PNVO001_FLOW_COEFFICIENT_MULTIPLIER
return AmesimPneumaticOrifice.from_mm2(
name=component.alias,
area_mm2=component.parameter_value(area_parameter),
flow_coefficient=flow_coefficient,
gas=gas,
opening=opening,
)
def _component_temperature(component: TestMqlResolvedComponent) -> float:
parameter = component.parameters.get("extemp")
if parameter is None or parameter.value is None:
return DEFAULT_TEST_MQL_TEMPERATURE_K
return parameter.value
@@ -0,0 +1,194 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from app.simulation.examples.test_mql.primitives.pneumatic import (
HELIUM_PNEUMATIC_GAS,
AmesimPneumaticGas,
)
from app.simulation.examples.test_mql.primitives.pneumatic_lines import (
AmesimPnl0001Pipe,
AmesimPnl0002Pipe,
AmesimPnl0003Pipe,
AmesimPnl00rPipe,
)
from app.simulation.examples.test_mql.line_parameters import (
TestMqlPnl0001Spec,
TestMqlPnl0002Spec,
TestMqlPnl0003Spec,
TestMqlPnl00rSpec,
load_test_mql_pnl0001_specs,
load_test_mql_pnl0002_specs,
load_test_mql_pnl0003_specs,
load_test_mql_pnl00r_specs,
)
TEST_MQL_PNL0001_D20_L1_LINEAR_CONDUCTANCE = 5.5636e-6
@dataclass(frozen=True)
class TestMqlPnl0001Assembly:
specs: tuple[TestMqlPnl0001Spec, ...]
lines: dict[str, AmesimPnl0001Pipe]
def spec(self, alias: str) -> TestMqlPnl0001Spec:
for spec in self.specs:
if spec.alias == alias:
return spec
raise KeyError(alias)
@dataclass(frozen=True)
class TestMqlPnl0002Assembly:
specs: tuple[TestMqlPnl0002Spec, ...]
lines: dict[str, AmesimPnl0002Pipe]
def spec(self, alias: str) -> TestMqlPnl0002Spec:
for spec in self.specs:
if spec.alias == alias:
return spec
raise KeyError(alias)
@dataclass(frozen=True)
class TestMqlPnl0003Assembly:
specs: tuple[TestMqlPnl0003Spec, ...]
lines: dict[str, AmesimPnl0003Pipe]
def spec(self, alias: str) -> TestMqlPnl0003Spec:
for spec in self.specs:
if spec.alias == alias:
return spec
raise KeyError(alias)
@dataclass(frozen=True)
class TestMqlPnl00rAssembly:
specs: tuple[TestMqlPnl00rSpec, ...]
lines: dict[str, AmesimPnl00rPipe]
def spec(self, alias: str) -> TestMqlPnl00rSpec:
for spec in self.specs:
if spec.alias == alias:
return spec
raise KeyError(alias)
def build_test_mql_pnl0001_assembly(
archive_path: str | Path,
*,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
) -> TestMqlPnl0001Assembly:
specs = load_test_mql_pnl0001_specs(archive_path)
lines = {
spec.alias: AmesimPnl0001Pipe(
name=spec.alias,
diameter_mm=spec.diameter_mm,
length_m=spec.length_m,
relative_roughness=spec.relative_roughness,
polytropic_constant=spec.polytropic_constant,
heat_transfer_coefficient=spec.heat_transfer_coefficient,
external_temperature_k=spec.external_temperature_k,
calibrated_linear_conductance=(
_test_mql_pnl0001_calibrated_linear_conductance(spec)
),
gas=gas,
p0=spec.initial_absolute_pressure_pa,
T0=spec.initial_temperature_k,
)
for spec in specs
}
return TestMqlPnl0001Assembly(specs=specs, lines=lines)
def _test_mql_pnl0001_calibrated_linear_conductance(
spec: TestMqlPnl0001Spec,
) -> float | None:
if spec.target_component.startswith("pn_c1_") and _matches_geometry(
spec, diameter_mm=20.0, length_m=1.0
):
return TEST_MQL_PNL0001_D20_L1_LINEAR_CONDUCTANCE
return None
def _matches_geometry(
spec: TestMqlPnl0001Spec,
*,
diameter_mm: float,
length_m: float,
) -> bool:
return (
abs(spec.diameter_mm - diameter_mm) < 1.0e-12
and abs(spec.length_m - length_m) < 1.0e-12
)
def build_test_mql_pnl0002_assembly(
archive_path: str | Path,
*,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
) -> TestMqlPnl0002Assembly:
specs = load_test_mql_pnl0002_specs(archive_path)
lines = {
spec.alias: AmesimPnl0002Pipe(
name=spec.alias,
diameter_mm=spec.diameter_mm,
length_m=spec.length_m,
relative_roughness=spec.relative_roughness,
polytropic_constant=spec.polytropic_constant,
heat_transfer_coefficient=spec.heat_transfer_coefficient,
external_temperature_k=spec.external_temperature_k,
gas=gas,
pctr_0=spec.initial_center_absolute_pressure_pa,
Tctr_0=spec.initial_center_temperature_k,
)
for spec in specs
}
return TestMqlPnl0002Assembly(specs=specs, lines=lines)
def build_test_mql_pnl0003_assembly(
archive_path: str | Path,
*,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
) -> TestMqlPnl0003Assembly:
specs = load_test_mql_pnl0003_specs(archive_path)
lines = {
spec.alias: AmesimPnl0003Pipe(
name=spec.alias,
diameter_mm=spec.diameter_mm,
length_m=spec.length_m,
relative_roughness=spec.relative_roughness,
polytropic_constant=spec.polytropic_constant,
heat_transfer_coefficient=spec.heat_transfer_coefficient,
external_temperature_k=spec.external_temperature_k,
gas=gas,
p1_0=spec.initial_absolute_pressure_1_pa,
T1_0=spec.initial_temperature_1_k,
p2_0=spec.initial_absolute_pressure_2_pa,
T2_0=spec.initial_temperature_2_k,
)
for spec in specs
}
return TestMqlPnl0003Assembly(specs=specs, lines=lines)
def build_test_mql_pnl00r_assembly(
archive_path: str | Path,
*,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
) -> TestMqlPnl00rAssembly:
specs = load_test_mql_pnl00r_specs(archive_path)
lines = {
spec.alias: AmesimPnl00rPipe(
name=spec.alias,
diameter_mm=spec.diameter_mm,
length_m=spec.length_m,
relative_roughness=spec.relative_roughness,
gas=gas,
)
for spec in specs
}
return TestMqlPnl00rAssembly(specs=specs, lines=lines)
@@ -0,0 +1,128 @@
from __future__ import annotations
from app.simulation.examples.test_mql.closure import TestMqlPneumaticChamberSegmentSpec
from app.simulation.examples.test_mql.topology import TestMqlCirTopology
def discover_fixed_chamber_segments(
topology: TestMqlCirTopology,
component_specs: list[dict[str, object]],
connection_specs: list[dict[str, object]],
) -> tuple[TestMqlPneumaticChamberSegmentSpec, ...]:
submodel_by_alias = {
str(component["alias"]): str(component["submodel"])
for component in component_specs
}
segments = []
for component in component_specs:
volume_alias = str(component["alias"])
if component["submodel"] != "PNCH023":
continue
orifice_contacts = []
for contact in topology.contacts_for(volume_alias):
other_alias, other_port = contact.other_endpoint(volume_alias)
if submodel_by_alias.get(other_alias) == "PNOR001":
orifice_contacts.append(
(
other_alias,
other_port,
contact.port_for(volume_alias),
)
)
if len(orifice_contacts) != 2:
raise ValueError(
f"{volume_alias} must contact exactly two PNOR001 orifices; "
f"found {len(orifice_contacts)}"
)
sides = [
_resolve_orifice_boundary(
orifice_alias=orifice_alias,
orifice_volume_port=orifice_volume_port,
volume_port=volume_port,
connection_specs=connection_specs,
submodel_by_alias=submodel_by_alias,
)
for orifice_alias, orifice_volume_port, volume_port in orifice_contacts
]
inlet_sides = [side for side in sides if side["role"] == "inlet"]
outlet_sides = [side for side in sides if side["role"] == "outlet"]
if len(inlet_sides) != 1 or len(outlet_sides) != 1:
raise ValueError(
f"{volume_alias} requires one inlet and one outlet topology side"
)
inlet = inlet_sides[0]
outlet = outlet_sides[0]
segments.append(
TestMqlPneumaticChamberSegmentSpec(
name=f"{volume_alias}_segment",
inlet_node_alias=inlet["node_alias"],
inlet_line_alias=inlet["line_alias"],
inlet_orifice_alias=inlet["orifice_alias"],
inlet_orifice_boundary_port=inlet["orifice_boundary_port"],
inlet_orifice_volume_port=inlet["orifice_volume_port"],
volume_alias=volume_alias,
volume_inlet_port=inlet["volume_port"],
volume_outlet_port=outlet["volume_port"],
outlet_orifice_alias=outlet["orifice_alias"],
outlet_orifice_volume_port=outlet["orifice_volume_port"],
outlet_orifice_boundary_port=outlet["orifice_boundary_port"],
outlet_line_alias=outlet["line_alias"],
outlet_node_alias=outlet["node_alias"],
)
)
return tuple(segments)
def _resolve_orifice_boundary(
*,
orifice_alias: str,
orifice_volume_port: str,
volume_port: str,
connection_specs: list[dict[str, object]],
submodel_by_alias: dict[str, str],
) -> dict[str, str]:
boundary_connections = []
for connection in connection_specs:
if (
connection["source_component"] == orifice_alias
and connection["source_port"] != orifice_volume_port
) or (
connection["target_component"] == orifice_alias
and connection["target_port"] != orifice_volume_port
):
boundary_connections.append(connection)
if len(boundary_connections) != 1:
raise ValueError(
f"{orifice_alias} must have exactly one non-volume boundary connection; "
f"found {len(boundary_connections)}"
)
connection = boundary_connections[0]
if connection["submodel"] != "PNL0001":
raise ValueError(
f"{orifice_alias} boundary must use PNL0001, got {connection['submodel']}"
)
if connection["target_component"] == orifice_alias:
role = "inlet"
node_alias = str(connection["source_component"])
orifice_boundary_port = str(connection["target_port"])
else:
role = "outlet"
node_alias = str(connection["target_component"])
orifice_boundary_port = str(connection["source_port"])
if submodel_by_alias.get(node_alias) != "PN3NODE2":
raise ValueError(
f"{orifice_alias} PNL0001 boundary must terminate at PN3NODE2, "
f"got {node_alias}"
)
return {
"role": role,
"node_alias": node_alias,
"line_alias": str(connection["alias"]),
"orifice_alias": orifice_alias,
"orifice_boundary_port": orifice_boundary_port,
"orifice_volume_port": orifice_volume_port,
"volume_port": volume_port,
}
@@ -0,0 +1 @@
"""Calibrated component primitives used only by the ``test_mql`` example."""
@@ -0,0 +1,168 @@
from __future__ import annotations
from dataclasses import dataclass
from math import pi
MM_TO_M = 1.0e-3
M_TO_MM = 1.0e3
M3_TO_CM3 = 1.0e6
M3_PER_S_TO_L_PER_MIN = 60_000.0
def circular_area(diameter_m: float) -> float:
if diameter_m < 0.0:
raise ValueError("diameter_m must be non-negative.")
return pi * diameter_m * diameter_m / 4.0
def mm_to_m(value: float) -> float:
return value * MM_TO_M
def m_to_mm(value: float) -> float:
return value * M_TO_MM
@dataclass(frozen=True)
class AmesimPistonGeometry:
"""Geometry relations used by AMESim PNRP17 pneumatic piston variables."""
piston_diameter_m: float
rod_diameter_m: float = 0.0
zero_length_m: float = 0.0
@property
def piston_area_m2(self) -> float:
return circular_area(self.piston_diameter_m)
@property
def rod_area_m2(self) -> float:
return circular_area(self.rod_diameter_m)
@property
def annulus_area_m2(self) -> float:
return self.piston_area_m2 - self.rod_area_m2
def chamber_length_m(self, port4_displacement_m: float, port5_displacement_m: float) -> float:
return self.zero_length_m + port5_displacement_m - port4_displacement_m
def chamber_length_mm(self, port4_displacement_m: float, port5_displacement_m: float) -> float:
return m_to_mm(self.chamber_length_m(port4_displacement_m, port5_displacement_m))
@property
def chamber_area_m2(self) -> float:
return self.annulus_area_m2
def chamber_volume_m3(self, port4_displacement_m: float, port5_displacement_m: float) -> float:
return self.chamber_area_m2 * self.chamber_length_m(
port4_displacement_m,
port5_displacement_m,
)
def chamber_volume_cm3(self, port4_displacement_m: float, port5_displacement_m: float) -> float:
return self.chamber_volume_m3(port4_displacement_m, port5_displacement_m) * M3_TO_CM3
def chamber_volume_rate_m3_s(self, port4_velocity_m_s: float, port5_velocity_m_s: float) -> float:
return self.chamber_area_m2 * (port5_velocity_m_s - port4_velocity_m_s)
def chamber_volume_rate_l_min(self, port4_velocity_m_s: float, port5_velocity_m_s: float) -> float:
return self.chamber_volume_rate_m3_s(
port4_velocity_m_s,
port5_velocity_m_s,
) * M3_PER_S_TO_L_PER_MIN
@dataclass(frozen=True)
class AmesimElasticEndstop:
"""Contact force part of AMESim LSTP00A elastic endstop."""
contact_stiffness_n_per_m: float
contact_damping_n_per_m_per_s: float = 0.0
gap0_m: float = 0.0
def penetration_m_from_gap_mm(self, gap_mm: float) -> float:
return max(-(mm_to_m(gap_mm) - self.gap0_m), 0.0)
def static_contact_force(self, gap_mm: float) -> float:
return self.contact_stiffness_n_per_m * self.penetration_m_from_gap_mm(gap_mm)
def contact_force(self, gap_mm: float, penetration_velocity_m_s: float = 0.0) -> float:
if self.penetration_m_from_gap_mm(gap_mm) <= 0.0:
return 0.0
damping_force = self.contact_damping_n_per_m_per_s * penetration_velocity_m_s
return max(self.static_contact_force(gap_mm) + damping_force, 0.0)
@dataclass(frozen=True)
class AmesimMassFrictionEndstops:
"""Parameter and observable helpers for AMESim MECMAS21 translation masses."""
mass_kg: float
lower_limit_m: float
upper_limit_m: float
lower_stiffness_n_per_m: float
upper_stiffness_n_per_m: float
lower_damping_n_per_m_per_s: float = 0.0
upper_damping_n_per_m_per_s: float = 0.0
viscous_friction_n_per_m_per_s: float = 0.0
coulomb_friction_n: float = 0.0
stiction_force_n: float = 0.0
windage_n_per_m2_per_s2: float = 0.0
def lower_penetration_m(self, displacement_m: float) -> float:
return max(self.lower_limit_m - displacement_m, 0.0)
def upper_penetration_m(self, displacement_m: float) -> float:
return max(displacement_m - self.upper_limit_m, 0.0)
def lower_static_force_magnitude(self, displacement_m: float) -> float:
return self.lower_stiffness_n_per_m * self.lower_penetration_m(displacement_m)
def upper_static_force_magnitude(self, displacement_m: float) -> float:
return self.upper_stiffness_n_per_m * self.upper_penetration_m(displacement_m)
def viscous_friction_force(self, velocity_m_s: float) -> float:
return -self.viscous_friction_n_per_m_per_s * velocity_m_s
def windage_force(self, velocity_m_s: float) -> float:
return -self.windage_n_per_m2_per_s2 * velocity_m_s * abs(velocity_m_s)
def dry_friction_force(self, velocity_m_s: float) -> float:
if velocity_m_s > 0.0:
return -self.coulomb_friction_n
if velocity_m_s < 0.0:
return self.coulomb_friction_n
return 0.0
def limit_contact_force(self, displacement_m: float, velocity_m_s: float) -> float:
lower_force = self.lower_static_force_magnitude(displacement_m)
if lower_force > 0.0:
lower_force += max(-self.lower_damping_n_per_m_per_s * velocity_m_s, 0.0)
upper_force = self.upper_static_force_magnitude(displacement_m)
if upper_force > 0.0:
upper_force += max(self.upper_damping_n_per_m_per_s * velocity_m_s, 0.0)
return lower_force - upper_force
def derivatives(
self,
*,
velocity_m_s: float,
displacement_m: float,
port_1_force_n: float = 0.0,
port_2_force_n: float = 0.0,
external_force_n: float = 0.0,
) -> tuple[float, float]:
total_force = (
port_1_force_n
+ port_2_force_n
+ external_force_n
+ self.viscous_friction_force(velocity_m_s)
+ self.windage_force(velocity_m_s)
+ self.dry_friction_force(velocity_m_s)
+ self.limit_contact_force(displacement_m, velocity_m_s)
)
return total_force / self.mass_kg, velocity_m_s
@@ -0,0 +1,437 @@
from __future__ import annotations
from dataclasses import dataclass
from math import pi, sqrt
from app.simulation.core.base import AlgebraicComponent, DynamicComponent
from app.simulation.core.medium import ThermodynamicProperties
from app.simulation.core.peng_robinson import HELIUM_PR, PengRobinsonFluid
from app.simulation.core.ports import PortState
from app.simulation.core.state import VolumeState
@dataclass(frozen=True)
class AmesimPneumaticGas:
"""Caloric constants plus Peng-Robinson EOS for AMESim pneumatic components."""
fluid: PengRobinsonFluid = HELIUM_PR
cp: float = 5193.0
cv: float = 3116.0
@property
def gamma(self) -> float:
return self.cp / self.cv
@property
def R_gas(self) -> float:
return self.fluid.specific_gas_constant
def density(self, pressure: float, temperature: float) -> float:
return self.fluid.density(pressure, temperature)
def pressure(self, density: float, temperature: float) -> float:
return self.fluid.pressure_from_density(temperature, density)
def specific_internal_energy(self, temperature: float) -> float:
return self.cv * temperature
def specific_enthalpy(self, temperature: float) -> float:
return self.cp * temperature
def specific_reference_enthalpy(
self,
temperature: float,
reference_temperature: float = 298.15,
) -> float:
return self.cp * (temperature - reference_temperature)
def reference_temperature_from_specific_enthalpy(
self,
specific_enthalpy: float,
reference_temperature: float = 298.15,
) -> float:
if self.cp <= 0.0:
raise ValueError("cp must be positive.")
return reference_temperature + specific_enthalpy / self.cp
def pressure_reference_enthalpy(
self,
pressure: float,
temperature: float,
reference_pressure: float = 101_300.0,
reference_temperature: float = 298.15,
) -> float:
return (
self.specific_reference_enthalpy(temperature, reference_temperature)
+ self.fluid.residual_specific_enthalpy(pressure, temperature)
- self.fluid.residual_specific_enthalpy(
reference_pressure,
reference_temperature,
)
)
def pressure_transport_enthalpy(
self,
pressure: float,
temperature: float,
reference_pressure: float = 101_300.0,
reference_temperature: float = 298.15,
) -> float:
"""Convert AMESim reference enthalpy to the absolute-energy state basis."""
return (
self.pressure_reference_enthalpy(
pressure,
temperature,
reference_pressure,
reference_temperature,
)
+ self.cp * reference_temperature
)
def temperature_from_internal_energy(self, specific_internal_energy: float) -> float:
if self.cv <= 0.0:
raise ValueError("cv must be positive.")
return specific_internal_energy / self.cv
HELIUM_PNEUMATIC_GAS = AmesimPneumaticGas()
def liters_to_m3(value: float) -> float:
return value * 1.0e-3
def m3_to_cm3(value: float) -> float:
return value * 1.0e6
def cm3_to_m3(value: float) -> float:
return value * 1.0e-6
def kg_to_g(value: float) -> float:
return value * 1.0e3
def mm2_to_m2(value: float) -> float:
return value * 1.0e-6
def diameter_mm_to_area_m2(diameter_mm: float) -> float:
diameter_m = diameter_mm * 1.0e-3
return pi * diameter_m * diameter_m / 4.0
class AmesimPneumaticVolume(DynamicComponent):
"""First-pass AMESim pneumatic control volume using helium PR pressure closure."""
def __init__(
self,
name: str,
volume: float,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
p0: float = 101_325.0,
T0: float = 293.15,
heat_transfer_coefficient: float = 0.0,
heat_transfer_area: float = 0.0,
external_temperature_k: float = 293.15,
) -> None:
if volume <= 0.0:
raise ValueError("volume must be positive.")
if heat_transfer_coefficient < 0.0:
raise ValueError("heat_transfer_coefficient must be non-negative.")
if heat_transfer_area < 0.0:
raise ValueError("heat_transfer_area must be non-negative.")
if external_temperature_k <= 0.0:
raise ValueError("external_temperature_k must be positive.")
super().__init__(name=name)
self.volume = volume
self.gas = gas
self.heat_transfer_coefficient = heat_transfer_coefficient
self.heat_transfer_area = heat_transfer_area
self.external_temperature = external_temperature_k
rho0 = gas.density(p0, T0)
m0 = rho0 * volume
U0 = m0 * gas.specific_internal_energy(T0)
self.state = VolumeState(m=m0, U=U0)
self.port_a = PortState()
self.port_b = PortState()
@classmethod
def from_liters(
cls,
name: str,
volume_liters: float,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
p0: float = 101_325.0,
T0: float = 293.15,
heat_transfer_coefficient: float = 0.0,
heat_transfer_area: float = 0.0,
external_temperature_k: float = 293.15,
) -> "AmesimPneumaticVolume":
return cls(
name=name,
volume=liters_to_m3(volume_liters),
gas=gas,
p0=p0,
T0=T0,
heat_transfer_coefficient=heat_transfer_coefficient,
heat_transfer_area=heat_transfer_area,
external_temperature_k=external_temperature_k,
)
def get_state_vector(self) -> list[float]:
return self.state.as_vector()
def set_state_vector(self, values: list[float]) -> None:
self.state = VolumeState.from_vector(values)
def volume_cm3(self) -> float:
return m3_to_cm3(self.volume)
def volume_rate_m3_s(self) -> float:
return 0.0
def thermal_energy_flow_w(self, temperature_k: float | None = None) -> float:
temperature = self.properties().T if temperature_k is None else temperature_k
return (
self.heat_transfer_coefficient
* self.heat_transfer_area
* (self.external_temperature - temperature)
)
def gas_mass_g(self) -> float:
return kg_to_g(self.state.m)
def pressure_gauge_pa(self, reference_pressure_pa: float = 101_300.0) -> float:
return self.properties().p - reference_pressure_pa
def properties(self) -> ThermodynamicProperties:
if self.state.m <= 0.0:
raise ValueError("volume mass must stay positive.")
T = self.gas.temperature_from_internal_energy(self.state.U / self.state.m)
rho = self.state.m / self.volume
p = self.gas.pressure(rho, T)
u = self.state.U / self.state.m
h = self.gas.specific_enthalpy(T)
self.port_a.p = p
self.port_a.h_outflow = h
self.port_b.p = p
self.port_b.h_outflow = h
return ThermodynamicProperties(p=p, T=T, rho=rho, u=u, h=h)
def derivatives(self, inlet_h: float, m_flow: float) -> VolumeState:
return VolumeState(
m=m_flow,
U=m_flow * inlet_h + self.thermal_energy_flow_w(),
)
def derivatives_from_two_connections(
self,
*,
port_a_m_flow: float,
connected_h_a: float,
port_b_m_flow: float,
connected_h_b: float,
internal_h: float,
volume_rate_m3_s: float | None = None,
) -> VolumeState:
properties = self.properties()
inlet_h_a = self.connection_inlet_enthalpy(
port_m_flow=port_a_m_flow,
connected_h=connected_h_a,
internal_h=internal_h,
)
inlet_h_b = self.connection_inlet_enthalpy(
port_m_flow=port_b_m_flow,
connected_h=connected_h_b,
internal_h=internal_h,
)
return VolumeState(
m=port_a_m_flow + port_b_m_flow,
U=(
port_a_m_flow * inlet_h_a
+ port_b_m_flow * inlet_h_b
+ self.thermal_energy_flow_w(properties.T)
- properties.p * (
self.volume_rate_m3_s()
if volume_rate_m3_s is None
else volume_rate_m3_s
)
),
)
class AmesimVariablePneumaticVolume(AmesimPneumaticVolume):
"""PNCH012-style volume with a dead volume plus an external moving volume."""
def __init__(
self,
name: str,
dead_volume: float,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
p0: float = 101_325.0,
T0: float = 293.15,
external_volume: float = 0.0,
heat_transfer_coefficient: float = 0.0,
heat_transfer_area: float = 0.0,
external_temperature_k: float = 293.15,
) -> None:
if dead_volume <= 0.0:
raise ValueError("dead_volume must be positive.")
if dead_volume + external_volume <= 0.0:
raise ValueError("total volume must be positive.")
self.dead_volume = dead_volume
self.external_volume = external_volume
self.external_volume_rate = 0.0
super().__init__(
name=name,
volume=dead_volume + external_volume,
gas=gas,
p0=p0,
T0=T0,
heat_transfer_coefficient=heat_transfer_coefficient,
heat_transfer_area=heat_transfer_area,
external_temperature_k=external_temperature_k,
)
@classmethod
def from_liters(
cls,
name: str,
dead_volume_liters: float,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
p0: float = 101_325.0,
T0: float = 293.15,
external_volume_liters: float = 0.0,
heat_transfer_coefficient: float = 0.0,
heat_transfer_area: float = 0.0,
external_temperature_k: float = 293.15,
) -> "AmesimVariablePneumaticVolume":
return cls(
name=name,
dead_volume=liters_to_m3(dead_volume_liters),
gas=gas,
p0=p0,
T0=T0,
external_volume=liters_to_m3(external_volume_liters),
heat_transfer_coefficient=heat_transfer_coefficient,
heat_transfer_area=heat_transfer_area,
external_temperature_k=external_temperature_k,
)
def volume_rate_m3_s(self) -> float:
return self.external_volume_rate
def set_external_volume_m3(
self,
external_volume: float,
external_volume_rate_m3_s: float = 0.0,
) -> None:
if self.dead_volume + external_volume <= 0.0:
raise ValueError("total volume must be positive.")
self.external_volume = external_volume
self.external_volume_rate = external_volume_rate_m3_s
self.volume = self.dead_volume + self.external_volume
class AmesimPneumaticOrifice(AlgebraicComponent):
"""First-pass PNOR001/PNVO001-style compressible helium orifice.
This is a calibrated placeholder boundary for the Python port. It preserves
AMESim-style area and coefficient inputs, but final parity must be checked
against AMESim CSV results before treating it as numerically equivalent.
"""
def __init__(
self,
name: str,
area: float,
flow_coefficient: float = 1.0,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
opening: float = 1.0,
) -> None:
if area < 0.0:
raise ValueError("area must be non-negative.")
if flow_coefficient < 0.0:
raise ValueError("flow_coefficient must be non-negative.")
super().__init__(name=name)
self.area = area
self.flow_coefficient = flow_coefficient
self.gas = gas
self.opening = opening
self.port_a = PortState()
self.port_b = PortState()
@classmethod
def from_mm2(
cls,
name: str,
area_mm2: float,
flow_coefficient: float = 1.0,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
opening: float = 1.0,
) -> "AmesimPneumaticOrifice":
return cls(
name=name,
area=mm2_to_m2(area_mm2),
flow_coefficient=flow_coefficient,
gas=gas,
opening=opening,
)
@property
def effective_area(self) -> float:
opening = min(max(self.opening, 0.0), 1.0)
return self.area * opening
def mass_flow(self, p_a: float, p_b: float, upstream_temperature: float) -> float:
if p_a == p_b or self.effective_area == 0.0 or self.flow_coefficient == 0.0:
return 0.0
if p_a > p_b:
return compressible_orifice_mass_flow(
upstream_pressure=p_a,
downstream_pressure=p_b,
upstream_temperature=upstream_temperature,
area=self.effective_area,
flow_coefficient=self.flow_coefficient,
gas=self.gas,
)
return -compressible_orifice_mass_flow(
upstream_pressure=p_b,
downstream_pressure=p_a,
upstream_temperature=upstream_temperature,
area=self.effective_area,
flow_coefficient=self.flow_coefficient,
gas=self.gas,
)
def compressible_orifice_mass_flow(
*,
upstream_pressure: float,
downstream_pressure: float,
upstream_temperature: float,
area: float,
flow_coefficient: float,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
) -> float:
if upstream_pressure <= 0.0 or downstream_pressure < 0.0:
raise ValueError("pressures must be non-negative and upstream pressure must be positive.")
if upstream_temperature <= 0.0:
raise ValueError("upstream_temperature must be positive.")
if area < 0.0 or flow_coefficient < 0.0:
raise ValueError("area and flow_coefficient must be non-negative.")
if downstream_pressure >= upstream_pressure or area == 0.0 or flow_coefficient == 0.0:
return 0.0
gamma = gas.gamma
pressure_ratio = max(downstream_pressure / upstream_pressure, 0.0)
critical_ratio = (2.0 / (gamma + 1.0)) ** (gamma / (gamma - 1.0))
coefficient = flow_coefficient * area * upstream_pressure / sqrt(gas.R_gas * upstream_temperature)
if pressure_ratio <= critical_ratio:
flow_function = sqrt(gamma) * (2.0 / (gamma + 1.0)) ** ((gamma + 1.0) / (2.0 * (gamma - 1.0)))
else:
term = pressure_ratio ** (2.0 / gamma) - pressure_ratio ** ((gamma + 1.0) / gamma)
flow_function = sqrt((2.0 * gamma / (gamma - 1.0)) * max(term, 0.0))
return coefficient * flow_function
@@ -0,0 +1,881 @@
from __future__ import annotations
from dataclasses import dataclass
from math import log10, pi, sqrt
from app.simulation.examples.test_mql.primitives.pneumatic import (
HELIUM_PNEUMATIC_GAS,
AmesimPneumaticGas,
compressible_orifice_mass_flow,
diameter_mm_to_area_m2,
)
from app.simulation.core.base import AlgebraicComponent, DynamicComponent
from app.simulation.core.medium import ThermodynamicProperties
from app.simulation.core.ports import PortState
from app.simulation.core.state import VolumeState
@dataclass(frozen=True)
class AmesimPnl0001Diagnostics:
mass_flow_kg_s: float
reynolds_number: float
gas_velocity_m_s: float
friction_factor: float
pressure_drop_pa: float
class _DarcyPipeResistanceMixin:
diameter: float
length: float
relative_roughness: float
area: float
def _mass_flow_for_pressure_drop(
self,
pressure_drop_pa: float,
*,
density: float,
temperature: float,
) -> float:
if pressure_drop_pa <= 0.0:
return 0.0
upper = 1.0e-9
while self._darcy_pressure_drop(
upper,
density=density,
temperature=temperature,
) < pressure_drop_pa:
upper *= 10.0
if upper > 1.0e3:
raise ValueError("unable to bracket pneumatic pipe resistance flow")
lower = 0.0
for _ in range(48):
middle = 0.5 * (lower + upper)
if self._darcy_pressure_drop(
middle,
density=density,
temperature=temperature,
) < pressure_drop_pa:
lower = middle
else:
upper = middle
return 0.5 * (lower + upper)
def pn2pipefr_mass_flow(
self,
*,
port_1_pressure_pa: float,
port_1_temperature_k: float,
port_2_pressure_pa: float,
port_2_temperature_k: float,
length: float | None = None,
) -> float:
pressure_difference = port_1_pressure_pa - port_2_pressure_pa
if pressure_difference == 0.0:
return 0.0
upstream_pressure = max(port_1_pressure_pa, port_2_pressure_pa)
downstream_pressure = min(port_1_pressure_pa, port_2_pressure_pa)
upstream_temperature = (
port_1_temperature_k
if pressure_difference > 0.0
else port_2_temperature_k
)
resistance_length = self.length if length is None else length
if resistance_length <= 0.0:
raise ValueError("length must be positive")
def target_flow(mass_flow_kg_s: float) -> float:
reynolds = self._reynolds_number(mass_flow_kg_s, upstream_temperature)
friction_factor = self._friction_factor(reynolds)
flow_coefficient = sqrt(
self.diameter / (resistance_length * friction_factor)
)
return compressible_orifice_mass_flow(
upstream_pressure=upstream_pressure,
downstream_pressure=downstream_pressure,
upstream_temperature=upstream_temperature,
area=self.area,
flow_coefficient=flow_coefficient,
gas=self.gas,
)
flow_coefficient = sqrt(self.diameter / (resistance_length * 0.02))
magnitude = compressible_orifice_mass_flow(
upstream_pressure=upstream_pressure,
downstream_pressure=downstream_pressure,
upstream_temperature=upstream_temperature,
area=self.area,
flow_coefficient=flow_coefficient,
gas=self.gas,
)
for _ in range(12):
next_magnitude = target_flow(magnitude)
if abs(next_magnitude - magnitude) <= max(1.0e-12, abs(magnitude) * 1.0e-9):
magnitude = next_magnitude
break
magnitude = 0.5 * (magnitude + next_magnitude)
return magnitude if pressure_difference > 0.0 else -magnitude
def _darcy_pressure_drop(
self,
mass_flow_kg_s: float,
*,
density: float,
temperature: float,
) -> float:
if mass_flow_kg_s == 0.0:
return 0.0
reynolds = self._reynolds_number(mass_flow_kg_s, temperature)
friction_factor = self._friction_factor(reynolds)
velocity = mass_flow_kg_s / (density * self.area)
magnitude = (
friction_factor
* (self.length / self.diameter)
* density
* velocity
* velocity
/ 2.0
)
return magnitude if mass_flow_kg_s > 0.0 else -magnitude
def _reynolds_number(self, mass_flow_kg_s: float, temperature: float) -> float:
viscosity = helium_dynamic_viscosity(temperature)
return 4.0 * abs(mass_flow_kg_s) / (pi * self.diameter * viscosity)
def _friction_factor(self, reynolds_number: float) -> float:
if reynolds_number <= 0.0:
return 64_000_000.0
laminar = 64.0 / reynolds_number
if reynolds_number <= 2_300.0:
return laminar
turbulent = 1.0 / (
-1.8
* log10(
(self.relative_roughness / 3.7) ** 1.11
+ 6.9 / reynolds_number
)
) ** 2
if reynolds_number >= 4_000.0:
return turbulent
fraction = (reynolds_number - 2_300.0) / 1_700.0
return laminar + fraction * (turbulent - laminar)
class AmesimPnl0001Pipe(_DarcyPipeResistanceMixin, DynamicComponent):
"""Physical first-pass implementation of AMESim ``PNL0001`` (C-R).
Port 2 owns the lumped gas storage. Port 1 is connected through a Darcy
resistance. Both connection mass flows use the simulation convention:
positive values enter the pipe storage.
AMESim's proprietary ``pn2pipefr`` utility is represented by an
optional calibrated linear conductance when a model-specific baseline
supports it; otherwise the component falls back to an auditable
Darcy-Weisbach law. Both paths preserve the real geometry, state count,
mass/energy balance, heat-transfer parameter, and observable diagnostics.
"""
def __init__(
self,
name: str,
*,
diameter_mm: float,
length_m: float,
relative_roughness: float,
polytropic_constant: float = 1.35,
heat_transfer_coefficient: float = 0.0,
external_temperature_k: float = 293.15,
calibrated_linear_conductance: float | None = None,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
p0: float = 101_325.0,
T0: float = 293.15,
) -> None:
if diameter_mm <= 0.0:
raise ValueError("diameter_mm must be positive")
if length_m <= 0.0:
raise ValueError("length_m must be positive")
if relative_roughness < 0.0:
raise ValueError("relative_roughness must be non-negative")
if polytropic_constant <= 0.0:
raise ValueError("polytropic_constant must be positive")
if heat_transfer_coefficient < 0.0:
raise ValueError("heat_transfer_coefficient must be non-negative")
if external_temperature_k <= 0.0:
raise ValueError("external_temperature_k must be positive")
if (
calibrated_linear_conductance is not None
and calibrated_linear_conductance <= 0.0
):
raise ValueError("calibrated_linear_conductance must be positive")
super().__init__(name=name)
self.diameter = diameter_mm * 1.0e-3
self.length = length_m
self.relative_roughness = relative_roughness
self.polytropic_constant = polytropic_constant
self.heat_transfer_coefficient = heat_transfer_coefficient
self.external_temperature = external_temperature_k
self.calibrated_linear_conductance = calibrated_linear_conductance
self.gas = gas
self.area = diameter_mm_to_area_m2(diameter_mm)
self.volume = self.area * self.length
self.heat_transfer_area = pi * self.diameter * self.length
rho0 = gas.density(p0, T0)
mass0 = rho0 * self.volume
self.state = VolumeState(
m=mass0,
U=mass0 * gas.specific_internal_energy(T0),
)
self.port_1 = PortState()
self.port_2 = PortState()
def get_state_vector(self) -> list[float]:
return self.state.as_vector()
def set_state_vector(self, values: list[float]) -> None:
self.state = VolumeState.from_vector(values)
def properties(self) -> ThermodynamicProperties:
if self.state.m <= 0.0:
raise ValueError("pipe mass must stay positive")
temperature = self.gas.temperature_from_internal_energy(
self.state.U / self.state.m
)
density = self.state.m / self.volume
pressure = self.gas.pressure(density, temperature)
properties = ThermodynamicProperties(
p=pressure,
T=temperature,
rho=density,
u=self.state.U / self.state.m,
h=self.gas.specific_enthalpy(temperature),
)
self.port_2.p = pressure
self.port_2.h_outflow = properties.h
return properties
def gas_mass_g(self) -> float:
return self.state.m * 1.0e3
def resistance_mass_flow(
self,
*,
port_1_pressure_pa: float,
port_1_temperature_k: float,
) -> float:
"""Return mass flow from port 1 into the port-2 storage in kg/s."""
if port_1_pressure_pa <= 0.0:
raise ValueError("port_1_pressure_pa must be positive")
if port_1_temperature_k <= 0.0:
raise ValueError("port_1_temperature_k must be positive")
internal = self.properties()
pressure_difference = port_1_pressure_pa - internal.p
if pressure_difference == 0.0:
return 0.0
if self.calibrated_linear_conductance is not None:
return (
self.calibrated_linear_conductance
* pressure_difference
/ sqrt(internal.T)
)
upstream_pressure = max(port_1_pressure_pa, internal.p)
upstream_temperature = (
port_1_temperature_k if pressure_difference > 0.0 else internal.T
)
density = self.gas.density(upstream_pressure, upstream_temperature)
magnitude = self._mass_flow_for_pressure_drop(
abs(pressure_difference),
density=density,
temperature=upstream_temperature,
)
return magnitude if pressure_difference > 0.0 else -magnitude
def diagnostics(
self,
*,
mass_flow_kg_s: float,
temperature_k: float | None = None,
) -> AmesimPnl0001Diagnostics:
properties = self.properties()
temperature = temperature_k or properties.T
reynolds = self._reynolds_number(mass_flow_kg_s, temperature)
friction_factor = self._friction_factor(reynolds)
velocity = mass_flow_kg_s / (properties.rho * self.area)
pressure_drop = self._darcy_pressure_drop(
mass_flow_kg_s,
density=properties.rho,
temperature=temperature,
)
return AmesimPnl0001Diagnostics(
mass_flow_kg_s=mass_flow_kg_s,
reynolds_number=reynolds,
gas_velocity_m_s=velocity,
friction_factor=friction_factor,
pressure_drop_pa=pressure_drop,
)
def darcy_pressure_drop_for_state(
self,
*,
mass_flow_kg_s: float,
pressure_pa: float,
temperature_k: float,
) -> float:
if pressure_pa <= 0.0:
raise ValueError("pressure_pa must be positive")
if temperature_k <= 0.0:
raise ValueError("temperature_k must be positive")
density = self.gas.density(pressure_pa, temperature_k)
return self._darcy_pressure_drop(
mass_flow_kg_s,
density=density,
temperature=temperature_k,
)
def derivatives_from_connections(
self,
*,
port_1_m_flow: float,
connected_h_1: float,
port_2_m_flow: float,
connected_h_2: float,
) -> VolumeState:
internal = self.properties()
# Default first-pass PNL0001 behavior uses the historical internal-energy
# approximation. AMESim-specific transport-enthalpy corrections are kept
# behind derivatives_from_transport_enthalpy_connections so they can be
# applied only where validated against baseline data.
inlet_u_1 = (
connected_h_1 / self.gas.gamma
if port_1_m_flow > 0.0
else internal.u
)
inlet_u_2 = (
connected_h_2 / self.gas.gamma
if port_2_m_flow > 0.0
else internal.u
)
heat_flow = (
self.heat_transfer_coefficient
* self.heat_transfer_area
* (self.external_temperature - internal.T)
)
return VolumeState(
m=port_1_m_flow + port_2_m_flow,
U=port_1_m_flow * inlet_u_1 + port_2_m_flow * inlet_u_2 + heat_flow,
)
def derivatives_from_transport_enthalpy_connections(
self,
*,
port_1_m_flow: float,
connected_h_1: float,
port_2_m_flow: float,
connected_h_2: float,
) -> VolumeState:
internal = self.properties()
inlet_h_1 = connected_h_1 if port_1_m_flow > 0.0 else internal.h
inlet_h_2 = connected_h_2 if port_2_m_flow > 0.0 else internal.h
heat_flow = (
self.heat_transfer_coefficient
* self.heat_transfer_area
* (self.external_temperature - internal.T)
)
return VolumeState(
m=port_1_m_flow + port_2_m_flow,
U=port_1_m_flow * inlet_h_1 + port_2_m_flow * inlet_h_2 + heat_flow,
)
class AmesimPnl0003Pipe(_DarcyPipeResistanceMixin, DynamicComponent):
"""First-pass AMESim ``PNL0003`` (C-R-C) pipe.
The two pipe-end compliances are represented as equal half-volume gas
stores connected by the same auditable Darcy resistance used for PNL0001.
Center flow is positive from port 1 storage to port 2 storage.
"""
state_size = 4
def __init__(
self,
name: str,
*,
diameter_mm: float,
length_m: float,
relative_roughness: float,
polytropic_constant: float = 1.35,
heat_transfer_coefficient: float = 0.0,
external_temperature_k: float = 293.15,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
p1_0: float = 101_325.0,
T1_0: float = 293.15,
p2_0: float = 101_325.0,
T2_0: float = 293.15,
) -> None:
if diameter_mm <= 0.0:
raise ValueError("diameter_mm must be positive")
if length_m <= 0.0:
raise ValueError("length_m must be positive")
if relative_roughness < 0.0:
raise ValueError("relative_roughness must be non-negative")
if polytropic_constant <= 0.0:
raise ValueError("polytropic_constant must be positive")
if heat_transfer_coefficient < 0.0:
raise ValueError("heat_transfer_coefficient must be non-negative")
if external_temperature_k <= 0.0:
raise ValueError("external_temperature_k must be positive")
super().__init__(name=name)
self.diameter = diameter_mm * 1.0e-3
self.length = length_m
self.relative_roughness = relative_roughness
self.polytropic_constant = polytropic_constant
self.heat_transfer_coefficient = heat_transfer_coefficient
self.external_temperature = external_temperature_k
self.gas = gas
self.area = diameter_mm_to_area_m2(diameter_mm)
self.volume = self.area * self.length
self.compliance_volume = self.volume / 2.0
self.heat_transfer_area = pi * self.diameter * self.length
self.state_1 = self._initial_state(p1_0, T1_0)
self.state_2 = self._initial_state(p2_0, T2_0)
self.port_1 = PortState()
self.port_2 = PortState()
def _initial_state(self, pressure: float, temperature: float) -> VolumeState:
rho = self.gas.density(pressure, temperature)
mass = rho * self.compliance_volume
return VolumeState(
m=mass,
U=mass * self.gas.specific_internal_energy(temperature),
)
def get_state_vector(self) -> list[float]:
return [*self.state_1.as_vector(), *self.state_2.as_vector()]
def set_state_vector(self, values: list[float]) -> None:
if len(values) != 4:
raise ValueError("PNL0003 state vector requires four values")
self.state_1 = VolumeState.from_vector(values[:2])
self.state_2 = VolumeState.from_vector(values[2:])
def properties_1(self) -> ThermodynamicProperties:
properties = self._properties(self.state_1)
self.port_1.p = properties.p
self.port_1.h_outflow = properties.h
return properties
def properties_2(self) -> ThermodynamicProperties:
properties = self._properties(self.state_2)
self.port_2.p = properties.p
self.port_2.h_outflow = properties.h
return properties
def _properties(self, state: VolumeState) -> ThermodynamicProperties:
if state.m <= 0.0:
raise ValueError("pipe mass must stay positive")
temperature = self.gas.temperature_from_internal_energy(state.U / state.m)
density = state.m / self.compliance_volume
pressure = self.gas.pressure(density, temperature)
return ThermodynamicProperties(
p=pressure,
T=temperature,
rho=density,
u=state.U / state.m,
h=self.gas.specific_enthalpy(temperature),
)
def gas_mass_g(self) -> float:
return (self.state_1.m + self.state_2.m) * 1.0e3
def resistance_mass_flow(self) -> float:
"""Return center mass flow from port 1 storage to port 2 storage."""
port_1 = self.properties_1()
port_2 = self.properties_2()
pressure_difference = port_1.p - port_2.p
if pressure_difference == 0.0:
return 0.0
upstream = port_1 if pressure_difference > 0.0 else port_2
magnitude = self._mass_flow_for_pressure_drop(
abs(pressure_difference),
density=upstream.rho,
temperature=upstream.T,
)
return magnitude if pressure_difference > 0.0 else -magnitude
def diagnostics(
self,
*,
mass_flow_kg_s: float,
temperature_k: float | None = None,
) -> AmesimPnl0001Diagnostics:
port_1 = self.properties_1()
port_2 = self.properties_2()
temperature = temperature_k or (port_1.T if mass_flow_kg_s >= 0.0 else port_2.T)
density = port_1.rho if mass_flow_kg_s >= 0.0 else port_2.rho
reynolds = self._reynolds_number(mass_flow_kg_s, temperature)
friction_factor = self._friction_factor(reynolds)
velocity = mass_flow_kg_s / (density * self.area)
pressure_drop = self._darcy_pressure_drop(
mass_flow_kg_s,
density=density,
temperature=temperature,
)
return AmesimPnl0001Diagnostics(
mass_flow_kg_s=mass_flow_kg_s,
reynolds_number=reynolds,
gas_velocity_m_s=velocity,
friction_factor=friction_factor,
pressure_drop_pa=pressure_drop,
)
def derivatives_from_connections(
self,
*,
port_1_m_flow: float,
connected_h_1: float,
port_2_m_flow: float,
connected_h_2: float,
) -> tuple[VolumeState, VolumeState]:
port_1 = self.properties_1()
port_2 = self.properties_2()
center_flow = self.resistance_mass_flow()
heat_flow_each = (
self.heat_transfer_coefficient
* self.heat_transfer_area
* (self.external_temperature - 0.5 * (port_1.T + port_2.T))
/ 2.0
)
port_1_external_h = self.connection_inlet_enthalpy(
port_m_flow=port_1_m_flow,
connected_h=connected_h_1,
internal_h=port_1.h,
)
port_2_external_h = self.connection_inlet_enthalpy(
port_m_flow=port_2_m_flow,
connected_h=connected_h_2,
internal_h=port_2.h,
)
port_1_center_h = self.connection_inlet_enthalpy(
port_m_flow=-center_flow,
connected_h=port_2.h,
internal_h=port_1.h,
)
port_2_center_h = self.connection_inlet_enthalpy(
port_m_flow=center_flow,
connected_h=port_1.h,
internal_h=port_2.h,
)
return (
VolumeState(
m=port_1_m_flow - center_flow,
U=(
port_1_m_flow * port_1_external_h
- center_flow * port_1_center_h
+ heat_flow_each
),
),
VolumeState(
m=port_2_m_flow + center_flow,
U=(
port_2_m_flow * port_2_external_h
+ center_flow * port_2_center_h
+ heat_flow_each
),
),
)
class AmesimPnl0002Pipe(_DarcyPipeResistanceMixin, DynamicComponent):
"""First-pass AMESim ``PNL0002`` (R-C-R) pipe.
The center compliance owns the gas state. Positive connection mass flows
enter that center storage from each external port.
"""
state_size = 2
def __init__(
self,
name: str,
*,
diameter_mm: float,
length_m: float,
relative_roughness: float,
polytropic_constant: float = 1.35,
heat_transfer_coefficient: float = 0.0,
external_temperature_k: float = 293.15,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
pctr_0: float = 101_325.0,
Tctr_0: float = 293.15,
) -> None:
if diameter_mm <= 0.0:
raise ValueError("diameter_mm must be positive")
if length_m <= 0.0:
raise ValueError("length_m must be positive")
if relative_roughness < 0.0:
raise ValueError("relative_roughness must be non-negative")
if polytropic_constant <= 0.0:
raise ValueError("polytropic_constant must be positive")
if heat_transfer_coefficient < 0.0:
raise ValueError("heat_transfer_coefficient must be non-negative")
if external_temperature_k <= 0.0:
raise ValueError("external_temperature_k must be positive")
super().__init__(name=name)
self.diameter = diameter_mm * 1.0e-3
self.length = length_m
self.relative_roughness = relative_roughness
self.polytropic_constant = polytropic_constant
self.heat_transfer_coefficient = heat_transfer_coefficient
self.external_temperature = external_temperature_k
self.gas = gas
self.area = diameter_mm_to_area_m2(diameter_mm)
self.volume = self.area * self.length
self.heat_transfer_area = pi * self.diameter * self.length
self._resistance_length = self.length / 2.0
rho0 = gas.density(pctr_0, Tctr_0)
mass0 = rho0 * self.volume
self.state = VolumeState(
m=mass0,
U=mass0 * gas.specific_internal_energy(Tctr_0),
)
self.port_1 = PortState()
self.port_2 = PortState()
def get_state_vector(self) -> list[float]:
return self.state.as_vector()
def set_state_vector(self, values: list[float]) -> None:
self.state = VolumeState.from_vector(values)
def properties(self) -> ThermodynamicProperties:
if self.state.m <= 0.0:
raise ValueError("pipe mass must stay positive")
temperature = self.gas.temperature_from_internal_energy(
self.state.U / self.state.m
)
density = self.state.m / self.volume
pressure = self.gas.pressure(density, temperature)
properties = ThermodynamicProperties(
p=pressure,
T=temperature,
rho=density,
u=self.state.U / self.state.m,
h=self.gas.specific_enthalpy(temperature),
)
self.port_1.p = pressure
self.port_1.h_outflow = properties.h
self.port_2.p = pressure
self.port_2.h_outflow = properties.h
return properties
def gas_mass_g(self) -> float:
return self.state.m * 1.0e3
def port_mass_flow(
self,
*,
port_pressure_pa: float,
port_temperature_k: float,
) -> float:
"""Return mass flow from an external port into the center storage."""
if port_pressure_pa <= 0.0:
raise ValueError("port_pressure_pa must be positive")
if port_temperature_k <= 0.0:
raise ValueError("port_temperature_k must be positive")
center = self.properties()
pressure_difference = port_pressure_pa - center.p
if pressure_difference == 0.0:
return 0.0
upstream_pressure = max(port_pressure_pa, center.p)
upstream_temperature = (
port_temperature_k if pressure_difference > 0.0 else center.T
)
density = self.gas.density(upstream_pressure, upstream_temperature)
magnitude = self._mass_flow_for_resistance_pressure_drop(
abs(pressure_difference),
density=density,
temperature=upstream_temperature,
)
return magnitude if pressure_difference > 0.0 else -magnitude
def _mass_flow_for_resistance_pressure_drop(
self,
pressure_drop_pa: float,
*,
density: float,
temperature: float,
) -> float:
original_length = self.length
self.length = self._resistance_length
try:
return self._mass_flow_for_pressure_drop(
pressure_drop_pa,
density=density,
temperature=temperature,
)
finally:
self.length = original_length
def diagnostics(
self,
*,
mass_flow_kg_s: float,
temperature_k: float | None = None,
) -> AmesimPnl0001Diagnostics:
properties = self.properties()
temperature = temperature_k or properties.T
reynolds = self._reynolds_number(mass_flow_kg_s, temperature)
friction_factor = self._friction_factor(reynolds)
velocity = mass_flow_kg_s / (properties.rho * self.area)
original_length = self.length
self.length = self._resistance_length
try:
pressure_drop = self._darcy_pressure_drop(
mass_flow_kg_s,
density=properties.rho,
temperature=temperature,
)
finally:
self.length = original_length
return AmesimPnl0001Diagnostics(
mass_flow_kg_s=mass_flow_kg_s,
reynolds_number=reynolds,
gas_velocity_m_s=velocity,
friction_factor=friction_factor,
pressure_drop_pa=pressure_drop,
)
def derivatives_from_connections(
self,
*,
port_1_m_flow: float,
connected_h_1: float,
port_2_m_flow: float,
connected_h_2: float,
) -> VolumeState:
center = self.properties()
inlet_h_1 = self.connection_inlet_enthalpy(
port_m_flow=port_1_m_flow,
connected_h=connected_h_1,
internal_h=center.h,
)
inlet_h_2 = self.connection_inlet_enthalpy(
port_m_flow=port_2_m_flow,
connected_h=connected_h_2,
internal_h=center.h,
)
heat_flow = (
self.heat_transfer_coefficient
* self.heat_transfer_area
* (self.external_temperature - center.T)
)
return VolumeState(
m=port_1_m_flow + port_2_m_flow,
U=port_1_m_flow * inlet_h_1 + port_2_m_flow * inlet_h_2 + heat_flow,
)
class AmesimPnl00rPipe(_DarcyPipeResistanceMixin, AlgebraicComponent):
"""First-pass AMESim ``PNL00R`` (R) pipe resistance."""
def __init__(
self,
name: str,
*,
diameter_mm: float,
length_m: float,
relative_roughness: float,
gas: AmesimPneumaticGas = HELIUM_PNEUMATIC_GAS,
) -> None:
if diameter_mm <= 0.0:
raise ValueError("diameter_mm must be positive")
if length_m <= 0.0:
raise ValueError("length_m must be positive")
if relative_roughness < 0.0:
raise ValueError("relative_roughness must be non-negative")
super().__init__(name=name)
self.diameter = diameter_mm * 1.0e-3
self.length = length_m
self.relative_roughness = relative_roughness
self.gas = gas
self.area = diameter_mm_to_area_m2(diameter_mm)
self.port_1 = PortState()
self.port_2 = PortState()
def mass_flow(
self,
*,
port_1_pressure_pa: float,
port_1_temperature_k: float,
port_2_pressure_pa: float,
port_2_temperature_k: float,
) -> float:
"""Return mass flow from port 1 to port 2 in kg/s."""
if port_1_pressure_pa <= 0.0 or port_2_pressure_pa <= 0.0:
raise ValueError("port pressures must be positive")
if port_1_temperature_k <= 0.0 or port_2_temperature_k <= 0.0:
raise ValueError("port temperatures must be positive")
pressure_difference = port_1_pressure_pa - port_2_pressure_pa
if pressure_difference == 0.0:
return 0.0
upstream_pressure = max(port_1_pressure_pa, port_2_pressure_pa)
upstream_temperature = (
port_1_temperature_k
if pressure_difference > 0.0
else port_2_temperature_k
)
density = self.gas.density(upstream_pressure, upstream_temperature)
magnitude = self._mass_flow_for_pressure_drop(
abs(pressure_difference),
density=density,
temperature=upstream_temperature,
)
return magnitude if pressure_difference > 0.0 else -magnitude
def diagnostics(
self,
*,
mass_flow_kg_s: float,
pressure_pa: float,
temperature_k: float,
) -> AmesimPnl0001Diagnostics:
density = self.gas.density(pressure_pa, temperature_k)
reynolds = self._reynolds_number(mass_flow_kg_s, temperature_k)
friction_factor = self._friction_factor(reynolds)
velocity = mass_flow_kg_s / (density * self.area)
pressure_drop = self._darcy_pressure_drop(
mass_flow_kg_s,
density=density,
temperature=temperature_k,
)
return AmesimPnl0001Diagnostics(
mass_flow_kg_s=mass_flow_kg_s,
reynolds_number=reynolds,
gas_velocity_m_s=velocity,
friction_factor=friction_factor,
pressure_drop_pa=pressure_drop,
)
def helium_dynamic_viscosity(temperature_k: float) -> float:
"""Sutherland approximation centered on the test_mql initial condition."""
if temperature_k <= 0.0:
raise ValueError("temperature_k must be positive")
reference_temperature = 293.15
reference_viscosity = 2.0e-5
sutherland_constant = 79.4
return (
reference_viscosity
* (temperature_k / reference_temperature) ** 1.5
* (reference_temperature + sutherland_constant)
/ (temperature_k + sutherland_constant)
)
+20 -2
View File
@@ -5,8 +5,7 @@ from datetime import UTC, datetime
from pathlib import Path
from app.simulation.paths import PROJECT_ROOT, SIMULATION_RUNS_DIR
from PythonModels.scripts.run_test_mql import format_test_mql_summary
from PythonModels.systems.test_mql import (
from app.simulation.examples.test_mql.system import (
TestMqlRunConfig,
TestMqlSimulationResult,
TestMqlSystem,
@@ -29,6 +28,25 @@ class TestMqlRunResult:
summary_path: Path
def format_test_mql_summary(system: TestMqlSystem) -> str:
snapshot = system.snapshot()
lines = [
"Model: test_mql",
f"Source archive: {system.archive_path}",
f"Components: {snapshot.component_count}",
f"Connections: {snapshot.connection_count}",
f"Continuous states in AMESim modelinfo: {snapshot.continuous_state_count}",
f"Discrete states in AMESim modelinfo: {snapshot.discrete_state_count}",
"Global parameters:",
]
for name, value in sorted(snapshot.global_parameters.items()):
lines.append(f" - {name}: {value}")
lines.append("Component submodels:")
for name, count in sorted(snapshot.submodel_counts.items()):
lines.append(f" - {name}: {count}")
return "\n".join(lines) + "\n"
def _default_run_output_dir() -> Path:
timestamp = datetime.now(UTC).strftime("test_mql_%Y%m%d_%H%M%S_%f")
return SIMULATION_RUNS_DIR / timestamp
@@ -0,0 +1,80 @@
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from app.simulation.examples.test_mql.baseline import (
TestMqlBaselineRun,
run_test_mql_baseline_passthrough,
)
from app.simulation.paths import (
AMESIM_TEST_MQL_ARCHIVE_PATH,
SIMULATION_RUNS_DIR,
)
@dataclass(frozen=True)
class TestMqlBaselinePathConfig:
archive_path: Path = field(
default_factory=lambda: AMESIM_TEST_MQL_ARCHIVE_PATH
)
output_dir: Path | None = None
@dataclass(frozen=True)
class TestMqlBaselineExecutionConfig:
write_summary: bool = True
data_paths: tuple[str, ...] | None = None
@dataclass(frozen=True)
class TestMqlBaselineScriptConfig:
paths: TestMqlBaselinePathConfig = field(default_factory=TestMqlBaselinePathConfig)
execution: TestMqlBaselineExecutionConfig = field(default_factory=TestMqlBaselineExecutionConfig)
def _default_output_dir() -> Path:
timestamp = datetime.now(UTC).strftime("test_mql_baseline_%Y%m%d_%H%M%S_%f")
return SIMULATION_RUNS_DIR / timestamp
def format_test_mql_baseline_summary(run: TestMqlBaselineRun) -> str:
return "\n".join(
[
"Model: test_mql",
"Mode: AMESim baseline passthrough",
f"Samples: {run.sample_count}",
f"Output schema signals: {run.output_schema.signal_count}",
f"Compared signals: {run.signal_count}",
f"Observation bindings: {run.observation_catalog.binding_count}",
f"Max absolute error: {run.comparison.max_abs_error}",
f"Max relative error: {run.comparison.max_rel_error}",
]
) + "\n"
def run_test_mql_baseline(config: TestMqlBaselineScriptConfig | None = None):
config = config or TestMqlBaselineScriptConfig()
run = run_test_mql_baseline_passthrough(
config.paths.archive_path,
data_paths=config.execution.data_paths,
)
output_dir = config.paths.output_dir or _default_output_dir()
if config.execution.write_summary:
output_dir.mkdir(parents=True, exist_ok=True)
(output_dir / "test_mql_baseline_summary.txt").write_text(
format_test_mql_baseline_summary(run),
encoding="utf-8",
)
return run, output_dir
def main() -> None:
run, output_dir = run_test_mql_baseline()
print(format_test_mql_baseline_summary(run), end="")
print(f"Output directory: {output_dir}")
if __name__ == "__main__":
main()
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,18 @@
from __future__ import annotations
from dataclasses import dataclass
from app.simulation.solvers.solver import (
SolveIVPConfig as _BaseSolveIVPConfig,
integrate_ode,
)
@dataclass(frozen=True)
class SolveIVPConfig(_BaseSolveIVPConfig):
"""Solver defaults used by the calibrated ``test_mql`` example."""
atol: float = 1e-10
__all__ = ["SolveIVPConfig", "integrate_ode"]
@@ -0,0 +1,78 @@
from __future__ import annotations
from dataclasses import dataclass
from app.simulation.core.base import Component, DynamicComponent
@dataclass(frozen=True)
class Connection:
source_component: str
source_port: str
target_component: str
target_port: str
class SimulationNetwork:
"""Container for components, topology, and state-vector bookkeeping."""
def __init__(self, name: str) -> None:
self.name = name
self.components: dict[str, Component] = {}
self.connections: list[Connection] = []
def add_component(self, component: Component) -> None:
if component.name in self.components:
raise ValueError(f"Duplicate component name: {component.name}")
self.components[component.name] = component
def connect(
self,
source_component: str,
source_port: str,
target_component: str,
target_port: str,
) -> None:
self.connections.append(
Connection(
source_component=source_component,
source_port=source_port,
target_component=target_component,
target_port=target_port,
)
)
def dynamic_components(self) -> list[DynamicComponent]:
return [
component
for component in self.components.values()
if isinstance(component, DynamicComponent)
]
def initial_state_vector(self) -> list[float]:
values: list[float] = []
for component in self.dynamic_components():
values.extend(component.get_state_vector())
return values
def apply_state_vector(self, values: list[float]) -> None:
cursor = 0
for component in self.dynamic_components():
next_cursor = cursor + component.state_size
component.set_state_vector(values[cursor:next_cursor])
cursor = next_cursor
if cursor != len(values):
raise ValueError("State vector length does not match dynamic components.")
def summary(self) -> str:
lines = [f"Network: {self.name}", "Components:"]
for name, component in self.components.items():
lines.append(f" - {name}: {component.__class__.__name__}")
lines.append("Connections:")
for conn in self.connections:
lines.append(
f" - {conn.source_component}.{conn.source_port}"
f" -> {conn.target_component}.{conn.target_port}"
)
return "\n".join(lines)
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,118 @@
from __future__ import annotations
import re
import tarfile
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class TestMqlComponentContact:
component_a: str
port_a: str
component_b: str
port_b: str
def other_endpoint(self, component_alias: str) -> tuple[str, str]:
if component_alias == self.component_a:
return self.component_b, self.port_b
if component_alias == self.component_b:
return self.component_a, self.port_a
raise KeyError(component_alias)
def port_for(self, component_alias: str) -> str:
if component_alias == self.component_a:
return self.port_a
if component_alias == self.component_b:
return self.port_b
raise KeyError(component_alias)
@dataclass(frozen=True)
class TestMqlCirTopology:
component_contacts: tuple[TestMqlComponentContact, ...]
def contacts_for(self, component_alias: str) -> tuple[TestMqlComponentContact, ...]:
return tuple(
contact
for contact in self.component_contacts
if component_alias in (contact.component_a, contact.component_b)
)
def load_test_mql_cir_topology(
archive_path: str | Path,
*,
cir_member: str = "test_mql_.cir",
) -> TestMqlCirTopology:
with tarfile.open(archive_path) as archive:
cir_file = archive.extractfile(cir_member)
if cir_file is None:
raise ValueError(f"Missing AMESim circuit member: {cir_member}")
cir_text = cir_file.read().decode("latin1")
root = ET.fromstring(_topology_only_xml(cir_text))
components = root.findall(".//COMPS_LIST/COMP")
aliases = tuple(_required_text(component, "ALIAS") for component in components)
contacts: dict[
tuple[tuple[int, int], tuple[int, int]],
TestMqlComponentContact,
] = {}
directed_contacts: set[tuple[tuple[int, int], tuple[int, int]]] = set()
for component_index, component in enumerate(components):
ports = component.findall("./COMP_PORTS_LIST/COMP_PORT")
for port_index, port in enumerate(ports):
if port.findtext("PORT_CONNECT") != "1":
continue
for connection in port.findall("./CONNECT_LIST/CONNECT"):
target_index = int(_required_text(connection, "CONNECT_ENTITY_NUM"))
target_port_index = int(_required_text(connection, "CONNECT_ENTITY_PORT"))
if target_index < 0 or target_index >= len(components):
raise ValueError(f"Component contact references unknown entity {target_index}")
target_ports = components[target_index].findall("./COMP_PORTS_LIST/COMP_PORT")
if target_port_index < 0 or target_port_index >= len(target_ports):
raise ValueError(
f"Component contact references unknown port {target_port_index} "
f"on {aliases[target_index]}"
)
endpoint = (component_index, port_index)
target_endpoint = (target_index, target_port_index)
directed_contacts.add((endpoint, target_endpoint))
key = tuple(sorted((endpoint, target_endpoint)))
first, second = key
contacts[key] = TestMqlComponentContact(
component_a=aliases[first[0]],
port_a=f"port_{first[1] + 1}",
component_b=aliases[second[0]],
port_b=f"port_{second[1] + 1}",
)
for endpoint, target_endpoint in directed_contacts:
if (target_endpoint, endpoint) not in directed_contacts:
raise ValueError(
"AMESim component contact is not reciprocal: "
f"{endpoint} -> {target_endpoint}"
)
return TestMqlCirTopology(component_contacts=tuple(contacts.values()))
def _topology_only_xml(cir_text: str) -> str:
# AMESim expressions inside SUBMODEL contain unescaped && and <= operators.
# Topology lives outside those blocks, so omit them before XML parsing.
return re.sub(
r"<SUBMODEL>.*?</SUBMODEL>",
"<SUBMODEL />",
cir_text,
flags=re.DOTALL,
)
def _required_text(element: ET.Element, child_name: str) -> str:
value = element.findtext(child_name)
if value is None:
raise ValueError(f"Missing AMESim circuit element: {child_name}")
return value