ClaudeandClaude Opus 5.5
d930323479
FixedPatchStepper(observe_window_exchange=True), which requires observe_raw_exchange, allocates a second per-link FP64 ledger per body level. While a window is open, the unchanged BlockAccumulateExchange kernel also accumulates into it after each fine hop. Small API: start_exchange_window() zeroes and opens at the current completed step; read_exchange_window(body_id, level=) is non-destructive and reports impulse with rates over the integer window duration. initialize/ _initialize_raw and step failure invalidate; remap builds a new instance with no open window; Jet amplitude updates may stay in-window. Default disabled: no allocation and no extra launch; last_substep/last_base_step unchanged. summarize_exchange gains window='window' with duration. Host contract tests (mocked driver) and a GPU gate against summed per-base reads with a reordering-bound FP64 tolerance. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
46 lines
2.0 KiB
Python
46 lines
2.0 KiB
Python
"""Host reduction contract for the optional observation-window exchange ledger."""
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from CelerisLab.lbm.raw_exchange import summarize_exchange
|
|
|
|
|
|
def reduce(rows, window, duration=None, h=.5, time_end=10):
|
|
return summarize_exchange(rows, np.array([0, 0, 1]), np.array([False, True, False]),
|
|
body_id=0, level=1, h=h, time_end=time_end, window=window,
|
|
reference_center=(0., 0.), duration=duration)
|
|
|
|
|
|
class ExchangeWindowReduction(unittest.TestCase):
|
|
def test_window_duration_scaling(self):
|
|
rows = np.array([[1., 2., 3., 4.], [.5, -1., 0., 1.], [9., 9., 9., 9.]])
|
|
out = reduce(rows, "window", duration=4)
|
|
np.testing.assert_array_equal(out["total"]["impulse"], [1.5, 1.])
|
|
np.testing.assert_array_equal(out["total"]["force"], [1.5 / 4, 1. / 4])
|
|
self.assertEqual(out["total"]["torque"], 3. / 4)
|
|
self.assertEqual(out["total"]["mass_rate"], 5. / 4)
|
|
np.testing.assert_array_equal(out["aperture"]["impulse"], [.5, -1.])
|
|
self.assertEqual((out["time_start"], out["time_end"]), (6., 10))
|
|
self.assertEqual(out["sample_count"], 8) # duration / h fine substeps
|
|
self.assertEqual(out["window"], "window")
|
|
|
|
def test_existing_windows_unchanged(self):
|
|
rows = np.array([[1., 2., 3., 4.], [0., 0., 0., 0.], [0., 0., 0., 0.]])
|
|
base = reduce(rows, "last_base_step")
|
|
self.assertEqual((base["time_start"], base["sample_count"]), (9., 2))
|
|
np.testing.assert_array_equal(base["total"]["force"], [1., 2.])
|
|
sub = reduce(rows, "last_substep")
|
|
self.assertEqual(sub["sample_count"], 1)
|
|
|
|
def test_duration_validation(self):
|
|
rows = np.zeros((3, 4))
|
|
for bad in (None, 0, -1, 1.5, True):
|
|
with self.subTest(duration=bad), self.assertRaises(ValueError):
|
|
reduce(rows, "window", duration=bad)
|
|
with self.assertRaises(ValueError):
|
|
reduce(rows, "last_base_step", duration=2)
|
|
|
|
|
|
if __name__ == "__main__": unittest.main()
|