"""Rehearse registration in an isolated source tree; never register the demo in production. Run with the repository Python environment and a fresh --output-dir. The final sandbox can be served with uvicorn for a real browser check. No API is mocked. """ from __future__ import annotations import argparse from dataclasses import replace from hashlib import sha256 import json import os from pathlib import Path import shutil import subprocess import sys ROOT = Path(__file__).resolve().parents[2] MODEL = '''from app.simulation.core.base import AlgebraicComponent from app.simulation.core.catalog import ComponentDisplaySpec, PortDisplaySpec from app.simulation.core.metadata import ParameterDefinition, ResultVariableDefinition from app.simulation.core.ports import PortDefinition class DemoRamp(AlgebraicComponent): MODEL_TYPE = 'registration_demo_ramp' MODEL_VERSION = '1.0.0' PORTS = (PortDefinition.signal('out', nominal_role='output'),) PARAMETERS = ( ParameterDefinition('offset', 2.0, label='初始值'), ParameterDefinition('amplitude', 3.0, label='幅值'), ParameterDefinition('duration', 1.0, label='时长', quantity='time', unit='s', minimum=0.0, minimum_exclusive=True), ) RESULT_VARIABLES = (ResultVariableDefinition('y', '输出', 'dimensionless', '', 'signal', 10),) DISPLAY = ComponentDisplaySpec(label='注册演练斜坡信号', library_id='registration_demo', category_id='signals', symbol='amesim_step0', ports=(PortDisplaySpec('out', 'right', order=10),), order=10) EQUATIONS = () def __init__(self, name, medium, *, offset, amplitude, duration): super().__init__(name) self.set_parameter_values(dict(offset=offset, amplitude=amplitude, duration=duration)) self.offset, self.amplitude, self.duration = offset, amplitude, duration self.out = self.register_declared_port('out') @classmethod def create(cls, *, name, medium, parameters): return cls(name, medium, **parameters) ''' LIBRARY = '''from app.simulation.core.catalog import ComponentLibrarySpec, ComponentCategorySpec LIBRARY = ComponentLibrarySpec(id='registration_demo', label='组件注册演练库', version='1.0.0', source_package='app.simulation.components.registration_demo', categories=(ComponentCategorySpec(id='signals', label='信号元件', order=10),), models=('app.simulation.components.registration_demo.signals.ramp:DemoRamp',), order=300) ''' STAGES = ['unlisted', 'catalog_only', 'native_discovery_only', 'contract_only', 'lowered_without_module_export', 'complete'] def write(root, name, value): path = root / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(value, encoding='utf-8', newline='\n') def edit(root, name, before, after): path = root / name value = path.read_text(encoding='utf-8') if value.count(before) != 1: raise RuntimeError(f'Rehearsal source anchor changed: {name}: {before!r}') write(root, name, value.replace(before, after)) def worker(sandbox, output, stage): sys.path.insert(0, str(sandbox)) from app.simulation.registry import COMPONENT_MODEL_REGISTRY, build_component_catalog from app.simulation.native_codegen.compiler import compile_native_program from app.simulation.native_codegen.build import build_native from app.simulation.native_codegen.input import project_xml from app.simulation.native_codegen.runner import execute_native from app.simulation.backends import simulation_config from app.main import compile_system_xml_network from app.system_xml import validate_system_xml_document record = {'stage': stage, 'registered': 'registration_demo_ramp' in COMPONENT_MODEL_REGISTRY} if stage == 'unlisted': assert not record['registered'] return record catalog = build_component_catalog() model = next(c for lib in catalog['libraries'] for c in lib['components'] if c['modelType'] == 'registration_demo_ramp') record['catalog'] = model project = {'projectSchemaVersion': 1, 'name': 'registration-demo', 'simulation': {'t_start': 0, 't_stop': 1, 'step': .1, 'max_step': .05, 'method': 'BDF'}, 'edges': [], 'nodes': [{'id': 'ramp_1', 'type': 'simulationComponent', 'position': {'x': 200, 'y': 200}, 'data': {'label': '注册演练斜坡信号', 'componentType': model['type'], 'modelType': model['modelType'], 'modelVersion': model['modelVersion'], 'symbol': model['symbol'], # Project snapshots use the browser's normalized port shape, not raw catalog objects. 'ports': [{k: v for k, v in port.items() if k in ('name', 'kind', 'domain', 'nominalRole', 'side') or (k == 'positiveFlowDirection' and v == 'intoComponent')} for port in model['ports']], 'parameters': {p['name']: p['default'] for p in model['parameters']}, 'parameterUnits': {'duration': 's'}, 'rotation': 0, 'mirrored': False}}]} xml = project_xml(project) report = validate_system_xml_document(xml) assert report.valid, report.as_dict() write(output, 'project.json', json.dumps(project, ensure_ascii=False, indent=2)) write(output, 'catalog.json', json.dumps(catalog, ensure_ascii=False, indent=2)) (output / 'input.xml').write_bytes(xml) # The browser requires all displayed ports to be connected, unlike the # backend's warning-only standalone signal fixture. import copy browser_project = copy.deepcopy(project) for index, (kind, name) in enumerate((('amesim_forc', 'force_1'), ('amesim_mecmas21', 'mass_1'), ('amesim_f000', 'zero_1')), 1): item = next(c for lib in catalog['libraries'] for c in lib['components'] if c['modelType'] == kind) node = copy.deepcopy(project['nodes'][0]) node.update(id=name, position={'x': 200 + index * 220, 'y': 200}) node['data'].update(label=name, componentType=kind, modelType=kind, modelVersion=item['modelVersion'], symbol=item['symbol'], ports=[{k: v for k, v in port.items() if k in ('name', 'kind', 'domain', 'nominalRole', 'side') or (k == 'positiveFlowDirection' and v == 'intoComponent')} for port in item['ports']], parameters={param['name']: param['default'] for param in item['parameters']}, parameterUnits={}) if kind == 'amesim_mecmas21': node['data']['parameters'].update(stoptype=4, useFriction=1, mass=10, x0=0, v0=0) browser_project['nodes'].append(node) for index, (a, pa, b, pb) in enumerate((('ramp_1', 'out', 'force_1', 'res'), ('force_1', 'port_2', 'mass_1', 'port_2'), ('mass_1', 'port_1', 'zero_1', 'port_1'))): browser_project['edges'].append(dict(id=f'edge_{index}', source=a, sourceHandle=pa, target=b, targetHandle=pb, data={'isContactEdge': False})) browser_report = validate_system_xml_document(project_xml(browser_project)) assert browser_report.valid, browser_report.as_dict() write(output, 'browser-project.json', json.dumps(browser_project, ensure_ascii=False, indent=2)) network = compile_system_xml_network(report.document) try: program = compile_native_program(network) record['generated'] = True build = build_native(program, cache_dir=output / 'cache') except (ValueError, RuntimeError) as exc: record['error'] = str(exc) if stage == 'complete': raise expected = {'catalog_only': 'no native contract', 'native_discovery_only': 'version does not match', 'contract_only': 'output mapping incomplete', 'lowered_without_module_export': 'undefined reference'}[stage] assert expected.lower() in str(exc).lower(), str(exc) return record try: assert stage == 'complete' assert not build.cache_hit assert 'demo_signal' in build.details['selectedModules'] record['firstCompleteBuild'] = build.details # Earlier failed link may warm shared objects. record['runs'] = [] for method in ('RK45', 'BDF'): config = replace(simulation_config(report.document.simulation), method=method) result = execute_native(build, config, .1, run_dir=output / method) assert result['success'] and result['simulatedUntil'] == 1 errors = [abs(y - (2 + 3*t)) for t, y in zip(result['series']['time'], result['series']['ramp_1.y'])] assert max(errors) < 1e-12 assert result['series']['ramp_1.y'] == result['series']['ramp_1.out.signal'] record['runs'].append({'method': method, 'samples': len(errors), 'maxAnalyticError': max(errors)}) cached = build_native(program, cache_dir=output / 'cache') try: assert cached.cache_hit and cached.details['objectCompilations'] == 0 record['warmBuild'] = cached.details finally: cached.close() project['nodes'][0]['data']['parameters']['duration'] = 2 changed_report = validate_system_xml_document(project_xml(project)) changed = compile_native_program(compile_system_xml_network(changed_report.document)) new_build = build_native(changed, cache_dir=output / 'cache') try: assert new_build.details['objectCompilations'] == 1 assert new_build.details['objectCacheHits'] > 0 result = execute_native(new_build, config, .1, run_dir=output / 'parameter-change') assert result['success'] and abs(result['final']['ramp_1.y'] - 3.5) < 1e-12 record['parameterBuild'] = new_build.details finally: new_build.close() invalid = {'wrong_version': xml.replace(b'modelVersion="1.0.0"', b'modelVersion="9.0.0"')} # Build semantic mutations from the parsed tree; serializer whitespace # and float spellings are not an input contract. import xml.etree.ElementTree as ET tree = ET.fromstring(xml) comp = tree.find('./Components/Component') duration = comp.find("Parameter[@name='duration']") duration.set('value', '0') invalid['zero_duration'] = ET.tostring(tree) comp.remove(duration) invalid['missing_parameter'] = ET.tostring(tree) record['rejections'] = {} for name, payload in invalid.items(): invalid_report = validate_system_xml_document(payload) assert not invalid_report.valid, name record['rejections'][name] = invalid_report.as_dict() record['sourceSha256'] = sha256(program.source.encode()).hexdigest() return record finally: build.close() def prepare(output): output.mkdir(parents=True, exist_ok=False) sandbox = output / 'sandbox' for name in ('app', 'native', 'schemas'): shutil.copytree(ROOT / name, sandbox / name, ignore=shutil.ignore_patterns('__pycache__', '*.pyc', 'data')) if (ROOT / 'frontend/dist').exists(): shutil.copytree(ROOT / 'frontend/dist', sandbox / 'frontend/dist') write(sandbox, 'app/simulation/components/registration_demo/__init__.py', '') write(sandbox, 'app/simulation/components/registration_demo/signals/__init__.py', '') write(sandbox, 'app/simulation/components/registration_demo/signals/ramp.py', MODEL) write(sandbox, 'app/simulation/components/registration_demo/library.py', LIBRARY) rows = [] for stage in STAGES: if stage == 'catalog_only': edit(sandbox, 'app/simulation/registry.py', 'ENABLED_COMPONENT_LIBRARIES = (', 'ENABLED_COMPONENT_LIBRARIES = (\n "app.simulation.components.registration_demo.library:LIBRARY",') elif stage == 'native_discovery_only': edit(sandbox, 'app/simulation/native_codegen/extended.py', 'for entry in (*a.models, *e.models):', 'from app.simulation.components.registration_demo.library import LIBRARY as demo\n for entry in (*a.models, *e.models, *demo.models):') elif stage == 'contract_only': edit(sandbox, 'app/simulation/native_codegen/contracts.py', 'SUPPORTED_VERSIONS = {', "SUPPORTED_VERSIONS = {\n 'registration_demo_ramp': '1.0.0',") elif stage == 'lowered_without_module_export': edit(sandbox, 'app/simulation/native_codegen/extended.py', " elif c.model_type == 'amesim_ud00':", " elif c.model_type == 'registration_demo_ramp':\n put(c, 'y', f'native_demo_ramp(t,{num(c.offset)},{num(c.amplitude)},{num(c.duration)})')\n elif c.model_type == 'amesim_ud00':") edit(sandbox, 'native/include/kernels.h', '#include ', '#include \ndouble native_demo_ramp(double t, double offset, double amplitude, double duration);') write(sandbox, 'native/components/modules/demo_signal.c', '#include "kernels.h"\ndouble native_demo_ramp(double t, double offset, double amplitude, double duration) { return offset + amplitude * t / duration; }\n') elif stage == 'complete': edit(sandbox, 'app/simulation/native_codegen/modules.py', 'EXPORTS = {', "EXPORTS = {\n 'demo_signal': frozenset({'native_demo_ramp'}),") edit(sandbox, 'native/components/kernels.c', '#define NATIVE_COMPONENT_AMALGAMATION 1', '#define NATIVE_COMPONENT_AMALGAMATION 1\n#include "../components/modules/demo_signal.c"') edit(sandbox, 'app/simulation/native_codegen/jacobian.py', '_FUNCTIONS = frozenset({', "_FUNCTIONS = frozenset({\n 'native_demo_ramp',") record_path = output / f'{stage}.json' proc = subprocess.run([sys.executable, '-B', str(Path(__file__).resolve()), '--worker', stage, '--sandbox', str(sandbox), '--output-dir', str(output)], capture_output=True, text=True, encoding='utf-8', timeout=180) write(output, f'{stage}.log', proc.stdout + proc.stderr) if proc.returncode: raise RuntimeError(f'{stage} failed; see {output / (stage + ".log")}') rows.append(json.loads(record_path.read_text(encoding='utf-8'))) summary = {'baselineHead': subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=ROOT, text=True).strip(), 'platform': sys.platform, 'windowsExecuted': os.name == 'nt', 'stages': rows, 'exampleScope': 'Stateless signal source, third library, new C module, existing symbol and parameter controls. No new physical domain, dynamic ports or nontrivial Jacobian coloring.', 'sandbox': str(sandbox), 'productionModelRegistered': False} write(output, 'summary.json', json.dumps(summary, ensure_ascii=False, indent=2)) print(json.dumps({'output': str(output), 'stages': len(rows), 'complete': True})) if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--output-dir', type=Path, required=True) parser.add_argument('--worker', choices=STAGES) parser.add_argument('--sandbox', type=Path) args = parser.parse_args() out = args.output_dir.resolve() if args.worker: result = worker(args.sandbox.resolve(), out, args.worker) write(out, f'{args.worker}.json', json.dumps(result, ensure_ascii=False, indent=2)) else: prepare(out)