from __future__ import annotations import unittest from app.main import ( ReactFlowProjectPayload, build_reactflow_system_xml, compile_reactflow_network, ) from app.system_xml import validate_system_xml_document from app.simulation.components.amesim.gases import ( AMESIM_BUILTIN_AIR_GAS_INDEX, AMESIM_DEFAULT_GAS_INDEX, AmesimGasDefinition, AmesimGasRegistry, default_amesim_gas_registry, normalize_amesim_defined_gas_index, normalize_amesim_gas_index, ) from app.simulation.components.amesim.media import AmesimIdealAirMedium from app.simulation.core.medium import IdealGasMedium from app.simulation.registry import COMPONENT_MODEL_REGISTRY from tests.test_generic_system_xml_simulation import component_node, physical_edge from tests.test_system_xml_protocol import physical_port GAS_INDEXED_MODEL_TYPES = ( "amesim_pnch023", "amesim_pnch012", "amesim_pnrp17", "amesim_pnor001", "amesim_pnvo001_fixed", "amesim_pnvo001", "amesim_pnl00r", "amesim_pnl0001", "amesim_pnl0002", "amesim_pnl0003", ) def pneumatic_pipe_node( component_id: str, *, gi: float, ) -> dict[str, object]: return component_node( component_id, "amesim_pnl00r", [ physical_port("port_1", "bidirectional", "left"), physical_port("port_2", "bidirectional", "right"), ], {"diam": 0.01, "le": 1.0, "rr": 1.0e-5, "gi": gi}, ) def pneumatic_chamber_node( component_id: str, *, gi: float | None = None, volume: float = 0.01, pressure: float = 200000.0, temperature: float = 300.0, ) -> dict[str, object]: parameters = { "cvol": volume, "kth": 0.0, "sth": 0.1, "extemp": 293.15, "p0": pressure, "T0": temperature, } if gi is not None: parameters["gi"] = gi return component_node( component_id, "amesim_pnch023", [ physical_port("port_1", "bidirectional", "left"), physical_port("port_2", "bidirectional", "right"), ], parameters, ) def ideal_air_medium_node( component_id: str, *, gi: float, property_model: float | None = None, ) -> dict[str, object]: parameters = {"gi": gi} if property_model is not None: parameters["property_model"] = property_model return component_node( component_id, "amesim_ideal_air_medium", [], parameters, ) def two_slot_registry() -> tuple[AmesimGasRegistry, IdealGasMedium, IdealGasMedium]: future_helium = IdealGasMedium( name="FutureHeliumStub", R_gas=2077.26439404998, cp_ref=5193.0, viscosity_ref=1.96e-5, sutherland_constant=79.4, ) registry = AmesimGasRegistry( ( AmesimGasDefinition( gi=2, label="Future helium", medium=future_helium, fluid_type=12, eos_type=6, ), ) ) air = registry.resolve(AMESIM_BUILTIN_AIR_GAS_INDEX) return registry, air, future_helium class AmesimGasRegistryTests(unittest.TestCase): def test_all_gas_sensitive_models_expose_the_same_gi_contract(self) -> None: medium = IdealGasMedium() for model_type in GAS_INDEXED_MODEL_TYPES: with self.subTest(model_type=model_type): spec = COMPONENT_MODEL_REGISTRY[model_type] parameter = spec.parameter_by_name["gi"] self.assertIs(spec.parameters[0], parameter) self.assertEqual(parameter.default, 0.0) self.assertEqual(parameter.minimum, 0.0) self.assertEqual(parameter.maximum, 99.0) self.assertEqual(parameter.editor, "amesimGasReference") component = spec.create(f"{model_type}_1", medium, {}) self.assertEqual(component.gi, AMESIM_DEFAULT_GAS_INDEX) with self.assertRaisesRegex(ValueError, "gi must be an integer"): spec.create(f"{model_type}_fractional", medium, {"gi": 1.5}) def test_default_slot_resolves_to_air(self) -> None: registry = default_amesim_gas_registry() medium = registry.resolve(AMESIM_DEFAULT_GAS_INDEX) self.assertIsInstance(medium, AmesimIdealAirMedium) self.assertEqual(medium.name, "AMESimAirIdealGas") self.assertEqual(medium.SUBSTANCE_ID, "air") self.assertEqual(medium.PROPERTY_METHOD_ID, "ideal_gas") self.assertEqual(medium.R_gas, 287.0) self.assertEqual(medium.cp_ref, 1005.0) self.assertEqual(medium.cv, 718.0) def test_gi_normalization_rejects_fractional_and_out_of_range_values(self) -> None: self.assertEqual(normalize_amesim_gas_index(0.0), 0) self.assertEqual(normalize_amesim_gas_index(1.0), 1) with self.assertRaisesRegex(ValueError, "integer"): normalize_amesim_gas_index(1.5) with self.assertRaisesRegex(ValueError, "between 0 and 99"): normalize_amesim_gas_index(-1) with self.assertRaisesRegex(ValueError, "between 0 and 99"): normalize_amesim_gas_index(100) with self.assertRaisesRegex(ValueError, "reserved"): normalize_amesim_defined_gas_index(0) self.assertEqual(normalize_amesim_defined_gas_index(1), 1) def test_unknown_slot_is_not_silently_replaced_with_air(self) -> None: registry = default_amesim_gas_registry() with self.assertRaisesRegex(ValueError, "gi=2.*not defined"): registry.resolve(2, component_name="pipe_1") project = ReactFlowProjectPayload( name="undefined-gas", nodes=[pneumatic_pipe_node("pipe_1", gi=2.0)], ) with self.assertRaisesRegex(ValueError, "gi=2.*pipe_1.*not defined"): compile_reactflow_network(project) def test_custom_registry_can_define_future_helium_slot(self) -> None: registry, air, future_helium = two_slot_registry() self.assertIs(registry.resolve(0), air) self.assertIs(registry.resolve(2), future_helium) self.assertEqual(registry.definitions[2].fluid_type, 12) self.assertEqual(registry.definitions[2].eos_type, 6) def test_builtin_air_slot_cannot_be_overridden(self) -> None: registry = default_amesim_gas_registry() with self.assertRaisesRegex(ValueError, "gi=0.*reserved"): registry.register( AmesimGasDefinition( gi=0, label="Invalid override", medium=IdealGasMedium(name="NotBuiltinAir"), ) ) self.assertEqual(registry.resolve(0).name, "AMESimAirIdealGas") def test_air_medium_definition_is_a_zero_port_compile_time_component(self) -> None: spec = COMPONENT_MODEL_REGISTRY["amesim_ideal_air_medium"] self.assertEqual(spec.display.category_id, "media") self.assertEqual(spec.display.role, "amesimGasMediumDefinition") self.assertEqual(spec.display.label, "空气介质定义") self.assertEqual(spec.ports, ()) self.assertEqual(spec.parameters[0].minimum, 1.0) property_model = spec.parameter_by_name["property_model"] self.assertIs(spec.parameters[1], property_model) self.assertEqual(property_model.default, 0.0) self.assertEqual(property_model.editor, "amesimGasPropertyModel") self.assertEqual( [(option.value, option.label) for option in property_model.options], [(0, "理想气体")], ) project = ReactFlowProjectPayload( name="project-air-slot", nodes=[ pneumatic_pipe_node("pipe_1", gi=1.0), ideal_air_medium_node("air_properties", gi=1.0), ], ) network = compile_reactflow_network(project) self.assertNotIn("air_properties", network.components) self.assertEqual( network.components["pipe_1"].medium.name, "AMESimAirIdealGas", ) def test_air_definition_dispatches_the_selected_property_model(self) -> None: spec = COMPONENT_MODEL_REGISTRY["amesim_ideal_air_medium"] definition = spec.create( "air_properties", IdealGasMedium(), {"gi": 3.0, "property_model": 0.0}, ) gas_definition = definition.gas_definition() self.assertEqual( definition.parameter_values, {"gi": 3.0, "property_model": 0.0}, ) self.assertEqual(gas_definition.label, "空气（理想气体）") self.assertEqual(gas_definition.fluid_type, 2) self.assertEqual(gas_definition.eos_type, 1) self.assertIsInstance(gas_definition.medium, AmesimIdealAirMedium) self.assertEqual( gas_definition.medium.PROPERTY_METHOD_ID, "ideal_gas", ) with self.assertRaisesRegex(ValueError, "property_model.*at most 0"): spec.create( "unsupported_air_properties", IdealGasMedium(), {"gi": 4.0, "property_model": 1.0}, ) def test_project_medium_definition_may_appear_after_its_references(self) -> None: project = ReactFlowProjectPayload( name="two-phase-medium-resolution", nodes=[ pneumatic_chamber_node("chamber_1", gi=7.0), ideal_air_medium_node("air_properties", gi=7.0), ], ) network = compile_reactflow_network(project) self.assertEqual( network.components["chamber_1"].medium.name, "AMESimAirIdealGas", ) def test_medium_definition_and_reference_survive_system_xml_round_trip( self, ) -> None: original = ReactFlowProjectPayload( name="medium-xml-round-trip", nodes=[ ideal_air_medium_node( "air_properties", gi=4.0, property_model=0.0, ), pneumatic_pipe_node("pipe_1", gi=4.0), ], ) xml_bytes = build_reactflow_system_xml(original) report = validate_system_xml_document(xml_bytes) self.assertTrue(report.valid, report.issues) self.assertIsNotNone(report.document) assert report.document is not None self.assertEqual(report.document.medium_reference_version, "1") components = { component.id: component for component in report.document.components } self.assertEqual( components["air_properties"].model_type, "amesim_ideal_air_medium", ) self.assertEqual( {item.name: item.value for item in components["air_properties"].parameters}, {"gi": 4.0, "property_model": 0.0}, ) self.assertEqual( {item.name: item.value for item in components["pipe_1"].parameters}[ "gi" ], 4.0, ) parsed_project = ReactFlowProjectPayload( **report.document.as_project_data() ) network = compile_reactflow_network(parsed_project) self.assertNotIn("air_properties", network.components) self.assertIsInstance( network.components["pipe_1"].medium, AmesimIdealAirMedium, ) def test_duplicate_project_medium_indices_are_rejected(self) -> None: project = ReactFlowProjectPayload( name="duplicate-medium-index", nodes=[ ideal_air_medium_node("air_properties_1", gi=1.0), ideal_air_medium_node("air_properties_2", gi=1.0), ], ) with self.assertRaisesRegex( ValueError, "air_properties_2.*gi=1.*already defined", ): compile_reactflow_network(project) def test_medium_definition_cannot_claim_builtin_slot_zero(self) -> None: project = ReactFlowProjectPayload( name="reserved-medium-index", nodes=[ideal_air_medium_node("air_properties", gi=0.0)], ) with self.assertRaisesRegex(ValueError, "gi.*at least 1"): compile_reactflow_network(project) def test_default_compile_uses_air_before_state_initialization(self) -> None: project = ReactFlowProjectPayload( name="default-air", nodes=[pneumatic_chamber_node("chamber_1")], ) network = compile_reactflow_network(project) chamber = network.components["chamber_1"] self.assertEqual(chamber.medium.name, "AMESimAirIdealGas") self.assertAlmostEqual( chamber.state.m, chamber.medium.density(200000.0, 300.0) * 0.01, ) def test_gi_is_resolved_before_dynamic_state_initialization(self) -> None: registry, _, future_helium = two_slot_registry() project = ReactFlowProjectPayload( name="selected-medium-initialization", nodes=[pneumatic_chamber_node("chamber_1", gi=2.0)], ) network = compile_reactflow_network( project, amesim_gas_registry=registry, ) chamber = network.components["chamber_1"] self.assertIs(chamber.medium, future_helium) self.assertAlmostEqual( chamber.state.m, future_helium.density(200000.0, 300.0) * 0.01, ) self.assertNotAlmostEqual( chamber.state.m, IdealGasMedium().density(200000.0, 300.0) * 0.01, ) def test_selected_medium_is_injected_into_the_whole_pneumatic_circuit( self, ) -> None: registry, _, future_helium = two_slot_registry() project = ReactFlowProjectPayload( name="future-helium-circuit", nodes=[ pneumatic_pipe_node("pnl_1", gi=2.0), component_node( "tank_1", "tank", [physical_port("port_a", "inlet", "left")], {"volume": 0.1, "p0": 100000.0, "T0": 300.0}, ), ], edges=[ physical_edge("edge-1", "pnl_1", "port_2", "tank_1", "port_a"), ], ) network = compile_reactflow_network( project, amesim_gas_registry=registry, ) self.assertIs(network.components["pnl_1"].medium, future_helium) self.assertIs(network.components["tank_1"].medium, future_helium) def test_connected_circuit_rejects_conflicting_gas_indices(self) -> None: registry, _, _ = two_slot_registry() project = ReactFlowProjectPayload( name="conflicting-gases", nodes=[ pneumatic_pipe_node("air_pipe", gi=0.0), pneumatic_pipe_node("helium_pipe", gi=2.0), ], edges=[ physical_edge( "edge-1", "air_pipe", "port_2", "helium_pipe", "port_1", ), ], ) with self.assertRaisesRegex( ValueError, "conflicting AMESim gas definitions", ): compile_reactflow_network( project, amesim_gas_registry=registry, ) def test_disconnected_circuits_may_use_different_gas_indices(self) -> None: registry, air, future_helium = two_slot_registry() project = ReactFlowProjectPayload( name="separate-gases", nodes=[ pneumatic_pipe_node("air_pipe", gi=0.0), pneumatic_pipe_node("helium_pipe", gi=2.0), ], ) network = compile_reactflow_network( project, amesim_gas_registry=registry, ) self.assertIs(network.components["air_pipe"].medium, air) self.assertIs(network.components["helium_pipe"].medium, future_helium) if __name__ == "__main__": unittest.main()