body: host-only translating-circle state (S1, no kernels)
MovingCircle keeps the applied offset and an optional pending target apart, validates targets against the applied offset (box, Euclidean 0.5-cell step, speed, no sliding hold) and commits only after upload, step and sync succeed; any failure after the upload attempt is fail-closed. Simulation.add_body (translation=...) declares one; moving circles are excluded from the packed static links, unsupported modes and swept-band placement are rejected at initialize, and every positive-step run raises before any device write until the moving kernels exist. Host tests include a refill-donor enumeration. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
c986343a15
commit
305ab2409b
5 files changed
+370
-3
No files matched your search
@@ -0,0 +1,131 @@
|
||||
"""Translating-circle host contract, S1 (docs/moving_bodies.md 4a, 5, 7).
|
||||
|
||||
Host doubles only: no kernels exist yet, so every positive-step run with a
|
||||
moving body must be rejected before any device write.
|
||||
"""
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from CelerisLab.body.motion import (MAX_STEP, R_MIN, MovingCircle, Translation,
|
||||
enumerate_donor_coverage)
|
||||
|
||||
BAD = (float("nan"), float("inf"))
|
||||
|
||||
|
||||
def circle(box=(-2., 2., -3., 3.), speed=.1):
|
||||
return MovingCircle((50., 40.), 10., Translation(box, speed))
|
||||
|
||||
|
||||
def test_translation_and_radius_validation():
|
||||
for box in ((0., 1., -1., 1.)[:3], (.5, 1., -1., 1.), (-1., 1., BAD[0], 1.)):
|
||||
with pytest.raises(ValueError):
|
||||
Translation(box, .1)
|
||||
for speed in (0., MAX_STEP + 1e-9) + BAD:
|
||||
with pytest.raises(ValueError):
|
||||
Translation((0., 0., -1., 1.), speed)
|
||||
with pytest.raises(ValueError):
|
||||
MovingCircle((0., 0.), R_MIN - .1, Translation((0., 0., -1., 1.), .1))
|
||||
|
||||
|
||||
def test_targets_validate_against_applied_and_last_wins():
|
||||
c = circle()
|
||||
c.set_target((0., .3), (0., .05))
|
||||
c.set_target((0., .5), (0., .05)) # second setter replaces, checked vs applied (0, 0)
|
||||
assert c.pending == ((0., .5), (0., .05))
|
||||
for offset, velocity in (((0., .51), (0., 0.)), # Euclidean bound
|
||||
((.36, .36), (0., 0.)), # |dc|_2 = .509
|
||||
((0., 3.1), (0., 0.)), # outside box
|
||||
((0., .1), (0., .11)), # speed bound
|
||||
((0., 0.), (0., .01)), # hold with sliding velocity
|
||||
((BAD[0], 0.), (0., 0.)), ((0., 0.), (0., BAD[1]))):
|
||||
with pytest.raises(ValueError):
|
||||
c.set_target(offset, velocity)
|
||||
assert c.pending == ((0., .5), (0., .05)) # invalid replacement keeps the old target
|
||||
c.set_target((.35, .35), (0., 0.)) # |dc|_2 = .495 inside the bound
|
||||
assert c.applied == (0., 0.)
|
||||
|
||||
|
||||
def test_record_commit_and_hold():
|
||||
c = circle()
|
||||
assert c.record() == ((50., 40.), (50., 40.), (0., 0.)) # hold: no translation velocity
|
||||
c.set_target((0., .25), (0., .05))
|
||||
assert c.record() == ((50., 40.), (50., 40.25), (0., .05))
|
||||
uploads = []
|
||||
c.advance(uploads.append, lambda: None, lambda: None)
|
||||
assert uploads == [((50., 40.), (50., 40.25), (0., .05))]
|
||||
assert c.applied == (0., .25) and c.pending is None # offset-only commit
|
||||
c.advance(uploads.append, lambda: None, lambda: None) # no new target: hold step
|
||||
assert uploads[-1] == ((50., 40.25), (50., 40.25), (0., 0.)) and c.applied == (0., .25)
|
||||
c.set_target((0., 0.), (0., -.05)) # return to zero offset is a real target
|
||||
assert c.pending == ((0., 0.), (0., -.05))
|
||||
c.advance(uploads.append, lambda: None, lambda: None)
|
||||
assert c.applied == (0., 0.) and c.pending is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stage", ("upload", "step", "synchronize"))
|
||||
def test_failure_after_upload_attempt_is_fail_closed(stage):
|
||||
c = circle()
|
||||
c.set_target((0., .25), (0., .05))
|
||||
|
||||
def call(name):
|
||||
def run(*_):
|
||||
if name == stage:
|
||||
raise RuntimeError(f"{name} failed")
|
||||
return run
|
||||
with pytest.raises(RuntimeError, match=f"{stage} failed"):
|
||||
c.advance(call("upload"), call("step"), call("synchronize"))
|
||||
assert c.failed and c.applied == (0., 0.)
|
||||
for later in (c.record, lambda: c.set_target((0., .1), (0., 0.)),
|
||||
lambda: c.advance(*(lambda *_: None,) * 3)):
|
||||
with pytest.raises(RuntimeError, match="unknown"):
|
||||
later()
|
||||
|
||||
|
||||
def test_band_covers_offset_box_with_margin():
|
||||
assert circle().band() == (50 - 2 - 10 - 2, 50 + 2 + 10 + 2, 40 - 3 - 10 - 2, 40 + 3 + 10 + 2)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("radius", (R_MIN, 4.3, 7.5, 12.0))
|
||||
def test_refill_donor_sets_nonempty_within_bounds(radius):
|
||||
# Evidence for R >= R_MIN and |dc|_2 <= MAX_STEP on a fixed sample, not a proof.
|
||||
assert enumerate_donor_coverage(radius) == 0
|
||||
|
||||
|
||||
pytest.importorskip("pycuda.driver")
|
||||
from CelerisLab import simulation as module # noqa: E402
|
||||
from tests.unit.test_simulation_lifecycle import make_simulation # noqa: E402
|
||||
|
||||
|
||||
def moving_simulation(monkeypatch, *, moving=True):
|
||||
sim = make_simulation(monkeypatch)
|
||||
obj = SimpleNamespace(motion=circle() if moving else None, obj_id=0,
|
||||
center=(50., 40.), radius=10.)
|
||||
writes = []
|
||||
sim._failed, sim._initialized = None, True
|
||||
sim.bodies = SimpleNamespace(objects=[obj], count=1, jets=[],
|
||||
has_pending_edit=lambda: False,
|
||||
_upload_action_async=lambda s: writes.append("action"),
|
||||
zero_obs_async=lambda s: writes.append("obs"))
|
||||
sim.stepper = SimpleNamespace(step=lambda *a, **k: writes.append("step"))
|
||||
return sim, writes
|
||||
|
||||
|
||||
def test_positive_steps_rejected_before_device_writes(monkeypatch):
|
||||
sim, writes = moving_simulation(monkeypatch)
|
||||
sim.bodies.objects[0].motion.set_target((0., .1), (0., .01))
|
||||
for call in (lambda: sim.run(1), lambda: sim.run(5), lambda: sim.step(),
|
||||
lambda: sim.run(3, checkpoint_interval=1)):
|
||||
with pytest.raises(RuntimeError, match="moving kernels"):
|
||||
call()
|
||||
with pytest.raises(RuntimeError, match="moving kernels"): # nominal at rest as well
|
||||
sim.bodies.objects[0].motion.pending = None
|
||||
sim.run(1)
|
||||
assert writes == [] and sim.bodies.objects[0].motion.applied == (0., 0.)
|
||||
|
||||
|
||||
def test_moving_body_rejects_topology_and_checkpoint(monkeypatch):
|
||||
sim, _ = moving_simulation(monkeypatch)
|
||||
for call in (sim.sync_bodies, sim.save_checkpoint, lambda: sim.load_checkpoint("x.h5")):
|
||||
with pytest.raises(ValueError, match="translating circles"):
|
||||
call()
|
||||
Reference in new issue
Block a user