"""CPU contracts for Fig. 2 acquisition; no CUDA or simulation import required.""" import importlib.util import json from pathlib import Path import sys import tempfile import types import unittest from unittest.mock import patch import numpy as np ROOT = Path(__file__).resolve().parents[2] SPEC = importlib.util.spec_from_file_location('fig2_collection_tested', ROOT/'research/fig2_collection/run.py') driver = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(driver) METRIC_SPEC = importlib.util.spec_from_file_location('fig2_collection_metrics', ROOT/'research/fig2_materials/metrics.py') metrics = importlib.util.module_from_spec(METRIC_SPEC) METRIC_SPEC.loader.exec_module(metrics) class CollectionContracts(unittest.TestCase): def test_frozen_clocks_and_windows(self): for scene, release, count, first in [('karman', 4576000, 400, 240), ('steady', 2048000, 400, 240), ('taylor', 1024000, 150, 0)]: plan = driver.schedule(scene) self.assertEqual((plan['release_clock'], plan['intervals'], plan['field_start']), (release, count, first)) self.assertEqual(plan['si'], 3200) self.assertEqual((400-240)*3200*driver.U/driver.D, 64.) self.assertEqual(400*3200*driver.U/driver.D, 160.) self.assertEqual(150*3200*driver.U/driver.D, 60.) def test_smoke_has_no_physical_timing(self): for scene in ('karman', 'steady', 'taylor'): plan = driver.schedule(scene, True) self.assertEqual((plan['si'], plan['intervals'], plan['field_start']), (32, 3, 0)) self.assertEqual(plan['release_clock'], 64 if scene == 'taylor' else 32) def test_zero_is_exact_physical_zero_despite_bias(self): for scene in ('karman', 'steady', 'taylor'): np.testing.assert_array_equal(driver.wall_surface(scene, 'zero'), np.zeros(3)) np.testing.assert_array_equal(driver.wall_surface(scene, 'target'), np.zeros(3)) np.testing.assert_allclose(driver.wall_surface('karman', 'controlled', [.5, -.5, 1]), [.02, -.04, .06]) np.testing.assert_allclose(driver.wall_surface('taylor', 'controlled', [.5, -.5, 1]), [.01, -.03, .04]) np.testing.assert_allclose(driver.wall_surface('steady', 'controlled'), [0, -.0255, .0255]) with self.assertRaises(ValueError): driver.wall_surface('karman', 'controlled', [0, 0, np.nan]) def test_structure_hard_stops(self): driver.check_structure(40., 0., 40.) for y, v, old in [(40.01, 0, 40), (0, .026, 0), (.026, 0, 0), (np.nan, 0, 0)]: with self.subTest(y=y, v=v), self.assertRaises(RuntimeError): driver.check_structure(y, v, old) def test_fluid_nonfinite_fails_nonfluid_only_zero_filled(self): flags = np.array([[1, 257, 2]], np.uint16) fields = dict(rho=np.array([[1., 1., np.nan]]), ux=np.array([[0., .1, np.nan]]), uy=np.array([[0., -.1, np.inf]])) clean, retained, fluid = driver.clean_field(fields, flags) np.testing.assert_array_equal(fluid, [[True, True, False]]) np.testing.assert_array_equal(retained, flags) self.assertTrue(np.isfinite(clean).all()) self.assertEqual(clean[0, 0, 0], 0.) # zero flow remains fluid fields['ux'][0, 1] = np.nan with self.assertRaisesRegex(RuntimeError, 'FLUID'): driver.clean_field(fields, flags) def test_memory_admission_uses_available_not_free(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory)/'meminfo' path.write_text('MemFree: 1 kB\nMemAvailable: 16777216 kB\n') self.assertEqual(driver.available_memory(path), driver.MIN_AVAILABLE) with patch.object(driver, 'available_memory', return_value=driver.MIN_AVAILABLE-1): with self.assertRaises(MemoryError): driver.admit_memory() self.assertIsNone(driver.admit_memory(True)) def test_roi_bounds_match_full_downstream_support(self): roi = metrics.roi_slices((1022, 2559), D=40, xs=1600, y0=510.5, downstream_solid_extent=1272.) self.assertEqual(roi, (slice(311, 711), slice(1360, 2161))) def test_same_clock_error_and_mean_are_distinct(self): shape = (2, 2, 2, 3) target = np.zeros(shape, np.float32) changed = target.copy() changed[0, 0], changed[1, 0] = driver.U, -driver.U role = lambda fields: dict(roi=fields, mask=np.ones((2, 3), bool), times=np.array([3200, 6400])) roles = dict(target=role(target), zero=role(changed), controlled=role(target)) result, mask = driver.score_roles(roles, metrics) self.assertAlmostEqual(result['zero']['mean_instantaneous_E'], 1., places=7) self.assertEqual(result['zero']['E_of_fixed_window_mean'], 0.) roles['zero']['times'] = np.array([6400, 9600]) with self.assertRaisesRegex(ValueError, 'clocks'): driver.score_roles(roles, metrics) roles['zero']['times'] = roles['target']['times'] roles['zero']['mask'][0, 0] = False with self.assertRaisesRegex(ValueError, 'entirely common fluid'): driver.score_roles(roles, metrics) def test_native_archive_verified_raw_retained(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory)/'native.npy' rows = np.lib.format.open_memmap(path, mode='w+', dtype=np.float64, shape=(9, len(driver.NATIVE_COLUMNS))) rows[:] = np.arange(rows.size).reshape(rows.shape) report = driver.compress_native(path, rows, 7) self.assertTrue(report['verified']) self.assertTrue(path.exists()) with np.load(path.with_suffix('.npz')) as data: np.testing.assert_array_equal(data['rows'], rows[:7]) def test_readonly_import_reuses_actual_offsets_and_functions(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root/'legacy_test/core').mkdir(parents=True) # Package __init__ must never be executed by the isolated source loader. for folder in (root, root/'legacy_test', root/'legacy_test/core'): (folder/'__init__.py').write_text("raise RuntimeError('application import forbidden')\n") (root/'legacy_test/core/dtw_metrics.py').write_text('calc_dtw_sim = None\ncalc_lag = None\n') (root/'acquisition.py').write_text( 'from .legacy_test.core.dtw_metrics import calc_dtw_sim, calc_lag\n' 'VORTEX_EVENT_OFFSETS = (-20,-15,-10,-5,0,5,10,15)\n') (root/'legacy_test/acquire.py').write_text( "raise RuntimeError('heavy imports forbidden')\n" 'def _vortex_event_channel(case_name):\n return 3, "center_sensor_uy"\n' 'def _select_vortex_event(sensors, case_name):\n' ' return dict(offsets=VORTEX_EVENT_OFFSETS, channel=_vortex_event_channel(case_name))\n') helper = driver.load_helpers(root/'acquisition.py') result = helper.select_vortex_event(None, 'vortex_taylor') self.assertEqual(result['offsets'], (-20,-15,-10,-5,0,5,10,15)) self.assertEqual(result['channel'], (3, 'center_sensor_uy')) self.assertFalse(list(root.rglob('*.pyc'))) def test_geometry_and_shared_translation_envelope(self): calls, translations = [], [] def translation(*args, **kwargs): item = types.SimpleNamespace(args=args, kwargs=kwargs) translations.append(item) return item def add_body(*args, **kwargs): calls.append((args, kwargs)) return len(calls)-1 motion = types.ModuleType('CelerisLab.body.motion') motion.Translation = translation with patch.dict(sys.modules, {'CelerisLab.body.motion': motion}): probes, bodies = driver.add_scene(types.SimpleNamespace(add_body=add_body), 'karman', moving=True) self.assertEqual(calls[0][0], ('circle', (400.,510.5),40.)) self.assertEqual(probes, [1,2,3]) self.assertEqual(bodies, [4,5,6]) self.assertEqual(translations[0].args, ((0.,0.,-40.,40.), .025)) self.assertTrue(translations[0].kwargs['shared']) self.assertTrue(all(calls[i][1]['translation'] is translations[0] for i in bodies)) calls.clear() with patch.dict(sys.modules, {'CelerisLab.body.motion': motion}): driver.add_scene(types.SimpleNamespace(add_body=add_body), 'steady', pinball=False) self.assertEqual(len(calls), 3) self.assertTrue(all(call[0][0] == 'sensor' for call in calls)) def test_taylor_operator_called_once_inside_context_at_response_zero(self): events = [] class Context: def __enter__(self): events.append('enter') def __exit__(self, *args): events.append('exit') initializer = types.ModuleType('CelerisLab.lbm.initializers') initializer.add_vortex = lambda *args: events.append(args) field = object() sim = types.SimpleNamespace(field=field, stepper=types.SimpleNamespace(step_count=1024000), ctx=types.SimpleNamespace(activate=Context), stream=types.SimpleNamespace(synchronize=lambda: events.append('sync'))) with patch.dict(sys.modules, {'CelerisLab.lbm.initializers': initializer}): receipt = driver.inject_taylor(sim) self.assertEqual(events, ['enter', 'sync', (field, (600.,510.5),80.,.000075,'taylor'), 'sync', 'exit']) self.assertEqual((receipt['response_clock'], receipt['absolute_clock'], receipt['count']), (0,1024000,1)) class SelectionContracts(unittest.TestCase): def test_taylor_release_uses_committed_ids_not_staged_minus_one(self): reads=[] class Sim: def __init__(self,*args): self.bodies=types.SimpleNamespace(count=0) self.stepper=types.SimpleNamespace(step_count=0) self.initialized=False;self.pending=0 def __enter__(self):return self def __exit__(self,*args):pass def add_body(self,*args,**kwargs): if self.initialized:self.pending+=1;return -1 index=self.bodies.count;self.bodies.count+=1;return index def initialize(self):self.initialized=True def sync_bodies(self):self.bodies.count+=self.pending;self.pending=0 def run(self,n,**kwargs):self.stepper.step_count+=n def read_force(self,body): if body<0:raise IndexError('staged ID') reads.append(body);return [0.,float(body),0.] pkg=types.ModuleType('CelerisLab');pkg.Simulation=Sim motion=types.ModuleType('CelerisLab.body.motion');motion.Translation=None base=types.SimpleNamespace(warmup=lambda sim,n,*args:sim.run(n), capture_seed=lambda sim:(np.zeros(18,np.float32),np.ones(18,np.float32), np.ones(2,np.uint16),sim.stepper.step_count)) with tempfile.TemporaryDirectory() as directory,patch.dict(sys.modules,{'CelerisLab':pkg,'CelerisLab.body.motion':motion}): seed,force,_=driver.make_release('taylor',driver.schedule('taylor',True),'unused',base, Path(directory),0,60,None) self.assertEqual(reads,[3,4,5]) self.assertEqual(force,12.) self.assertEqual(seed[3],64) def test_recorded_clock_must_equal_observed_solver(self): driver.check_clock(types.SimpleNamespace(step_count=4579200),4576000,3200) with self.assertRaisesRegex(RuntimeError,'observed solver clock'): driver.check_clock(types.SimpleNamespace(step_count=4579201),4576000,3200) @staticmethod def mean(ux, uy, indices): return dict(mean_ux=ux[indices].mean(axis=0), mean_uy=uy[indices].mean(axis=0)) def test_karman_missing_cycles_preserves_fixed_mean_and_last(self): helper = types.SimpleNamespace(accumulate_mean_fields=self.mean) def no_crossing(*args, **kwargs): raise ValueError('at least two accepted rising crossings are required') helper.assign_periodic_phase = no_crossing cache = np.arange(24, dtype=np.float32).reshape(3, 2, 2, 2) arrays, metadata, event = driver.select_fields('karman', dict(smoke=False), helper, np.array([3200, 6400, 9600]), np.zeros((3, 6)), cache) self.assertEqual(metadata['periodic_phase_status'], 'insufficient_crossings') self.assertNotIn('cycle_mean_ux', arrays) np.testing.assert_array_equal(arrays['ux'][0], cache[-1, 0]) np.testing.assert_allclose(arrays['fixed_window_mean_ux'], cache[:, 0].mean(axis=0)) def test_taylor_common_controlled_event_not_reselected(self): def forbidden(*args): raise AssertionError('other roles must not select their own event') helper = types.SimpleNamespace(accumulate_mean_fields=self.mean, select_vortex_event=forbidden) cache = np.zeros((150, 2, 2, 2), np.float32) event = dict(field_indices=np.array([20,25,30,35,40,45,50,55])) arrays, meta, returned = driver.select_fields('taylor', dict(smoke=False), helper, np.arange(1,151)*3200, np.zeros((150,6)), cache, event) self.assertIs(returned, event) np.testing.assert_array_equal(arrays['selected_index'], event['field_indices']) self.assertEqual(len(arrays['ux']), 8) def test_smoke_selection_has_no_event_or_phase_claim(self): helper = types.SimpleNamespace(accumulate_mean_fields=self.mean) for scene in ('karman', 'steady', 'taylor'): arrays, metadata, event = driver.select_fields(scene, dict(smoke=True), helper, np.array([32,64,96]), np.zeros((3,6)), np.zeros((3,2,2,2), np.float32)) self.assertEqual(metadata['selection'], 'smoke_last_snapshot_not_physical') self.assertIsNone(event) if __name__ == '__main__': unittest.main()