157 lines
8.1 KiB
Python
157 lines
8.1 KiB
Python
"""Supply errors are rejected at UI-independent model and native entry points."""
|
|
import unittest
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from app.main import ReactFlowProjectPayload, build_reactflow_system_xml, compile_reactflow_network
|
|
from app.simulation.core.port_computation import (
|
|
PortSupplyError, port_supply_issue, reference_supply_issues,
|
|
)
|
|
from app.simulation.native_codegen.compiler import compile_native_program
|
|
from app.simulation.native_codegen.extended import compile_extended_program
|
|
from app.simulation.registry import COMPONENT_MODEL_REGISTRY, build_component_catalog
|
|
from app.simulation.core.medium import IdealGasMedium
|
|
from app.simulation.systems.network import Connection, Endpoint, SimulationNetwork
|
|
from app.system_xml import validate_system_xml_document
|
|
from tests.test_amesim_pneumatic_node_xml import amesim_pn3node_project
|
|
|
|
|
|
class Circuit:
|
|
def __init__(self):
|
|
self.net = SimulationNetwork('port_supply')
|
|
|
|
def add(self, model, name):
|
|
spec = COMPONENT_MODEL_REGISTRY[model]
|
|
component = spec.create(name, IdealGasMedium(), {p.name: p.default for p in spec.parameters})
|
|
self.net.add_component(component)
|
|
return component
|
|
|
|
def chamber(self, name):
|
|
return self.add('amesim_pnch023', name)
|
|
|
|
def connect(self, first, first_port, second, second_port):
|
|
self.net.connect(first.name, first_port, second.name, second_port)
|
|
|
|
|
|
def port(model, name):
|
|
return next(p for p in COMPONENT_MODEL_REGISTRY[model].ports if p.name == name)
|
|
|
|
|
|
class PortComputationTests(unittest.TestCase):
|
|
def test_catalog_declares_both_sides_without_changing_flow_or_state_contract(self):
|
|
for spec in COMPONENT_MODEL_REGISTRY.values():
|
|
for p in spec.ports:
|
|
if p.domain == 'pneumatic':
|
|
self.assertIsNotNone(p.computation, spec.model_type)
|
|
self.assertEqual([v.name for v in p.variables],
|
|
['p', 'm_flow', 'h_outflow', 'volume', 'volume_flow'])
|
|
reference = port('amesim_p4node2', 'port_2')
|
|
self.assertEqual(reference.computation.inputs, ('p', 'T'))
|
|
self.assertEqual(reference.computation.outputs, ('m_flow', 'H_flow'))
|
|
self.assertEqual(reference.nominal_role, 'bidirectional')
|
|
self.assertEqual(port('amesim_p4node2', 'port_1').computation.reference_port, 'port_2')
|
|
catalog = build_component_catalog()
|
|
declared = next(c for lib in catalog['libraries'] for c in lib['components']
|
|
if c['modelType'] == 'amesim_p4node2')['ports']
|
|
self.assertEqual(declared[1]['computation'], reference.computation.as_dict())
|
|
|
|
def test_reference_accepts_storage_and_branch_accepts_flow_in_both_edge_orders(self):
|
|
for node in ('amesim_pn3node2', 'amesim_p4node2'):
|
|
for model, name in [('amesim_pnl0001', 'port_2'), ('amesim_pnl0003', 'port_1'),
|
|
('amesim_pnch023', 'port_1'), ('tank', 'port_a')]:
|
|
a, b = port(node, 'port_2'), port(model, name)
|
|
self.assertIsNone(port_supply_issue(a, b))
|
|
self.assertIsNone(port_supply_issue(b, a))
|
|
for model, name in [('amesim_pnvo001', 'port_2'), ('amesim_pnl0001', 'port_1'),
|
|
('amesim_pnl0002', 'port_2'), ('amesim_pnpl01', 'port_1')]:
|
|
a, b = port(node, 'port_1'), port(model, name)
|
|
self.assertIsNone(port_supply_issue(a, b))
|
|
self.assertIsNone(port_supply_issue(b, a))
|
|
|
|
def test_wrong_reference_and_wrong_branch_report_specific_missing_variables(self):
|
|
for node in ('amesim_pn3node2', 'amesim_p4node2'):
|
|
a, b = port(node, 'port_2'), port('amesim_pnvo001', 'port_2')
|
|
for first, second in [(a, b), (b, a)]:
|
|
issue = port_supply_issue(first, second, 'left.port_2', 'right.port_2')
|
|
self.assertEqual(issue.code, 'CONNECTION_VARIABLE_SUPPLY_MISSING')
|
|
self.assertIn('温度', issue.message)
|
|
self.assertIn('压力', issue.message)
|
|
self.assertIn('left.port_2', issue.message)
|
|
issue = port_supply_issue(port(node, 'port_1'), port('amesim_pnl0003', 'port_1'))
|
|
self.assertIn('质量流率', issue.message)
|
|
self.assertIn('能量流率', issue.message)
|
|
self.assertIsNotNone(port_supply_issue(a, port('amesim_pnpl01', 'port_1')))
|
|
|
|
def test_equation_connections_are_not_mistaken_for_fixed_reference_errors(self):
|
|
for a, b in [(port('amesim_pnl00r', 'port_1'), port('amesim_pnor001', 'port_1')),
|
|
(port('amesim_pnl0003', 'port_2'), port('amesim_pnl0001', 'port_2')),
|
|
(port('tee', 'port_in'), port('orifice', 'port_b'))]:
|
|
self.assertIsNone(port_supply_issue(a, b))
|
|
|
|
def test_saved_port_metadata_cannot_override_registered_supply(self):
|
|
payload = amesim_pn3node_project().model_dump()
|
|
for edge in payload['edges']:
|
|
if edge['target'] == 'node_1':
|
|
edge['targetHandle'] = 'port_2'
|
|
if edge['source'] == 'node_1' and edge['sourceHandle'] == 'port_2':
|
|
edge['sourceHandle'] = 'port_1'
|
|
for node in payload['nodes']:
|
|
for p in node['data']['ports']:
|
|
p['computation'] = {'mode': 'equation', 'inputs': [], 'outputs': ['p', 'T']}
|
|
project = ReactFlowProjectPayload.model_validate(payload)
|
|
for entry in (build_reactflow_system_xml, compile_reactflow_network):
|
|
with self.assertRaisesRegex(ValueError, 'CONNECTION_VARIABLE_SUPPLY_MISSING'):
|
|
entry(project)
|
|
|
|
def test_xml_validation_reports_the_edge_before_native_execution(self):
|
|
root = ET.fromstring(build_reactflow_system_xml(amesim_pn3node_project()))
|
|
for endpoint in root.findall('./Connections/Connection/Endpoint'):
|
|
if endpoint.get('component') == 'node_1':
|
|
if endpoint.get('port') in ('port_1', 'port_2'):
|
|
endpoint.set('port', 'port_2' if endpoint.get('port') == 'port_1' else 'port_1')
|
|
report = validate_system_xml_document(ET.tostring(root))
|
|
self.assertFalse(report.valid)
|
|
errors = [i for i in report.issues if i.code == 'CONNECTION_VARIABLE_SUPPLY_MISSING']
|
|
self.assertEqual(len(errors), 2)
|
|
self.assertTrue(all('Connection[' in i.path for i in errors))
|
|
|
|
def test_native_compilers_recheck_manually_inserted_connections(self):
|
|
b = Circuit()
|
|
b.add('amesim_p4node2', 'node')
|
|
b.add('amesim_pnvo001', 'valve')
|
|
b.net.connections.append(Connection('bad', 'physical', 'pneumatic',
|
|
Endpoint('node', 'port_2'), Endpoint('valve', 'port_2')))
|
|
for compiler in (compile_native_program, compile_extended_program):
|
|
with self.assertRaisesRegex(PortSupplyError, 'node.port_2.*温度'):
|
|
compiler(b.net)
|
|
|
|
def test_reference_chains_need_a_real_origin(self):
|
|
b = Circuit()
|
|
a = b.add('amesim_pn3node2', 'a')
|
|
c = b.add('amesim_p4node2', 'b')
|
|
tank = b.chamber('tank')
|
|
b.connect(a, 'port_2', c, 'port_1')
|
|
b.connect(c, 'port_2', tank, 'port_1')
|
|
b.net.validate_port_supplies()
|
|
b.net.connections.pop()
|
|
with self.assertRaisesRegex(PortSupplyError, 'REFERENCE_SUPPLY_UNCONNECTED'):
|
|
b.net.validate_port_supplies()
|
|
b.connect(c, 'port_2', a, 'port_1')
|
|
with self.assertRaisesRegex(PortSupplyError, 'REFERENCE_SUPPLY_CYCLE'):
|
|
b.net.validate_port_supplies()
|
|
|
|
def test_deep_reference_chain_does_not_require_python_recursion(self):
|
|
reference = port('amesim_pn3node2', 'port_2')
|
|
branch = port('amesim_pn3node2', 'port_1')
|
|
ports, adjacency = {}, {}
|
|
# Longer than Python's default recursion limit; resolved paths are reused.
|
|
for i in range(1100):
|
|
ports[str(i), 'port_1'] = branch
|
|
ports[str(i), 'port_2'] = reference
|
|
adjacency[str(i), 'port_2'] = (str(i+1), 'port_1')
|
|
ports['1100', 'port_1'] = port('tank', 'port_a')
|
|
self.assertEqual(reference_supply_issues(ports, adjacency), [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|