From 62c525e73532a974663be2e112af671fa868b972 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 08:27:35 +0000 Subject: [PATCH] Jets: freeze origins, guard set_body, fine-boundary hit helper Sector/Tube/Radial store origin as a validated float 2-tuple, so caller lists/arrays cannot change a registered definition. Simulation.set_body rejects nonzero motion for bodies with Jet-driven links once initialized (before initialize the sync-time stationary check applies). prepare_fine_boundary accepts jets in the same coarse cell-edge frame as fine objects, records read-only FP64 edge-frame hits (FineBoundary2D .curved_hits, default None) and rejects empty or overlapping selections before CUDA. FixedPatchStepper wiring is not included. Co-Authored-By: Claude Opus 5.5 --- src/CelerisLab/body/jets.py | 17 +++++++++-- src/CelerisLab/lbm/refinement.py | 16 +++++++++-- src/CelerisLab/simulation.py | 5 ++++ tests/unit/test_jets.py | 49 ++++++++++++++++++++++++++++++++ 4 files changed, 82 insertions(+), 5 deletions(-) diff --git a/src/CelerisLab/body/jets.py b/src/CelerisLab/body/jets.py index b32ffaa..5223cd3 100644 --- a/src/CelerisLab/body/jets.py +++ b/src/CelerisLab/body/jets.py @@ -32,6 +32,15 @@ def _finite(*values): raise ValueError("Jet parameters must be finite") +def _freeze_origin(spec): + """Store ``origin`` as a validated float 2-tuple (no caller-owned list/array).""" + origin = tuple(float(v) for v in spec.origin) + if len(origin) != 2: + raise ValueError("Jet origin must be a 2D point") + _finite(*origin) + object.__setattr__(spec, "origin", origin) + + @dataclass(frozen=True) class Sector: """Annular sector about ``origin``: |angle offset| < half_width, r_min <= r <= r_max.""" @@ -42,7 +51,8 @@ class Sector: r_min: float = 0.0 def __post_init__(self): - _finite(*self.origin, self.angle, self.half_width, self.r_max, self.r_min) + _freeze_origin(self) + _finite(self.angle, self.half_width, self.r_max, self.r_min) if not (0 < self.half_width <= math.pi and 0 <= self.r_min <= self.r_max): raise ValueError("Sector requires 0 < half_width <= pi and 0 <= r_min <= r_max") @@ -63,7 +73,8 @@ class Tube: half_width: float def __post_init__(self): - _finite(*self.origin, self.axis_angle, self.length, self.half_width) + _freeze_origin(self) + _finite(self.axis_angle, self.length, self.half_width) if self.length <= 0 or self.half_width <= 0: raise ValueError("Tube requires positive length and half_width") @@ -82,7 +93,7 @@ class Radial: inward: bool = True def __post_init__(self): - _finite(*self.origin) + _freeze_origin(self) def direction(self, hits): d = hits - np.asarray(self.origin, np.float64) diff --git a/src/CelerisLab/lbm/refinement.py b/src/CelerisLab/lbm/refinement.py index fc30a9f..fb25211 100644 --- a/src/CelerisLab/lbm/refinement.py +++ b/src/CelerisLab/lbm/refinement.py @@ -153,6 +153,7 @@ class FineBoundary2D: transfer_fine_cells: np.ndarray transfer_coarse_cells: np.ndarray transfer_reads: np.ndarray + curved_hits: np.ndarray | None = None # FP64 cell-edge-frame hits, row-aligned def _transfer_footprint(patch): @@ -183,15 +184,20 @@ def _transfer_footprint(patch): return fine, coarse, _readonly(sorted(reads), np.int64) -def prepare_fine_boundary(cfg, patch, objects, *, wall_aware=False): +def prepare_fine_boundary(cfg, patch, objects, *, wall_aware=False, jets=()): """Project fixed circles and reject boundary/transfer overlap before CUDA. Input centers/radii are in coarse physical coordinates (dx=dt=1). SurfaceActuation retains its lattice-velocity meaning: input coarse units are converted into independent fine-local objects with dx=dt=.5. Original objects are not retained or mutated. No runtime geometry updates. + ``jets`` use the same coarse cell-edge frame as the objects; their + selection is checked against true fine hits here. Jet amplitudes need no + rescaling (lattice velocity scales by dt/dx = 1 under acoustic 2:1). """ if not objects: + if jets: + raise ValueError("Fine jets need fine boundary objects") return None checked = replace(cfg) checked.validate() @@ -286,8 +292,14 @@ def prepare_fine_boundary(cfg, patch, objects, *, wall_aware=False): # coalescence; rows 2/ny-3 are ordinary bulk cells only read as donors. if wall_aware and (y <= 1 or y >= cfg.ny-2): raise ValueError("Transfer footprint touches the coarse wall reconstruction stencil") + from ..body.jets import fine_hits_to_edge_frame, jet_link_shapes + hits = fine_hits_to_edge_frame([(link.hit_x, link.hit_y) for link in links], + patch.fine_origin, dx) + if jets: + jet_link_shapes(jets, hits) # empty selection/overlap rejected before CUDA return FineBoundary2D(tuple(local), _readonly(flags, np.uint16), tuple(_readonly(a, a.dtype) for a in pack_cut_links_to_soa(links)), _readonly(sorted(boundary), np.uint32), _readonly(sorted(fine), np.uint32), - _readonly(sorted(coarse), np.uint32), reads) + _readonly(sorted(coarse), np.uint32), reads, + _readonly(hits, np.float64)) diff --git a/src/CelerisLab/simulation.py b/src/CelerisLab/simulation.py index 33cff47..18c6f12 100644 --- a/src/CelerisLab/simulation.py +++ b/src/CelerisLab/simulation.py @@ -260,6 +260,11 @@ class Simulation: proposed.update({k: float(v) for k, v in (("omega", omega), ("vx", vx), ("vy", vy)) if v is not None}) obj.validate_surface_actuation(self.lbm_cfg, proposed) + cl = self.field.curved + if (self._initialized and cl.jet_id is not None + and np.any(cl.body_id[cl.jet_id >= 0] == id) + and any(proposed.get(k, 0.) != 0 for k in ("vx", "vy", "omega"))): + raise ValueError("Jet-driven bodies must stay stationary") changed = False if omega is not None: obj.state.omega = float(omega) diff --git a/tests/unit/test_jets.py b/tests/unit/test_jets.py index 4a56c1e..0ec354c 100644 --- a/tests/unit/test_jets.py +++ b/tests/unit/test_jets.py @@ -221,3 +221,52 @@ def test_manager_rejects_overlapping_jets_and_moving_bodies(device): bodies.jets.append(a) with pytest.raises(ValueError, match="stationary"): bodies.sync_to_gpu(field) + + +def test_origins_are_frozen_float_pairs(): + point = [6., 5.] + array = np.array([6., 5.]) + jets = [Jet(Sector(point, 0., .5, 4.), Radial(array)), Jet(Tube(array, 0., 1., 1.), Parallel(0.))] + point[0], array[0] = 100., 100. + assert jets[0].region.origin == (6., 5.) and jets[0].field.origin == (6., 5.) + assert jets[1].region.origin == (6., 5.) + assert all(type(v) is float for v in jets[0].region.origin) + for bad in ((1., 2., 3.), (1., float("nan"))): + with pytest.raises(ValueError): + Radial(bad) + + +def test_set_body_rejects_motion_of_jet_driven_body(device): + from CelerisLab.simulation import Simulation + jet = Jet(Sector(CENTER, 0., .6, 3.), Radial(CENTER, inward=False), "quartic", 2e-3) + field, bodies = jet_case(device, [jet]) + sim = SimpleNamespace(bodies=bodies, field=field, lbm_cfg=field.cfg, _initialized=True) + with pytest.raises(ValueError, match="stationary"): + Simulation.set_body(sim, 0, omega=.01) + assert bodies.get(0).state.omega == 0 + Simulation.set_body(sim, 0, omega=0.) # stationary updates stay allowed + + +def test_fine_boundary_keeps_edge_frame_hits_and_checks_jets(): + from CelerisLab.body.objects import SimObject + from CelerisLab.body.geometry.circle import CircleGeometry + from CelerisLab.config import LBMConfig + from CelerisLab.lbm import refinement + cfg = LBMConfig(nx=24, ny=24, viscosity=.05) + cfg.validate() + patch = refinement.FixedPatch2D.from_config(cfg, (4, 4, 20, 20)) + C, R = (12.25, 12.25), 2.15 # coarse cell-edge frame, as fine_objects + body = SimObject(0, geometry=CircleGeometry(*C, R), center=C, radius=R) + plan = refinement.prepare_fine_boundary(cfg, patch, [body]) + assert plan.curved_hits.shape == (len(plan.curved_host[0]), 2) + assert not plan.curved_hits.flags.writeable + np.testing.assert_allclose(np.hypot(*(plan.curved_hits - C).T), R, rtol=0, atol=1e-12) + jet = Jet(Sector(C, 0., .5, R + .01), Radial(C, inward=False)) + refinement.prepare_fine_boundary(cfg, patch, [body], jets=[jet]) + away = Jet(Sector((2., 2.), 0., .1, .5), Parallel(0.)) + with pytest.raises(ValueError, match="selects no"): + refinement.prepare_fine_boundary(cfg, patch, [body], jets=[away]) + with pytest.raises(ValueError, match="overlaps"): + refinement.prepare_fine_boundary(cfg, patch, [body], jets=[jet, jet]) + with pytest.raises(ValueError, match="need fine boundary objects"): + refinement.prepare_fine_boundary(cfg, patch, [], jets=[jet])