test: add optional public macro CV observations

tests/unit/test_jet_circle_cv.py

@@ -16,6 +16,7 @@ from run_jet_circle_cv import (
     case,
     state,
     cv,
+    cv_from_macros,
     aperture_probe,
     measured_aperture_flux,
     verify,
@@ -28,6 +29,39 @@ def eq(rho, u, v):
 
 
 class Host(unittest.TestCase):
+    def test_public_macros_pressure_advection_and_cropped_frames(self):
+        for h, origin, shape, rect in (
+            (.5, (0., 0.), (48, 64), (5., 4., 25., 20.)),
+            (.5, (.25, .5), (48, 64), (5.25, 4., 25.25, 20.)),
+            (.25, (.375, .75), (96, 128), (5.125, 4., 25.125, 20.)),
+        ):
+            y, x = np.indices(shape)
+            x = origin[0] + h * (x + .5)
+            k, u, v = 1e-5, .02, -.01
+            rho = 1 + 3 * k * x  # p=(rho-1)/3=k*x
+            ux, uy, fluid = np.full(shape, u), np.full(shape, v), np.ones(shape, bool)
+            result = cv_from_macros(rho, ux, uy, fluid, origin, rect, .004, h)
+            expected = 320 * np.array([3*k*u, k + 3*k*u*u, 3*k*u*v])
+            np.testing.assert_allclose(result[1], expected, rtol=1e-11, atol=1e-13)
+            np.testing.assert_allclose(result, cv(eq(rho, u, v), fluid, origin, rect, .004, h), atol=1e-12)
+            x0, y0, x1, y1 = np.rint((np.array(rect)-np.tile(origin, 2))/h).astype(int)
+            selection = np.s_[y0-2:y1+2, x0-2:x1+2]
+            cropped_origin = np.asarray(origin) + h*np.array([x0-2, y0-2])
+            cropped = cv_from_macros(rho[selection], ux[selection], uy[selection],
+                                    fluid[selection], cropped_origin, rect, .004, h)
+            np.testing.assert_array_equal(cropped, result)
+
+    def test_public_macros_full_normal_viscous_stress(self):
+        y, x = np.indices((48, 64))
+        x = (x + .5) * .5
+        rho, ux, uy, fluid = np.ones_like(x), 1e-4*x*x, np.zeros_like(x), np.ones(x.shape, bool)
+        args = (rho, ux, uy, fluid, (0., 0.), (5., 4., 25., 20.))
+        _, viscous = cv_from_macros(*args, .004)
+        _, inviscid = cv_from_macros(*args, 0.)
+        np.testing.assert_allclose(viscous-inviscid, [0, -4*.004*1e-4*320, 0], atol=1e-13)
+        with self.assertRaisesRegex(ValueError, "matching 2D"):
+            cv_from_macros(rho, ux, uy[:-1], fluid, (0, 0), (5, 4, 25, 20), .004)
+
     def test_descriptor_and_uniform_closed_flux(self):
         from CelerisLab.lbm.descriptors import D2Q9_EX, D2Q9_EY
 

tests/unit/test_refined_dfg.py

@@ -521,6 +521,49 @@ def test_public_jet_source_and_pilot_separation():
         r.verify_source(r.DEFAULT_SOURCE, "MRT", "last_base_step", "fixed")
 
 
+def test_cv_snapshot_common_units_and_window_endpoints():
+    records = []
+    for refined, shape, origin, diameter, impulse in (
+        (True, (240, 568), (38., 22.), 40, [4., -8.]),
+        (False, (330, 1761), (.25, .5), 80, [16., -32.]),
+    ):
+        rho, ux, uy, fluid = np.ones(shape), np.full(shape, .01), np.zeros(shape), np.ones(shape, bool)
+        begin, n = round(290*diameter/r.SPEED), round(.02*diameter/r.SPEED)
+        record = r.cv_snapshot(rho, ux, uy, fluid, origin, refined, begin, n, diameter, impulse)
+        records.append(record)
+        np.testing.assert_allclose(record["outward_flux"], 0, atol=1e-12)
+        assert r.cv_selected(begin, diameter, "full") and not r.cv_selected(begin-n, diameter, "full")
+        end = round(300*diameter/r.SPEED)
+        assert r.cv_selected(end, diameter, "full") and not r.cv_selected(end+n, diameter, "full")
+        assert r.cv_selected(n, diameter, "pilot")
+        assert sum(r.cv_selected(step, diameter, "full") for step in range(begin-n, end+n+1, n)) == 501
+    for key in ("time_root", "interval_start_root", "interval_dt_root", "interval_impulse_root", "inventory"):
+        np.testing.assert_allclose(records[0][key], records[1][key], rtol=0, atol=1e-10)
+    assert records[0]["interval_dt_root"] == 80 and np.isclose(records[0]["tstar"], 290.02)
+    np.testing.assert_array_equal(records[0]["interval_impulse_root"], [4, -8])
+
+
+def test_cv_pilot_mode_and_helper_provenance():
+    from tempfile import TemporaryDirectory
+
+    manifest = r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", "fixed", True)
+    assert "tests/validation/run_jet_circle_cv.py" in manifest["sha256"]
+    with TemporaryDirectory() as tmp:
+        path = Path(tmp)
+        old = pilot_fixture(path, jet_scenario="fixed")
+        with unittest.TestCase().assertRaisesRegex(ValueError, "CV observation mode"):
+            r.verify_pilot(path, old, "MRT", "window", "fixed", True)
+        for relative in ("plan.json", "uniform/summary.json", "refined/summary.json",
+                         "uniform/ready.json", "refined/ready.json"):
+            file = path / relative
+            record = json.loads(file.read_text())
+            record.update(observe_cv=True, cv_samples=250)
+            if relative == "plan.json":
+                record["source_manifest"] = manifest
+            file.write_text(json.dumps(record))
+        assert r.verify_pilot(path, manifest, "MRT", "window", "fixed", True) == 3606
+
+
 if __name__ == "__main__":
     suite = unittest.TestSuite(
         unittest.FunctionTestCase(fn)