"""CPU-only science, geometry, units, and source tests; no PyCUDA import.""" import sys from pathlib import Path import unittest import numpy as np ROOT = Path(__file__).resolve().parents[2] REPO = ROOT sys.path.insert(0, str(ROOT / "tests/validation")) sys.path.insert(0, str(REPO / "src")) from run_jet_circle_cv import ( CX, CY, W, case, state, cv, cv_from_macros, aperture_probe, measured_aperture_flux, verify, ) def eq(rho, u, v): cu = CX[:, None, None] * u + CY[:, None, None] * v return W[:, None, None] * rho * (1 + 3 * cu + 4.5 * cu**2 - 1.5 * (u * u + v * 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 np.testing.assert_array_equal(CX, D2Q9_EX) np.testing.assert_array_equal(CY, D2Q9_EY) rho = np.full((48, 64), 1.01) g = eq(rho, 0.023, -0.017) inventory, flux = cv(g, np.ones((48, 64), bool), (0, 0), (5, 4, 25, 20), 0.0072) np.testing.assert_allclose( inventory, 320 * 1.01 * np.array([1, 0.023, -0.017]), atol=1e-12 ) np.testing.assert_allclose(flux, 0, atol=1e-14) def test_affine_velocity_independent_divergence(self): y, x = np.mgrid[:48, :64] x = (x + 0.5) * 0.5 y = (y + 0.5) * 0.5 a = 0.001 g = eq(np.ones_like(x), a * x, -a * y) _, flux = cv(g, np.ones(x.shape, bool), (0, 0), (5, 4, 25, 20), 0.0072) np.testing.assert_allclose( flux, [0, a * a * 320 * 15, a * a * 320 * 12], atol=1e-13 ) def test_quadratic_viscous_flux_sign_and_units(self): y, x = np.mgrid[:48, :64] y = (y + 0.5) * 0.5 a = 0.0001 nu = 0.0072 g = eq(np.ones_like(y), a * y * y, np.zeros_like(y)) _, flux = cv(g, np.ones(y.shape, bool), (0, 0), (5, 4, 25, 20), nu) np.testing.assert_allclose(flux, [0, -2 * nu * a * 320, 0], atol=2e-14) def test_geometry_adapter_match_and_apertures(self): from CelerisLab.config import LBMConfig from CelerisLab.body.geometry.circle import CircleGeometry from CelerisLab.body.objects import SimObject from CelerisLab.body.jets import Jet, Sector, Radial, jet_link_shapes from CelerisLab.lbm.refinement import FixedPatch2D, prepare_fine_boundary from CelerisLab.lbm.descriptors import SOLID c = case() center = np.array(c["center"]) r = c["diameter"] / 2 cfg = LBMConfig(nx=c["nx"], ny=c["ny"], viscosity=c["nu"]) cfg.validate() patch = FixedPatch2D.from_config(cfg, c["bounds"]) obj = SimObject(0, CircleGeometry(*center, r), center=tuple(center), radius=r) jets = [ Jet( Sector(tuple(center), a, np.pi / 12, r + 0.1), Radial(tuple(center), inward=False), "uniform", s * c["jet_speed"], ) for a, s in [(np.pi / 2, 1), (3 * np.pi / 2, -1)] ] host = prepare_fine_boundary(cfg, patch, [obj], jets=jets) shape, ids = jet_link_shapes(jets, host.curved_hits) self.assertTrue((ids == 0).any() and (ids == 1).any() and (ids < 0).any()) self.assertGreater(host.curved_host[2].min(), 0.010001) self.assertFalse(np.any(host.curved_host[-1])) links = CircleGeometry(*(2 * center - 0.5), 2 * r).build_cut_links( 2 * c["nx"], 2 * c["ny"] ) np.testing.assert_array_equal([v.dir for v in links], host.curved_host[1]) np.testing.assert_allclose( [v.q for v in links], host.curved_host[2], atol=2e-7, rtol=0 ) np.testing.assert_allclose( np.array([[v.hit_x, v.hit_y] for v in links]) * 0.5 + 0.25, host.curved_hits, atol=2e-12, ) fluid = (host.flags.reshape(patch.fine_shape) & SOLID) == 0 g = state((9, *patch.fine_shape), c["u"]) g[:, ~fluid] = np.nan for contour in c["contours"]: cv(g, fluid, patch.fine_origin, contour, c["nu"]) probe = aperture_probe(fluid, patch.fine_origin, center, r) measured = measured_aperture_flux(g, probe) np.testing.assert_allclose( measured, 0, atol=1e-14 ) # x-directed flow has zero flux on symmetric top/bottom arcs print( "GEOMETRY", dict( links=len(links), aperture_counts=[int((ids == i).sum()) for i in [0, 1]], q_min=float(host.curved_host[2].min()), q_max=float(host.curved_host[2].max()), fallbacks=int(np.count_nonzero(host.curved_host[-1])), sample_distances=[p[-1] for p in probe], ), ) def test_diffusive_scaling(self): a, b = case(1), case(2) self.assertEqual(a["re_d"], b["re_d"]) self.assertEqual(a["nu"], b["nu"]) self.assertEqual(a["final_t_star"], b["final_t_star"]) self.assertEqual(a["steps"] * 4, b["steps"]) def test_source_pin(self): """Historical integrity: the manifest matches its pinned commit, read with git.""" from run_jet_circle_cv import verify_historical from source_provenance import HistoryUnavailable try: verify_historical(REPO) except HistoryUnavailable as error: self.skipTest(f"historical integrity NOT checked: {error}") def test_pinned_mode_still_rejects_changed_source(self): """Negative control: the frozen-source guard refuses a tree that differs from PIN.""" import json import run_jet_circle_cv as v manifest = json.loads((v.HERE / "data/jet_circle_cv/source_manifest.json").read_text()) import hashlib current = {f: hashlib.sha256((REPO / f).read_bytes()).hexdigest() for f in manifest["files"] if (REPO / f).exists()} if current == manifest["files"]: self.skipTest("working tree equals PIN; no changed source to reject") with self.assertRaisesRegex(RuntimeError, "Source hash mismatch"): verify(REPO) def test_current_checkout_provenance(self): from run_jet_circle_cv import current_checkout record = current_checkout(REPO) self.assertEqual(record["source_mode"], "current_checkout") self.assertIn("not the frozen", record["source_capture"]) self.assertEqual(len(record["src_tree"]), 40) self.assertTrue(record["src_sha256"]) if __name__ == "__main__": unittest.main()