"""Independent stage/one-ULP checks and isolated cyclic UD00 execution.""" from __future__ import annotations from bisect import bisect_right import math from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch from app.simulation.config import SolveIVPConfig from app.simulation.native_codegen import build as builder, result_storage from app.simulation.native_codegen.compiler import compile_native_program from app.simulation.native_codegen.runner import execute_native from tests.test_native_catalog import Circuit ROOT = Path(__file__).resolve().parents[1] HARNESS = r''' #include "kernels.h" #include #include int main(void) { char mode; double t,start,end,data[24]; int stages,cyclic,count; while(scanf(" %c %lf %lf %lf %d %d %d",&mode,&t,&start,&end,&stages,&cyclic,&count)==7) { for(int i=0;i<24;i++) if(scanf("%lf",&data[i])!=1) return 2; if(mode=='V') { printf("%.17g %.17g\n",native_signal(t,start,stages,cyclic,data), native_signal_break(t,end,start,stages,cyclic,data)); } else { for(int i=0;it)) return 3; t=b; } } } return 0; } ''' def stage_data(starts, ends, durations): return [*starts, *([0.] * (8-len(starts))), *ends, *([0.] * (8-len(ends))), *durations, *([0.] * (8-len(durations)))] class NativeSignalBoundaryTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.tmp = tempfile.TemporaryDirectory(prefix='native-signal-boundaries-') cls.addClassCleanup(cls.tmp.cleanup) cls.root = Path(cls.tmp.name) compiler, _, _ = builder.toolchain() source = cls.root / 'probe.c' source.write_text(HARNESS, encoding='ascii') cls.exe = cls.root / ('probe.exe' if sys.platform == 'win32' else 'probe') builder._command([compiler, *builder.COMPILER_FLAGS, '-I', str(ROOT / 'native/include'), str(source), str(ROOT / 'native/components/modules/signal.c'), '-lm', '-o', str(cls.exe)], log=[]) def probe(self, data, stages, queries, *, start=0., cyclic=True, end=1e12, events=0): lines = [' '.join(map(str, ['E' if events else 'V', t, start, end, stages, int(cyclic), events or 1, *data])) for t in queries] run = subprocess.run([str(self.exe)], input='\n'.join(lines)+'\n', capture_output=True, text=True, timeout=15, check=True) rows = [list(map(float, line.split())) for line in run.stdout.splitlines()] self.assertEqual(len(rows), len(queries) * (events or 1)) return rows def test_mql_cyclic_boundaries_and_adjacent_floats(self): # Freeze the known event times independently of signal evaluation. boundaries = [.8, 10.8, 11.600000000000001, 21.6, 22.400000000000002, 32.400000000000006, 33.2, 43.2, 44., 54.] data = stage_data([1e17, 49000.], [1e17, 49000.], [.8, 10.]) queries = [t for boundary in boundaries for t in (math.nextafter(boundary, -math.inf), boundary, math.nextafter(boundary, math.inf))] rows = self.probe(data, 2, [*queries, 32.4]) for i, boundary in enumerate(boundaries): before, after = (1e17, 49000.) if i % 2 == 0 else (49000., 1e17) left, at, right = rows[3*i:3*i+3] with self.subTest(boundary=boundary): self.assertEqual([left[0], at[0], right[0]], [before, after, after]) self.assertEqual(left[1], boundary) self.assertGreater(at[1], boundary) if i+1 < len(boundaries): self.assertEqual(at[1], boundaries[i+1]) # A genuinely earlier sample must not be pulled onto the event. self.assertEqual(rows[-1][0], 49000.) def test_multistage_long_cycle_sequence_has_no_skipped_or_duplicate_events(self): data = stage_data(list(range(1, 9)), list(range(1, 9)), [.1, .2, .3, .4, .5, .6, .7, .8]) rows = self.probe(data, 8, [0.], events=8000) last = 0. for i, (boundary, left, at, right) in enumerate(rows): old, new = i % 8 + 1, (i+1) % 8 + 1 self.assertGreater(boundary, last, (i, boundary)) self.assertEqual((left, at, right), (old, new, new), (i, boundary)) last = boundary self.assertAlmostEqual(last, 3600., places=9) def test_offset_start_and_small_period_boundaries(self): for start, scale in ((.37, 1.), (-1.25, 1.), (1e6, 1e-5)): data = stage_data([1., 2., 3.], [1., 2., 3.], [.1*scale, .2*scale, .3*scale]) rows = self.probe(data, 3, [start], start=start, events=900) for i, (boundary, left, at, right) in enumerate(rows): with self.subTest(start=start, scale=scale, event=i): self.assertEqual((left, at, right), (i % 3 + 1, (i+1) % 3 + 1, (i+1) % 3 + 1), boundary) def test_ramps_delayed_start_and_noncyclic_final_value(self): data = stage_data([0., 10., -10.], [2., 12., -7.], [.25, .5, .75]) times = [0., 1.25, 1.375, 1.5, 1.75, 2., 2.375, 2.75, 2.875, 3.] expected = [0., 0., 1., 10., 11., -10., -8.5, 0., 1., 10.] actual = self.probe(data, 3, times, start=1.25) self.assertEqual([row[0] for row in actual], expected) once = self.probe(data, 3, [2.75, 5.], start=1.25, cyclic=False, end=10.) self.assertEqual(once, [[-7., 10.], [-7., 10.]]) # A simulation beginning after the signal's start can enter any cycle. shifted = self.probe(data, 3, [-2., -1.875, -1.75, -.5], start=-2.) self.assertEqual([row[0] for row in shifted], [0., 1., 10., 0.]) def test_zero_duration_stages_and_zero_total_duration_keep_existing_semantics(self): data = stage_data([-100., 10., -300., 20., -500.], [-100., 10., -300., 20., -500.], [0., .5, 0., .5, 0.]) rows = self.probe(data, 5, [-1., 0., .5, 1., 1.5], end=10.) self.assertEqual(rows, [[-100., 0.], [10., .5], [20., 1.], [10., 1.5], [20., 2.]]) self.assertEqual(self.probe(data, 5, [1.], cyclic=False, end=10.), [[-500., 10.]]) zero = stage_data([2., 9.], [4., 11.], [0., 0.]) self.assertEqual(self.probe(zero, 2, [-1., 0., 50.], end=100.), [[2., 0.], [11., 100.], [11., 100.]]) # A trailing zero stage must not create a second, differently rounded # representation of the cycle endpoint or leak its instantaneous value. fractional = stage_data([-100., 10., -300., 20., -500.], [-100., 10., -300., 20., -500.], [0., .8, 0., 10., 0.]) for i, (boundary, left, at, right) in enumerate(self.probe(fractional, 5, [0.], events=2000)): old, new = (10., 20.) if i % 2 == 0 else (20., 10.) self.assertEqual((left, at, right), (old, new, new), (i, boundary)) def test_single_stage_sawtooth_and_queries_out_of_order(self): data = stage_data([2.], [6.], [2.]) times = [50., 2., -1., 1., 3., 4., 1.] expected = [2., 2., 2., 4., 4., 2., 4.] self.assertEqual([row[0] for row in self.probe(data, 1, times)], expected) def test_isolated_cyclic_component_runs_both_solvers_and_replays_boundaries(self): b = Circuit() b.add('amesim_ud00', 'signal', start1=1e17, end1=1e17, t1=.8, start2=49000., end2=49000., t2=10., nstages=2, iscyclic=1) with patch.object(builder, 'CACHE', self.root / 'build-cache'), \ patch.object(result_storage, 'RESULT_ROOT', self.root / 'results'): build = builder.build_native(compile_native_program(b.net)) self.addCleanup(build.close) outputs = [] for method in ('BDF', 'RK45'): result = execute_native(build, SolveIVPConfig(t_stop=60., method=method, max_step=1e30, rtol=1e-8), .01, run_dir=self.root / method) self.assertTrue(result['success'], result['message']) self.assertEqual(result['simulatedUntil'], 60.) series = result['series'] self.assertGreater(len(series['time']), 6001) self.assertTrue(set(i*.01 for i in range(6001)).issubset(series['time'])) self.assertTrue(all(a < b for a, b in zip(series['time'], series['time'][1:]))) values = series['signal.out.signal'] self.assertEqual(set(values), {1e17, 49000.}) self.assertEqual(values[series['time'].index(44.)], 49000.) self.assertEqual(values[series['time'].index(54.)], 1e17) known_boundaries = [.8, 10.8, 11.600000000000001, 21.6, 22.400000000000002, 32.400000000000006, 33.2, 43.2, 44., 54., 54.8] # A correct replay alone would not prove that integration # stopped and restarted at each of the eleven time events. self.assertEqual(result['solverStarts'], len(known_boundaries) + 1) expected = [49000. if bisect_right(known_boundaries, t) % 2 else 1e17 for t in series['time']] self.assertEqual(values, expected) for boundary in known_boundaries: self.assertIn(boundary, series['time']) self.assertIn(math.nextafter(boundary, -math.inf), series['time']) self.assertEqual(result['outputSemantics']['version'], 2) outputs.append(values) self.assertEqual(*outputs) if __name__ == '__main__': unittest.main()