Compare commits

..
Author SHA1 Message Date
ClaudeandClaude Opus 5.5 18f018befe README: several regions are refinements[0], not bounds
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 02:02:38 +00:00
ClaudeandClaude Opus 5.5 ec9b3d721c P1 (12/12): one FixedPatchStepper construction and initialize API
Removes the legacy entry points now that every caller is migrated:
FixedPatchStepper(cfg, *, context, refinements, coarse_boundary,
level_blocks, level_objects, level_jets, observe_*, explosion) and
initialize(*level_g). Deleted: positional bounds, fine_block, base_block,
fine_objects, jets, child, child_block, child_objects, child_jets,
child_rectangles_in_parent, initialize(coarse_g/fine_g/child_g),
_legacy_levels and _initial_levels. Docstring, README and docs describe
the level lists. Compatibility views (patch, fine_boundary, ...) and the
PeriodicPatchStepper alias are unchanged (separate follow-up).

Error -> rule table (legacy keyword check -> level-list rule now firing):
- several patches without fine_block/base_block -> _check_levels
  'several patches require fine_block and base_block' (unchanged rule)
- base_block without fine_block -> 'a blocked level requires blocked
  refined levels below it'
- nested child without all blocks -> 'three or more levels must all be
  blocked' / 'a blocked level requires ...'
- fine_objects/jets with a nested child ('only on L2') -> 'bodies/Jets are
  supported only on the finest level of a nested hierarchy'
- child_objects/child_jets without a nested child -> 'level_objects/
  level_jets needs one entry per level'
- scalar child with several L1 patches ('scalar child requires one parent
  patch') -> hierarchy 'a scalar child requires a scalar parent'
- child vs child_rectangles_in_parent mutually exclusive -> no level-list
  equivalent (cannot be expressed); covered by the new explicit
  RemovedLegacyAPI test: removed keywords and positional bounds raise
  TypeError before any context use, initialize has only *level_g.
The InitialLevels test of the removed adapter is replaced by that test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 02:02:29 +00:00
ClaudeandClaude Opus 5.5 84b76f49cc P1 (11/n): multipatch GPU/CPU tests use the level-list constructor
One local levels(bounds, blocks, objects, jets, **rest) builds the
two-level lists; run_steps and every rejection case use it. All five
before-CUDA rejections, the margin and one-jet checks keep their inputs.
Assertions unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 02:00:59 +00:00
ClaudeandClaude Opus 5.5 050d12b76c P1 (10/n): blocked-fine GPU/CPU tests use the level-list constructor
Dense root + blocked fine level is level_blocks=[None, block]; bodies go
to level_objects[1]. Both before-CUDA rejections (non-dividing and
malformed fine block) keep their inputs. Assertions unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 02:00:15 +00:00
ClaudeandClaude Opus 5.5 ec3e3b928a P1 (9/n): refinement follow-up tests pass refinements=[bounds]
FixedPatchStepper and the PeriodicPatchStepper alias are called with
refinements=[bounds] instead of the positional legacy bounds (same
_legacy_levels result). Assertions, including the before-CUDA and
four-cell/omega/layout rejections, unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:59:56 +00:00
ClaudeandClaude Opus 5.5 2be80324ad P1 (8/n): three-level GPU/CPU tests use the level-list constructor
KW becomes levels() (scalar L1 + scalar L2, per-level blocks, objects,
Jets). All rejection cases are kept per level. Two cases change branch:
L2 bodies/Jets with no nested child (legacy 'child_objects/child_jets
require a nested child') are now an entry for a level that does not
exist ('needs one entry per level'); several L1 patches + scalar child
now meets 'a scalar child requires a scalar parent'. Both still reject
before CUDA. Assertions unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:59:37 +00:00
ClaudeandClaude Opus 5.5 0fcd4ac4cb P1 (7/n): MRT, shifted-jet and blocked-base GPU tests migrated
test_refinement_mrt_gpu, test_shifted_jets_gpu and
test_refinement_blocked_base pass refinements/level_* lists. The three
blocked-base rejection cases keep their inputs per level; base-only
blocks now meet the level-list rule (blocked refined levels below) where
the legacy keyword check fired before. Assertions unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:58:57 +00:00
ClaudeandClaude Opus 5.5 af0566db98 P1 (6/n): exchange-window, raw-exchange, Jet and pinned-remap GPU tests migrated
test_exchange_window, test_raw_exchange_gpu (one fine(block) helper for
the repeated circle+jet construction), test_refinement_jets,
test_multiple_pinned_remap_gpu and test_pinned_body_remap_gpu pass
refinements/level_* lists. Dense fine levels stay level_blocks=[None,
None|block] as before. Assertions unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:58:15 +00:00
ClaudeandClaude Opus 5.5 0b1b3d938f P1 (5/n): six GPU validation tests use the level-list constructor
test_refinement_gpu (incl. PeriodicPatchStepper alias calls),
test_channel_root_gpu, test_fine_surface_gpu, test_shifted_remap_gpu,
test_runtime_child_remap_gpu and test_refinement_three_level_multipatch_gpu
pass refinements/level_* lists; scalar and set children map to the second
refinement entry exactly as _legacy_levels did. Assertions unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:48:09 +00:00
ClaudeandClaude Opus 5.5 5167c79378 P1 (4/n): unit tests and examples use the level-list constructor
test_shifted_jets, test_shifted_storage, test_raw_exchange and
test_channel_root pass refinements/level_* lists; examples
fixed_patch_jet and fixed_periodic_patch pass refinements=[BOUNDS]
instead of the positional bounds. Assertions kept; the shifted-storage
case base_block-without-fine_block now expects the level-list rule
'blocked refined levels below' (same constraint).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:46:52 +00:00
ClaudeandClaude Opus 5.5 aedddc35b1 P1 (3/n): three-level multipatch unit tests use the level-list constructor
KW (legacy base/fine/child blocks + child_rectangles_in_parent) becomes
levels(): refinements=[PARENTS, CHILDREN], level_blocks, level_objects,
level_jets. Every negative case is kept and expressed per level (child
rectangles, per-level blocks); the L1-body case now expects the level-list
rule's message 'only on the finest level' instead of the legacy keyword
message 'only on L2' (same constraint).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:46:06 +00:00
ClaudeandClaude Opus 5.5 7821c6ad71 P1 (2/n): runners and example use the level-list constructor
run_refined_dfg, run_dfg_re20, time_window_dfg, time_refined_dfg,
run_jet_circle_cv and examples/fixed_patch_circle pass refinements /
level_blocks / level_objects / level_jets instead of bounds, base_block,
fine_block, fine_objects and jets. One-to-one with _legacy_levels: same
rectangle (FixedPatch2D frame), same blocks, bodies and Jets.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:35:04 +00:00
ClaudeandClaude Opus 5.5 392314e1ae P1 (1/n): remap_child builds its candidate through the level-list API
The only in-library caller of the legacy constructor keywords now passes
refinements=[L1 bounds, L2 set], level_blocks=[base, fine, child],
level_objects=[(), (), child_objects], level_jets=[(), (), jets] -- the
exact lists _legacy_levels produced for it. Legacy keywords stay until
every caller is migrated (final P1 commit).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:34:33 +00:00
ClaudeandClaude Opus 5.5 f3f1e3d588 PeriodicBlockStepper.step: mid-batch launch failure test; async-error docstring note
Review follow-up (non-blocking items): a launch failure on the second of
three steps fails closed without the batch wait, keeps step_count at its
entry value and blocks download; the docstring now states, as for
FixedPatchStepper, that async faults surface at the single wait.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:31:09 +00:00
ClaudeandClaude Opus 5.5 3648ad70ef PeriodicBlockStepper.step(n): wait on the stream once per call
Same change as the reviewed FixedPatchStepper end-sync (cherry-picked
ad1d82c+512dcc9 above): the n steps are enqueued in order and the host
waits once; step(0) launches nothing; step_count advances after the wait
and keeps its entry value on failure. Launches, buffer roles and bits are
those of n step(1) calls. Mocked-driver host test covers the single
wait, launch/role equality with n step(1), step(0) and failure.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:25:16 +00:00
ClaudeandClaude Opus 5.5 5399c9c0a6 step(n) docstring: raw-exchange ledger holds the final interval only
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:24:36 +00:00
ClaudeandClaude Opus 5.5 54be33e006 FixedPatchStepper.step(n): wait on the stream once per call
Enqueue all n base intervals in order on the one stream and synchronize
once before returning (step(0): no launch, no wait). step_count, the raw
exchange ledger and an open window are published only after that wait;
any failure leaves step_count at its entry value and invalidates both
ledgers. Async device errors are no longer attributed to an interval;
the docstring says to use step(1) loops to localize one. Launch order and
buffers are unchanged. Adds a mocked-driver host test of the contract.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:24:36 +00:00
ClaudeandCodex 750e8a483e test: DFG Re20 steady rule as agreed (10 % range, 2 % slope, window-mean verdict)
Supersedes 71221837. Samples every t* 0.5 (interval-mean forces), trailing
t* 50..60 window (21 samples); range <= 10 % and |slope*window| <= 2 % of
each official band width. Verdict on the window mean with the distance to
the nearest band edge. Our engineering rule, frozen before any run.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 869d60d78e)
2026-10-20 21:59:15 +08:00
ClaudeandCodex 6a03f463bc test: DFG Re20 steady rule per review (1 % band, range and slope, 0.5 s samples)
Samples every 0.5 s at synchronized root steps; trailing 5 s window (11
samples); range and |slope*window| each at most 1 % of the official band
width. States the fluid-on-cylinder force sign, reports cd_positive,
defaults to the refined case and uses the review's label.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 71221837f9)
2026-10-20 21:59:15 +08:00
ClaudeandCodex 46dc6ba086 test: add DFG 2D-1 Re20 runner against official reference intervals
Reuses the pinned Re100 runner and helpers unchanged; sets only the
lattice viscosity for Re 20. Reports Cd, Cl and p_diff (dp * 0.04 for
U = 0.2 m/s), a steady rule frozen before any run (window ranges at most
a tenth of each official interval width) and no interval verdict when
not steady. Declares the pressure-NEQ outlet and Zou-He inlet deviations.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 9b8ff90609)
2026-10-20 21:59:15 +08:00
ClaudeandCodex e1efc18274 test: follow the recursive _advance in three-level feed checks
baba5e0 replaced _advance_child with _advance(index, parent_buffer) driven
from _hop; the two feed-order tests still patched the removed method.
Patch _advance and record only index 2 (L1 hop feeding L2); assertions,
expected order and the stale-feed negative control are unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 3f47436073)
2026-10-20 21:57:57 +08:00
ClaudeandClaude Opus 5.5 9660894a1b docs: checkpoint save/load scope and read_surface_exchange wording
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-20 13:38:30 +00:00
ClaudeandClaude Opus 5.5 ba3772e3b8 docs: record cut-link ownership and read_surface_exchange changes from 87c3a31
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-20 13:29:44 +00:00
CodexandFrank14f b4692e6ec4 test: select single-case resolution and compact refinement 2026-10-19 23:55:26 +08:00
CodexandFrank14f 5731c5171b test: add optional public macro CV observations 2026-10-19 22:43:22 +08:00
CodexandFrank14f 05de02ee60 test: parameterize matched public MRT Jet diagnostics 2026-10-19 22:43:22 +08:00
CodexandFrank14f 87c3a3165e fix: validate passive boundary ownership and expose Jet exchange 2026-10-19 22:43:22 +08:00
60 changed files with 1798 additions and 3251 deletions
+13 -12
View File
@@ -112,8 +112,9 @@ from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
cfg = LBMConfig(nx=32, ny=24, viscosity=.1)
body = SimObject(0, geometry=CircleGeometry(16.0, 12.0, 2.15), center=(16.0, 12.0), radius=2.15,
surface_actuation=SurfaceActuation(0.7, 0.85, 0.002)) # angle, half-width, speed
with CudaContext(0) as ctx, FixedPatchStepper(cfg, (8, 6, 24, 18), context=ctx,
fine_objects=[body], fine_block=(16, 20)) as run:
with CudaContext(0) as ctx, FixedPatchStepper(cfg, context=ctx, refinements=[(8, 6, 24, 18)],
level_blocks=[None, (16, 20)],
level_objects=[(), (body,)]) as run:
rest = np.asarray(compute_feq_d2q9(1.0, 0.0, 0.0), np.float32)[:, None, None]
# (9, ny, nx) postcollision populations per level; non-leaf cells are discarded.
run.initialize(np.broadcast_to(rest, (9, *run.patch.coarse_shape)),
@@ -133,19 +134,19 @@ Scope and limits:
- Bodies must be stationary circles whose whole boundary stencil lies in the fine
active region, clear of the coarse/fine transfer cells; violations are rejected
before any GPU allocation.
- `fine_block=(by, bx)` stores the fine level in fixed-size blocks; `by`/`bx` must
- `level_blocks=[None, (by, bx)]` stores the fine level in fixed-size blocks; `by`/`bx` must
divide the WHOLE fine allocation, i.e. `2*(y1-y0)+8` by `2*(x1-x0)+8` including the
four-cell ghost band. It gives results identical to dense storage and is
infrastructure for future block AMR, not a speedup.
- Several regions: pass a sequence of `(x0, y0, x1, y1)` rectangles as
`bounds`, with both `fine_block` and `base_block`. Transfers are built once
`refinements[0]`, with both levels blocked (`level_blocks`). Transfers are built once
from their union; adjacent, overlapping and concave unions are accepted when
their actual dependencies close. There is no fixed rectangle-gap rule.
The fine level is then one level-wide grid `(2*ny, 2*nx)`:
`run.patch.fine_shape` gives it, `initialize`/`download` use it, and only the
grid-aligned tiles intersecting the union plus its four-cell ghost band are stored.
Block sizes must divide the level-wide grid; rectangle edges need not align
with tiles. Order of rectangles does not matter. `fine_objects`/`jets` work
with tiles. Order of rectangles does not matter. L1 bodies/Jets (`level_objects[1]`/`level_jets[1]`) work
as for one patch: each circle (same coarse cell-edge frame) must lie wholly
inside the union's active region with
its boundary stencil clear of transfer cells; a jet may not select links of two
@@ -156,8 +157,8 @@ Scope and limits:
cfg = LBMConfig(nx=32, ny=16, viscosity=.1)
rest = np.asarray(compute_feq_d2q9(1.0, 0.0, 0.0), np.float32)[:, None, None]
with CudaContext(0) as ctx, FixedPatchStepper(
cfg, ((4, 4, 12, 12), (16, 4, 24, 12)), context=ctx,
fine_block=(4, 4), base_block=(4, 4)) as run:
cfg, context=ctx, refinements=[((4, 4, 12, 12), (16, 4, 24, 12))],
level_blocks=[(4, 4), (4, 4)]) as run:
run.initialize(np.broadcast_to(rest, (9, *run.patch.coarse_shape)),
np.broadcast_to(rest, (9, *run.patch.fine_shape))) # (9, 32, 64)
run.step(32)
@@ -166,12 +167,12 @@ Scope and limits:
```
- Legacy three-level scalar chain: add `child=(x0, y0, x1, y1)` in
**L1 window cells** (the patch's fine grid including its 4-cell ghost band) and
`child_block`, with `fine_block` and `base_block`. L2 must sit inside L1's
all three `level_blocks` entries. L2 must sit inside L1's
ACTIVE cells, away from L1's ghost band and transfer footprint (checked before
any GPU allocation). `initialize` and `download` take/return three arrays; each
level is NaN on its non-ACTIVE (and child-covered) cells. One coarse step = 2 L1
steps = 4 L2 steps. One L1 patch only. Optional stationary circles and Jets
live on L2 via `child_objects` and `child_jets`; `fine_objects`/`jets` retain
live on L2 via `level_objects[2]` and `level_jets[2]`; L1 bodies/Jets retain
their L1 meaning and must be empty in this mode. All object and Jet geometry
uses the **L0 edge frame**, not the child bounds' L1 window frame. L2 has
dx=dt=0.25, so prescribed lattice velocity is unchanged. Each whole body and
@@ -188,9 +189,9 @@ Scope and limits:
jet = Jet(Tube((6.5, 8.), 0., 3., .3), Parallel(0.), amplitude=.002)
rest = np.asarray(compute_feq_d2q9(1.0, 0.0, 0.0), np.float32)[:, None, None]
with CudaContext(0) as ctx, FixedPatchStepper(
cfg, (4, 4, 12, 12), context=ctx, fine_block=(4, 4), base_block=(4, 4),
child=(8, 8, 16, 16), child_block=(4, 4),
child_objects=[body], child_jets=[jet]) as run:
cfg, context=ctx, refinements=[(4, 4, 12, 12), (8, 8, 16, 16)],
level_blocks=[(4, 4)] * 3, level_objects=[(), (), (body,)],
level_jets=[(), (), (jet,)]) as run:
shapes = (run.patch.coarse_shape, run.patch.fine_shape, run.child_patch.fine_shape)
run.initialize(*(np.broadcast_to(rest, (9, *s)) for s in shapes)) # 16², 24², 24²
run.step(4)
+35
View File
@@ -0,0 +1,35 @@
# Compatibility and API changes
Changes that can make a previously accepted setup or call behave differently.
Each entry names the commit, the affected entry points and the observable result.
## 87c3a31: cut-link ownership is validated for every body
`validate_cut_link_ownership` (`body/preprocess/flag_overlay.py`) now runs for
passive bodies as well as Jet/aperture bodies. Every cut link must have an
active fluid receiver (FLUID set, SOLID clear), and each `(receiver, direction)`
population may have only one writer.
Rejected with `ValueError`, before flags, populations or device buffers change:
1. Overlapping or touching passive solids whose cut links write the same
population ("Boundary requires unique cut-link writers").
2. A body whose cut-link receiver is a non-fluid cell, e.g. a channel wall row
or the inlet/outlet column ("Cut links must have active fluid receivers").
Both were previously accepted silently for passive bodies; the interpolated
boundary then had undefined results. The rules apply to the cut links of all
bodies together, so a passive body overlapping a Jet body is rejected as well.
The check runs in `Simulation.initialize`, runtime body sync (`sync_bodies`),
checkpoint save and checkpoint load (`common/checkpoint.py`). Load validates
before restoring anything, so a checkpoint whose saved geometry violates either
rule no longer restores; rebuild the geometry with separated bodies.
## 87c3a31: `Simulation.read_surface_exchange` accepts Jet-driven bodies
`read_surface_exchange(id)` previously required `SurfaceActuation` on the body.
It now also accepts bodies that own Jet aperture links, including Jets with zero
or negative amplitude (the aperture test is geometric). Bodies with neither
`SurfaceActuation` nor aperture links still raise
`ValueError("Body has no surface actuation")`. The value is unchanged: the latest
step's wall/aperture exchange, not total nozzle force.
+1 -1
View File
@@ -1,6 +1,6 @@
# Blocked AMR root channel increment
Opt in with `FixedPatchStepper(..., coarse_boundary="channel", base_block=(by,bx), fine_block=(fy,fx))`, or the existing level-list form. Every level must be blocked. The supported boundary is the existing uniform D2Q9/SRT/FP32 double-buffer channel: local Zou-He west inlet, pressure-NEQ east outlet, stationary halfway bounce-back y walls. Uniform and parabolic inlet profiles and shifted/unshifted storage are supported; the existing configuration guards still reject MRT/LES and incompatible storage. This does not change the external DFG MRT reference.
Opt in with `FixedPatchStepper(..., coarse_boundary="channel", refinements=[bounds], level_blocks=[(by,bx), (fy,fx)])`. Every level must be blocked. The supported boundary is the existing uniform D2Q9/SRT/FP32 double-buffer channel: local Zou-He west inlet, pressure-NEQ east outlet, stationary halfway bounce-back y walls. Uniform and parabolic inlet profiles and shifted/unshifted storage are supported; the existing configuration guards still reject MRT/LES and incompatible storage. This does not change the external DFG MRT reference.
`step/block_channel.cu` is a small root-only kernel. Fine kernels, transfer kernels and recursive hop scheduling are unchanged. Global coordinates and the global profile height/velocity are separate from block-local storage strides. Corner ownership and collision phases match `OneStep`:
+6 -6
View File
@@ -16,8 +16,8 @@ Keep the existing scalar `child=(x0,y0,x1,y1)` for a single scalar parent patch.
Its coordinates retain their original meaning: half-open integer indices in the
allocated L1 window, including its ghost band.
The new keyword `child_rectangles_in_parent=[...]` is mutually exclusive with
`child`. It accepts a scalar or a set parent and interprets every rectangle in
A rectangle list as the second `refinements` entry (formerly the keyword
`child_rectangles_in_parent=[...]`) accepts a scalar or a set parent and interprets every rectangle in
the actual allocated L1 array, including ghost bands and holes. It never infers
region-local coordinates. The physical root cell edge corresponding to an L1
index `(x,y)` is `parent.fine_origin + (x,y)/2`:
@@ -47,9 +47,9 @@ tile edges.
```python
run = FixedPatchStepper(
cfg, [(4, 4, 20, 28), (24, 4, 44, 28)], context=ctx,
base_block=(4, 4), fine_block=(4, 4), child_block=(4, 4),
child_rectangles_in_parent=[(16, 16, 32, 40), (56, 16, 80, 40)],
cfg, context=ctx,
refinements=[[(4, 4, 20, 28), (24, 4, 44, 28)], [(16, 16, 32, 40), (56, 16, 80, 40)]],
level_blocks=[(4, 4)] * 3,
)
```
@@ -57,7 +57,7 @@ This example uses a 48x32 root. `initialize` and `download` use three Q-major
dense arrays. Public arrays and masks still scale with the full allocated
extent; sparse device buffers do not remove that host-memory limit.
Stationary whole-circle `child_objects` and `child_jets` use root edge
Stationary whole-circle L2 bodies and Jets (`level_objects[2]`, `level_jets[2]`) use root edge
coordinates and a single L2 boundary object. Body IDs are physical object-list
positions, not region IDs or caller `obj_id`. Complete boundary stencils must
avoid the transfer footprint. Change a child Jet with
+49
View File
@@ -0,0 +1,49 @@
# DFG 2D-1 (Re 20) against the official reference intervals
`tests/validation/run_dfg_re20.py` runs the steady DFG 2D-1 case on the refined
channel root (D40 root, D80 wide fine patch) and on the uniform D80 grid. Geometry,
inflow, initial state, layout and observation are those of the pinned Re100 runner
(`run_refined_dfg.py`, `re100_helpers.py`, both unchanged); only the lattice
viscosity is set for Re 20 (root tau 0.56, fine 0.62, uniform 0.62, lattice speed 0.01).
Label: "compared against the official reference bands", not a full benchmark
certification: the outlet is pressure-NEQ (`neq_extrap`), not do-nothing, and the
inlet is local Zou-He.
## Quantities
- Cd = 2 Fx / (rho U^2 D), Cl = 2 Fy / (rho U^2 D), interval means over each sample.
- p_diff = p(0.15, 0.2) - p(0.25, 0.2), linear extrapolation to the exact surface
points (`re100_helpers.pressure_probes`). The helper returns dp / (rho U^2) in
scaled units; p_diff = dp * 0.04 for U = 0.2 m/s, rho = 1. Both are reported.
- Official intervals (Schaefer & Turek 1996): Cd [5.57, 5.59], Cl [0.0104, 0.0110],
p_diff [0.1172, 0.1176].
## Bounds and steady rule (frozen before any run)
- End time t* = 60 (30 s physical): 240,000 refined root steps, 480,000 uniform steps.
- One sample every t* = 0.5 (0.25 s) at the end of a synchronized root step; Cd and
Cl are interval means. Trailing window t* 50..60 inclusive (21 samples).
- Steady rule (our engineering choice, not official): in the window, for each of Cd,
Cl, p_diff, max - min <= 10 % and |least-squares slope x window| <= 2 % of its
official interval width (range 2e-3 / 6e-5 / 4e-5; slope 4e-4 / 1.2e-5 / 8e-6).
Absolute limits, so the small Cl is never a divisor. Never relaxed after a run.
- The verdict uses the window mean; the distance to the nearest interval edge is
reported for each quantity (negative outside).
- Force sign: fluid on cylinder, x along the inflow, y up from the bottom wall; Cd > 0
and the official Cl > 0 (`cd_positive` is reported).
- Not steady at t* = 60: status `NOT_STEADY`, no interval verdict. Health limits
(Mach <= 0.2, rho in [0.5, 1.5]) stop the run with `UNHEALTHY`.
- Steady: `WITHIN_INTERVALS` or `OUTSIDE_INTERVALS`, with the per-quantity verdict.
## Run
```
python tests/validation/run_dfg_re20.py --output <new-dir> --plan
CUDA_VISIBLE_DEVICES=<n> python tests/validation/run_dfg_re20.py --case both \
--collision SRT --output <new-dir> --max-wall-seconds <cap>
```
The uniform D80 control separates method error from refinement error. `--plan` writes `plan.json` only. A GPU run writes `plan.json` and `result.json`
(per case: lattice parameters, every sample, window ranges, status, verdict).
Host contract tests: `tests/unit/test_dfg_re20.py`.
+59
View File
@@ -56,3 +56,62 @@ Root must independently review and admit any execution. An admitted paired pilot
the existing watchdog, device selection, `--gpu-gates-approved`, positive
`--max-wall-seconds` and a new output directory, adding only `--exchange-window` to the
selected collision's pilot command. No GPU or remote action was performed for this patch.
## Public Jet diagnostic option
`--collision MRT --exchange-window --jet-scenario zero|fixed|reverse` selects
unshifted FP32, no LES, constant transfer on both existing public backends. Default
`none` and the historical manifests remain unchanged. `zero` registers both Jets at
zero amplitude and retains every existing DFG verdict, including pressure. `fixed`
uses outward radial parabolic sectors at pi/2 and 3pi/2, half-width pi/12, peaks
(+.001,-.001). Both use the existing D40/D80 geometry and U=.01. Sector origins are
(80.5,81) in root edge coordinates and (160,160.5) in uniform node coordinates;
radial extent is [0,1.005R]. `reverse` keeps those peaks through t*=300, updates both
public setters at that completed boundary, and continues to 305 with opposite signs.
Fixed diagnostic windows are 250–260, 260–270, 270–280, 280–290 and 290–300; reversal
adds 300–305. Histories retain exact interval endpoints and root/uniform lattice
impulses. The refined window uses total exchange; uniform uses public `read_force`
batch averages. Driven windows report mean/min/max/RMS Cd/Cl without applying DFG
reference or periodicity gates. Exit zero means healthy execution, not qualification;
`matched_case_pass` remains false. No independent CV or stress-load claim is made.
This opt-in records current source/helper hashes with the existing manifest format,
checkout revision/dirty state and runner hash. Pilot/full must match that snapshot
and the exact scenario. It does not relabel current source as historical evidence.
Use the existing bounded plan/pilot/full commands with the three option flags above;
full requires the matching completed t*=5 paired pilot. `step_seconds` includes force
readback; observation/overhead and total wall time are separate, not kernel throughput.
Add `--observe-cv` to record independent scalar control-volume samples from the
already downloaded public macros. The pilot exercises this path at t*=.02..5;
full records 501 endpoints at 290..300. Each `cv_history.jsonl` row contains actual
root time, preceding interval start/duration/impulse, and two inventories/outward
fluxes ordered [mass,Px,Py]. Exclude the first row's preceding impulse: that row
supplies the initial inventory. Uniform native impulse converts by .5² to root
units; refined ledger impulse is already in root units. Explicit native-aligned
contours differ by .25 root cells in x and are recorded in the plan. Cropping
retains the two-cell face-gradient stencils. No fields or per-link arrays are saved.
`cv_observation_seconds` is the CV portion of observation/overhead time. Helper hash
and the observation flag must match pilot/full. Time integration, temporal
coarsening and any physical assessment remain separate offline work.
## Bounded resolution and placement controls
Defaults remain `--case both --uniform-diameter 80 --placement wide`. Single-case
runs use `--case uniform` or `--case refined` and never report a matched-case pass.
Their own reference/periodicity or driven-diagnostic results remain available.
A single zero-Jet full run exits successfully only when its periodicity and the
unchanged complete DFG reference verdict pass; driven completion is not qualification.
The D160 option requires `--case uniform --uniform-diameter 160` and an explicit
public Jet scenario. U and Jet speeds stay .01 and ±.001; viscosity is .016,
grid 3521×658, center (320,320.5), radius80. CV uses root spacing/time factor .25,
origin (.375,.75), impulse factor .0625, root viscosity .004, and x-face shift .125.
`--case refined --placement compact` selects root bounds (48,48,128,128), fine
168×168 and origin (46,46), with the same D40/fineD80 body/Jet resolution. Its CV
contours are (52,52,112,112) and (50,50,120,120); complete face-gradient stencils
are clear of both solids and the transfer footprint. Compact is refined-only.
CV metadata gives each grid's spacing/origin/contours explicitly. Case, uniform
diameter and placement must match the newly completed pilot in every receipt.
-300
View File
@@ -159,303 +159,3 @@ theoretical stress-consistency certification or production-kernel change.
## Repository integration
The four Python harness files remain together under tests/validation/taylor_green, so runtime provenance hashes precisely this harness directory. Formatting changes produce a new harness identity; preparation and integrated receipts must not be mixed. All analytic definitions, numerical floors, assessment gates, sample times, and source-import assertions are unchanged. Production source is unchanged from the cd3 snapshot; these periodic tests do not waive the separate channel GPU gate.
## Oblique shear wave (test-only diagnostic, not GPU-qualified)
`shear_wave.py` reuses the frozen oracle's constants, cell centers and
geometry. `oracle.py` is byte-identical, and `test_shear_wave.py` pins its
eec4677 SHA-256. `run_case.py --flow shear_wave` selects the shear wave;
the default flow and its receipt are the unchanged Taylor–Green path.
### Convention
Integer mode (2,1) on the L=N periodic root domain:
- the wavevector `k = (2*pi/L)*(2,1)` points 26.565° from +x;
- mode (1,2) would point 63.435°, and a wrong-mode test catches a swap.
Polarization is `E = (-1,2)/sqrt(5)`, perpendicular to k:
u = U*E*sin(k.x + phi)*exp(-nu*|k|^2*t), phi = 0, p = const, rho0 = 1
`nu = U*N/100`, the Taylor–Green Re100 viscosity at the same N. This is an
exact incompressible Navier–Stokes solution. The initial condition is FP64
second-order equilibrium at the actual leaf centers, with no
non-equilibrium part.
### Physical caveats
1. *Scope of the cubic-term argument.* The second-order D2Q9 equilibrium
lacks the cubic third moment `rho*u_a*u_b*u_g` of the Maxwellian (Dellar,
https://people.maths.ox.ac.uk/dellar/papers/Galilean.pdf, eq. 3). At
Navier–Stokes order this produces a stress error proportional to
`d_g(rho u_a u_b u_g)`.
For the ideal constant-density, zero-background transverse wave, k.E = 0
makes that contracted derivative vanish. That is the only claim made.
It does not remove other discrete or finite-Mach errors. It is not
guaranteed once AMR introduces density, longitudinal or higher-mode
contamination. The analytic NS field itself is exact.
2. *Startup.* Equilibrium initialization is not the discrete shear
eigenmode (Mei et al., https://doi.org/10.1016/j.compfluid.2005.08.008).
- The decay rate is a free-intercept least-squares fit of ln R over the
samples with t* >= 0.12, where `R = hypot(a_hat, b_hat)`.
- The intercept cancels only a constant amplitude prefactor, and only
once a single exponential dominates.
- Per-interval rates and the maximum log-fit residual are kept, so a
lingering transient stays visible.
- Late sampling alone does not show that transients have gone.
### Metrics
Computed on ACTIVE leaf cells, each weighted by its area `4**-level` in
root-cell units. The partition must cover the domain exactly once, and one
global pressure gauge is used.
- `a_hat`, `b_hat`: weighted least-squares fit of `u.E` onto sin/cos, using
the 2x2 Gram matrix of the actual composite quadrature. A wrong Fourier
mode need not project to exactly zero on this quadrature.
- `amplitude_error = R/A_exact - 1`.
- `phase_drift = atan2(b_hat, a_hat)`, in radians, reported separately.
- Rates are positive (minus the fitted slope).
- Per root step they are compared with `nu*|k|^2`. Uniform 2N uses its
own `nu = 0.0768` at N=256 and doubled step counts.
- In t* they are compared with `5*(2*pi)^2/Re = 1.97392088`.
- `rate_error = |rate/exact - 1|`.
- `velocity_l2`, `residual_l2` (velocity left after removing the fitted
mode) and `longitudinal_l2` (the `u.k_hat` part), all relative to the
exact-field norm. A longitudinal perturbation also raises the
full-vector residual and the velocity L2.
- `spurious_pressure_rms`: the raw gauged rms of `p = (rho-1)/3`.
- `spurious_pressure_scaled`: that rms divided by `A_exact(t)^2/4`, a
continuum reference scale. It is *not* the discrete, globally gauged
Taylor–Green norm on the composite grid. Exact pressure is constant, so
no relative-to-exact claim is made. Max and per-level values are kept.
### Preregistered screens (`shear_wave.assess`)
Run set: uniform N, uniform 2N, and refined depth-2 runs at root N.
- `eps(N) = (|k| h_root)^2 = 5*(2*pi/N)^2`: dimensionless, coefficient
one, 0.01205 at N=128. This is a chosen screen, not a proven error bound.
- Absolute screens:
- rate_error, max |phase_drift| (radians) and max residual_l2 must each
be <= eps;
- velocity_l2 <= 0.05, mass <= 1e-5, spurious_pressure_scaled <= 0.02.
- Budgets keep the `assess.py` form, `refined <= max(2*control_N, floor)`:
- floors are 0.002 for rate and residual, 0.02 for pressure;
- the comparative pressure budget is always redundant:
`max(2*control, 0.02)` is never below the 0.02 absolute ceiling;
- raw ratios to both the root and finest controls, including raw
pressure rms, are always reported.
- Every record is validated before any screen is applied:
- exact scheduled steps and t*, with no duplicates;
- every numeric field finite, including list fields and density;
- error quantities nonnegative and rho_min > 0;
- signed quantities (amplitudes, `amplitude_error`, `phase_drift`,
per-interval rate errors) keep their sign.
- Outcomes are kept distinct:
- INVALID_CONTROL: a control or refined run is missing, a control is
incomplete or invalid, or the records come from mixed sources;
- INVALID_RUN: a refined record fails validation;
- CONTROL_SCREEN_FAIL: complete, valid controls fail their own screens.
Each refined run's `own_screens_pass` is still reported, but
`controlled_pass` is false, so it is never a controlled-comparison pass;
- PASS/FAIL: every refined run's `controlled_pass` (own screens and
budgets against valid, passing controls).
- Finite modal extinction (late R = 0) is a physical failure, not
malformed data. The run gets `rate_fit_valid=false` with null fit
quantities, and its other metrics are kept. Nothing is clipped and no
rate is invented. The result is CONTROL_SCREEN_FAIL for a control, or
FAIL for a refined run.
- Screens are never retuned.
### Limitations
One flow, one phase and one rectangle, SRT only. The rectangle's vertical
faces span 0.625 tangential periods and its horizontal faces 1.5, so this
is not placement robustness. Two resolutions show a trend, not an order.
### Receipt assessment and runner integrity
- Fixed four-case N128 contract:
`python assess_shear_wave.py U128.json U256.json R128_constant.json R128_projected.json --output report.json --expected-revision <sha>`.
It reuses the shared provenance, build, workload and sample-time checks
from `assess_projected_pairs.py`, and adds shear metadata checks (flow,
mode, polarization, angle, phase, eps, limits, disabled early trace, and
pinned `shear_wave.py` hash).
- Missing, extra, duplicate or mixed-source records give `INVALID_SET`,
as does any failed shared check.
- Otherwise the result comes from `shear_wave.assess`, which separates
`INVALID_CONTROL` from `CONTROL_SCREEN_FAIL`.
- `run_case.verify_source(..., flow="shear_wave")` requires the imported
`shear_wave.py` to be the selected harness file with the pinned
`SHEAR_WAVE_SHA256`. The default TG flow does not import it.
- Sample and READY finiteness use `finite_row`, which accepts scalar and
list diagnostics. For TG's all-scalar rows it behaves the same as before.
- Merge note: native 92b8afe adds a `scaled_n256` receipt block. When the
two are merged, keep both blocks and make `--flow shear_wave` and
`scaled_n256` mutually exclusive. Not merged here.
### Scaled shear diagnostic (opt-in `scaled_shear_n256`)
Runner: `run_case.py --flow shear_wave --diagnostic scaled_shear_n256 --re 100`
with one of these four cases:
- `--n 256 --depth 1` (U256)
- `--n 512 --depth 1` (U512)
- `--n 256 --depth 2 --explosion constant` (constant R256)
- `--n 256 --depth 2 --explosion projected_linear` (projected R256)
Everything physical is unchanged from N128:
- `rectangles(256,2)` covers the same physical box;
- same t* schedule (16N root steps);
- `nu = U*N/100`;
- same `eps(N)`, limits and floors.
`scaled_n256` (Taylor–Green, from native 92b8afe) and `scaled_shear_n256`
each require their own flow, so they are mutually exclusive. N512 is
accepted only through one of the two diagnostics. The shear diagnostic
takes no early-pressure trace.
Assessment:
python assess_shear_wave.py U256.json U512.json R256_constant.json R256_projected.json \
--diagnostic scaled_shear_n256 --reference-files <four de5b0cc N128 receipts> \
--output report.json --expected-revision <sha>
- The scaled N256/512 case, build and workload checks come from the
92b8afe assessor. Its TG-only diagnostic identity and early-sampling
checks are not used. Instead every record must carry
`diagnostic=scaled_shear_n256`, and N128 records must carry none.
- The reference N128 receipts are hash-pinned to the audited de5b0cc
evidence (commit cba6bca). They are never relabelled as fresh controls.
- The report always includes, for each case:
- the raw signed rate bias `rate/exact - 1` and the interval errors;
- each refined run's signed excess over *each* uniform control (rate
bias, residual, raw pressure rms);
- the N128→N256 signed changes and ratios. The ratio is null when the
N128 value is zero.
- These are descriptive. Passing or failing another floor-controlled
screen is not the conclusion, and a two-resolution ratio is not a
convergence order.
- Portability: the receipt's recomputed `eps = |k|^2` is compared within
4 ULPs (`close_ulps`), not by exact equality. A cross-runtime replay
differed by one ULP: 0.012047856934923534 on Horsea vs
0.012047856934923532 elsewhere. Source and file hashes, case identity,
the other metadata and every scientific threshold stay exact.
### Nested shear preset (opt-in `nested_shear_n128`)
A fixed three-level, two-region preset, defined in `shear_nested.py`. The
wave, phase, Re100/U=0.03, initialization, five sample times, eps(root N),
limits and floors are the same as before.
- Refinements, given as parent-frame `(x0, y0, x1, y1)` rectangles on
level-wide, zero-origin unions:
- L0→L1: `(16,16,60,112)`, `(68,24,112,120)`
- L1→L2: `(44,44,108,212)`, `(148,60,212,228)`
- Leaves `[7936, 12288, 86016]`, hops `[2048, 4096, 8192]`, work
771227648 per refined case; 3723493376 for all four cases.
- Level ν `[.0384, .0768, .1536]`, τ `[.6152, .7304, .9608]`.
Cases (`run_case.py --flow shear_wave --diagnostic nested_shear_n128 --re 100`):
- `--n 128 --depth 1`
- `--n 512 --depth 1` — the finest control is 4N, not 2N
- `--n 128 --depth 3`
- `--n 128 --depth 3 --explosion projected_linear [--final-snapshot]`
`--final-snapshot` is allowed only for the projected depth-3 case. It writes
`<output>.final_macro.npz`, containing:
- ACTIVE leaves packed as `level`/`i`/`j`;
- `rho`, `ux`, `uy` from the FP64 physical populations at the final root
synchronization;
- embedded metadata: exact time, phase, mode and config.
It contains no populations, restart state or time series. Its SHA-256 and
size go into the receipt. A file over 8 MiB is removed and reported, never
kept. All old paths stay scalar-only.
Assessment: `assess_shear_wave.py U128.json U512.json R128c.json R128p.json
--diagnostic nested_shear_n128 --output report.json --expected-revision <sha>`.
- It reuses the case-agnostic provenance checks from the TG assessor, plus
exact nested scales, builds, patch, workload, metadata and the pinned
`shear_nested.py` hash.
- The physics is `shear_wave.assess` generalized only in its depth-3 / 4N
shape contract; the metrics are unchanged.
- `shear_wave.py` is byte-identical (81d5c8ee…), so historical receipt
identities and verdicts are untouched.
- This preset changes placement and refined volume as well as depth. It
gives no isolated cost of an extra level and no resolution trend.
### Depth-2 twin of the nested preset (opt-in `twin_shear_n128`)
`shear_twin.py` uses the same two L0→L1 boxes as `shear_nested.py`, with no
L2. Leaves are `[7936, 33792]`, hops `[2048, 4096]`, and work is 154664960 per
refined case (611319808 for all four).
The campaign is four fresh runs:
`run_case.py --flow shear_wave --diagnostic twin_shear_n128 --re 100` with
- `--n 128 --depth 1`
- `--n 256 --depth 1` (the finest control is 2N)
- `--n 128 --depth 2`
- `--n 128 --depth 2 --explosion projected_linear`
There is no snapshot option. Older exact controls serve only as consistency
checks; they are not reused as references.
Assessment uses the same preset contract as the nested preset, through
`assess_shear_wave.py ... --diagnostic twin_shear_n128`:
- the physics is the unchanged depth-2 / 2N `shear_wave.assess`;
- receipts must pin `shear_twin.py`, `shear_nested.py` and `shear_wave.py`;
- `shear_nested.py` stays byte-identical (fa6cfe87…), so the 335624c nested
receipts still validate.
Labels:
- *Twin vs nested* is the effect of adding this nested L2 refinement.
Removing L2 also changes the finest spacing and refined volume, so it does
not isolate abstract depth.
- *Twin vs the single-box depth-2 runs* is a placement/geometry contrast at
equal depth.
- Neither contrast is additive by construction.
- The thin-band/acoustic explanation for the nested L1 pressure is a
hypothesis, not established.
### Signed interface-probe traces (opt-in `--probe-trace`, mechanism diagnostic)
`shear_probes.py` (pinned) uses native's validated 24-target layout through
box 1, in root-128 physical coordinates:
- a horizontal line y = 56, crossing the vertical faces x = 16 and 22;
- a vertical line x = 38, crossing the horizontal faces y = 16 and 22;
- normal coordinates 12.5, 14.5, 15.5, 16.25, 17.25, 19.25, 20.75, 21.75,
22.125, 23.125, 24.125, 26.125.
Each target maps to the nearest ACTIVE leaf center. The deterministic tie
rule is: distance, then finer level, then smaller j, then smaller i. This
gives 24 unique cells in every case, and nested has 3 L0 + 5 L1 + 4 L2 per
transect. Maximum displacement from target to center is 0.625 root cells in
U128 (0.177 in U512).
Allowed cases:
- nested U128, nested U512, nested projected depth 3;
- twin projected depth 2.
The option can't be combined with `--final-snapshot`.
Sampling:
- The receipt stores each probe's leaf level, `i`/`j`, actual cell center
and displacement from its target, all in root-physical units (N=128).
- Grids differ between cases, so centers are *not* collocated. Analytic
fields are evaluated at each actual center (`shear_probes.residuals`), and
comparisons across cases are between residuals at nearby but distinct
centers.
- Times are root-physical steps 0–128 and 1920–2048. U512 is sampled every 4
lattice steps to match.
- Each sample stores `rho`, `ux`, `uy` from the FP64 physical populations,
POST-collision G at root synchronization only. No substep content is
observed, and grid-scale alternation can alias.
What's kept and enforced:
- The five qualification samples stay unchanged.
- No field dumps. The trace is validated for exact schedule, finiteness and
a 600 kB cap. If invalid it is dropped and an error is recorded, keeping
the scalar receipt.
- There are no new pass gates; this is a mechanism diagnostic only.
+2 -2
View File
@@ -90,8 +90,8 @@ def main():
# The caller owns the CUDA context; the stepper must be closed first (the
# nested with-statement does that).
with CudaContext(0) as context, FixedPatchStepper(
cfg, BOUNDS, context=context, coarse_boundary=args.mode,
fine_objects=[body], fine_block=fine_block) as run:
cfg, context=context, coarse_boundary=args.mode, refinements=[BOUNDS],
level_blocks=[None, fine_block], level_objects=[(), (body,)]) as run:
p = run.patch
# Rest equilibrium is already postcollision G, so no initial collision is
# needed. Non-leaf entries (covered, ghost, solid) are discarded by initialize.
+2 -2
View File
@@ -148,8 +148,8 @@ def main():
samples=[], status='running', last_synchronized_coarse_step=0)
start, run = time.monotonic(), None
try:
with CudaContext(0) as context, FixedPatchStepper(cfg, BOUNDS, context=context,
coarse_boundary='slot_reservoir') as run:
with CudaContext(0) as context, FixedPatchStepper(cfg, context=context,
coarse_boundary='slot_reservoir', refinements=[BOUNDS]) as run:
p = run.patch
masks = [np.ones(p.coarse_shape, bool), p.fine_kind == ACTIVE]
masks[0][[0, -1]] = False
+1 -1
View File
@@ -48,7 +48,7 @@ def main():
nx=16, ny=16, viscosity=.1, velocity=1e-3, rho=1.0,
lattice_model="D2Q9", collision="SRT", streaming="double_buffer",
store_precision="FP32", ddf_shifting=False, les_enabled=False)
with CudaContext(0) as context, PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context) as run:
with CudaContext(0) as context, PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)]) as run:
p = run.patch
coarse = shear_equilibrium(p.coarse_shape, (0, 0), 1.0, cfg.ny)
fine = shear_equilibrium(p.fine_shape, p.fine_origin, .5, cfg.ny)
+2 -5
View File
@@ -216,11 +216,8 @@ def surface_link_velocity(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
enabled = [obj for obj in objects if obj.surface_actuation is not None]
if not enabled and not jets:
return None, None
from ..lbm.descriptors import FLUID, SOLID
if np.any((flags[fluid_idx] & FLUID) == 0) or np.any(flags[fluid_idx] & SOLID):
raise ValueError("Surface cut links must have active fluid receivers")
if len(np.unique(fluid_idx.astype(np.uint64)*cfg.nq+dir)) != len(dir):
raise ValueError("Surface boundary requires unique cut-link writers")
from .preprocess.flag_overlay import validate_cut_link_ownership
validate_cut_link_ownership(flags, fluid_idx, dir, cfg.nq)
surface_velocity = np.zeros((len(dir), 2), np.float32)
surface_aperture = np.zeros(len(dir), bool)
for obj in enabled:
+17 -12
View File
@@ -27,7 +27,7 @@ import pycuda.driver as cuda
from .objects import SimObject
from .registry import BodyRegistry
from .preprocess.flag_overlay import merge_flag_masks
from .preprocess.flag_overlay import merge_flag_masks, validate_cut_link_ownership
from ..cuda.compiler_v2 import obs_layout
from ..cuda.obs_layout import ObsLayout
from ..lbm.descriptors import (
@@ -369,9 +369,9 @@ class ObjectManager:
Args:
objects: List of SimObject instances to process.
domain_flags: Optional uint16 flags for donor-cell classification.
domain_flags: Optional uint16 flags for receiver/donor classification.
sensor_cell_counts_out: Int array to fill with per-object sensor
cell counts. If provided, ``fill(0)`` is called first.
cell counts after the full geometry has passed validation.
hits_out: List that receives one FP64 (n, 2) array of true hit
points row-aligned with the cut-link columns (for Jets).
@@ -394,8 +394,8 @@ class ObjectManager:
active_fluid = ((domain_flags & FLUID) != 0) & ((domain_flags & SOLID) == 0)
force_owners: set[int] = set()
if sensor_cell_counts_out is not None:
sensor_cell_counts_out.fill(0)
sensor_counts = (np.zeros_like(sensor_cell_counts_out)
if sensor_cell_counts_out is not None else None)
ez = None
if self.cfg and self.cfg.is_d3q19:
@@ -430,9 +430,9 @@ class ObjectManager:
if len(cells) > 0:
s_cells.append(cells)
s_ids.append(ids)
if (sensor_cell_counts_out is not None
and 0 <= obj.obj_id < sensor_cell_counts_out.size):
sensor_cell_counts_out[obj.obj_id] = int(len(cells))
if (sensor_counts is not None
and 0 <= obj.obj_id < sensor_counts.size):
sensor_counts[obj.obj_id] = int(len(cells))
if obj.is_force_region and hasattr(obj, 'get_sensor_list'):
cells, ids = obj.get_sensor_list(self.nx, self.ny, self.nz)
keep = active_fluid[cells]
@@ -461,8 +461,9 @@ class ObjectManager:
r_fr_ids = np.concatenate(fr_ids) if fr_ids else np.zeros(0, dtype=np.int32)
r_hit = np.concatenate(cl_hit) if cl_hit else np.zeros((0, 2))
if hits_out is not None:
hits_out.append(r_hit)
# Passive walls need the same ownership guarantees as driven walls.
# Overlapping solids are not a supported union-boundary operation.
validate_cut_link_ownership(domain_flags, r_idx, r_dir, self.nq)
if self.jets or any(obj.surface_actuation is not None for obj in objects):
# Reject unresolved apertures/ownership before sync or restart mutates state.
from ..lbm.curved_links import CurvedLinkSoA
@@ -470,6 +471,11 @@ class ObjectManager:
links.assign_host(r_idx, r_dir, r_q, r_bid, r_rx, r_ry, r_rz, r_fallback)
links.assign_surface(objects, self.cfg, domain_flags, jets=self.jets, hits=r_hit)
if sensor_cell_counts_out is not None:
sensor_cell_counts_out[:] = sensor_counts
if hits_out is not None:
hits_out.append(r_hit)
return (
r_idx, r_dir, r_q, r_bid,
r_rx, r_ry, r_rz, r_fallback,
@@ -487,8 +493,6 @@ class ObjectManager:
the caller has already prepared and uploaded the final flag field
for this geometry, such as during initialization.
"""
self._telemetry_field = field
next_flags = self.build_flags(field.build_channel_flags()) if rebuild_flags else field.flag
hits = []
@@ -499,6 +503,7 @@ class ObjectManager:
fr_cells, fr_obj_id,
) = self.build_compact_lists(domain_flags=next_flags, hits_out=hits)
self._telemetry_field = field
if rebuild_flags:
field.flag = next_flags
field.upload_flags()
+10 -1
View File
@@ -1,6 +1,6 @@
# CelerisLab/body/preprocess/flag_overlay.py
"""
Flag overlay: merge per-object flag masks onto a clean domain base.
Flag overlay and cut-link ownership checks against the merged domain.
"""
from __future__ import annotations
@@ -12,6 +12,15 @@ import numpy as np
from ...lbm.descriptors import FLUID, SOLID, SENSOR_FLAG, FRC_REGION
def validate_cut_link_ownership(flags, fluid_idx, directions, nq):
"""Reject solid receivers and multiple writers to one population."""
if np.any((flags[fluid_idx] & FLUID) == 0) or np.any(flags[fluid_idx] & SOLID):
raise ValueError("Cut links must have active fluid receivers")
writers = fluid_idx.astype(np.uint64) * nq + directions
if len(np.unique(writers)) != len(directions):
raise ValueError("Boundary requires unique cut-link writers")
def merge_flag_masks(
base: np.ndarray,
masks: List[np.ndarray],
+14 -4
View File
@@ -204,11 +204,21 @@ class PeriodicBlockStepper:
raise
def step(self, n=1):
"""Advance n steps: halo refresh, then core pull/collide, per step."""
"""Advance n steps: halo refresh, then core pull/collide, per step.
All n steps are enqueued in order on the one stream and the host waits
once before returning (same launches and bits as n step(1) calls).
Asynchronous device errors surface at that single wait, so they are not
attributed to the step that caused them; call step(1) repeatedly to
localize one. On any failure the instance is failed and step_count keeps
its entry value.
"""
if isinstance(n, bool) or not isinstance(n, Integral) or n < 0:
raise ValueError("n must be a nonnegative integer")
with self._lock, self.ctx.activate():
self._require("ready")
if n == 0:
return
kw = dict(block=self._block, stream=self._stream)
try:
self._status = "advancing"
@@ -224,9 +234,9 @@ class PeriodicBlockStepper:
self._step(self._current, self._spare, np.uint32(self.layout.halo),
self._tables.get("flags", np.uintp(0)), np.uintp(0), # no kind
grid=self._grid, **kw)
self._stream.synchronize()
self._current, self._spare = self._spare, self._current
self._step_count += 1
self._current, self._spare = self._spare, self._current # host handles only
self._stream.synchronize()
self._step_count += n
self._status = "ready"
except BaseException:
self._status = "failed"
+81 -142
View File
@@ -65,52 +65,8 @@ def _level_name(level):
return ("base", "fine", "child")[level] if level < 3 else f"level {level}"
def _legacy_levels(bounds, *, fine_block, base_block, child, child_block, child_rectangles_in_parent,
fine_objects, jets, child_objects, child_jets):
"""Normalise the original two/three-level keywords to per-level lists (original errors)."""
if child is not None and child_rectangles_in_parent is not None:
raise ValueError("child and child_rectangles_in_parent are mutually exclusive")
nested = child is not None or child_rectangles_in_parent is not None
bounds = tuple(bounds)
if not is_rectangle(bounds) and (fine_block is None or base_block is None):
raise ValueError("several patches require fine_block and base_block")
fine_objects, jets = tuple(fine_objects), tuple(jets)
child_objects, child_jets = tuple(child_objects), tuple(child_jets)
if not nested and (child_objects or child_jets):
raise ValueError("child_objects/child_jets require a nested child")
if base_block is not None and fine_block is None:
raise ValueError("base_block requires fine_block")
refinements, blocks = [bounds], [base_block, fine_block]
objects, level_jets = [(), fine_objects], [(), jets]
if nested:
if child is not None and not is_rectangle(bounds):
raise ValueError("scalar child requires one parent patch; use child_rectangles_in_parent")
if fine_block is None or base_block is None or child_block is None:
raise ValueError("a nested child needs fine_block, base_block and child_block")
if fine_objects or jets:
raise ValueError("a nested child supports bodies/Jets only on L2; use child_objects/child_jets")
refinements.append(tuple(child) if child is not None else tuple(child_rectangles_in_parent))
blocks.append(child_block)
objects.append(child_objects)
level_jets.append(child_jets)
return refinements, blocks, objects, level_jets
def _initial_levels(level_g, coarse_g, fine_g, child_g):
"""One tuple of per-level arrays from initialize()'s positional or legacy named form."""
if coarse_g is None and fine_g is None:
if child_g is not None and len(level_g) != 2:
raise TypeError("child_g needs exactly the coarse and fine arrays before it")
return tuple(level_g) + (() if child_g is None else (child_g,))
if level_g:
raise TypeError("pass level arrays positionally or as coarse_g/fine_g/child_g, not both")
if coarse_g is None or fine_g is None:
raise TypeError("coarse_g and fine_g are both required in the named form")
return (coarse_g, fine_g) + (() if child_g is None else (child_g,))
def _check_levels(refinements, blocks, objects, jets):
"""Candidate limits shared by the list and the legacy entry points."""
"""Candidate limits of the level lists, checked before any device work."""
depth = len(refinements) + 1
if depth < 2:
raise ValueError("at least one refinement level is required")
@@ -134,7 +90,14 @@ def _check_levels(refinements, blocks, objects, jets):
class FixedPatchStepper:
"""D2Q9/SRT/FP32, one unforced internal 2:1 patch in an x-periodic box.
"""D2Q9 SRT/MRT FP32 fixed 2:1 refinement hierarchy, built from level lists.
``refinements[l]`` gives level l+1 in level-l integer cells: one rectangle
(x0, y0, x1, y1) (FixedPatch2D, a fine window with its own origin) or a
sequence of rectangles (FixedPatchSet2D, a dependency-validated union on one
level-wide grid, public fine arrays (2*ny, 2*nx), NaN outside ACTIVE cells);
see lbm.hierarchy. ``level_blocks``, ``level_objects`` and ``level_jets``
hold one entry per level, L0 included (L0 bodies/Jets are not supported).
``coarse_boundary`` is ``periodic`` (default) or ``slot_reservoir``:
a bottom on-site slot and top fixed-density reservoir from ``cfg``.
@@ -145,68 +108,53 @@ class FixedPatchStepper:
actual channel closure footprint. Channel remapping is not supported.
Periodic mode does not apply the wall/slot boundary fields in ``cfg``.
Outer walls remain coarse; fine halo donors must be ordinary fluid.
``fine_objects`` accepts fixed circles in coarse physical coordinates;
velocities are in coarse lattice units (dx=dt=1). Independent local copies
are projected at fine cell centers. Their complete stencil must be active
and disjoint from the actual two-hop transfer footprint.
These use the edge/cell-center frame (coarse centers I+.5), whereas ordinary Simulation objects use the integer-node frame.
``initialize(coarse_g, fine_g)`` accepts physical Q-major (9,ny,nx) arrays
already in postcollision phase. It does not collide or build channel flags.
Every ``step()`` publishes only after both levels complete on one stream.
On a device/launch failure the instance becomes unusable: close and recreate
it, rather than pretending the partially advanced fine buffer rolled back.
Bodies are fixed circles in root (L0) physical coordinates; velocities are
in root lattice units (dx=dt=1). Independent local copies are projected at
fine cell centers; their complete stencil must be active and disjoint from
the actual two-hop transfer footprint. These use the edge/cell-center frame
(coarse centers I+.5), whereas ordinary Simulation objects use the
integer-node frame. Jets (body.jets.Jet) use the same frame and drive cut
links selected at their true hits; ``set_jet_amplitude(index, amplitude,
level=l)`` re-uploads the per-link velocity column only.
``initialize(*level_g)`` accepts one physical Q-major (9, ny, nx) array per
level, already in postcollision phase. It does not collide or build channel
flags. Every ``step()`` publishes only after all levels complete on one
stream. On a device/launch failure the instance becomes unusable: close and
recreate it, rather than pretending a partially advanced buffer rolled back.
A ``level_blocks`` entry (by, bx) stores that level in blocks.cu storage:
the ENTIRE level shape, including the four-cell patch ghost band, is
partitioned by a non-periodic BlockLayout2D with one-cell same-level halos.
Patch kinds stay per cell (OUTER invalid, INNER stream only, ACTIVE
collide); halos are gathered before each hop. Public arrays keep the dense
shapes; only the allocation differs. A blocked level requires blocked
refined levels below it; several L1 rectangles and depth >= 3 require all
levels blocked. Blocked bodies are stationary circles, optionally
surface-actuated: prepare_fine_boundary decides legality on the dense grid,
then cut links are staged receiver-local per hop: halo -> BFL -> step, same
buffer. A blocked parent gathers halos before receiver-local coalescence
staging and pull/collision; covered cells are poisoned on every step.
``fine_block=(by, bx)`` (opt-in) stores the fine level in
blocks.cu storage: the ENTIRE fine_shape, including the four-cell patch
ghost band, is partitioned by a non-periodic BlockLayout2D with one-cell
same-level halos. Patch kinds stay per cell (OUTER invalid, INNER stream
only, ACTIVE collide); halos are gathered before each fine hop. Public
arrays keep the dense (9, ny, nx) fine shape; only the allocation differs.
Blocked fine_objects are stationary circles, optionally surface-actuated:
prepare_fine_boundary decides legality on the dense fine grid, then cut
links are staged receiver-local per hop: halo -> BFL -> step, same buffer.
``base_block=(by, bx)`` also blocks the parent; it requires ``fine_block``.
Parent halos precede receiver-local coalescence staging and pull/collision;
covered cells are explicitly poisoned on every step. Public shapes stay dense.
``observe_raw_exchange=True`` records blocked physical cut-link exchange
without changing populations. read_raw_exchange reports the last complete
base interval or last substep in base-lattice units per unit depth. This
is boundary momentum reaction, not a validated nozzle/stress load. Dense
body layouts are not supported by this new diagnostic in this candidate.
body levels are not supported by this diagnostic.
``observe_window_exchange=True`` (requires observe_raw_exchange) adds a second
per-link FP64 ledger accumulated by the same kernel between
start_exchange_window() and read_exchange_window(); disabled by default.
``jets`` (body.jets.Jet) use the same edge frame as ``fine_objects`` and
drive fine cut links selected at their true hits; ``set_jet_amplitude``
re-uploads the per-link velocity column only.
Several patches: pass ``bounds`` as a sequence of (x0, y0, x1, y1)
rectangles (FixedPatchSet2D, dependency-validated union). This requires
fine_block and base_block; fine_objects/jets are projected once onto the
level-wide fine grid (each circle inside the union, body ids = object
positions, jet and writer checks level-wide), and
uses one level-wide fine grid (2*ny, 2*nx): public fine arrays have that
shape, NaN outside the patches' ACTIVE cells. The step sequence is the
blocked one above, with unique union transfers in one launch each.
Three-level separated unions: ``child_rectangles_in_parent=[...]`` is
mutually exclusive with scalar ``child`` and accepts either parent form.
Coordinates index the allocated L1 array, including ghost bands and holes:
root edge = patch.fine_origin + L1 index / 2. A parent set has origin zero;
a scalar parent retains its shifted window origin. The child set's public
extent is twice the allocated L1 shape, even though storage is sparse.
Each pair requires gap >= 4 in its own parent-cell units. All regions share
one buffer pair per level and the fixed 1:2:4 schedule; L2-only bodies/Jets
below apply unchanged, with object IDs assigned over the whole level.
Three-level single chain: ``child`` bounds use L1 window cells; all three
levels must be blocked. ``child_objects`` and ``child_jets`` use the L0
edge frame, with L2 dx=dt=.25 and unchanged lattice velocity. L1 bodies
and Jets remain unsupported in a chain. Every whole circle and its BFL
stencil must fit strictly away from the L2 transfer footprint. L2 runs
halo -> BFL -> step four times per base step. Runtime child Jet changes
require ``set_jet_amplitude(index, amplitude, level=2)``.
Nested levels: a child set's coordinates index the allocated parent array,
including ghost bands and holes: root edge = parent fine_origin + index / 2
(zero origin for a parent set; a scalar parent keeps its shifted window
origin). A scalar child requires a scalar parent. Each pair requires gap >= 4
in its own parent-cell units. All regions share one buffer pair per level
and the fixed schedule of 2**l hops per base step at level l. Bodies/Jets are supported only on the
finest level of a nested hierarchy (L2 dx=dt=.25 in root units, unchanged
lattice velocity); every whole circle and its BFL stencil must fit strictly
away from that level's transfer footprint.
``explosion="projected_linear"`` selects experimental transverse linear
reconstruction on each adjacent pair. Default "constant" is unchanged.
@@ -214,43 +162,21 @@ class FixedPatchStepper:
bodies/Jets or LES; this backend supplies no forcing. Geometry is generic,
but union/multilevel GPU and matched-physics qualification remain pending.
Level lists (any depth): ``refinements[l]`` gives level l+1 in level-l
integer cells (scalar window or rectangle list, see lbm.hierarchy);
``level_blocks``/``level_objects``/``level_jets`` hold one entry per level,
L0 included. They are mutually exclusive with the legacy keywords above,
which are normalised to the same lists. One recursive hop advances every
level: a child explodes from its parent's current hop source, takes two
hops, then the parent gathers halos, stages restriction, applies BFL and
steps. Depth >= 3 must be fully blocked with bodies/Jets on the finest level.
One recursive hop advances every level: a child explodes from its parent's
current hop source, takes two hops, then the parent gathers halos, stages
restriction, applies BFL and steps.
"""
def __init__(self, cfg, bounds=None, *, context, coarse_boundary="periodic", fine_objects=(),
fine_block=None, base_block=None, jets=(), child=None, child_block=None,
child_objects=(), child_jets=(), child_rectangles_in_parent=None,
refinements=None, level_blocks=None, level_objects=None, level_jets=None,
def __init__(self, cfg, *, context, refinements, coarse_boundary="periodic",
level_blocks=None, level_objects=None, level_jets=None,
observe_raw_exchange=False, observe_window_exchange=False, explosion="constant"):
if HALO != 4:
raise RuntimeError("refinement.cu requires a four-cell fine halo")
if coarse_boundary not in ("periodic", "slot_reservoir", "channel"):
raise ValueError("coarse_boundary must be periodic, slot_reservoir or channel")
if refinements is None:
if bounds is None:
raise ValueError("bounds or refinements is required")
if level_blocks is not None or level_objects is not None or level_jets is not None:
raise ValueError("level_blocks/level_objects/level_jets require refinements")
refinements, level_blocks, level_objects, level_jets = _legacy_levels(
bounds, fine_block=fine_block, base_block=base_block, child=child,
child_block=child_block, child_rectangles_in_parent=child_rectangles_in_parent,
fine_objects=fine_objects, jets=jets, child_objects=child_objects,
child_jets=child_jets)
else:
if (bounds is not None or child is not None or child_rectangles_in_parent is not None
or fine_block is not None or base_block is not None or child_block is not None
or tuple(fine_objects) or tuple(jets) or tuple(child_objects) or tuple(child_jets)):
raise ValueError("refinements and the legacy patch keywords are mutually exclusive")
depth = len(refinements) + 1
level_blocks = [None] * depth if level_blocks is None else list(level_blocks)
level_objects = [()] * depth if level_objects is None else list(level_objects)
level_jets = [()] * depth if level_jets is None else list(level_jets)
depth = len(refinements) + 1
level_blocks = [None] * depth if level_blocks is None else list(level_blocks)
level_objects = [()] * depth if level_objects is None else list(level_objects)
level_jets = [()] * depth if level_jets is None else list(level_jets)
level_objects = [tuple(o) for o in level_objects]
level_jets = [tuple(j) for j in level_jets]
level_blocks = [None if b is None else tuple(b) for b in level_blocks]
@@ -540,13 +466,13 @@ class FixedPatchStepper:
if self._ddf_shifting and any(layout is None for layout in self._layouts):
raise ValueError("shifted populations require all-blocked levels")
def initialize(self, *level_g, coarse_g=None, fine_g=None, child_g=None):
def initialize(self, *level_g):
"""Upload physical postcollision G, encoding before any FP32 rounding.
Shifted input should be physical FP64 to retain small perturbations.
Public initialization resets the step count; non-leaves are discarded.
"""
given = _initial_levels(level_g, coarse_g, fine_g, child_g)
given = tuple(level_g)
with self._lock, self.ctx.activate():
self._require("new", "ready")
self._check_storage()
@@ -670,10 +596,9 @@ class FixedPatchStepper:
old, new, source, base_count=count, level_steps=(count, 2*count, 4*count),
phase="postcollision_g", **options)
candidate = type(self)(
cfg, pairs[0].bounds, context=self.ctx, base_block=base_block,
fine_block=fine_block, child_block=child_block,
child_rectangles_in_parent=new.child.bounds,
child_objects=child_objects, child_jets=jets,
cfg, context=self.ctx, refinements=[pairs[0].bounds, new.child.bounds],
level_blocks=[base_block, fine_block, child_block],
level_objects=[(), (), child_objects], level_jets=[(), (), jets],
observe_raw_exchange=self._observe_raw_exchange,
observe_window_exchange=self._observe_window, explosion=self._explosion)
try:
@@ -822,27 +747,41 @@ class FixedPatchStepper:
grid=((pairs + self._block[0] - 1) // self._block[0], 1), **kw)
def step(self, n=1):
"""Advance G→G by n base intervals: level l takes 2**l hops per interval."""
"""Advance G→G by n base intervals: level l takes 2**l hops per interval.
All n intervals are enqueued in order on the one stream, and the host waits
once, before returning; step(0) launches and waits for nothing. On return
the device work is complete, step_count and an open window cover all n
intervals, and the raw-exchange ledger holds the final interval only.
Asynchronous device errors surface at that single wait, so they are not
attributed to the interval that caused them; call step(1) repeatedly to
localize one (same launches, same bits). On any failure the instance is
failed, the exchange ledger and window are invalid, and step_count keeps
its value from entry to the failing call.
"""
if isinstance(n, bool) or not isinstance(n, Integral) or n < 0:
raise ValueError("n must be a nonnegative integer")
with self._lock, self.ctx.activate():
self._require("ready")
if n == 0:
return
root = self._levels[0]
kw = dict(block=self._block, stream=self._stream)
try:
self._status = "advancing"
self._exchange_valid = False
for _ in range(n):
if self._observe_raw_exchange:
self._exchange_valid = False
if self._observe_raw_exchange: # stream-ordered: holds the last interval only
for b in self._level_boundaries:
if b is not None:
cuda.memset_d8_async(b.interval_integral, 0,
len(b.tables["cell"]) * 32, self._stream)
self._hop(0, root.current, root.spare, **kw)
self._stream.synchronize()
root.current, root.spare = root.spare, root.current
self._step_count += 1
self._exchange_valid = self._observe_raw_exchange
root.current, root.spare = root.spare, root.current # host handles only
self._stream.synchronize()
self._step_count += n
self._exchange_valid = self._observe_raw_exchange
self._status = "ready"
except BaseException:
self._window_valid = False
+17 -8
View File
@@ -336,13 +336,19 @@ class Simulation:
def read_surface_exchange(self, id: int):
"""Latest step's wall/aperture (Fx,Fy,Tz) exchange, not total nozzle force.
Supports SurfaceActuation and Jet-driven bodies, including zero-amplitude
apertures. Passive bodies without an aperture are rejected.
Unlike read_body this is not accumulated/time-averaged. It resets after
geometry sync or checkpoint load; advance before interpreting it.
"""
if self.bodies.get(id).surface_actuation is None:
obj = self.bodies.get(id)
curved = self.field.curved
has_aperture = (curved.aperture is not None
and np.any(curved.aperture & (curved.body_id == id)))
if obj.surface_actuation is None and not has_aperture:
raise ValueError("Body has no surface actuation")
self.stream.synchronize()
return self.field.curved.surface_exchange(id)
return curved.surface_exchange(id)
@context_bound
def read_body(self, id: int, *,
@@ -451,18 +457,21 @@ class Simulation:
@context_bound
def initialize(self):
"""Initialize flow field and sync objects to GPU."""
# Recompile if objects were added after construction.
self._assert_allocation_contract()
# Validate the complete geometry before flags, populations or modules
# change. Compact-list construction publishes no outputs on failure.
next_flags = self.bodies.build_flags(self.field.build_channel_flags())
if self.bodies.jets and self.bodies.count == 0:
raise ValueError("Jets need bodies with cut links")
self.bodies.build_compact_lists_for(self.bodies.objects, next_flags)
# Recompile if objects were added after construction.
if self._read_generated_n_objects() != self.bodies.count:
self.recompile()
self.field.flag = self.field.build_channel_flags()
if self.bodies.count > 0:
self.field.flag = self.bodies.build_flags(self.field.flag)
self.field.flag = next_flags
self.field.upload_flags()
if self.bodies.jets and self.bodies.count == 0:
raise ValueError("Jets need bodies with cut links")
self.stepper.initialize()
if self.bodies.count > 0:
self.bodies.sync_to_gpu(self.field, rebuild_flags=False)
+102
View File
@@ -0,0 +1,102 @@
"""Host contract of PeriodicBlockStepper.step(n): one stream wait per call (mocked driver)."""
import ast
from contextlib import nullcontext
from pathlib import Path
import types
from unittest.mock import Mock
import unittest
import numpy as np
import CelerisLab.lbm.block_stepper as b
from CelerisLab.config import LBMConfig
from CelerisLab.lbm.blocks import BlockLayout2D
def host_module():
"""block_stepper with pycuda/compiler replaced by mocks; no GPU context."""
path = Path(b.__file__)
tree = ast.parse(path.read_text())
tree.body = [node for node in tree.body if not (
isinstance(node, ast.Import) and any(a.name == "pycuda.driver" for a in node.names)
or isinstance(node, ast.ImportFrom) and node.module == "cuda")]
module = types.ModuleType("CelerisLab.lbm._test_host_block_stepper")
module.__package__ = "CelerisLab.lbm"
module.__dict__.update(cuda=Mock(), compiler=Mock())
exec(compile(tree, str(path), "exec"), module.__dict__)
return module
def backend():
host, log = host_module(), []
host.cuda.mem_get_info.return_value = (1 << 40, 1 << 40)
host.cuda.mem_alloc.side_effect = lambda size: Mock(size=size) # distinct buffers
module = Mock()
module.get_global.return_value = (Mock(), 20)
module.get_function.side_effect = lambda name: Mock(
side_effect=lambda *a, **k: log.append((name, a[:2])))
host.compiler.load_module.return_value = module
cfg = LBMConfig(nx=16, ny=16, viscosity=.1)
run = host.PeriodicBlockStepper(cfg, BlockLayout2D.build(16, 16, 8, 8),
context=Mock(activate=nullcontext, sm_arch="sm_70"))
run._stream.synchronize.side_effect = lambda: log.append(("sync", ()))
run.initialize(np.full((9, 16, 16), 1 / 9))
log.clear()
return run, log
class BlockStepSync(unittest.TestCase):
def test_step_n_waits_once_and_matches_n_single_steps(self):
many, log_many = backend()
many.step(3)
single, log_single = backend()
for _ in range(3):
single.step(1)
names = [name for name, _ in log_many]
self.assertEqual(names.count("sync"), 1)
self.assertEqual(names[-1], "sync")
self.assertEqual(names.count("BlockStep"), 3)
# Same launch order and buffer roles as three step(1) calls.
strip = lambda log: [(n, tuple(id(x) for x in a)) for n, a in log if n != "sync"]
roles = lambda run, log: [(n, tuple({id(run._current): "c", id(run._spare): "s"}.get(x, "-")
for x in ids)) for n, ids in strip(log)]
self.assertEqual([n for n, _ in strip(log_many)], [n for n, _ in strip(log_single)])
self.assertEqual(roles(many, log_many), roles(single, log_single))
self.assertIsNot(many._current, many._spare)
self.assertEqual((many.step_count, single.step_count), (3, 3))
def test_step_zero_launches_and_waits_for_nothing(self):
run, log = backend()
run.step(0)
self.assertEqual((log, run.step_count), ([], 0))
def test_failure_keeps_entry_step_count(self):
run, log = backend()
run.step(1)
run._stream.synchronize.side_effect = RuntimeError("device fault")
with self.assertRaises(RuntimeError):
run.step(2)
self.assertEqual((run.step_count, run._status), (1, "failed"))
def test_launch_failure_mid_batch_fails_closed_without_waiting(self):
run, log = backend()
calls = []
step = run._step
def failing(*args, **kwargs):
calls.append(1)
if len(calls) == 2:
raise RuntimeError("launch failure")
return step(*args, **kwargs)
run._step = failing
with self.assertRaises(RuntimeError):
run.step(3)
self.assertNotIn("sync", [name for name, _ in log])
self.assertEqual((run.step_count, run._status), (0, "failed"))
with self.assertRaises(RuntimeError):
run.download()
if __name__ == "__main__":
unittest.main()
+2 -2
View File
@@ -80,8 +80,8 @@ def test_evolved_ghost_snapshot_and_physical_download():
host.cuda.memcpy_dtoh.side_effect = lambda a, ptr: np.copyto(a, storage[id(ptr)])
context = Mock(activate=nullcontext, sm_arch="sm_70")
cfg = LBMConfig(nx=24, ny=20, viscosity=.1, velocity=.025)
backend = host.FixedPatchStepper(cfg, (6, 6, 18, 14), context=context,
coarse_boundary="channel", base_block=(4, 6), fine_block=(4, 4))
backend = host.FixedPatchStepper(cfg, context=context, coarse_boundary="channel",
refinements=[(6, 6, 18, 14)], level_blocks=[(4, 6), (4, 4)])
try:
weights = np.array([4/9]+[1/9]*4+[1/36]*4)[:, None, None]
arrays = [np.broadcast_to(weights, level.shape).copy() for level in backend._levels]
+79
View File
@@ -0,0 +1,79 @@
"""CPU-only contract tests for run_dfg_re20. Not CUDA or DFG acceptance evidence."""
import sys
from pathlib import Path
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
import run_dfg_re20 as v # noqa: E402
import run_refined_dfg as rr # noqa: E402
h, _ = rr.helpers(rr.DEFAULT_SOURCE)
def rows(values, tstar_end=v.TSTAR_MAX):
n = int(round(tstar_end / v.SAMPLE_TSTAR))
return [dict(tstar=(i + 1) * v.SAMPLE_TSTAR, **values(i)) for i in range(n)]
class Re20Contract(unittest.TestCase):
def test_lattice_parameters_and_bounds(self):
root, uniform = v.lattice_parameters(40), v.lattice_parameters(80)
self.assertAlmostEqual(root["tau"], 0.56, places=12)
self.assertAlmostEqual(uniform["tau"], 0.62, places=12)
self.assertEqual((root["steps"], root["interval"]), (240000, 2000))
self.assertEqual((uniform["steps"], uniform["interval"]), (480000, 4000))
self.assertAlmostEqual(v.plan()["refined_fine_tau"], 0.62, places=12)
def test_config_only_changes_viscosity(self):
base = v._re100_config(h, 40, "SRT", "none")
cfg, body, p = v.config(h, 40, "SRT")
self.assertAlmostEqual(cfg["physics"]["viscosity"], 0.01 * 40 / 20, places=15)
self.assertEqual((p["re"], p["viscosity"]), (20.0, cfg["physics"]["viscosity"]))
base[0]["physics"]["viscosity"] = cfg["physics"]["viscosity"]
self.assertEqual(base[0], cfg)
self.assertEqual(base[1], body)
with self.assertRaisesRegex(ValueError, "no Jets"):
v.config(h, 40, "MRT", "zero")
def test_units(self):
s = v.sample([2.0, -0.5], 4, 40, 3.0)
den = rr.SPEED**2 * 40 / 2
self.assertAlmostEqual(s["cd"], 2.0 / 4 / den)
self.assertAlmostEqual(s["cl"], -0.5 / 4 / den)
self.assertAlmostEqual(s["p_diff"], 3.0 * 0.04)
self.assertEqual(s["dp_scaled"], 3.0)
def test_steady_and_verdict(self):
inside = dict(cd=5.58, cl=0.0106, p_diff=0.1174)
out = v.summarize(rows(lambda i: inside))
self.assertEqual((out["status"], out["steady"], out["window_samples"]),
("WITHIN_INTERVALS", True, 21))
self.assertTrue(out["cd_positive"])
self.assertAlmostEqual(out["edge_distance"]["cl"], 0.0002)
self.assertTrue(all(out["verdict"].values()))
low = v.summarize(rows(lambda i: dict(inside, cl=0.0100)))
self.assertEqual(low["status"], "OUTSIDE_INTERVALS")
self.assertFalse(low["verdict"]["cl"])
def test_slow_drift_gives_no_verdict(self):
# Range 2e-5 passes (limit 6e-5); slope * window 2e-5 exceeds 1.2e-5.
out = v.summarize(rows(lambda i: dict(cd=5.58, cl=0.0106 + 1e-6 * i, p_diff=0.1174)))
self.assertEqual((out["status"], out["verdict"]), ("NOT_STEADY", None))
self.assertLessEqual(out["ranges"]["cl"], out["range_limits"]["cl"])
self.assertGreater(out["drifts"]["cl"], out["slope_limits"]["cl"])
def test_noise_range(self):
quiet = v.summarize(rows(lambda i: dict(cd=5.58, cl=0.0106 + 2e-5 * (i % 2), p_diff=0.1174)))
self.assertTrue(quiet["steady"]) # within the range limit, no trend
noisy = v.summarize(rows(lambda i: dict(cd=5.58, cl=0.0106 + 1e-4 * (i % 2), p_diff=0.1174)))
self.assertEqual(noisy["status"], "NOT_STEADY")
self.assertGreater(noisy["ranges"]["cl"], noisy["range_limits"]["cl"])
def test_incomplete_run_gives_no_verdict(self):
out = v.summarize(rows(lambda i: dict(cd=5.58, cl=0.0106, p_diff=0.1174), tstar_end=30))
self.assertEqual((out["status"], out["verdict"]), ("INCOMPLETE", None))
if __name__ == "__main__":
unittest.main()
+34
View File
@@ -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
+49
View File
@@ -188,6 +188,55 @@ def jet_case(device, jets):
return field, bodies
def test_public_surface_exchange_for_jet_preserves_aperture_at_zero_and_reverse(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, _initialized=True,
stream=SimpleNamespace(synchronize=device.synchronize))
cl = field.curved
assert bodies.get(0).surface_actuation is None
aperture, velocity = cl.aperture, cl.surface_velocity.copy()
assert aperture.any() and (~aperture).any()
# Synthetic per-link exchanges test readback/splitting, not kernel physics.
values = np.arange(cl.count * 3, dtype=np.float32).reshape(-1, 3)
device.htod(cl.link_exchange_gpu, values)
for amplitude in (2e-3, 0., -1e-3):
Simulation.set_jet_amplitude(sim, 0, amplitude)
result = Simulation.read_surface_exchange(sim, 0)
assert cl.aperture is aperture
np.testing.assert_array_equal(aperture, cl.jet_id >= 0)
np.testing.assert_allclose(cl.surface_velocity, velocity * (amplitude / 2e-3))
for name, mask in (("wall", ~aperture), ("aperture", aperture)):
np.testing.assert_array_equal(result[name], values[mask].sum(axis=0, dtype=np.float64))
def test_public_surface_exchange_keeps_legacy_and_passive_guards(device):
from CelerisLab.simulation import Simulation
field = make_field(device)
bodies = manager(field, [circle(SurfaceActuation(0., .6, .002), radius=2.)])
bodies.sync_to_gpu(field)
sim = SimpleNamespace(bodies=bodies, field=field,
stream=SimpleNamespace(synchronize=device.synchronize))
for value in Simulation.read_surface_exchange(sim, 0).values():
np.testing.assert_array_equal(value, 0.)
field.curved.free_gpu_columns()
with pytest.raises(RuntimeError, match="diagnostic is allocated"):
Simulation.read_surface_exchange(sim, 0)
# An exchange buffer elsewhere in the scene must not admit a passive body.
for jets in ([], [Jet(Sector(CENTER, 0., .6, 3.), Radial(CENTER))]):
field, bodies = jet_case(device, jets)
passive = bodies.add(circle(center=(2., 3.), radius=.6))
bodies.sync_to_gpu(field)
sim.bodies, sim.field = bodies, field
with pytest.raises(ValueError, match="Body has no surface actuation"):
Simulation.read_surface_exchange(sim, passive)
if not jets:
with pytest.raises(ValueError, match="Body has no surface actuation"):
Simulation.read_surface_exchange(sim, 0)
def test_manager_passes_true_hits_and_reuploads_only_velocity(device):
# CPU memory double only: checks host/device bookkeeping, not GPU numerics.
jet = Jet(Sector(CENTER, 0., .6, 3.), Radial(CENTER, inward=False), "quartic", 2e-3)
+3 -2
View File
@@ -157,8 +157,9 @@ class RawExchange(unittest.TestCase):
host = host_module()
ctx = Mock()
with self.assertRaisesRegex(ValueError, 'blocked body'):
host.FixedPatchStepper(LBMConfig(nx=32, ny=32, viscosity=.1), (4, 4, 28, 28),
context=ctx, fine_objects=[body(radius=2.15)], observe_raw_exchange=True)
host.FixedPatchStepper(LBMConfig(nx=32, ny=32, viscosity=.1), context=ctx,
refinements=[(4, 4, 28, 28)],
level_objects=[(), (body(radius=2.15),)], observe_raw_exchange=True)
ctx.activate.assert_not_called()
def test_all_depth_weights(self):
+174 -4
View File
@@ -146,23 +146,26 @@ def test_full_requires_completed_pilot():
r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE, "MRT", "window"))
def pilot_fixture(path, observation_mode="window"):
def pilot_fixture(path, observation_mode="window", jet_scenario="none", case_selection="both", uniform_diameter=80, placement="wide"):
"""Synthetic contract fixture only; never a physical result or run receipt."""
manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")
manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario)
if observation_mode == "window" else
json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text()))
selection = dict(case_selection=case_selection, uniform_diameter=uniform_diameter, placement=placement)
(path / "plan.json").write_text(
json.dumps(
dict(
source_manifest=manifest,
**selection,
collision="MRT",
jet_scenario=jet_scenario,
observation_mode=observation_mode,
bounds_root_edge=list(r.BOUNDS),
bounds_root_edge=list(r.PLACEMENTS[placement]),
runner_sha256=r.runner_hash(),
)
)
)
for case, diameter in (("uniform", 80), ("refined", 40)):
for case, diameter in r.selected_cases(case_selection, uniform_diameter):
dest = path / case
dest.mkdir()
interval = round(0.02 * diameter / r.SPEED)
@@ -189,7 +192,9 @@ def pilot_fixture(path, observation_mode="window"):
json.dumps(
dict(
case=case,
**selection,
collision="MRT",
jet_scenario=jet_scenario,
observation_mode=observation_mode,
mode="pilot",
status="pilot_complete",
@@ -206,7 +211,9 @@ def pilot_fixture(path, observation_mode="window"):
json.dumps(
dict(
case=case,
**selection,
collision="MRT",
jet_scenario=jet_scenario,
observation_mode=observation_mode,
mode="pilot",
step=interval,
@@ -451,6 +458,169 @@ def test_window_contract_rejection():
assert not backend.events
def test_public_jet_configuration_and_geometry():
for diameter, center, radius in ((40, (80.5, 81), 20), (80, (160, 160.5), 40)):
cfg, _, _ = r.config(h, diameter, "MRT", "fixed")
assert not cfg["method"]["ddf_shifting"] and not cfg["method"]["les"]["enabled"]
jets = r.public_jets(center, radius, "fixed")
assert [j.amplitude for j in jets] == [.001, -.001]
assert all(j.field.inward is False and j.profile == "parabolic" for j in jets)
assert all(j.region.r_max == radius * 1.005 for j in jets)
assert all(j.amplitude == 0 for j in r.public_jets(center, radius, "zero"))
cfg, _, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT", "fixed")
assert not cfg.ddf_shifting and len(host.boundary_cells) > 0
assert r.public_jets((0, 0), 1, "none") == ()
with unittest.TestCase().assertRaises(ValueError):
r.config(h, 40, "SRT", "fixed")
def test_public_jet_reversal_boundary_and_windows():
from types import SimpleNamespace
calls = []
handle = SimpleNamespace(set_jet_amplitude=lambda i, a: calls.append((i, a)))
for diameter in (40, 80):
boundary = round(250 * diameter / r.SPEED)
assert r.jet_interval(handle, "fixed", boundary - 3, 80, diameter) == (3, None)
switch = round(300 * diameter / r.SPEED)
assert r.jet_interval(handle, "reverse", switch - 3, 80, diameter) == (3, None)
assert not calls
n, event = r.jet_interval(handle, "reverse", switch, 80, diameter)
assert n == 80 and event["step"] == switch and event["tstar"] == 300
assert calls == [(0, -.001), (1, .001)]
calls.clear()
assert r.jet_interval(handle, "reverse", switch + 80, 80, diameter) == (80, None)
assert r.jet_interval(handle, "fixed", switch, 80, diameter) == (80, None)
assert not calls
rows = [dict(tstar=t, tstar_force=t - .5, cd=2.0, cl=-.1) for t in range(251, 306)]
windows = r.jet_diagnostics(rows, "reverse")
assert len(windows) == 6 and all(w["complete"] for w in windows)
assert all(w["cd"]["mean"] == 2 and np.isclose(w["cl"]["mean"], -.1) for w in windows)
assert windows[-1]["covered_tstar"] == 5
with unittest.TestCase().assertRaisesRegex(ValueError, "crosses"):
r.jet_diagnostics([dict(tstar=250.5, tstar_force=250, cd=1, cl=0)], "fixed")
assert r.plan(h, "MRT", "window", "reverse")["modes"]["full_tstar"] == 305
def test_public_jet_source_and_pilot_separation():
from tempfile import TemporaryDirectory
with TemporaryDirectory() as tmp:
path = Path(tmp)
manifest = pilot_fixture(path, jet_scenario="fixed")
assert manifest["source_capture"].startswith("current checkout")
assert len(manifest["checkout_revision"]) == 40
assert r.verify_pilot(path, manifest, "MRT", "window", "fixed") == 3606.0
for scenario in ("none", "zero", "reverse"):
with unittest.TestCase().assertRaisesRegex(ValueError, "Jet scenario"):
r.verify_pilot(path, manifest, "MRT", "window", scenario)
for relative in ("uniform/summary.json", "refined/ready.json"):
receipt = path / relative
record = json.loads(receipt.read_text())
receipt.write_text(json.dumps(dict(record, jet_scenario="zero")))
with unittest.TestCase().assertRaises(ValueError):
r.verify_pilot(path, manifest, "MRT", "window", "fixed")
receipt.write_text(json.dumps(record))
with unittest.TestCase().assertRaises(ValueError):
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.]),
(False, (658, 3521), (.375, .75), 160, [64., -128.]),
):
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"):
for record in records[1:]:
np.testing.assert_allclose(records[0][key], record[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
def test_compact_geometry_and_cv_stencils():
from CelerisLab.lbm.refinement import ACTIVE
from CelerisLab.lbm.descriptors import SOLID
_, pair, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT", "fixed", "compact")
assert pair.fine_shape == (168, 168) and tuple(pair.fine_origin) == (46, 46)
fluid = (pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)
rho, ux, uy = np.ones(pair.fine_shape), np.full(pair.fine_shape, .01), np.zeros(pair.fine_shape)
record = r.cv_snapshot(rho, ux, uy, fluid, pair.fine_origin, True, 1160000, 80, 40, [1, 2], "compact")
np.testing.assert_allclose(record["outward_flux"], 0, atol=1e-12)
stencil = np.zeros(pair.fine_shape, bool)
for rect in r.cv_contours(True, 40, "compact"):
x0, y0, x1, y1 = np.rint((rect - np.tile(pair.fine_origin, 2))/.5).astype(int)
stencil[y0-2:y1+2, x0-2:x0+2] = stencil[y0-2:y1+2, x1-2:x1+2] = True
stencil[y0-2:y0+2, x0-2:x1+2] = stencil[y1-2:y1+2, x0-2:x1+2] = True
assert fluid[stencil].all() and not stencil.ravel()[host.transfer_fine_cells].any()
def test_d160_config_and_single_case_receipts():
from tempfile import TemporaryDirectory
from unittest.mock import patch
cfg, _, p = r.config(h, 160, "MRT", "fixed")
assert (p["nx"], p["ny"], p["center"], p["radius"]) == (3521, 658, [320, 320.5], 80)
assert p["viscosity"] == .016 and p["mean_speed"] == .01 and not cfg["method"]["ddf_shifting"]
assert r.uniform_origin(160) == (.375, .75)
np.testing.assert_array_equal(r.cv_contours(False, 160), r.CV_CONTOURS + [.125, 0, .125, 0])
for scenario, expected_exit in (("fixed", 0), ("zero", 2)):
with TemporaryDirectory() as tmp:
path = Path(tmp)
manifest = pilot_fixture(path, jet_scenario=scenario, case_selection="uniform", uniform_diameter=160)
assert r.verify_pilot(path, manifest, "MRT", "window", scenario, False, "uniform", 160) == 1803
for selection, diameter, placement in (("both", 160, "wide"), ("uniform", 80, "wide"), ("uniform", 160, "compact")):
with unittest.TestCase().assertRaises(ValueError):
r.verify_pilot(path, manifest, "MRT", "window", scenario, False, selection, diameter, placement)
# Synthetic dispatch/result check: no CUDA imports or execution.
output = path / "full"
argv = ["runner", "--collision", "MRT", "--exchange-window", "--jet-scenario", scenario,
"--case", "uniform", "--uniform-diameter", "160", "--mode", "full",
"--pilot-results", str(path), "--output", str(output), "--max-wall-seconds", "10", "--gpu-gates-approved"]
result = dict(case="uniform", healthy=True, periodic=scenario == "zero", jet_diagnostic_windows=[],
status="driven_diagnostics_complete" if scenario == "fixed" else "periodic",
reference_comparison=dict(reference_agreement=False))
with patch.object(sys, "argv", argv), patch.dict(r.os.environ, CUDA_VISIBLE_DEVICES="mock-no-gpu"), \
patch.object(r, "run_case", return_value=result) as run:
assert r.main() == expected_exit and run.call_count == 1
comparison = json.loads((output / "comparison.json").read_text())
assert comparison["matched_case_pass"] is False and "refined_vs_uniform" not in comparison
assert comparison["status"].startswith("single_case_")
if __name__ == "__main__":
suite = unittest.TestSuite(
unittest.FunctionTestCase(fn)
@@ -173,8 +173,10 @@ def bodies_and_jets(speed=.002):
return bodies,jets
KW=dict(base_block=(4,4),fine_block=(4,4),child_block=(4,4),
child_rectangles_in_parent=CHILDREN)
def levels(parents=PARENTS,children=CHILDREN,blocks=((4,4),)*3,objects=((),(),()),jets=((),(),())):
"""Level-list constructor keywords for the L0 set, L1 set and L2 set used here."""
return dict(refinements=[parents,children],level_blocks=list(blocks),
level_objects=list(objects),level_jets=list(jets))
def mock_backend(*, children=CHILDREN, bodies=False, fail_compile=False, fail_upload=False):
@@ -201,9 +203,8 @@ def mock_backend(*, children=CHILDREN, bodies=False, fail_compile=False, fail_up
module.cuda.memcpy_htod.side_effect=upload
objects,jets=bodies_and_jets() if bodies else ((),())
try:
run=module.FixedPatchStepper(CFG,PARENTS,context=context,
child_objects=objects,child_jets=jets,
**dict(KW,child_rectangles_in_parent=children))
run=module.FixedPatchStepper(CFG,context=context,
**levels(children=children,objects=((),(),objects),jets=((),(),jets)))
except RuntimeError:
for allocation in allocations:
allocation.free.assert_called_once_with()
@@ -214,28 +215,29 @@ def mock_backend(*, children=CHILDREN, bodies=False, fail_compile=False, fail_up
class ConstructorAndSchedule(unittest.TestCase):
def test_invalid_inputs_fail_before_context_or_allocations(self):
context=Mock()
cases=[dict(child=(8,8,16,16)),dict(child_rectangles_in_parent=[]),
dict(child_rectangles_in_parent=[(16,16,32)]),
dict(child_rectangles_in_parent=[(16.5,16,32,40)]),
dict(child_rectangles_in_parent=[(16,16,32,24),(16,27,32,55)]),
dict(child_rectangles_in_parent=[(40,16,48,40)]),
dict(child_rectangles_in_parent=[(8,16,32,40)]),
dict(child_block=None),dict(fine_block=None),dict(base_block=None),
dict(child_block=(5,4)),dict(fine_block=(5,4)),
dict(child_block=(4,)),dict(child_rectangles_in_parent=(16,16,32,40))]
b=(4,4)
cases=[dict(children=(8,8,16,16)),dict(children=[]),
dict(children=[(16,16,32)]),
dict(children=[(16.5,16,32,40)]),
dict(children=[(16,16,32,24),(16,27,32,55)]),
dict(children=[(40,16,48,40)]),
dict(children=[(8,16,32,40)]),
dict(blocks=(b,b,None)),dict(blocks=(b,None,b)),dict(blocks=(None,b,b)),
dict(blocks=(b,b,(5,4))),dict(blocks=(b,(5,4),b)),
dict(blocks=(b,b,(4,))),dict(children=(16,16,32,40))]
for changes in cases:
with self.subTest(changes=changes),self.assertRaises((ValueError,TypeError)):
host.FixedPatchStepper(CFG,PARENTS,context=context,**dict(KW,**changes))
host.FixedPatchStepper(CFG,context=context,**levels(**changes))
with self.assertRaisesRegex(ValueError,'margin'):
host.FixedPatchStepper(CFG,[(1,4,20,28),(23,4,44,28)],context=context,**KW)
host.FixedPatchStepper(CFG,context=context,**levels(parents=[(1,4,20,28),(23,4,44,28)]))
objects,jets=bodies_and_jets()
with self.assertRaisesRegex(ValueError,'only on L2'):
host.FixedPatchStepper(CFG,PARENTS,context=context,fine_objects=objects,**KW)
with self.assertRaisesRegex(ValueError,'only on the finest level'):
host.FixedPatchStepper(CFG,context=context,**levels(objects=((),objects,())))
from CelerisLab.body.jets import Jet,Tube,Parallel
spanning=Jet(Tube((10.5,14.),0.,25.,.3),Parallel(0.),amplitude=0.)
with self.assertRaisesRegex(ValueError,'one jet'):
host.FixedPatchStepper(CFG,PARENTS,context=context,child_objects=objects,
child_jets=[spanning],**KW)
host.FixedPatchStepper(CFG,context=context,
**levels(objects=((),(),objects),jets=((),(),[spanning])))
context.activate.assert_not_called()
host.cuda.mem_alloc.assert_not_called()
@@ -313,21 +315,23 @@ class ConstructorAndSchedule(unittest.TestCase):
class InitialLevels(unittest.TestCase):
"""initialize() entry adapter: positional per-level arrays or the legacy names, never both."""
class RemovedLegacyAPI(unittest.TestCase):
"""The pre-level-list constructor/initialize forms are gone and fail loudly."""
def test_forms(self):
a, b, c = object(), object(), object()
f = host._initial_levels
self.assertEqual(f((a, b), None, None, None), (a, b))
self.assertEqual(f((a, b, c), None, None, None), (a, b, c))
self.assertEqual(f((a, b), None, None, c), (a, b, c)) # old initialize(c, f, child_g=g)
self.assertEqual(f((), a, b, None), (a, b))
self.assertEqual(f((), a, b, c), (a, b, c))
for args in (((a,), None, None, c), ((a, b, c), None, None, c), ((a,), a, b, None),
((), a, None, None), ((), None, b, c)):
with self.subTest(args=args), self.assertRaises(TypeError):
f(*args)
def test_removed_constructor_forms_raise_type_error(self):
import inspect
context = Mock()
with self.assertRaises(TypeError): # positional bounds
host.FixedPatchStepper(CFG, PARENTS, context=context)
removed = dict(bounds=PARENTS, fine_block=(4, 4), base_block=(4, 4), fine_objects=(),
jets=(), child=(8, 8, 16, 16), child_block=(4, 4), child_objects=(),
child_jets=(), child_rectangles_in_parent=CHILDREN)
for name, value in removed.items():
with self.subTest(keyword=name), self.assertRaises(TypeError):
host.FixedPatchStepper(CFG, context=context, **levels(), **{name: value})
context.activate.assert_not_called()
self.assertEqual(list(inspect.signature(host.FixedPatchStepper.initialize).parameters),
["self", "level_g"]) # no coarse_g/fine_g/child_g
if __name__=='__main__':
+5 -6
View File
@@ -54,14 +54,13 @@ class ShiftedJets(unittest.TestCase):
(Jet(Tube((10.,14.),0.,26.,.3),Parallel(0.)),))]
for objects,jets in cases:
with self.subTest(objects=len(objects),jets=len(jets)),self.assertRaises(ValueError):
module.FixedPatchStepper(CFG,backend.PARENTS,context=None,
base_block=(4,4),fine_block=(4,4),child_block=(4,4),
child_rectangles_in_parent=A,child_objects=objects,child_jets=jets)
module.FixedPatchStepper(CFG,context=None,refinements=[backend.PARENTS,A],
level_blocks=[(4,4)]*3,level_objects=[(),(),objects],level_jets=[(),(),jets])
for blocks in ((None,None,None),((4,4),None,(4,4))):
with self.assertRaises(ValueError):
module.FixedPatchStepper(CFG,backend.PARENTS,context=None,
base_block=blocks[0],fine_block=blocks[1],child_block=blocks[2],
child_rectangles_in_parent=A,child_objects=(body(),),child_jets=(jet(),))
module.FixedPatchStepper(CFG,context=None,refinements=[backend.PARENTS,A],
level_blocks=list(blocks),level_objects=[(),(),(body(),)],
level_jets=[(),(),(jet(),)])
module.cuda.mem_alloc.assert_not_called()
module.compiler.compile_kernel.assert_not_called()
+4 -3
View File
@@ -129,9 +129,10 @@ class ShiftedStorage(unittest.TestCase):
def test_dense_or_mixed_shifted_rejected_before_compilation(self):
module=backend.host_module()
cfg=replace(backend.CFG,ddf_shifting=True)
for kwargs in ({},dict(base_block=(4,4)),dict(fine_block=(4,4))):
with self.assertRaisesRegex(ValueError,'all-blocked|base_block requires fine_block'):
module.FixedPatchStepper(cfg,(4,4,20,28),context=None,**kwargs)
for blocks in ([None,None],[(4,4),None],[None,(4,4)]):
with self.assertRaisesRegex(ValueError,'all-blocked|blocked refined levels below'):
module.FixedPatchStepper(cfg,context=None,refinements=[(4,4,20,28)],
level_blocks=blocks)
module.compiler.compile_kernel.assert_not_called()
def test_uniform_block_codec_and_raw_snapshot(self):
+75
View File
@@ -238,6 +238,81 @@ def test_overlapping_force_regions_rejected(device):
bodies.build_compact_lists(bodies.build_flags(field.flag))
@pytest.mark.parametrize("second_x,error", [(14., "unique cut-link writers"),
(17., "active fluid receivers")])
def test_passive_overlap_rejected_before_publishing_lists(second_x, error):
cfg = LBMConfig(nx=32, ny=24)
cfg.validate()
bodies = ObjectManager(cfg.nx, cfg.ny, cfg.nz, cfg.nq, cfg)
for x in (14., second_x):
bodies.add(obj(x=x, y=12., r=4.3))
bodies.add(obj(x=4., y=4., r=2., is_sensor=True))
counts = np.full(bodies.count, 91, np.int32)
hits = []
flags = bodies.build_flags(np.full(cfg.nx * cfg.ny, FLUID, np.uint16))
for domain_flags in (flags, None):
with pytest.raises(ValueError, match=error):
bodies.build_compact_lists_for(bodies.objects, domain_flags,
sensor_cell_counts_out=counts, hits_out=hits)
np.testing.assert_array_equal(counts, 91)
assert hits == []
@pytest.mark.parametrize("centers", [((10., 12.), (22., 12.)),
((10.5, 12.), (19.5, 12.))])
def test_separated_passive_bodies_keep_valid_writers(centers):
# The second case has a subcell physical gap with a resolved fluid column.
cfg = LBMConfig(nx=32, ny=24)
cfg.validate()
bodies = ObjectManager(cfg.nx, cfg.ny, cfg.nz, cfg.nq, cfg)
for x, y in centers:
bodies.add(obj(x=x, y=y, r=4.3))
flags = bodies.build_flags(np.full(cfg.nx * cfg.ny, FLUID, np.uint16))
indices, directions, _, ids, *_ = bodies.build_compact_lists(flags)
assert set(ids) == {0, 1}
assert np.all(flags[indices] & FLUID) and not np.any(flags[indices] & SOLID)
assert len(np.unique(indices.astype(np.uint64) * cfg.nq + directions)) == len(indices)
@pytest.mark.parametrize("entry", ["sync", "initialize", "runtime_edit"])
@pytest.mark.parametrize("second_x", [5., 6.])
def test_passive_overlap_failure_keeps_live_state(device, entry, second_x):
from CelerisLab.simulation import Simulation
field = make_field(device)
bodies = manager(field, [obj(), obj(x=9., y=3., r=1., is_sensor=True)])
bodies.sync_to_gpu(field)
overlapping = obj(x=second_x)
if entry == "runtime_edit":
bodies.stage_add(overlapping)
else:
bodies.add(overlapping)
objects = bodies.objects
flags, curved, counts = field.flag.copy(), field.curved.fluid_idx.copy(), bodies.sensor_cell_counts.copy()
action = bodies.action.copy()
buffers = [(block, bytes(block.data), block.freed) for block in device.blocks]
writes = device.writes
sim = SimpleNamespace(bodies=bodies, field=field, lbm_cfg=field.cfg,
_initialized=entry == "runtime_edit",
_assert_allocation_contract=lambda: None,
stream=SimpleNamespace(synchronize=device.synchronize))
with pytest.raises(ValueError, match="active fluid receivers|unique cut-link writers"):
if entry == "sync":
bodies.sync_to_gpu(field)
elif entry == "initialize":
Simulation.initialize(sim)
else:
Simulation.sync_bodies(sim)
assert bodies.objects == objects
assert bodies.has_pending_edit() == (entry == "runtime_edit")
assert sim._initialized == (entry == "runtime_edit")
assert device.writes == writes and len(device.blocks) == len(buffers)
assert all(bytes(block.data) == data and block.freed == freed for block, data, freed in buffers)
np.testing.assert_array_equal(field.flag, flags)
np.testing.assert_array_equal(field.curved.fluid_idx, curved)
np.testing.assert_array_equal(bodies.sensor_cell_counts, counts)
np.testing.assert_array_equal(bodies.action, action)
def test_action_upload_and_scratch_are_pinned(device):
field = make_field(device, streaming="esopull")
bodies = manager(field, [obj(is_force_region=True)])
+170
View File
@@ -0,0 +1,170 @@
"""Host contract of FixedPatchStepper.step(n): one stream wait per call (mocked driver).
All n base intervals are enqueued in order on the backend stream and the host
waits once before step returns. Launch order and buffer arguments must equal n
step(1) calls; completion metadata, raw-exchange and window ledgers, and failure
invalidation are published only after that wait.
"""
from contextlib import nullcontext
from unittest.mock import Mock
import unittest
import numpy as np
from CelerisLab.config import LBMConfig
from tests.unit.test_raw_exchange import body
from tests.unit.test_refinement_three_level_multipatch import host_module
BODY_LEVEL, HOPS = 3, 8 # body on L3: 2**3 hops per base interval
def backend(window=False):
"""Four-level host backend; every launch, memset and stream wait goes to one log."""
host = host_module()
log, allocations = [], []
def allocate(size):
allocations.append(Mock(size=size))
return allocations[-1]
module = Mock()
module.get_global.return_value = (Mock(), 20)
module.get_function.side_effect = lambda name: Mock(
side_effect=lambda *a, **k: log.append((name, a, k)))
host.compiler.load_module.return_value = module
host.cuda.mem_alloc.side_effect = allocate
host.cuda.memcpy_dtoh.side_effect = lambda array, pointer: array.fill(0)
host.cuda.memset_d8_async.side_effect = lambda *a: log.append(("memset", a, {}))
run = host.FixedPatchStepper(
LBMConfig(nx=48, ny=48, viscosity=.02), context=Mock(activate=nullcontext, sm_arch="sm_70"),
refinements=[[(6, 6, 26, 26)], [(20, 20, 44, 44)], [(56, 56, 72, 72)]],
level_blocks=[(4, 4)] * 4, level_objects=[(), (), (), (body(),)],
observe_raw_exchange=True, observe_window_exchange=window)
run._stream.synchronize.side_effect = lambda: log.append(("sync", (), {}))
run.initialize(*[np.full(level.shape, 1 / 9, np.float32) for level in run._levels])
log.clear()
return run, log, allocations
def signature(log, allocations):
"""Event names with device-pointer arguments replaced by allocation index."""
index = {id(a): i for i, a in enumerate(allocations)}
return [(name, tuple(index.get(id(a), "-") for a in args)) for name, args, _ in log]
def names(log):
return [name for name, _, _ in log]
class StepSync(unittest.TestCase):
def test_step1_waits_once_after_all_launches(self):
run, log, _ = backend()
run.step(1)
self.assertEqual(names(log).count("sync"), 1)
self.assertEqual(names(log)[-1], "sync")
self.assertEqual(names(log)[0], "memset")
self.assertEqual(names(log).count("BlockStep"), 1 + 2 + 4 + 8)
self.assertEqual(run.step_count, 1)
run.close()
def test_step_n_equals_n_step1_launches_with_one_wait(self):
single, single_log, single_alloc = backend()
for _ in range(3):
single.step(1)
batched, batched_log, batched_alloc = backend()
batched.step(3)
expected = [e for e in signature(single_log, single_alloc) if e[0] != "sync"]
got = signature(batched_log, batched_alloc)
self.assertEqual(got[-1][0], "sync")
self.assertEqual(names(batched_log).count("sync"), 1)
self.assertEqual(got[:-1], expected) # same order, same ping-pong buffers
self.assertEqual(names(batched_log).count("memset"), 3)
self.assertEqual(batched.step_count, single.step_count)
index = [{id(a): i for i, a in enumerate(alloc)} for alloc in (single_alloc, batched_alloc)]
self.assertEqual(index[1][id(batched._levels[0].current)],
index[0][id(single._levels[0].current)])
single.close()
batched.close()
def test_step0_launches_and_waits_for_nothing(self):
run, log, _ = backend()
run.step(2)
log.clear()
run.step(0)
self.assertEqual(log, [])
self.assertEqual(run.step_count, 2)
value = run.read_raw_exchange(0, level=BODY_LEVEL) # step(0) keeps the ledger valid
self.assertEqual((value["time_start"], value["time_end"]), (1, 2))
run.close()
def test_completion_metadata_covers_all_intervals(self):
run, _, _ = backend()
with self.assertRaisesRegex(RuntimeError, "completed"):
run.read_raw_exchange(0, level=BODY_LEVEL)
run.step(3)
value = run.read_raw_exchange(0, level=BODY_LEVEL)
self.assertEqual((value["time_start"], value["time_end"], value["sample_count"]),
(2, 3, HOPS)) # step(n) retains only the final base interval
run.step(2)
self.assertEqual(run.read_raw_exchange(0, level=BODY_LEVEL)["time_end"], 5)
run.close()
def test_window_accounting_across_batched_calls(self):
windows = []
for calls in ((1, 3), (4,), (1, 1, 1, 1)):
run, log, _ = backend(window=True)
run.step(1)
run.start_exchange_window()
log.clear()
for n in calls:
run.step(n)
self.assertEqual(names(log).count("BlockAccumulateExchange"), 2 * 4 * HOPS)
self.assertEqual(names(log).count("sync"), len(calls))
out = run.read_exchange_window(0, level=BODY_LEVEL)
windows.append((out["time_start"], out["time_end"], out["sample_count"]))
run.close()
self.assertEqual(windows, [(1., 5, 4 * HOPS)] * 3)
def test_failed_wait_publishes_nothing(self):
run, _, _ = backend(window=True)
run.step(2)
run.start_exchange_window()
run._stream.synchronize.side_effect = RuntimeError("device failure")
with self.assertRaisesRegex(RuntimeError, "device failure"):
run.step(3)
self.assertEqual(run.step_count, 2) # value at entry to the failing call
self.assertFalse(run._exchange_valid)
self.assertFalse(run._window_valid)
for read in (lambda: run.read_raw_exchange(0, level=BODY_LEVEL),
lambda: run.read_exchange_window(0, level=BODY_LEVEL), run.download,
lambda: run.step(1)):
with self.assertRaisesRegex(RuntimeError, "failed"):
read()
run._stream.synchronize.side_effect = None
run.close()
def test_launch_error_mid_batch_skips_wait_and_fails(self):
run, log, _ = backend()
run.step(1)
log.clear()
hop, calls = run._hop, []
def failing(index, *args, **kw):
if index == 0:
calls.append(1)
if len(calls) == 2:
raise RuntimeError("launch failure")
return hop(index, *args, **kw)
run._hop = failing
with self.assertRaisesRegex(RuntimeError, "launch failure"):
run.step(3)
self.assertNotIn("sync", names(log))
self.assertEqual(run.step_count, 1)
self.assertFalse(run._exchange_valid)
with self.assertRaisesRegex(RuntimeError, "failed"):
run.read_raw_exchange(0, level=BODY_LEVEL)
run.close()
if __name__ == "__main__":
unittest.main()
@@ -106,9 +106,10 @@ class ChannelRootGPU(unittest.TestCase):
cfg = LBMConfig(nx=48, ny=32, viscosity=.1, velocity=.015, ddf_shifting=shifted)
reference = None
for block in ((32,48), (8,8), (4,1)):
with FixedPatchStepper(cfg, (8,6,40,26), context=context,
coarse_boundary="channel", base_block=block, fine_block=(8,8),
fine_objects=(body,) if bodies else (), jets=(jet,) if bodies else ()) as backend:
with FixedPatchStepper(cfg, context=context, coarse_boundary="channel",
refinements=[(8,6,40,26)], level_blocks=[block, (8,8)],
level_objects=[(), (body,) if bodies else ()],
level_jets=[(), (jet,) if bodies else ()]) as backend:
arrays = [state(*level.shape[1:]) for level in backend._levels]
backend.initialize(*arrays)
backend.step(4)
@@ -25,8 +25,8 @@ EPS = np.finfo(np.float64).eps
def build(m, window=True, raw=True):
return FixedPatchStepper(config("periodic"), BOUNDS, context=m["context"], base_block=(3, 4),
fine_block=(16, 20), fine_objects=[circle()],
return FixedPatchStepper(config("periodic"), context=m["context"], refinements=[BOUNDS],
level_blocks=[(3, 4), (16, 20)], level_objects=[(), (circle(),)],
observe_raw_exchange=raw, observe_window_exchange=window)
@@ -109,10 +109,11 @@ class ExchangeWindowGPU(unittest.TestCase):
from test_raw_exchange_gpu import BOUNDS as RB, CFG, circle as rc, jet, state
from CelerisLab.cuda.context import CudaContext
with CudaContext(0) as ctx:
kw = dict(context=ctx, fine_block=(16, 20), fine_objects=[rc()], jets=[jet()],
kw = dict(context=ctx, refinements=[RB], level_blocks=[None, (16, 20)],
level_objects=[(), (rc(),)], level_jets=[(), (jet(),)],
observe_raw_exchange=True)
with FixedPatchStepper(CFG, RB, observe_window_exchange=True, **kw) as windowed, \
FixedPatchStepper(CFG, RB, **kw) as plain:
with FixedPatchStepper(CFG, observe_window_exchange=True, **kw) as windowed, \
FixedPatchStepper(CFG, **kw) as plain:
arrays = [state(level.shape) for level in windowed._levels]
windowed.initialize(*arrays)
plain.initialize(*arrays)
@@ -200,11 +201,11 @@ class ExchangeWindowShiftedMRTGPU(unittest.TestCase):
from CelerisLab.cuda.context import CudaContext
cfg = LBMConfig(nx=32, ny=24, viscosity=.1, collision="MRT", ddf_shifting=True)
with CudaContext(0) as ctx:
kw = dict(context=ctx, base_block=(12, 8), fine_block=(16, 20), fine_objects=[rc()],
observe_raw_exchange=True)
with FixedPatchStepper(cfg, RB, observe_window_exchange=True, **kw) as windowed, \
FixedPatchStepper(cfg, RB, observe_window_exchange=True, **kw) as split, \
FixedPatchStepper(cfg, RB, **kw) as plain:
kw = dict(context=ctx, refinements=[RB], level_blocks=[(12, 8), (16, 20)],
level_objects=[(), (rc(),)], observe_raw_exchange=True)
with FixedPatchStepper(cfg, observe_window_exchange=True, **kw) as windowed, \
FixedPatchStepper(cfg, observe_window_exchange=True, **kw) as split, \
FixedPatchStepper(cfg, **kw) as plain:
arrays = [state(level.shape) for level in windowed._levels]
for run in (windowed, split, plain):
run.initialize(*arrays)
@@ -140,7 +140,8 @@ def run_patch(speed, *, audit=True):
build = (patch.object(compiler, "compile_kernel", audit_compiler(compiler.compile_kernel))
if audit else nullcontext())
with build:
backend = implementation.FixedPatchStepper(cfg, BOUNDS, context=context, fine_objects=[source])
backend = implementation.FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
level_objects=[(), (source,)])
with backend, context.activate():
c, f = backend._levels
cl = backend._fine_curved
@@ -325,6 +326,7 @@ def test_cross_interface_donor_rejects_before_cuda():
context.activate.side_effect = AssertionError("must reject before CUDA activation")
with patch.object(cuda, "mem_alloc") as allocate:
with pytest.raises(ValueError, match="donor crosses"):
implementation.FixedPatchStepper(cfg, BOUNDS, context=context, fine_objects=[source])
implementation.FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
level_objects=[(), (source,)])
context.activate.assert_not_called()
allocate.assert_not_called()
@@ -48,9 +48,8 @@ def make(context):
for (x, y), amplitude in zip(centers, (.003, -.002))
)
run = FixedPatchStepper(
CFG, PARENTS, context=context, base_block=(4, 4), fine_block=(4, 4),
child_block=(4, 4), child_rectangles_in_parent=START,
child_objects=objects, child_jets=jets,
CFG, context=context, refinements=[PARENTS, START], level_blocks=[(4, 4)] * 3,
level_objects=[(), (), objects], level_jets=[(), (), jets],
)
return run, objects
@@ -25,9 +25,8 @@ B = ((16,16,32,40), (60,16,76,40))
def make(ctx):
body = SimObject(71, geometry=CircleGeometry(12.,14.,.8),center=(12.,14.),radius=.8)
jet = Jet(Tube((10.5,14.),0.,3.,.3),Parallel(0.),'parabolic',.003)
run = FixedPatchStepper(CFG,PARENTS,context=ctx,base_block=(4,4),fine_block=(4,4),
child_block=(4,4),child_rectangles_in_parent=A,
child_objects=(body,),child_jets=(jet,))
run = FixedPatchStepper(CFG,context=ctx,refinements=[PARENTS,A],level_blocks=[(4,4)]*3,
level_objects=[(),(),(body,)],level_jets=[(),(),(jet,)])
return run,(body,)
@@ -90,14 +90,18 @@ def census(run, index, buffer):
return totals(dense, leaf), external, independent
def fine(block):
"""Two-level circle+jet construction on BOUNDS; block None keeps the dense fine level."""
return dict(refinements=[BOUNDS], level_blocks=[None, block],
level_objects=[(), (circle(),)], level_jets=[(), (jet(),)])
class RawExchangeGPU(unittest.TestCase):
def test_row_ledger_interval_jet_switch_and_no_population_effect(self):
with CudaContext(0) as ctx:
for block in ((32, 40), (16, 20), (8, 10)):
with FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=block,
fine_objects=[circle()], jets=[jet()], observe_raw_exchange=True) as measured, \
FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=block,
fine_objects=[circle()], jets=[jet()]) as plain:
with FixedPatchStepper(CFG, context=ctx, **fine(block), observe_raw_exchange=True) as measured, \
FixedPatchStepper(CFG, context=ctx, **fine(block)) as plain:
arrays = [state(l.shape) for l in measured._levels]
measured.initialize(*arrays)
plain.initialize(*arrays)
@@ -146,9 +150,8 @@ class RawExchangeGPU(unittest.TestCase):
"""Same fine resolution/initial data with dense storage: a numerical
reference for cut-link rows, not a uniform-whole-domain convergence study."""
with CudaContext(0) as ctx:
with FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_objects=[circle()], jets=[jet()]) as dense, \
FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=(16, 20),
fine_objects=[circle()], jets=[jet()], observe_raw_exchange=True) as blocked:
with FixedPatchStepper(CFG, context=ctx, **fine(None)) as dense, \
FixedPatchStepper(CFG, context=ctx, **fine((16, 20)), observe_raw_exchange=True) as blocked:
arrays = [state(l.shape) for l in blocked._levels]
dense.initialize(*arrays)
blocked.initialize(*arrays)
@@ -229,8 +232,7 @@ class RawExchangeGPU(unittest.TestCase):
layout = BlockLayout2D.build(48, 64, 8, 8)
with CudaContext(0) as ctx:
with PeriodicBlockStepper(fine_cfg, layout, context=ctx, objects=[obj]) as uniform, \
FixedPatchStepper(CFG, BOUNDS, context=ctx, fine_block=(16, 20),
fine_objects=[circle()], jets=[jet()], observe_raw_exchange=True) as amr:
FixedPatchStepper(CFG, context=ctx, **fine((16, 20)), observe_raw_exchange=True) as amr:
amr.initialize(*(state(l.shape) for l in amr._levels))
uniform.initialize(state((9, 48, 64)))
b = amr._level_boundaries[1]
@@ -83,11 +83,10 @@ def mocked_backend(mode="periodic", fail_alloc=None):
class BlockedBaseCPU(unittest.TestCase):
def test_reject_before_cuda(self):
context = Mock(activate=Mock(side_effect=AssertionError("CUDA reached")))
for kwargs in ({"base_block": (12, 8)},
{"base_block": (7, 8), "fine_block": FINE_BLOCK},
{"base_block": (12,), "fine_block": FINE_BLOCK}):
with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
FixedPatchStepper(config("periodic"), BOUNDS, context=context, **kwargs)
for blocks in ([(12, 8), None], [(7, 8), FINE_BLOCK], [(12,), FINE_BLOCK]):
with self.subTest(blocks=blocks), self.assertRaises(ValueError):
FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
level_blocks=blocks)
context.activate.assert_not_called()
def test_all_arrival_rows_donors_and_kind_masks(self):
@@ -193,9 +192,10 @@ class BlockedBaseCPU(unittest.TestCase):
def test_launch_order_stream_wall_constants_and_buffer_swaps(self):
with mocked_backend("slot_reservoir") as m:
backend = FixedPatchStepper(config("slot_reservoir"), BOUNDS, context=m["context"],
coarse_boundary="slot_reservoir", fine_block=FINE_BLOCK,
base_block=(3, 4), fine_objects=[circle(actuation(.002))])
backend = FixedPatchStepper(config("slot_reservoir"), context=m["context"],
coarse_boundary="slot_reservoir", refinements=[BOUNDS],
level_blocks=[(3, 4), FINE_BLOCK],
level_objects=[(), (circle(actuation(.002)),)])
try:
backend.initialize(state(NY, NX, 1), state(*backend.patch.fine_shape, 2))
coarse, fine = backend._levels
@@ -236,16 +236,16 @@ class BlockedBaseCPU(unittest.TestCase):
for fail_at in (1, 6, 10):
with self.subTest(fail_at=fail_at), mocked_backend(fail_alloc=fail_at) as m:
with self.assertRaisesRegex(MemoryError, "injected allocation"):
FixedPatchStepper(config("periodic"), BOUNDS, context=m["context"],
fine_block=FINE_BLOCK, base_block=(3, 4))
FixedPatchStepper(config("periodic"), context=m["context"], refinements=[BOUNDS],
level_blocks=[(3, 4), FINE_BLOCK])
self.assertEqual(len(m["allocations"]), fail_at)
for allocation in m["allocations"]:
allocation.free.assert_called_once_with()
def test_failed_parent_stage_cannot_publish_or_resume(self):
with mocked_backend() as m:
backend = FixedPatchStepper(config("periodic"), BOUNDS, context=m["context"],
fine_block=FINE_BLOCK, base_block=(3, 4))
backend = FixedPatchStepper(config("periodic"), context=m["context"], refinements=[BOUNDS],
level_blocks=[(3, 4), FINE_BLOCK])
backend.initialize(state(NY, NX, 1), state(*backend.patch.fine_shape, 2))
m["functions"]["BlockCoalesce"].side_effect = RuntimeError("injected parent stage failure")
current = backend._levels[0].current
@@ -289,9 +289,10 @@ class BlockedBaseGPU(unittest.TestCase):
for base_block in (None, *layouts):
label = f"{mode} width={width} speed={speed} base={base_block}"
snapshots = []
with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
fine_block=FINE_BLOCK, base_block=base_block,
fine_objects=() if speed is None else [circle(actuation(speed))]) as backend:
with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
refinements=[BOUNDS], level_blocks=[base_block, FINE_BLOCK],
level_objects=[(), () if speed is None
else (circle(actuation(speed)),)]) as backend:
backend.initialize(*arrays)
for step in (0, 1, 2, 8):
if step == 1 and base_block is not None:
@@ -105,9 +105,10 @@ class BlockedFineCPU(unittest.TestCase):
return Mock(activate=Mock(side_effect=AssertionError("CUDA reached")), sm_arch="sm_70")
def test_rejections_before_cuda(self):
for kw in ({"fine_block": (7, 7)}, {"fine_block": (16,)}):
for kw in ({"level_blocks": [None, (7, 7)]}, {"level_blocks": [None, (16,)]}):
with self.subTest(kw=kw), self.assertRaises(ValueError):
FixedPatchStepper(config("periodic"), BOUNDS, context=self.forbidden_context(), **kw)
FixedPatchStepper(config("periodic"), context=self.forbidden_context(),
refinements=[BOUNDS], **kw)
def test_single_block_layout_allocates_no_empty_halo_tables(self):
# A 1x1 non-periodic fine layout has zero halo pairs; CUDA rejects mem_alloc(0).
@@ -125,8 +126,8 @@ class BlockedFineCPU(unittest.TestCase):
patch.object(compiler, "compile_kernel"), patch.object(compiler, "cleanup_build_dir"), \
patch.object(compiler, "load_module", return_value=module), \
patch.object(cuda, "mem_alloc", side_effect=alloc), patch.object(cuda, "memcpy_htod"):
backend = FixedPatchStepper(config("periodic"), BOUNDS, context=context,
fine_block=FINE_BLOCKS[0])
backend = FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
level_blocks=[None, FINE_BLOCKS[0]])
self.assertEqual(len(backend.fine_layout.halo_dst), 0)
self.assertIsNone(backend._levels[1].halo_dst)
backend._gather_halo(backend._levels[1], Mock()) # no launch without pairs
@@ -177,8 +178,9 @@ class BlockedFineCPU(unittest.TestCase):
patch.object(cuda, "mem_alloc", side_effect=lambda size: Mock()), \
patch.object(cuda, "memcpy_htod"), \
patch.object(curved_links, "CurvedLinkSoA", side_effect=AssertionError("dense SoA built")):
backend = FixedPatchStepper(config("periodic"), BOUNDS, context=context,
fine_objects=objects, fine_block=FINE_BLOCKS[1])
backend = FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
level_blocks=[None, FINE_BLOCKS[1]],
level_objects=[(), tuple(objects)])
backend.initialize(state(NY, NX, 1), state(*backend.patch.fine_shape, 2))
return backend, calls
@@ -289,8 +291,9 @@ class BlockedFineGPU(unittest.TestCase):
results = []
with CudaContext(0) as context:
for fine_block in (None, *FINE_BLOCKS):
with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
fine_block=fine_block) as backend:
with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
refinements=[BOUNDS],
level_blocks=[None, fine_block]) as backend:
backend.initialize(g_coarse, g_fine)
backend.step(STEPS)
self.assertEqual(backend.step_count, STEPS)
@@ -318,8 +321,9 @@ class BlockedFineGPU(unittest.TestCase):
results = []
with CudaContext(0) as context:
for fine_block in (None, *FINE_BLOCKS):
with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
fine_objects=[circle()], fine_block=fine_block) as backend:
with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
refinements=[BOUNDS], level_blocks=[None, fine_block],
level_objects=[(), (circle(),)]) as backend:
backend.initialize(g_coarse, g_fine)
backend.step(STEPS)
results.append((backend.download(), raw(backend)))
@@ -346,9 +350,9 @@ class BlockedFineGPU(unittest.TestCase):
results = []
with CudaContext(0) as context:
for fine_block in (None, *FINE_BLOCKS):
with FixedPatchStepper(cfg, BOUNDS, context=context, coarse_boundary=mode,
fine_objects=[circle(actuation(speed))],
fine_block=fine_block) as backend:
with FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
refinements=[BOUNDS], level_blocks=[None, fine_block],
level_objects=[(), (circle(actuation(speed)),)]) as backend:
backend.initialize(g_coarse, g_fine)
backend.step(STEPS)
results.append((backend.download(), raw(backend)))
@@ -94,7 +94,8 @@ class FaultGuardsCPU(unittest.TestCase):
("slot_reservoir", (12, 4, 20, 21))):
with self.subTest(mode=mode, bounds=bounds):
with self.assertRaises(ValueError):
implementation.FixedPatchStepper(cfg, bounds, context=context, coarse_boundary=mode)
implementation.FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
refinements=[bounds])
context.activate.assert_not_called()
# Both exact channel limits and the original periodic two-cell margin
# must reach context activation, with no allocation or device required.
@@ -102,7 +103,8 @@ class FaultGuardsCPU(unittest.TestCase):
("periodic", (2, 2, 6, 5))):
with self.subTest(mode=mode, bounds=bounds):
with self.assertRaisesRegex(RuntimeError, "validated before CUDA"):
implementation.FixedPatchStepper(cfg, bounds, context=context, coarse_boundary=mode)
implementation.FixedPatchStepper(cfg, context=context, coarse_boundary=mode,
refinements=[bounds])
def fake_backend(self):
backend = PeriodicPatchStepper.__new__(PeriodicPatchStepper)
@@ -154,12 +156,12 @@ class FaultGuardsCPU(unittest.TestCase):
context = Mock(activate=nullcontext, sm_arch="sm_70")
with patch.object(implementation, "HALO", 2):
with self.assertRaisesRegex(RuntimeError, "four-cell"):
PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context)
PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)])
p = implementation.FixedPatch2D.from_config(cfg, (4, 4, 12, 12))
with patch.object(cuda, "Stream"), patch.object(implementation.FixedPatch2D, "from_config",
return_value=replace(p, tau_coarse=.9)):
with self.assertRaisesRegex(RuntimeError, "omega/tau"):
PeriodicPatchStepper(cfg, p.bounds, context=context)
PeriodicPatchStepper(cfg, context=context, refinements=[p.bounds])
module = Mock()
module.get_global.return_value = (Mock(), 16)
with patch.object(cuda, "Stream"), patch.object(compiler, "generate_config"), \
@@ -167,7 +169,7 @@ class FaultGuardsCPU(unittest.TestCase):
patch.object(compiler, "load_module", return_value=module), \
patch.object(cuda, "memcpy_htod") as upload:
with self.assertRaisesRegex(RuntimeError, "layout mismatch"):
PeriodicPatchStepper(cfg, p.bounds, context=context)
PeriodicPatchStepper(cfg, context=context, refinements=[p.bounds])
upload.assert_not_called()
@@ -241,7 +243,7 @@ class NormalBuildGPU(unittest.TestCase):
max_error = max_bc_error = 0.0
with CudaContext(0) as context:
with patch.object(compiler, "generate_config", side_effect=record_config):
backend = implementation.FixedPatchStepper(cfg, bounds, context=context,
backend = implementation.FixedPatchStepper(cfg, context=context, refinements=[bounds],
coarse_boundary="slot_reservoir")
with backend:
self.assertEqual(len(generated), 2)
@@ -311,7 +313,8 @@ class NormalBuildGPU(unittest.TestCase):
for a in (initial.coarse_f, initial.fine_f)))
max_pop = max_drift = max_moment = 0.0
with CudaContext(0) as context, PeriodicPatchStepper(
LBMConfig(nx=16, ny=16, viscosity=.1), (4, 4, 12, 12), context=context) as backend:
LBMConfig(nx=16, ny=16, viscosity=.1), context=context,
refinements=[(4, 4, 12, 12)]) as backend:
backend.initialize(*postcollision(state, spec, maps))
before = totals(backend.download(), maps)
for step in range(1, 9):
@@ -342,7 +345,7 @@ class NormalBuildGPU(unittest.TestCase):
fine_mask[4:-4, 4:-4] = True
initial[1][:, ~fine_mask] = np.nan
expected = scalar_interval(*[a.astype(np.float64) for a in initial], bounds)
with CudaContext(0) as context, PeriodicPatchStepper(cfg, bounds, context=context) as backend:
with CudaContext(0) as context, PeriodicPatchStepper(cfg, context=context, refinements=[bounds]) as backend:
self.assertEqual(backend.patch.fine_shape, (14, 16))
p = backend.patch
allocated = 2 * 9 * (17*13+16*14) * 4 + sum(a.nbytes for a in
@@ -84,13 +84,13 @@ class CoupledGPU(unittest.TestCase):
cases = (("dense_unshifted", False, {}),
("blocked_shifted", True,
dict(base_block=(4, 4), fine_block=(6, 6))))
dict(level_blocks=[(4, 4), (6, 6)])))
with CudaContext(0) as context:
for label, shifted, storage in cases:
with self.subTest(storage=label):
cfg = LBMConfig(nx=16, ny=16, viscosity=.0192,
ddf_shifting=shifted)
with FixedPatchStepper(cfg, (4, 4, 12, 12), context=context,
with FixedPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)],
**storage) as backend:
self.assertAlmostEqual(backend.patch.tau_coarse, spec.tau_coarse)
self.assertAlmostEqual(backend.patch.tau_fine, spec.tau_fine)
@@ -148,7 +148,7 @@ class CoupledGPU(unittest.TestCase):
maps = ref.build_maps(spec)
with CudaContext(0) as context:
with PeriodicPatchStepper(LBMConfig(nx=16, ny=16, viscosity=.1),
(4, 4, 12, 12), context=context) as backend:
context=context, refinements=[(4, 4, 12, 12)]) as backend:
backend.initialize(*postcollision(ref.make_state(spec), spec, maps))
coarse, fine = backend._levels
@@ -187,7 +187,7 @@ class CoupledGPU(unittest.TestCase):
with CudaContext(0) as context:
with patch.object(compiler, "compile_kernel", audit_compile):
backend = PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context)
backend = PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)])
with backend:
# Production allocations: four DDF arrays and four compact tables.
device_bytes = 2 * 9 * (16**2 + 24**2) * 4
@@ -270,7 +270,7 @@ class CoupledGPU(unittest.TestCase):
backend.download()
# Compile and execute the actual non-audited entry as well: one step.
with PeriodicPatchStepper(cfg, (4, 4, 12, 12), context=context) as plain:
with PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)]) as plain:
rest = postcollision(ref.make_state(spec), spec, maps)
plain.initialize(*rest)
plain.step()
@@ -41,8 +41,9 @@ def mocked(objects, jets, fine_block):
patch.object(compiler, "load_module", return_value=module), \
patch.object(cuda, "mem_alloc", side_effect=lambda size: Mock()), \
patch.object(cuda, "memcpy_htod", htod):
backend = FixedPatchStepper(config("periodic"), BOUNDS, context=context,
fine_objects=objects, fine_block=fine_block, jets=jets)
backend = FixedPatchStepper(config("periodic"), context=context, refinements=[BOUNDS],
level_blocks=[None, fine_block], level_objects=[(), tuple(objects)],
level_jets=[(), tuple(jets)])
return backend, htod
@@ -149,8 +150,9 @@ class JetStepperCPU(unittest.TestCase):
def test_jet_overlapping_legacy_aperture_rejected_before_cuda(self):
with patch.object(cuda, "Stream", side_effect=AssertionError("CUDA touched")):
with self.assertRaisesRegex(ValueError, "SurfaceActuation"):
FixedPatchStepper(config("periodic"), BOUNDS, context=Mock(),
fine_objects=[circle(actuation(SPEED))], jets=[jet()])
FixedPatchStepper(config("periodic"), context=Mock(), refinements=[BOUNDS],
level_objects=[(), (circle(actuation(SPEED)),)],
level_jets=[(), (jet(),)])
def run(mode, fine_block, *, objects, jets, switch=None):
@@ -158,8 +160,9 @@ def run(mode, fine_block, *, objects, jets, switch=None):
cfg = config(mode)
patch_ = FixedPatch2D.from_config(cfg, BOUNDS)
with CudaContext(0) as context, FixedPatchStepper(
cfg, BOUNDS, context=context, coarse_boundary=mode, fine_objects=objects,
fine_block=fine_block, jets=jets) as backend:
cfg, context=context, coarse_boundary=mode, refinements=[BOUNDS],
level_blocks=[None, fine_block], level_objects=[(), tuple(objects)],
level_jets=[(), tuple(jets)]) as backend:
backend.initialize(state(NY, NX, 1), state(*patch_.fine_shape, 2))
if switch is None:
backend.step(STEPS)
@@ -42,8 +42,9 @@ class RefinementMRTGPU(unittest.TestCase):
collision="MRT", ddf_shifting=shifted)
reference = None
for block in ((32, 48), (8, 8), (4, 1)):
with FixedPatchStepper(cfg, (8, 6, 40, 26), context=context, coarse_boundary=boundary,
base_block=block, fine_block=(8, 8), fine_objects=(body,)) as backend:
with FixedPatchStepper(cfg, context=context, coarse_boundary=boundary,
refinements=[(8, 6, 40, 26)], level_blocks=[block, (8, 8)],
level_objects=[(), (body,)]) as backend:
backend.initialize(*[state(*level.shape[1:]) for level in backend._levels])
backend.step(8)
raw = backend._download_raw()
@@ -63,8 +64,8 @@ class RefinementMRTGPU(unittest.TestCase):
results = {}
layouts = ((None, None), *((base, FINE_BLOCK) for base in (None, *BASE_BLOCKS)))
for base_block, fine_block in layouts:
with FixedPatchStepper(cfg, BOUNDS, context=context, fine_block=fine_block,
base_block=base_block) as backend:
with FixedPatchStepper(cfg, context=context, refinements=[BOUNDS],
level_blocks=[base_block, fine_block]) as backend:
backend.initialize(*[fixture.state(*level.shape[1:], seed) for seed, level in enumerate(backend._levels)])
backend.step(4)
results[base_block, fine_block] = raw(backend)
@@ -75,9 +76,9 @@ class RefinementMRTGPU(unittest.TestCase):
def test_shifted_all_blocked_three_level_mrt_smoke(self):
"""Per-level shear rate omega_l = 1/(3*2**l*nu + 1/2); finite ACTIVE leaves."""
cfg = LBMConfig(nx=16, ny=16, viscosity=.1, collision="MRT", ddf_shifting=True)
kw = dict(fine_block=(4, 4), base_block=(4, 4), child=(8, 8, 16, 16), child_block=(4, 4))
kw = dict(refinements=[(4, 4, 12, 12), (8, 8, 16, 16)], level_blocks=[(4, 4)] * 3)
with CudaContext(0) as context:
with FixedPatchStepper(cfg, (4, 4, 12, 12), context=context, **kw) as backend:
with FixedPatchStepper(cfg, context=context, **kw) as backend:
self.assertEqual(len(backend._level_cfgs), 3)
for level, level_cfg in enumerate(backend._level_cfgs):
checked = replace(level_cfg)
@@ -26,6 +26,12 @@ FACE = ((4, 4, 8, 12), (12, 4, 16, 12))
S1 = ((4, 4, 12, 12), (20, 4, 28, 12))
def levels(bounds, blocks=(None, None), objects=(), jets=(), **rest):
"""Two-level constructor keywords: one L1 rectangle or set, per-level blocks, L1 bodies/Jets."""
return dict(refinements=[bounds], level_blocks=list(blocks),
level_objects=[(), tuple(objects)], level_jets=[(), tuple(jets)], **rest)
def level_fine(patch_set, windows):
"""Level-wide fine array with each patch window filled from `windows`."""
out = np.full((9, *patch_set.fine_shape), np.nan, np.float32)
@@ -45,21 +51,21 @@ class MultipatchCPU(unittest.TestCase):
def test_rejections_before_cuda(self):
cfg = config("periodic")
context = object() # never activated: every case must fail first
for kw in (dict(), dict(fine_block=(8, 4)), dict(base_block=(4, 4)),
dict(fine_block=(8, 4), base_block=(4, 4), fine_objects=[circle()]),
dict(fine_block=(5, 4), base_block=(4, 4))):
for kw in (dict(), dict(blocks=(None, (8, 4))), dict(blocks=((4, 4), None)),
dict(blocks=((4, 4), (8, 4)), objects=[circle()]),
dict(blocks=((4, 4), (5, 4)))):
with self.subTest(kw=kw), self.assertRaises(ValueError):
FixedPatchStepper(cfg, ((4, 4, 8, 12), (14, 4, 18, 12)), context=context, **kw)
FixedPatchStepper(cfg, context=context, **levels(((4, 4, 8, 12), (14, 4, 18, 12)), **kw))
with self.assertRaisesRegex(ValueError, "margin"):
FixedPatchStepper(cfg, ((1, 4, 8, 12), (10, 4, 14, 12)), context=context,
fine_block=(8, 4), base_block=(4, 4))
FixedPatchStepper(cfg, context=context,
**levels(((1, 4, 8, 12), (10, 4, 14, 12)), blocks=((4, 4), (8, 4))))
from CelerisLab.body.jets import Jet, Parallel, Tube
cfg32 = LBMConfig(nx=32, ny=16, viscosity=.1)
span = Jet(Tube((5., 8.), 0., 22., .6), Parallel(0.), amplitude=0.)
with self.assertRaisesRegex(ValueError, "one jet"): # same obj_id cannot bypass it
FixedPatchStepper(cfg32, ((2, 2, 14, 14), (18, 2, 30, 14)), context=context,
fine_block=(4, 4), base_block=(4, 4),
fine_objects=[body(8., 8.), body(24., 8.)], jets=[span])
FixedPatchStepper(cfg32, context=context,
**levels(((2, 2, 14, 14), (18, 2, 30, 14)), blocks=((4, 4), (4, 4)),
objects=[body(8., 8.), body(24., 8.)], jets=[span]))
class MultipatchGPU(unittest.TestCase):
@@ -67,7 +73,7 @@ class MultipatchGPU(unittest.TestCase):
def run_steps(self, context, cfg, bounds, coarse, fine, **kw):
out = []
with FixedPatchStepper(cfg, bounds, context=context, **kw) as backend:
with FixedPatchStepper(cfg, context=context, **levels(bounds, **kw)) as backend:
backend.initialize(coarse, fine)
for step in self.STEPS:
backend.step(step - backend.step_count)
@@ -82,7 +88,7 @@ class MultipatchGPU(unittest.TestCase):
p = FixedPatch2D.from_config(cfg, BOUNDS)
s = FixedPatchSet2D.from_config(cfg, (BOUNDS,))
coarse, fine = state(NY, NX, 1), state(*p.fine_shape, 2)
kw = dict(coarse_boundary=mode, fine_block=(8, 4), base_block=(3, 4))
kw = dict(coarse_boundary=mode, blocks=((3, 4), (8, 4)))
old = self.run_steps(context, cfg, BOUNDS, coarse, fine, **kw)
new = self.run_steps(context, cfg, (BOUNDS,), coarse, level_fine(s, [fine]), **kw)
for step, a, b in zip(self.STEPS, old, new):
@@ -99,7 +105,7 @@ class MultipatchGPU(unittest.TestCase):
one = FixedPatchSet2D.from_config(cfg16, (S1[0],))
two = FixedPatchSet2D.from_config(cfg32, S1)
coarse, fine = state(16, 16, 3), state(*one.patches[0].fine_shape, 4)
kw = dict(fine_block=(4, 4), base_block=(4, 4))
kw = dict(blocks=((4, 4), (4, 4)))
with CudaContext(0) as context:
ref = self.run_steps(context, cfg16, (S1[0],), coarse, level_fine(one, [fine]), **kw)
tiled = np.concatenate([coarse, coarse], axis=2)
@@ -118,7 +124,7 @@ class MultipatchGPU(unittest.TestCase):
s = FixedPatchSet2D.from_config(cfg, FACE)
coarse = state(16, 24, 5)
fine = level_fine(s, [state(*p.fine_shape, 6 + i) for i, p in enumerate(s.patches)])
kw = dict(fine_block=(4, 4), base_block=(4, 4))
kw = dict(blocks=((4, 4), (4, 4)))
leaf = s.fine_kind == ACTIVE
support = s.fine_layout((4, 4)).owned
outer = support & (s.fine_kind == OUTER_GHOST)
@@ -164,7 +170,7 @@ class MultipatchBodiesGPU(unittest.TestCase):
def run_steps(self, context, cfg, bounds, coarse, fine, switch=None, **kw):
out = []
with FixedPatchStepper(cfg, bounds, context=context, **kw) as backend:
with FixedPatchStepper(cfg, context=context, **levels(bounds, **kw)) as backend:
backend.initialize(coarse, fine)
for step in self.STEPS:
if switch is not None and step == 8:
@@ -182,8 +188,7 @@ class MultipatchBodiesGPU(unittest.TestCase):
p = FixedPatch2D.from_config(cfg, BOUNDS)
s = FixedPatchSet2D.from_config(cfg, (BOUNDS,))
coarse, fine = state(NY, NX, 1), state(*p.fine_shape, 2)
kw = dict(coarse_boundary=mode, fine_block=(8, 4), base_block=(3, 4),
fine_objects=[circle()])
kw = dict(coarse_boundary=mode, blocks=((3, 4), (8, 4)), objects=[circle()])
old = self.run_steps(context, cfg, BOUNDS, coarse, fine, **kw)
new = self.run_steps(context, cfg, (BOUNDS,), coarse, level_fine(s, [fine]), **kw)
for step, a, b in zip(self.STEPS, old, new):
@@ -202,17 +207,17 @@ class MultipatchBodiesGPU(unittest.TestCase):
two = FixedPatchSet2D.from_config(cfg32, self.PATCHES)
coarse, fine = state(16, 16, 3), state(*one.patches[0].fine_shape, 4)
jet = lambda x: Jet(Tube((x, 8.), 0., 6., .6), Parallel(0.), amplitude=.002)
kw = dict(fine_block=(4, 4), base_block=(4, 4))
kw = dict(blocks=((4, 4), (4, 4)))
with CudaContext(0) as context:
for jets in ((), (jet(5.),)):
ref = self.run_steps(context, cfg16, self.PATCHES[:1], coarse, level_fine(one, [fine]),
switch=[(0, -.001)] if jets else None,
fine_objects=[body(8., 8.)], jets=jets, **kw)
objects=[body(8., 8.)], jets=jets, **kw)
both = jets and (jet(5.), jet(21.))
got = self.run_steps(context, cfg32, self.PATCHES, np.concatenate([coarse, coarse], 2),
level_fine(two, [fine, fine]),
switch=[(0, -.001), (1, -.001)] if jets else None,
fine_objects=[body(8., 8., 5), body(24., 8., 5)],
objects=[body(8., 8., 5), body(24., 8., 5)],
jets=both or (), **kw)
for step, a, b in zip(self.STEPS, ref, got):
label = f"jets={len(jets)} step {step}"
@@ -27,7 +27,16 @@ from test_refinement_blocked_base import raw
CFG = LBMConfig(nx=16, ny=16, viscosity=.1)
L1, L2 = (4, 4, 12, 12), (8, 8, 16, 16)
KW = dict(fine_block=(4, 4), base_block=(4, 4), child=L2, child_block=(4, 4))
B = (4, 4)
def levels(l1=L1, l2=L2, blocks=(B, B, B), objects=((), (), ()), jets=((), (), ())):
"""Level-list keywords for the scalar L1 patch with one scalar L2 child."""
return dict(refinements=[l1, l2], level_blocks=list(blocks),
level_objects=list(objects), level_jets=list(jets))
KW = levels()
def child_body(spec=None):
@@ -52,26 +61,29 @@ def mocked_child():
patch.object(compiler, "load_module", return_value=module), \
patch.object(cuda, "mem_alloc", side_effect=lambda size: Mock()), \
patch.object(cuda, "memcpy_htod"):
return FixedPatchStepper(CFG, L1, context=context, child_objects=[child_body()],
child_jets=[child_jet()], **KW)
return FixedPatchStepper(CFG, context=context,
**levels(objects=((), (), (child_body(),)), jets=((), (), (child_jet(),))))
class ThreeLevelCPU(unittest.TestCase):
def test_rejections_before_cuda(self):
context = object() # never activated: every case must fail first
bad = (dict(KW, child_block=None), dict(KW, fine_block=None),
dict(KW, fine_objects=[circle()]), dict(KW, child=(4, 8, 16, 16)),
dict(KW, child_block=(5, 4)))
bad = (levels(blocks=(B, B, None)), levels(blocks=(B, None, B)),
levels(objects=((), (circle(),), ())), levels(l2=(4, 8, 16, 16)),
levels(blocks=(B, B, (5, 4))))
for kw in bad:
with self.subTest(kw=kw), self.assertRaises(ValueError):
FixedPatchStepper(CFG, L1, context=context, **kw)
for kw in (dict(child_objects=[child_body()]), dict(child_jets=[child_jet()]),
dict(KW, child_jets=[child_jet()])):
FixedPatchStepper(CFG, context=context, **kw)
# L2 bodies/Jets without an L2 level (one entry too many), and an L2 Jet without a body.
two_level = dict(refinements=[L1], level_blocks=[None, None])
for kw in (dict(two_level, level_objects=[(), (), (child_body(),)]),
dict(two_level, level_jets=[(), (), (child_jet(),)]),
levels(jets=((), (), (child_jet(),)))):
with self.assertRaises(ValueError):
FixedPatchStepper(CFG, L1, context=context, **kw)
with self.assertRaises(ValueError): # several L1 patches + child: not yet
FixedPatchStepper(LBMConfig(nx=32, ny=16, viscosity=.1), ((4, 4, 12, 12), (20, 4, 28, 12)),
context=context, **KW)
FixedPatchStepper(CFG, context=context, **kw)
with self.assertRaises(ValueError): # several L1 patches + scalar child: not yet
FixedPatchStepper(LBMConfig(nx=32, ny=16, viscosity=.1), context=context,
**levels(l1=((4, 4, 12, 12), (20, 4, 28, 12))))
def test_four_halo_bfl_steps_and_fresh_middle_feeds(self):
@@ -80,11 +92,12 @@ class ThreeLevelCPU(unittest.TestCase):
f, g = run._levels[1:]
b = g.boundary
events, feeds = [], []
advance, gather = run._advance_child, run._gather_halo
def record_feed(buffer, **kw):
self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
feeds.append(buffer)
advance(buffer, **kw)
advance, gather = run._advance, run._gather_halo
def record_feed(index, buffer, **kw): # index 2: L1 hop feeds L2
if index == 2:
self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
feeds.append(buffer)
advance(index, buffer, **kw)
def record_halo(level, buffer, **kw):
if level is g:
events.append(("halo", buffer))
@@ -95,7 +108,7 @@ class ThreeLevelCPU(unittest.TestCase):
self.assertIs(kind, g.kind)
events.append(("step", src))
g.kernel.side_effect = record_step
with patch.object(run, "_advance_child", side_effect=record_feed), \
with patch.object(run, "_advance", side_effect=record_feed), \
patch.object(run, "_gather_halo", side_effect=record_halo):
run.step()
self.assertEqual(feeds, [f.current, f.spare])
@@ -105,8 +118,8 @@ class ThreeLevelCPU(unittest.TestCase):
self.assertGreater(len(np.unique(b.tables["block"])), 1)
# Negative control: reusing L1's old source at the midpoint must fail.
feeds.clear()
with patch.object(run, "_advance_child", side_effect=lambda buffer, **kw:
record_feed(f.current, **kw)):
with patch.object(run, "_advance", side_effect=lambda index, buffer, **kw:
record_feed(index, f.current if index == 2 else buffer, **kw)):
with self.assertRaises(AssertionError):
run.step()
run.close()
@@ -155,7 +168,7 @@ class ThreeLevelGPU(unittest.TestCase):
def test_equilibrium_is_kept_on_every_level(self):
from CelerisLab.cuda.context import CudaContext
feq = np.asarray(compute_feq_d2q9(1.0, .03, -.02), np.float32)[:, None, None]
with CudaContext(0) as ctx, FixedPatchStepper(CFG, L1, context=ctx, **KW) as run:
with CudaContext(0) as ctx, FixedPatchStepper(CFG, context=ctx, **KW) as run:
shapes = ((16, 16), run.patch.fine_shape, run.child_patch.fine_shape)
run.initialize(*(np.broadcast_to(feq, (9, *s)).copy() for s in shapes))
run.step(8)
@@ -168,15 +181,16 @@ class ThreeLevelGPU(unittest.TestCase):
def test_nonequilibrium_masks_and_fresh_middle_feeds(self):
from CelerisLab.cuda.context import CudaContext
with CudaContext(0) as ctx, FixedPatchStepper(CFG, L1, context=ctx, **KW) as run:
with CudaContext(0) as ctx, FixedPatchStepper(CFG, context=ctx, **KW) as run:
p1, p2 = run.patch, run.child_patch
run.initialize(state(16, 16, 1), state(*p1.fine_shape, 2), state(*p2.fine_shape, 3))
f = run._levels[1]
feeds, original = [], run._advance_child
def record(buffer, **kw):
feeds.append(int(buffer))
original(buffer, **kw)
with patch.object(run, "_advance_child", side_effect=record):
feeds, original = [], run._advance
def record(index, buffer, **kw): # index 2: L1 hop feeds L2
if index == 2:
feeds.append(int(buffer))
original(index, buffer, **kw)
with patch.object(run, "_advance", side_effect=record):
cur, spare = int(f.current), int(f.spare)
run.step(1)
self.assertEqual(feeds, [cur, spare]) # current G, then the new G of hop 1
@@ -203,8 +217,9 @@ class ThreeLevelBodyGPU(unittest.TestCase):
CFG, velocity=0., wall_normal_velocity=.001, wall_slot_width=4,
y_wall_bc="on_site_velocity", top_wall_bc="reservoir")
with CudaContext(0) as ctx, FixedPatchStepper(
cfg, L1, context=ctx, coarse_boundary=mode, child_objects=[child_body(spec)],
child_jets=jets, **dict(KW, child_block=block)) as run:
cfg, context=ctx, coarse_boundary=mode,
**levels(blocks=(B, B, block), objects=((), (), (child_body(spec),)),
jets=((), (), tuple(jets)))) as run:
run.initialize(*(state(*level.shape[1:], i + 1) for i, level in enumerate(run._levels)))
snapshots = []
for stop in (0, 1, 2, 4, 8):
@@ -78,10 +78,10 @@ class ThreeLevelMultipatchGPU(unittest.TestCase):
def run_case(self, ctx, cfg=CFG, parents=PARENTS, children=CHILDREN,
mode='periodic', blocks=((4, 4), (4, 4)), objects=(), drive=(), switch=False,
scalar_child=None):
kwargs = dict(child_rectangles_in_parent=children) if scalar_child is None else dict(child=scalar_child)
with FixedPatchStepper(config(cfg, mode), parents, context=ctx, coarse_boundary=mode,
base_block=(4, 4), fine_block=blocks[0], child_block=blocks[1],
child_objects=objects, child_jets=drive, **kwargs) as run:
child = children if scalar_child is None else scalar_child
with FixedPatchStepper(config(cfg, mode), context=ctx, coarse_boundary=mode,
refinements=[parents, child], level_blocks=[(4, 4), blocks[0], blocks[1]],
level_objects=[(), (), objects], level_jets=[(), (), drive]) as run:
frame = frames(run)
run.initialize(*(physical_state(level.shape[1:], *f)
for level, f in zip(run._levels, frame)))
@@ -26,9 +26,8 @@ FP64_ATOL = 3e-6
def make(ctx):
return FixedPatchStepper(CFG, PARENTS, context=ctx, base_block=(4, 4),
fine_block=(4, 4), child_block=(4, 4),
child_rectangles_in_parent=A)
return FixedPatchStepper(CFG, context=ctx, refinements=[PARENTS, A],
level_blocks=[(4, 4)] * 3)
def initial(run, uniform):
@@ -136,8 +136,8 @@ class ShiftedJetsGPU(unittest.TestCase):
Parallel(0.),amplitude=.013)
for rho in (.99,1.01):
with self.subTest(rho=rho),CudaContext(0) as ctx:
with FixedPatchStepper(cfg,(8,6,24,18),context=ctx,base_block=(4,4),fine_block=(4,4),
fine_objects=(obj,),jets=(whole,),observe_raw_exchange=True) as run:
with FixedPatchStepper(cfg,context=ctx,refinements=[(8,6,24,18)],level_blocks=[(4,4)]*2,
level_objects=[(),(obj,)],level_jets=[(),(whole,)],observe_raw_exchange=True) as run:
run._initialize_raw(initial(run,rho,u=.013,v=0.))
expected=run._download_raw();ledger=Ledger(self,run,1)
self.assertTrue(run._level_boundaries[1].tables['aperture'].all())
@@ -153,7 +153,8 @@ class ShiftedJetsGPU(unittest.TestCase):
obj=circle((16.,12.),2.15);j=drive((16.,12.),2.15)
for rho in (.99,1.01):
with self.subTest(rho=rho),CudaContext(0) as ctx,ExitStack() as owned:
settings=dict(bounds=(8,6,24,18),context=ctx,base_block=(4,4),fine_block=(4,4),fine_objects=(obj,),jets=(j,))
settings=dict(context=ctx,refinements=[(8,6,24,18)],level_blocks=[(4,4)]*2,
level_objects=[(),(obj,)],level_jets=[(),(j,)])
measured=owned.enter_context(FixedPatchStepper(cfg,observe_raw_exchange=True,**settings))
plain=owned.enter_context(FixedPatchStepper(cfg,**settings))
measured._initialize_raw(initial(measured,rho));plain._initialize_raw(initial(plain,rho))
@@ -186,8 +187,9 @@ class ShiftedJetsGPU(unittest.TestCase):
parents=((4,4,20,28),(24,4,44,28));a=((16,16,32,40),(56,16,72,40));b=(a[0],(60,16,76,40))
obj=circle((12.,14.),.8);j=drive((12.,14.),.8)
with CudaContext(0) as ctx,ExitStack() as owned:
old=owned.enter_context(FixedPatchStepper(cfg,parents,context=ctx,base_block=(4,4),fine_block=(4,4),
child_block=(4,4),child_rectangles_in_parent=a,child_objects=(obj,),child_jets=(j,),observe_raw_exchange=True))
old=owned.enter_context(FixedPatchStepper(cfg,context=ctx,refinements=[parents,a],
level_blocks=[(4,4)]*3,level_objects=[(),(),(obj,)],level_jets=[(),(),(j,)],
observe_raw_exchange=True))
old._initialize_raw(initial(old,.99));ledger=Ledger(self,old,2)
for speed in (.002,-.001):old.set_jet_amplitude(0,speed,level=2);ledger.step()
before=old._download_raw();count=old.step_count
@@ -22,9 +22,8 @@ B=((20,16,36,40),)
def make(ctx):
return FixedPatchStepper(CFG,PARENTS,context=ctx,base_block=(4,4),
fine_block=(4,4),child_block=(4,4),
child_rectangles_in_parent=A)
return FixedPatchStepper(CFG,context=ctx,refinements=[PARENTS,A],
level_blocks=[(4,4)]*3)
def initial(run):
+222
View File
@@ -0,0 +1,222 @@
"""Steady DFG 2D-1 (Re 20): refined and uniform runs against the official reference intervals.
Not a full benchmark certification. Declared deviations from the official setup:
pressure-NEQ outlet (``neq_extrap``) instead of do-nothing, local Zou-He inlet.
Geometry, inflow profile, initial state, refined layout and observation reuse the
pinned Re100 runner (``run_refined_dfg``) and ``re100_helpers`` unchanged; only the
lattice viscosity is set for Re 20 (nu = U_lattice * D / 20 at the root level).
Units. The helpers use scaled DFG units with mean inflow speed 1, so
Cd = 2 Fx / (rho U^2 D) and Cl = 2 Fy / (rho U^2 D) are scale-free, and
``pressure_probes`` returns dp / (rho U^2). The official p_diff = p(0.15, 0.2) -
p(0.25, 0.2) uses U = 0.2 m/s, rho = 1, so p_diff = dp * 0.04. Both are reported;
only p_diff is compared with the official interval. Forces use the existing sign
convention: F is the force of the fluid on the cylinder, i.e. the refined raw
exchange impulse (window ledger, ``total.impulse``) or the uniform ``read_force``,
with x along the inflow and y up from the bottom wall. Cd must be positive, and the
official Cl is positive (centre 0.20 below the centreline 0.205), so a flipped sign
shows up as Cl outside its interval.
Sampling. One sample every SAMPLE_TSTAR (t* = U t / D; 0.25 s), taken at the end of
a completed root step with all levels synchronized; the step index is recorded. Cd
and Cl are means over the sample interval (window exchange ledger refined, per-batch
forces uniform), which filters acoustic and step-parity oscillation; p_diff is the
end-of-interval value at the exact surface points.
Steady rule (our engineering choice, frozen before any run; not official). Over the
trailing WINDOW_TSTAR (t* 50..60 inclusive, 21 samples), for each of Cd, Cl and
p_diff: range (max - min) <= 10 % and |least-squares slope * window| <= 2 % of the
width of its official interval. Absolute limits, so the small Cl is never a divisor;
the slope term bounds slow monotone drift that the range alone would pass. If the
rule fails at TSTAR_MAX the run reports "not steady" and gives no interval verdict;
limits are never relaxed after a run. The verdict uses the window mean, and the
distance of each mean to its nearest interval edge is reported (negative outside).
"""
import argparse
import json
import math
import os
from pathlib import Path
import sys
import time
from tempfile import TemporaryDirectory
import numpy as np
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import run_refined_dfg as rr # noqa: E402
RE = 20.0
U_PHYS = 0.2 # m/s mean inflow; rho = 1
TSTAR_MAX = 60.0 # 30 s physical
SAMPLE_TSTAR = 0.5 # 0.25 s
WINDOW_TSTAR = 10.0 # trailing 5 s, 21 samples
# Schaefer & Turek (1996), 2D-1 lower/upper bounds.
BANDS = dict(cd=(5.57, 5.59), cl=(0.0104, 0.0110), p_diff=(0.1172, 0.1176))
STEADY_RANGE = {k: (hi - lo) / 10 for k, (lo, hi) in BANDS.items()}
STEADY_SLOPE = {k: (hi - lo) / 50 for k, (lo, hi) in BANDS.items()}
DEVIATIONS = ("pressure-NEQ outlet (neq_extrap), not do-nothing", "local Zou-He inlet")
_re100_config = rr.config
def config(h, diameter, collision="SRT", jet_scenario="none"):
"""The pinned Re100 configuration with the lattice viscosity set for Re 20."""
if jet_scenario != "none":
raise ValueError("DFG 2D-1 runs have no Jets")
cfg, body, p = _re100_config(h, diameter, collision, jet_scenario)
nu = p["mean_speed"] * p["diameter"] / RE
cfg["physics"]["viscosity"] = nu
return cfg, body, dict(p, viscosity=nu, re=RE)
def lattice_parameters(diameter, speed=rr.SPEED):
nu = speed * diameter / RE
return dict(viscosity=nu, tau=0.5 + 3 * nu, steps=math.ceil(TSTAR_MAX * diameter / speed),
interval=round(SAMPLE_TSTAR * diameter / speed))
def sample(impulse, n, diameter, dp):
den = rr.SPEED**2 * diameter / 2
return dict(cd=float(impulse[0] / n / den), cl=float(impulse[1] / n / den),
dp_scaled=float(dp), p_diff=float(dp) * U_PHYS**2)
def summarize(rows):
"""Steady rule over the trailing window, then the interval verdict (or none)."""
if not rows or rows[-1]["tstar"] < TSTAR_MAX - 1e-9:
return dict(status="INCOMPLETE", steady=False, verdict=None)
window = [r for r in rows if r["tstar"] >= TSTAR_MAX - WINDOW_TSTAR - 1e-9]
t = np.array([r["tstar"] for r in window])
ranges = {k: max(r[k] for r in window) - min(r[k] for r in window) for k in BANDS}
drifts = {k: abs(np.polyfit(t, [r[k] for r in window], 1)[0]) * WINDOW_TSTAR for k in BANDS}
steady = all(ranges[k] <= STEADY_RANGE[k] and drifts[k] <= STEADY_SLOPE[k] for k in BANDS)
final = {k: float(np.mean([r[k] for r in window])) for k in BANDS}
edge = {k: min(final[k] - BANDS[k][0], BANDS[k][1] - final[k]) for k in BANDS}
result = dict(window_samples=len(window), ranges=ranges, drifts=drifts,
range_limits=STEADY_RANGE, slope_limits=STEADY_SLOPE, steady=steady,
window_mean=final, edge_distance=edge, bands=BANDS,
deviations=DEVIATIONS, cd_positive=final["cd"] > 0)
if not steady:
return dict(result, status="NOT_STEADY", verdict=None)
inside = {k: BANDS[k][0] <= final[k] <= BANDS[k][1] for k in BANDS}
return dict(result, status="WITHIN_INTERVALS" if all(inside.values()) else "OUTSIDE_INTERVALS",
verdict=inside)
def healthy(row):
return row["max_mach"] <= 0.2 and row["rho_min"] >= 0.5 and row["rho_max"] <= 1.5
def run_case(a, h, case):
from contextlib import ExitStack
from CelerisLab.cuda.context import CudaContext
from CelerisLab.lbm.descriptors import FLUID
from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
from CelerisLab.simulation import Simulation
refined = case == "refined"
cfgdict, bodydict, p = config(h, 40 if refined else 80, a.collision)
lp = lattice_parameters(p["diameter"])
rows = []
with ExitStack() as stack:
if refined:
cfg, pair, body, host = rr.host_geometry(a.source, h, a.collision)
context = stack.enter_context(CudaContext(0))
backend = stack.enter_context(FixedPatchStepper(
cfg, context=context, coarse_boundary="channel", refinements=[rr.PLACEMENTS["wide"]],
level_blocks=[rr.BASE_BLOCK, rr.FINE_BLOCK], level_objects=[(), (body,)],
observe_raw_exchange=True, observe_window_exchange=True))
from CelerisLab.lbm.channel import channel_masks
from CelerisLab.lbm.refinement import ACTIVE
from CelerisLab.lbm.descriptors import SOLID
masks = [channel_masks(pair.coarse_shape)[0] & ~pair.covered,
(pair.fine_kind == ACTIVE)
& ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)]
backend.initialize(rr.initial_state(pair.coarse_shape, (0.0, 0.0), 1.0, root=True),
rr.initial_state(pair.fine_shape, pair.fine_origin, 0.5))
probe_p = dict(p, center=[(80.5 - pair.fine_origin[0]) / 0.5 - 0.5,
(81.0 - pair.fine_origin[1]) / 0.5 - 0.5], radius=40.0)
else:
tmp = Path(stack.enter_context(TemporaryDirectory()))
(tmp / "lbm.json").write_text(json.dumps(cfgdict))
(tmp / "body.json").write_text(json.dumps(bodydict))
sim = stack.enter_context(Simulation(str(tmp / "lbm.json"), str(tmp / "body.json"),
device_id=0))
sim.initialize()
probe_p = p
fluid = (sim.field.flag.reshape(p["ny"], p["nx"]) & FLUID) != 0
begin = 0
while begin < lp["steps"]:
rr.check_deadline(a.deadline)
n = min(lp["interval"], lp["steps"] - begin)
if refined:
impulse = rr.advance_refined_interval(backend, n, a.deadline, "window")
fields = [rr.macros(g) for g in backend.download()]
rho, ux, uy = fields[1]
fluid = masks[1]
vals = [tuple(v[m] for v in f) for m, f in zip(masks, fields)]
else:
impulse, done = np.zeros(2), 0
while done < n:
batch = min(100, n - done)
sim.run(batch)
impulse += sim.read_force(0).astype(np.float64) * batch
done += batch
m = sim.get_macroscopic()
rho, ux, uy = [np.asarray(m[k], np.float64).reshape(p["ny"], p["nx"])
for k in ("rho", "ux", "uy")]
vals = [(rho[fluid], ux[fluid], uy[fluid])]
begin += n
row = dict(step=begin, tstar=begin * rr.SPEED / p["diameter"],
**sample(impulse, n, p["diameter"],
h.pressure_probes(rho, fluid, probe_p)["dp"]),
rho_min=float(min(v[0].min() for v in vals)),
rho_max=float(max(v[0].max() for v in vals)),
max_mach=float(max(np.sqrt(3 * (v[1]**2 + v[2]**2)).max() for v in vals)))
rr.ensure_finite(row)
rows.append(row)
if not healthy(row):
break
summary = summarize(rows) if healthy(rows[-1]) else dict(status="UNHEALTHY", steady=False,
verdict=None)
return dict(case=case, collision=a.collision, lattice=lp, rows=rows, **summary)
def plan(collision="SRT"):
return dict(re=RE, collision=collision, tstar_max=TSTAR_MAX, sample_tstar=SAMPLE_TSTAR,
window_tstar=WINDOW_TSTAR, bands=BANDS, range_limits=STEADY_RANGE, slope_limits=STEADY_SLOPE,
deviations=DEVIATIONS, pressure_scale=U_PHYS**2,
refined_root=lattice_parameters(40), refined_fine_tau=0.5 + 3 * 2 * rr.SPEED * 40 / RE,
uniform=lattice_parameters(80),
label="compared against the official reference bands (non-canonical neq_extrap outlet, zou_he_local inlet); not a full benchmark certification")
def main():
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--case", choices=["refined", "uniform", "both"], default="refined")
parser.add_argument("--collision", choices=["SRT", "MRT"], default="SRT")
parser.add_argument("--source", type=Path, default=rr.DEFAULT_SOURCE)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--plan", action="store_true", help="Write the frozen plan; no CUDA")
parser.add_argument("--max-wall-seconds", type=float)
a = parser.parse_args()
a.source = a.source.resolve()
a.output.mkdir(parents=True, exist_ok=False)
(a.output / "plan.json").write_text(json.dumps(plan(a.collision), indent=2, allow_nan=False))
if a.plan:
return
if not a.max_wall_seconds or a.max_wall_seconds <= 0 or not os.environ.get("CUDA_VISIBLE_DEVICES"):
raise SystemExit("GPU run needs --max-wall-seconds > 0 and explicit CUDA_VISIBLE_DEVICES")
h, _ = rr.helpers(a.source)
rr.config = config # host_geometry builds its root configuration through this name
a.deadline = time.monotonic() + a.max_wall_seconds
cases = ["refined", "uniform"] if a.case == "both" else [a.case]
results = [run_case(a, h, case) for case in cases]
(a.output / "result.json").write_text(json.dumps(results, indent=2, allow_nan=False))
if __name__ == "__main__":
main()
+23 -6
View File
@@ -74,7 +74,25 @@ def cv(g, fluid, origin, rectangle, nu, h=0.5):
Cauchy stress from centred velocity gradients, NOT postcollision non-equilibrium.
Rectangle faces lie on cell edges; fluid mask excludes the staircase cylinder.
"""
rho, mom, vel = moments(g)
return _cv_quantities(*moments(g), fluid, origin, rectangle, nu, h)
def cv_from_macros(rho, ux, uy, fluid, origin, rectangle, nu, h=0.5):
"""Same CV calculation from public macros, in one common length/time frame.
``origin`` is the lower cell edge, ``h`` the cell width, and ``nu`` uses
that frame. Returns [mass, Px, Py] inventory and outward total flux.
Crop fields/mask together and translate origin before calling if desired.
No boundary-exchange observation enters this independent estimator.
"""
rho = np.asarray(rho, np.float64)
if rho.ndim != 2 or any(np.shape(v) != rho.shape for v in (ux, uy, fluid)):
raise ValueError("CV macros and fluid mask must have matching 2D shapes")
vel = np.asarray([ux, uy], np.float64)
return _cv_quantities(rho, rho[None] * vel, vel, fluid, origin, rectangle, nu, h)
def _cv_quantities(rho, mom, vel, fluid, origin, rectangle, nu, h):
x0, y0, x1, y1 = np.rint((np.array(rectangle) - np.tile(origin, 2)) / h).astype(int)
np.testing.assert_allclose(
np.array([x0, y0, x1, y1]) * h + np.tile(origin, 2), rectangle, atol=1e-12
@@ -261,12 +279,11 @@ def run_case(c, ctx, output, provenance, continuation=False):
PeriodicBlockStepper(fine_cfg, layout, context=ctx, objects=[fobj]) as uniform,
FixedPatchStepper(
cfg,
c["bounds"],
context=ctx,
base_block=(8, 8),
fine_block=(8, 8),
fine_objects=[obj],
jets=jets,
refinements=[c["bounds"]],
level_blocks=[(8, 8), (8, 8)],
level_objects=[(), (obj,)],
level_jets=[(), tuple(jets)],
observe_raw_exchange=True,
) as amr,
):
+278 -41
View File
@@ -17,8 +17,109 @@ HERE = Path(__file__).resolve().parent
DEFAULT_SOURCE = HERE.parents[1]
BOUNDS = (40, 24, 320, 140)
BASE_BLOCK = (2, 881) # Exact 166 x 881 domain; no geometric padding.
FINE_BLOCK = (8, 8) # 240 x 568 including four fine halo cells.
FINE_BLOCK = (8, 8) # Wide 240 x 568 or compact 168 x 168, including fine halos.
SPEED = 0.01
JET_PEAK = 0.001
JET_WINDOWS = [(t, t + 10) for t in range(250, 300, 10)]
CV_CONTOURS = np.array([(48., 44., 160., 120.), (44., 36., 180., 128.)])
PLACEMENTS = {"wide": BOUNDS, "compact": (48, 48, 128, 128)}
COMPACT_CV_CONTOURS = np.array([(52., 52., 112., 112.), (50., 50., 120., 120.)])
def selected_cases(selection="both", uniform_diameter=80):
return [(case, diameter) for case, diameter in (("uniform", uniform_diameter), ("refined", 40))
if selection in ("both", case)]
def uniform_origin(diameter):
h = 40 / diameter
return (.5 - .5*h, 1 - h)
def cv_contours(refined, diameter, placement="wide"):
contours = CV_CONTOURS if placement == "wide" else COMPACT_CV_CONTOURS
return contours if refined else contours + [20/diameter, 0., 20/diameter, 0.]
def cv_selected(step, diameter, mode):
return mode == "pilot" or round(290 * diameter / SPEED) <= step <= round(300 * diameter / SPEED)
def cv_snapshot(rho, ux, uy, fluid, origin, refined, begin, n, diameter, impulse, placement="wide"):
"""Cropped public macros only; all inventories, fluxes and impulses use root units."""
from run_jet_circle_cv import cv_from_macros
rectangles = cv_contours(refined, diameter, placement)
h_cell = .5 if refined else 40 / diameter
origin = np.asarray(origin, dtype=float)
x0, y0, x1, y1 = np.rint((rectangles[-1] - np.tile(origin, 2)) / h_cell).astype(int)
selection = np.s_[y0-2:y1+2, x0-2:x1+2]
cropped_origin = origin + h_cell * np.array([x0-2, y0-2])
fields = [v[selection] for v in (rho, ux, uy, fluid)]
values = [cv_from_macros(*fields, cropped_origin, rect, .004, h=h_cell)
for rect in rectangles]
h_native = 1.0 if refined else 40 / diameter
return dict(step=begin+n, tstar=(begin+n)*SPEED/diameter,
time_root=(begin+n)*h_native, interval_start_root=begin*h_native,
interval_dt_root=n*h_native,
interval_impulse_root=(np.asarray(impulse)*h_native**2).tolist(),
inventory=[inventory.tolist() for inventory, _ in values],
outward_flux=[flux.tolist() for _, flux in values])
def public_jets(center, radius, scenario):
"""Same physical apertures; caller supplies its public geometry frame."""
if scenario == "none":
return ()
from CelerisLab.body.jets import Jet, Sector, Radial
peak = 0.0 if scenario == "zero" else JET_PEAK
return tuple(
Jet(Sector(center, angle, math.pi / 12, radius * 1.005),
Radial(center, inward=False), "parabolic", amplitude)
for angle, amplitude in ((math.pi / 2, peak), (3 * math.pi / 2, -peak))
)
def jet_interval(handle, scenario, begin, n, diameter):
"""Apply a reversal only at a completed interval boundary, before advancing."""
if scenario != "none":
boundaries = [round(t * diameter / SPEED) for t in (250, 260, 270, 280, 290, 300, 305)]
n = min([n] + [step - begin for step in boundaries if step > begin])
if scenario != "reverse":
return n, None
switch = round(300 * diameter / SPEED)
if begin < switch:
return min(n, switch - begin), None
if begin == switch:
handle.set_jet_amplitude(0, -JET_PEAK)
handle.set_jet_amplitude(1, JET_PEAK)
return n, dict(step=begin, tstar=300.0, amplitudes=[-JET_PEAK, JET_PEAK])
return n, None
def jet_diagnostics(rows, scenario):
"""Raw interval-weighted diagnostics only; no driven-flow qualification gate."""
windows = JET_WINDOWS + ([(300, 305)] if scenario == "reverse" else [])
result = []
for start, end in windows:
selected = [r for r in rows if start <= r["tstar_force"] < end]
if not selected:
continue
if any(2 * r["tstar_force"] - r["tstar"] < start - 1e-9
or r["tstar"] > end + 1e-9 for r in selected):
raise ValueError("Jet diagnostic interval crosses a window boundary")
weights = np.array([2 * (r["tstar"] - r["tstar_force"]) for r in selected])
record = dict(start_tstar=start, end_tstar=end, samples=len(selected),
covered_tstar=float(weights.sum()),
complete=bool(np.isclose(weights.sum(), end - start, rtol=0, atol=1e-9)))
for key in ("cd", "cl"):
values = np.array([r[key] for r in selected])
record[key] = dict(mean=float(np.average(values, weights=weights)),
min=float(values.min()), max=float(values.max()),
rms=float(np.sqrt(np.average(values**2, weights=weights))))
result.append(record)
return result
def helpers(source):
@@ -39,26 +140,34 @@ def helpers(source):
return h, periodic
def config(h, diameter, collision="SRT"):
def config(h, diameter, collision="SRT", jet_scenario="none"):
cfg, body, p = h.make_config(diameter, SPEED, collision, "neq_extrap", "zou_he_local")
cfg["method"]["outlet"]["srt_neq_damp"] = 0.5
if jet_scenario != "none":
if collision != "MRT":
raise ValueError("Public Jet scenarios require MRT")
cfg["method"]["ddf_shifting"] = False
return cfg, body, p
def plan(h, collision="SRT", observation_mode="last_base_step"):
_, _, c = config(h, 40, collision)
_, _, f = config(h, 80, collision)
x0, y0, x1, y1 = BOUNDS
def plan(h, collision="SRT", observation_mode="last_base_step", jet_scenario="none", uniform_diameter=80, placement="wide"):
_, _, c = config(h, 40, collision, jet_scenario)
_, _, f = config(h, uniform_diameter, collision, jet_scenario)
bounds = PLACEMENTS[placement]
x0, y0, x1, y1 = bounds
fine_active = 4 * (x1 - x0) * (y1 - y0)
coarse_uncovered = c["nx"] * c["ny"] - fine_active // 4
return dict(
status="prepared_not_run",
collision=collision,
jet_scenario=jet_scenario,
ddf_shifting=jet_scenario == "none",
observation_mode=observation_mode,
root_gpu_gates_required=True,
root=c,
uniform=f,
bounds_root_edge=list(BOUNDS),
bounds_root_edge=list(bounds),
fine_origin_root_edge=[x0 - 2, y0 - 2],
physical_patch_m=[
(x0 - 0.5) * c["dx"],
(y0 - 1) * c["dx"],
@@ -78,19 +187,31 @@ def plan(h, collision="SRT", observation_mode="last_base_step"):
physical_viscosity_m2_s=0.001,
finest_dx_m=f["dx"],
finest_dt_s=f["dt"],
refined_finest_dx_m=c["dx"]/2,
refined_finest_dt_s=c["dt"]/2,
approximate_leaf_cell_ratio=(coarse_uncovered + fine_active)
/ (f["nx"] * f["ny"]),
approximate_collision_work_ratio=(coarse_uncovered + 2 * fine_active)
/ (2 * f["nx"] * f["ny"]),
/ ((uniform_diameter/40) * f["nx"] * f["ny"]),
cost_caveat="Geometry-only estimates include ghosts/solids; exclude block halos, transfers, observation and launch overhead. Not measured speedup.",
sampling_offset="For origin (38,22), fine physical x=(j+75.5)*dx80 versus uniform i*dx80: half-fine-cell x offset. Fine physical y=(k+42.5)*dx80 equals uniform (l-.5)*dx80 at l=k+43: y samples align. Body, wall and source planes coincide.",
sampling_offset="Uniform nodes map to root edges X=.5+i*(40/D), Y=1+(j-.5)*(40/D). Refined centers are fine_origin_root_edge+(index+.5)*.5. Body, wall and source planes coincide; compare observables, not pointwise arrays.",
outlet=(
"SRT pressure NEQ, all nine fneq damped by .5, no backflow clamp; noncanonical DFG outlet"
if collision == "SRT" else
"MRT pressure NEQ, unknowns q2/6/8 extrapolated; no backflow clamp; noncanonical DFG outlet"
),
modes={"pilot_tstar": 5.0, "full_tstar": 300.0},
modes={"pilot_tstar": 5.0, "full_tstar": 305.0 if jet_scenario == "reverse" else 300.0},
observation_dense_tstar=0.02,
**(dict(jets=dict(
uniform_origin_node=f["center"], refined_origin_root_edge=[80.5, 81.0],
angles_rad=[math.pi / 2, 3 * math.pi / 2], half_width_rad=math.pi / 12,
radial_extent_over_radius=[0.0, 1.005],
field="Radial(outward)", profile="parabolic", condition="velocity",
initial_amplitudes=[0.0, 0.0] if jet_scenario == "zero" else [JET_PEAK, -JET_PEAK],
reversal_tstar=300.0 if jet_scenario == "reverse" else None,
diagnostic_windows_tstar=JET_WINDOWS + ([(300, 305)] if jet_scenario == "reverse" else []),
transfer="constant", driven_qualification="not assessed",
)) if jet_scenario != "none" else {}),
)
@@ -126,7 +247,7 @@ def initial_state(shape, origin, spacing, height=164.0, speed=SPEED, root=False)
return w * (1 + 3 * cu + 4.5 * cu * cu - 1.5 * velocity * velocity)
def host_geometry(source, h, collision="SRT"):
def host_geometry(source, h, collision="SRT", jet_scenario="none", placement="wide"):
"""Production donor/transfer and complete cut-link checks, without CUDA."""
from CelerisLab.config import load_lbm_config
from CelerisLab.lbm.refinement import FixedPatch2D, prepare_fine_boundary
@@ -139,12 +260,12 @@ def host_geometry(source, h, collision="SRT"):
from CelerisLab.body.geometry.circle import CircleGeometry
from CelerisLab.body.objects import SimObject
cfgdict, _, _ = config(h, 40, collision)
cfgdict, _, _ = config(h, 40, collision, jet_scenario)
with TemporaryDirectory() as tmp:
p = Path(tmp) / "cfg.json"
p.write_text(json.dumps(cfgdict))
cfg = load_lbm_config(str(p))
pair = FixedPatch2D.from_config(cfg, BOUNDS)
pair = FixedPatch2D.from_config(cfg, PLACEMENTS[placement])
layout = BlockLayout2D.build(cfg.ny, cfg.nx, *BASE_BLOCK)
validate_channel_config(cfg, [BASE_BLOCK, FINE_BLOCK])
targets = blocked_base_tables(pair, layout, np.argsort(pair.coalesce_coarse))[1]
@@ -153,18 +274,41 @@ def host_geometry(source, h, collision="SRT"):
body = SimObject(
0, geometry=CircleGeometry(80.5, 81.0, 20.0), center=(80.5, 81.0), radius=20.0
)
host = prepare_fine_boundary(cfg, pair, (body,), _shifted_blocked=True)
host = prepare_fine_boundary(
cfg, pair, (body,), jets=public_jets((80.5, 81.0), 20.0, jet_scenario),
_shifted_blocked=jet_scenario == "none",
)
return cfg, pair, body, host
def verify_source(source, collision="SRT", observation_mode="last_base_step"):
def verify_source(source, collision="SRT", observation_mode="last_base_step", jet_scenario="none", observe_cv=False):
if collision not in ("SRT", "MRT"):
raise ValueError("Matched collision must be SRT or MRT")
if observation_mode not in ("last_base_step", "window"):
raise ValueError("Unknown refined observation mode")
if observe_cv and jet_scenario == "none":
raise ValueError("--observe-cv requires an explicit public Jet scenario")
name = ("source_manifest_window.json" if observation_mode == "window" else
"source_manifest.json" if collision == "SRT" else "source_manifest_mrt.json")
expected = json.loads((HERE / "data/refined_dfg" / name).read_text())
if jet_scenario != "none":
if collision != "MRT" or observation_mode != "window":
raise ValueError("Public Jet scenarios require MRT and --exchange-window")
import subprocess
sys.path.insert(0, str(HERE))
from source_provenance import checkout_provenance
paths = set(expected["sha256"])
if observe_cv:
paths.add("tests/validation/run_jet_circle_cv.py")
paths.update(subprocess.check_output(
["git", "-C", str(source), "ls-files", "src"], text=True).splitlines())
return dict(
source_capture="current checkout, not historical DFG qualification",
**checkout_provenance(source), observation_mode=observation_mode,
sha256={name: hashlib.sha256((source / name).read_bytes()).hexdigest()
for name in sorted(paths)},
)
for name, digest in expected["sha256"].items():
if hashlib.sha256((source / name).read_bytes()).hexdigest() != digest:
raise RuntimeError("Source mismatch: " + name)
@@ -219,29 +363,32 @@ def run_case(a, h, periodic, case):
from contextlib import ExitStack
refined = case == "refined"
cfgdict, bodydict, p = config(h, 40 if refined else 80, a.collision)
cap = 5.0 if a.mode == "pilot" else 300.0
cfgdict, bodydict, p = config(h, 40 if refined else a.uniform_diameter, a.collision, a.jet_scenario)
cap = 5.0 if a.mode == "pilot" else (305.0 if a.jet_scenario == "reverse" else 300.0)
max_steps = math.ceil(cap * p["diameter"] / SPEED)
rows = []
healthy = True
step_seconds = 0.0
cv_seconds, cv_samples = 0.0, 0
amplitude_changes = []
started = time.monotonic()
check_deadline(a.deadline)
dest = a.output / case
dest.mkdir()
with ExitStack() as stack:
cv_out = stack.enter_context((dest / "cv_history.jsonl").open("w")) if a.observe_cv else None
if refined:
cfg, pair, body, host = host_geometry(a.source, h, a.collision)
cfg, pair, body, host = host_geometry(a.source, h, a.collision, a.jet_scenario, a.placement)
context = stack.enter_context(CudaContext(0))
backend = stack.enter_context(
FixedPatchStepper(
cfg,
BOUNDS,
context=context,
coarse_boundary="channel",
base_block=BASE_BLOCK,
fine_block=FINE_BLOCK,
fine_objects=(body,),
refinements=[PLACEMENTS[a.placement]],
level_blocks=[BASE_BLOCK, FINE_BLOCK],
level_objects=[(), (body,)],
level_jets=[(), public_jets((80.5, 81.0), 20.0, a.jet_scenario)],
observe_raw_exchange=True,
**({"observe_window_exchange": True} if a.observation_mode == "window" else {}),
)
@@ -274,6 +421,9 @@ def run_case(a, h, periodic, case):
sim = stack.enter_context(
Simulation(str(tmp / "lbm.json"), str(tmp / "body.json"), device_id=0)
)
for index, jet in enumerate(public_jets(p["center"], p["radius"], a.jet_scenario)):
if sim.add_jet(jet) != index:
raise RuntimeError("Unexpected public Jet registration index")
sim.initialize()
probe_p = p
fluid = (sim.field.flag.reshape(p["ny"], p["nx"]) & FLUID) != 0
@@ -290,6 +440,10 @@ def run_case(a, h, periodic, case):
else 0.5
)
n = min(max(1, round(width * p["diameter"] / SPEED)), max_steps - begin)
n, change = jet_interval(backend if refined else sim, a.jet_scenario,
begin, n, p["diameter"])
if change is not None:
amplitude_changes.append(change)
impulse = np.zeros(2)
tick = time.monotonic()
if refined:
@@ -359,6 +513,19 @@ def run_case(a, h, periodic, case):
/ SPEED,
**h.pressure_probes(rho, fluid, probe_p),
)
if a.jet_scenario != "none":
row.update(step_start=begin, tstar_start=begin * SPEED / p["diameter"],
interval_steps=n, impulse_x=float(impulse[0]),
impulse_y=float(impulse[1]))
if a.observe_cv and cv_selected(begin+n, p["diameter"], a.mode):
tick = time.monotonic()
record = cv_snapshot(rho, ux, uy, fluid,
pair.fine_origin if refined else uniform_origin(p["diameter"]),
refined, begin, n, p["diameter"], impulse, a.placement)
cv_out.write(json.dumps(record, allow_nan=False) + "\n")
cv_out.flush()
cv_samples += 1
cv_seconds += time.monotonic() - tick
check_deadline(a.deadline)
ensure_finite(row)
rows.append(row)
@@ -402,14 +569,24 @@ def run_case(a, h, periodic, case):
steps=begin,
startup_seconds=startup,
step_seconds=step_seconds,
observation_and_overhead_seconds=time.monotonic() - started - startup - step_seconds,
wall_seconds=time.monotonic() - started,
force="raw stationary-body link momentum reaction, per-unit-depth; average over entire observation interval; normalized in each base lattice",
mass_units="rho_reference * m^2 per unit depth",
flux_units="rho_reference * m^2/s per unit depth",
pressure="postcollision end-sample; physical pressure difference / (rho_reference * 1 m/s squared)",
suite_validated=False,
**periodic.assess(rows),
**(periodic.assess(rows) if a.jet_scenario in ("none", "zero") else
dict(periodic=False, checks={}, cycles=[])),
)
if a.jet_scenario != "none":
result.update(jet_diagnostic_windows=jet_diagnostics(rows, a.jet_scenario),
amplitude_changes=amplitude_changes,
pressure_classification=("DFG reference verdict retained, including pressure"
if a.jet_scenario == "zero" else "diagnostic only; no official DFG reference gate"))
if a.observe_cv:
result.update(cv_observation_seconds=cv_seconds, cv_samples=cv_samples,
cv_qualification="not assessed; scalar observations only")
if not healthy:
result["periodic"] = False
reference = json.loads(
@@ -425,7 +602,8 @@ def run_case(a, h, periodic, case):
else (
"pilot_complete"
if a.mode == "pilot"
else ("periodic" if result["periodic"] else "incomplete_periodicity")
else ("driven_diagnostics_complete" if a.jet_scenario in ("fixed", "reverse") else
"periodic" if result["periodic"] else "incomplete_periodicity")
)
)
(dest / "summary.json").write_text(json.dumps(result, indent=2, allow_nan=False))
@@ -464,27 +642,39 @@ def check_deadline(deadline):
)
def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_step"):
def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_step", jet_scenario="none", observe_cv=False,
case_selection="both", uniform_diameter=80, placement="wide"):
if path is None:
raise ValueError(
"Full comparison requires --pilot-results from a completed matched pilot"
)
previous = read_finite_json(path / "plan.json")
options = dict(case_selection=(case_selection, "both"), uniform_diameter=(uniform_diameter, 80), placement=(placement, "wide"))
if any(previous.get(k, default) != expected for k, (expected, default) in options.items()):
raise ValueError("Pilot case, resolution or placement differs")
if previous.get("observe_cv", False) != observe_cv:
raise ValueError("Pilot CV observation mode differs")
if previous.get("jet_scenario", "none") != jet_scenario:
raise ValueError("Pilot Jet scenario differs")
if previous.get("observation_mode") != observation_mode:
raise ValueError("Pilot observation mode differs")
if previous.get("collision") != collision:
raise ValueError("Pilot collision differs")
if previous["source_manifest"] != manifest or previous["bounds_root_edge"] != list(
BOUNDS
PLACEMENTS[placement]
):
raise ValueError("Pilot source or frozen geometry differs")
if previous.get("runner_sha256") != runner_hash():
raise ValueError("Pilot runner differs")
calibrated_seconds = 0.0
for case, diameter in (("uniform", 80), ("refined", 40)):
for case, diameter in selected_cases(case_selection, uniform_diameter):
summary = read_finite_json(path / case / "summary.json")
if (
summary.get("observation_mode") != observation_mode
any(summary.get(k, default) != expected for k, (expected, default) in options.items())
or summary.get("observe_cv", False) != observe_cv
or (observe_cv and summary.get("cv_samples") != 250)
or summary.get("jet_scenario", "none") != jet_scenario
or summary.get("observation_mode") != observation_mode
or summary.get("collision") != collision
or summary.get("case") != case
or summary.get("mode") != "pilot"
@@ -537,7 +727,10 @@ def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_st
raise ValueError("Unhealthy pilot observations")
ready = read_finite_json(path / case / "ready.json")
if (
ready.get("observation_mode") != observation_mode
any(ready.get(k, default) != expected for k, (expected, default) in options.items())
or ready.get("observe_cv", False) != observe_cv
or ready.get("jet_scenario", "none") != jet_scenario
or ready.get("observation_mode") != observation_mode
or ready.get("collision") != collision
or ready.get("case") != case
or ready.get("mode") != "pilot"
@@ -555,9 +748,9 @@ def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_st
or not 0 < stepping <= wall
):
raise ValueError("Invalid pilot calibration timings")
# Full physical duration is 60 times pilot; dense pilot sampling makes
# Full duration is 60 (or 61 with reversal) times pilot; dense sampling makes
# this conservative for the sparser early full observations. Not a guarantee.
calibrated_seconds += startup + 60 * (wall - startup)
calibrated_seconds += startup + (61 if jet_scenario == "reverse" else 60) * (wall - startup)
return 3 * calibrated_seconds
@@ -566,6 +759,13 @@ def main():
parser.add_argument("--exchange-window", action="store_true",
help="Opt in to refined window-ledger observations on the pinned candidate")
parser.add_argument("--collision", choices=["SRT", "MRT"], default="SRT")
parser.add_argument("--case", choices=["both", "uniform", "refined"], default="both")
parser.add_argument("--uniform-diameter", type=int, choices=[80, 160], default=80)
parser.add_argument("--placement", choices=list(PLACEMENTS), default="wide")
parser.add_argument("--jet-scenario", choices=["none", "zero", "fixed", "reverse"],
default="none", help="Opt in to public unshifted MRT Jets; reverse switches at t*=300 and ends at 305")
parser.add_argument("--observe-cv", action="store_true",
help="Record independent cropped-macro CV samples during pilot or full t*=290..300")
parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--mode", choices=["plan", "pilot", "full"], default="plan")
@@ -577,15 +777,26 @@ def main():
help="Required explicit campaign wall-time ceiling for GPU execution",
)
a = parser.parse_args()
if a.uniform_diameter == 160 and a.case != "uniform":
raise ValueError("D160 requires --case uniform; no matched-resolution claim")
if a.placement == "compact" and a.case != "refined":
raise ValueError("Compact placement requires --case refined")
if (a.uniform_diameter != 80 or a.placement != "wide") and a.jet_scenario == "none":
raise ValueError("Resolution/placement variants require an explicit public Jet scenario")
a.source = a.source.resolve()
a.observation_mode = "window" if a.exchange_window else "last_base_step"
manifest = verify_source(a.source, a.collision, a.observation_mode)
manifest = verify_source(a.source, a.collision, a.observation_mode, a.jet_scenario, a.observe_cv)
h, periodic = helpers(a.source)
from source_provenance import checkout_provenance
a.provenance = checkout_provenance(a.source)
a.provenance.update(jet_scenario=a.jet_scenario, ddf_shifting=a.jet_scenario == "none")
a.provenance.update(case_selection=a.case, uniform_diameter=a.uniform_diameter, placement=a.placement)
if a.observe_cv:
a.provenance["observe_cv"] = True
calibration = (
verify_pilot(a.pilot_results, manifest, a.collision, a.observation_mode)
verify_pilot(a.pilot_results, manifest, a.collision, a.observation_mode, a.jet_scenario, a.observe_cv,
a.case, a.uniform_diameter, a.placement)
if a.mode == "full" else None
)
if a.mode != "plan" and (
@@ -602,17 +813,31 @@ def main():
if a.output.exists() and any(a.output.iterdir()):
raise ValueError("Output must be empty/new")
a.output.mkdir(parents=True, exist_ok=True)
specification = plan(h, a.collision, a.observation_mode)
specification = plan(h, a.collision, a.observation_mode, a.jet_scenario, a.uniform_diameter, a.placement)
specification.update(
source_manifest=manifest,
**a.provenance,
gates=periodic.GATES,
reference_gates=periodic.REFERENCE_GATES,
gates=periodic.GATES if a.jet_scenario in ("none", "zero") else None,
reference_gates=periodic.REFERENCE_GATES if a.jet_scenario in ("none", "zero") else None,
runner_sha256=runner_hash(),
wall_time_budget_seconds=budget,
full_pilot_calibration_seconds=calibration,
)
host_geometry(a.source, h, a.collision)
if a.observe_cv:
specification["cv_observation"] = dict(
full_tstar=[290, 300], pilot_tstar=[.02, 5], spacing_tstar=.02,
refined_contours_root=cv_contours(True, 40, a.placement).tolist(),
uniform_contours_root=cv_contours(False, a.uniform_diameter, a.placement).tolist(),
uniform_grid_origin_root=uniform_origin(a.uniform_diameter),
refined_grid_origin_root=specification["fine_origin_root_edge"],
cell_width_root={"uniform": 40/a.uniform_diameter, "refined": .5},
viscosity_root=.004, components=["mass", "Px", "Py"],
units="root lattice per unit depth; outward flux per root time",
interval="Impulse belongs to the interval ending at each snapshot; first retained endpoint (pilot .02, full 290) supplies initial inventory only: exclude its preceding impulse",
qualification="not assessed; trapezoidal integration and time-coarsening are offline",
)
if a.case != "uniform":
host_geometry(a.source, h, a.collision, a.jet_scenario, a.placement)
(a.output / "plan.json").write_text(json.dumps(specification, indent=2))
if a.mode == "plan":
return 0
@@ -621,8 +846,9 @@ def main():
"GPU execution blocked: root GPU gates approval and explicit device selection required"
)
results = []
cases = selected_cases(a.case, a.uniform_diameter)
a.deadline = time.monotonic() + budget
for case in ("uniform", "refined"):
for case, _ in cases:
try:
result = run_case(a, h, periodic, case)
except TimeoutError as exc:
@@ -650,7 +876,12 @@ def main():
status="pilot_only" if a.mode == "pilot" else "incomplete",
matched_case_pass=False,
)
if len(results) == 2 and all(r["healthy"] and r["periodic"] for r in results):
if a.jet_scenario in ("fixed", "reverse"):
comparison.update(status=("pilot_only" if a.mode == "pilot" else "driven_diagnostics_complete")
if len(results) == len(cases) and all(r["healthy"] for r in results) else "invalid_state",
qualification="not assessed; execution completion is not physical acceptance",
jet_diagnostic_windows={r["case"]: r["jet_diagnostic_windows"] for r in results})
if a.case == "both" and len(results) == 2 and all(r["healthy"] and r["periodic"] for r in results):
comparison.update(
status="periodic_comparison",
refined_vs_uniform=periodic.compare(
@@ -660,14 +891,20 @@ def main():
comparison["matched_case_pass"] = comparison["refined_vs_uniform"][
"reference_agreement"
]
if a.case != "both":
comparison.update(status="single_case_" + results[0]["status"],
own_reference_comparison=results[0].get("reference_comparison"))
(a.output / "comparison.json").write_text(
json.dumps(comparison, indent=2, allow_nan=False)
)
return (
0
if len(results) == 2
if len(results) == len(cases)
and all(r["healthy"] for r in results)
and (a.mode == "pilot" or comparison["matched_case_pass"])
and (a.mode == "pilot" or comparison["matched_case_pass"]
or a.jet_scenario in ("fixed", "reverse")
or (a.case != "both" and results[0]["periodic"]
and results[0].get("reference_comparison", {}).get("reference_agreement", False)))
else 2
)
@@ -1,7 +1,6 @@
"""New-source four-case repeat of 11901a7; frozen oracle and gates are unchanged."""
import argparse
import hashlib
import json
import math
from pathlib import Path
@@ -9,25 +8,16 @@ from pathlib import Path
from assess import BASELINE_FACTOR, FLOORS
from oracle import LIMITS, TIMES, U, rectangles
from run_case import (INITIALIZATION, sample_steps, NUMERICAL_REVISION, SOURCE_SHA256,
BUILD_CAPTURE_SHA256, REFERENCE_HARNESS, SCALED_DIAGNOSTIC)
BUILD_CAPTURE_SHA256, REFERENCE_HARNESS)
# Host-generated snapshots of a135252 at the frozen local configurations.
# Host-generated snapshots of a135252 at the frozen two local configurations.
# These describe expected inputs; actual sidecars are captured before cleanup.
BUILD_SOURCES = {
128: "803177416b7599ab89cc5afd35cf4a28a673b2f4bb8e8a6a1abc0b6d77610873",
256: "17ec5e60f87b840d3d13b8778bc8a3aecf485469c5a061af46c5cc19b73396cf",
512: "4e85e38098858ec95c137d1b1e96846a58d9cf3cc242b349a3ad7c8b8e3ce0aa",
}
CASES = ((100, 128, 1, "constant"), (100, 256, 1, "constant"),
(100, 128, 2, "constant"), (100, 128, 2, "projected_linear"))
REFERENCE_REVISION = "2cf9853952e52f59bea63cf5c1d3a1cd76e53eed"
# Audited 2cf9853 scalar receipts, archived at evidence commit 90849a55cad4a0da9b72747da7bfe6eb75962a44.
REFERENCE_RECEIPTS = {
"uniform128": "724efafc267eaa59dcc59e8d4c16177b67f6c6a08fd2732a481833237bb19cee",
"uniform256": "3c312a99ebf747a135674dd481b77455a3d2c9a406ca388f30524b37dad9bbe9",
"constant128": "2ee28efd1e81b2e28b7902b6c4f95bb2c1ea4e5f0ec484638fa5ae83dca34bba",
"projected128": "ee67b1d4d90a048c0d3d1a16447c6eab3d89c94d151f791306e373f96dd8b963",
}
NUMERICS = dict(lattice_model="D2Q9", collision="SRT", store_precision="FP32",
ddf_shifting=True, streaming="double_buffer", les_enabled=False,
boundary="periodic", block_shape=[8, 8],
@@ -83,11 +73,7 @@ def valid_builds(record, n, depth):
return True
def assess(records, expected_revision=None, *, diagnostic=None):
if diagnostic not in (None, SCALED_DIAGNOSTIC):
raise ValueError("Unknown diagnostic")
base_n = 256 if diagnostic else 128
cases = tuple((re, n * base_n // 128, depth, scheme) for re, n, depth, scheme in CASES)
def assess(records, expected_revision=None):
indexed, duplicates = {}, []
for record in records:
key = tuple(record.get(k) for k in ("re_L", "n", "depth", "explosion"))
@@ -95,11 +81,9 @@ def assess(records, expected_revision=None, *, diagnostic=None):
duplicates.append(str(key))
else:
indexed[key] = record
missing = [str(k) for k in cases if k not in indexed]
checks = dict(complete_matrix=not missing, no_extra_cases=set(indexed) <= set(cases),
missing = [str(k) for k in CASES if k not in indexed]
checks = dict(complete_matrix=not missing, no_extra_cases=set(indexed) <= set(CASES),
no_duplicate_cases=not duplicates, finite_receipts=finite_tree(records))
if diagnostic or any("diagnostic" in r for r in records):
checks["diagnostic_identity"] = all(r.get("diagnostic") == diagnostic for r in records)
sources = [r.get("source", {}) for r in records]
checks["same_source_and_harness"] = bool(sources) and len({
json.dumps(s, sort_keys=True) for s in sources}) == 1
@@ -122,11 +106,9 @@ def assess(records, expected_revision=None, *, diagnostic=None):
) and len(set(devices)) == 1
enabled = [r.get("early_pressure", {}).get("enabled") for r in records]
checks["same_early_sampling"] = bool(enabled) and all(type(v) is bool for v in enabled) and len(set(enabled)) == 1
if diagnostic:
checks["diagnostic_early_sampling"] = all(v is True for v in enabled)
shared_valid = all(checks.values())
complete, case_valid, early = {}, {}, []
for key in cases:
for key in CASES:
if key not in indexed:
continue
r = indexed[key]
@@ -153,7 +135,7 @@ def assess(records, expected_revision=None, *, diagnostic=None):
(("module", "__init__.py"), ("compiler", "cuda/compiler_v2.py"), ("kernel_root", "lbm/kernels")))
expected_patch = [] if depth == 1 else rectangles(n, depth)
checks[label + " patch"] = json.dumps(r.get("refinements")) == json.dumps(expected_patch)
cells = [n*n] if depth == 1 else [17*n*n//32, 15*n*n//8]
cells = [n*n] if depth == 1 else [8704, 30720]
hops = [16*n * 2**l for l in range(depth)]
work = r.get("workload", {})
checks[label + " workload"] = (
@@ -186,22 +168,19 @@ def assess(records, expected_revision=None, *, diagnostic=None):
comparisons, scheme_results = [], {}
for scheme in ("constant", "projected_linear"):
key = (100, base_n, 2, scheme)
key = (100, 128, 2, scheme)
for metric, floor in FLOORS.items():
amr, coarse, fine = (worst(k, metric) for k in (key, cases[0], cases[1]))
amr, coarse, fine = (worst(k, metric) for k in (key, CASES[0], CASES[1]))
budget = None if coarse is None else max(BASELINE_FACTOR * coarse, floor)
budget = budget if finite(budget) else None
ratio = None if amr is None or fine is None else amr / max(fine, 1e-30)
passed = amr is not None and budget is not None and amr <= budget
checks[f"baseline {scheme}/{metric}"] = passed
comparisons.append(dict(explosion=scheme, metric=metric, amr=amr,
baseline_budget=budget, passed=passed,
**{f"uniform{base_n}": coarse, f"uniform{2*base_n}": fine,
f"amr_to_uniform{2*base_n}": ratio if finite(ratio) else None}))
if diagnostic:
comparisons[-1]["floor_dominates"] = None if coarse is None else floor >= BASELINE_FACTOR * coarse
uniform128=coarse, baseline_budget=budget, passed=passed,
uniform256=fine, amr_to_uniform256=ratio if finite(ratio) else None))
baseline_passed = all(checks[f"baseline {scheme}/{metric}"] for metric in FLOORS)
valid = shared_valid and all(case_valid.get(k, False) for k in (cases[0], cases[1], key))
valid = shared_valid and all(case_valid.get(k, False) for k in (CASES[0], CASES[1], key))
scheme_results[scheme] = dict(passed=valid and baseline_passed,
valid_case_and_controls=valid, baseline_passed=baseline_passed)
return dict(passed=all(checks.values()), missing=missing, duplicates=duplicates, checks=checks,
@@ -211,89 +190,19 @@ def assess(records, expected_revision=None, *, diagnostic=None):
floors=FLOORS, baseline_factor=BASELINE_FACTOR,
main_statistic="maximum of four post-initial samples at t*=.12,.24,.36,.48",
early_statistic="sampled pressure maximum only, t*=k*.001875, k=0..64; not an all-step or continuous maximum",
finest_control=f"uniform{2*base_n} and AMR/ uniform{2*base_n} ratios are descriptive, not acceptance gates",
finest_control="uniform256 and AMR/ uniform256 ratios are descriptive, not acceptance gates",
scope="2D periodic Re100 smooth-decay diagnostic; no new algorithm, convergence-order or general AMR qualification")
def load_reference(paths):
"""Use the four audited historical files verbatim, never relabel them fresh controls."""
blobs = [path.read_bytes() for path in paths]
hashes = [hashlib.sha256(blob).hexdigest() for blob in blobs]
if len(hashes) != 4 or set(hashes) != set(REFERENCE_RECEIPTS.values()):
raise ValueError("Historical N128 receipt hashes differ from the audited 2cf9853 evidence")
return [json.loads(blob) for blob in blobs]
def resolution_trend(records, reference, expected_revision):
"""Signed changes are descriptive; they never replace the frozen acceptance budgets."""
old = assess(reference, REFERENCE_REVISION)
new = assess(records, expected_revision, diagnostic=SCALED_DIAGNOSTIC)
# Numerical screen failures still have useful raw errors; only malformed inputs block trends.
valid = all(v for result in (old, new) for k, v in result["checks"].items()
if not k.startswith("baseline ") and not k.endswith(" absolute"))
valid = valid and all(maximum(r["rows"], m) is not None for r in reference + records for m in LIMITS)
oracle = reference[0]["source"]["harness_sha256"].get("oracle.py") if valid else None
valid = valid and bool(oracle) and records[0]["source"]["harness_sha256"].get("oracle.py") == oracle
if not valid:
return dict(valid=False, per_time=[], worst_post_initial=[], uniform256_bridge=[])
key = lambda r: (r["re_L"], r["n"], r["depth"], r["explosion"])
before, after = {key(r): r for r in reference}, {key(r): r for r in records}
def change(role, metric, low, high, **extra):
ratio = high / low if low else None
percent = 100*(ratio-1) if ratio is not None else None
return dict(role=role, metric=metric, n128_campaign_error=low, n256_campaign_error=high,
signed_change=high-low, new_to_old=ratio if finite(ratio) else None,
ratio_unavailable_reason=None if finite(ratio) else "zero historical error" if not low else "overflow",
change_percent=percent if finite(percent) else None, **extra)
per_time, worst, bridge = [], [], []
for old_key, role in zip(CASES, ("uniform_root", "uniform_fine", "constant", "projected_linear")):
re, n, depth, scheme = old_key
low, high = before[old_key], after[(re, 2*n, depth, scheme)]
for metric in LIMITS:
per_time.extend(change(role, metric, a[metric], b[metric], tstar=a["tstar"])
for a, b in zip(low["rows"], high["rows"]))
worst.append(change(role, metric, maximum(low["rows"][1:], metric),
maximum(high["rows"][1:], metric)))
for a, b in zip(before[CASES[1]]["rows"], after[CASES[1]]["rows"]):
bridge.extend(dict(metric=m, tstar=a["tstar"], historical=a[m], fresh=b[m], signed_change=b[m]-a[m])
for m in LIMITS)
low, high = before[CASES[1]], after[CASES[1]]
bridge_identity = dict(main_rows_identical=low["rows"] == high["rows"],
early_pressure_rows_identical=low["early_pressure"]["rows"] == high["early_pressure"]["rows"])
same_runtime = reference[0]["runtime_identity"] == records[0]["runtime_identity"]
return dict(valid=True, reference_revision=REFERENCE_REVISION, reference_receipt_sha256=REFERENCE_RECEIPTS,
reference_scheme_results=old["scheme_results"], per_time=per_time, worst_post_initial=worst,
uniform256_bridge=bridge, uniform256_bridge_identity=bridge_identity,
same_recorded_runtime=same_runtime,
trend_interpretation_ready=all(bridge_identity.values()) and same_runtime,
scope="Historical same-production/oracle reference versus fresh controls; same-placement two-resolution trend only, not convergence order or phase robustness")
if __name__ == "__main__":
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("files", type=Path, nargs="+")
p.add_argument("--output", type=Path, required=True)
p.add_argument("--expected-revision", required=True)
p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC])
p.add_argument("--reference-files", type=Path, nargs=4)
args = p.parse_args()
if args.output.exists():
p.error("Output already exists; choose a new path to retain prior evidence")
if bool(args.diagnostic) != bool(args.reference_files):
p.error("--diagnostic scaled_n256 requires --reference-files; references are diagnostic-only")
records = [json.loads(path.read_text()) for path in args.files]
result = assess(records, args.expected_revision, diagnostic=args.diagnostic)
if args.diagnostic:
try:
reference = load_reference(args.reference_files)
except ValueError as exc:
p.error(str(exc))
result["diagnostic"] = args.diagnostic
result["resolution_trend"] = resolution_trend(records, reference, args.expected_revision)
result["checks"]["valid_historical_reference_and_trend"] = result["resolution_trend"]["valid"]
result["passed"] = all(result["checks"].values())
result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision)
args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
print(json.dumps(result, indent=2, allow_nan=False))
raise SystemExit(0 if result["passed"] else 1)
@@ -1,336 +0,0 @@
"""Fixed four-case shear-wave contracts.
Default: N128 (U128, U256, constant R128, projected R128). Opt-in
--diagnostic scaled_shear_n256: N256 (U256, U512, constant R256, projected
R256), same physical rectangle, t* schedule, nu = U N / 100, eps(N), limits
and floors, plus a descriptive comparison with the hash-pinned N128 receipts.
Shared provenance, build, workload and sample-time checks are reused from
assess_projected_pairs (same cases, source pins and geometry); only the
Taylor-Green-specific checks are replaced by shear-wave metadata and physics.
"""
import argparse
import hashlib
import json
import math
from pathlib import Path
import assess_projected_pairs as tg
import shear_wave as sw
import shear_nested as nested
import shear_twin as twin
from run_case import (NESTED_SHEAR, SCALED_DIAGNOSTIC, SCALED_SHEAR, SHEAR_NESTED_SHA256,
SHEAR_TWIN_SHA256, SHEAR_WAVE_SHA256, TWIN_SHEAR, sample_steps)
# Audited de5b0cc N128 receipts (evidence commit cba6bca, shear-wave-de5b0cc zip).
REFERENCE_REVISION = "de5b0cc080794d0ce44d37f845aa8e8dbf8c2016"
REFERENCE_RECEIPTS = {
"uniform128": "d4a60fb806d04bfa8098a862cceba51e00b49979a6a250fef17a5fc8e3ea3255",
"uniform256": "e0b58e734724be5f82ee5d6df84395e3181b4e96e4246dd5a574ea23a8d12386",
"constant128": "c70d7cd14b1257a01cc49f8edfcc7685d430e5301e2f2d60ed426936f8b96c51",
"projected128": "980161458b9463af09a8e3178063fea0f4bbeb465329e1d3becdb0f224537b54",
}
ROLES = ("uniform_root", "uniform_fine", "constant", "projected_linear")
SHARED = ("complete_matrix", "no_extra_cases", "no_duplicate_cases", "finite_receipts",
"same_source_and_harness", "clean_a135252_numerical_source",
"same_recorded_runtime", "same_explicit_device", "same_early_sampling")
PER_CASE = ("complete_times", "scales", "actual_builds", "reference", "runtime_source",
"patch", "workload")
METADATA_ULPS = 4
def close_ulps(recorded, expected, ulps=METADATA_ULPS):
"""Portable check for recomputed float metadata only (e.g. eps = |k|^2).
The recorded value comes from another runtime/libm, so exact equality can
fail by one ULP (observed: 0.012047856934923534 Horsea vs ...532 here).
A few ULPs is far below any meaningful change; hashes, case identity and
scientific thresholds stay exact.
"""
return (isinstance(recorded, float) and math.isfinite(recorded)
and abs(recorded - expected) <= ulps * math.ulp(max(abs(recorded), abs(expected))))
def metadata_ok(record, n):
return (record.get("flow") == "shear_wave" and record.get("mode") == list(sw.MODE)
and record.get("polarization") == sw.E.tolist()
and record.get("angle_deg") == sw.ANGLE_DEG and record.get("phase") == sw.PHI
and close_ulps(record.get("eps"), sw.eps(n))
and record.get("numerics") == tg.NUMERICS
and record.get("initialization") == sw.INITIALIZATION
and record.get("limits") == sw.LIMITS
and record.get("early_pressure") == dict(enabled=False, rows=[],
sampled_max_pressure_l2=None)
and (record.get("source") or {}).get("shear_wave_sha256") == SHEAR_WAVE_SHA256)
def cases_for(scaled):
factor = 2 if scaled else 1
return tuple((re, n * factor, depth, scheme) for re, n, depth, scheme in tg.CASES)
def assess(records, expected_revision=None, *, scaled=False):
if not isinstance(records, list) or not all(isinstance(r, dict) for r in records):
raise ValueError("records must be a list of receipt objects")
# The scaled TG assessor supplies N256/512 cases, builds and workload; its
# TG-only diagnostic identity and early-sampling checks are not used here.
base_checks = tg.assess(records, expected_revision,
diagnostic=SCALED_DIAGNOSTIC if scaled else None)["checks"]
checks = {name: base_checks.get(name, False) for name in SHARED}
checks["shear_diagnostic_identity"] = all(
r.get("diagnostic") == (SCALED_SHEAR if scaled else None) for r in records)
cases = cases_for(scaled)
indexed = {}
for r in records:
indexed.setdefault(tuple(r.get(k) for k in ("re_L", "n", "depth", "explosion")), r)
for key in cases:
_, n, depth, scheme = key
label = f"{n}/{depth}/{scheme}"
for name in PER_CASE:
checks[f"{label} {name}"] = base_checks.get(f"{label} {name}", False)
checks[f"{label} shear_metadata"] = key in indexed and metadata_ok(indexed[key], n)
contract_valid = all(checks.values())
result = dict(checks=checks, contract_valid=contract_valid,
cases=[list(k) for k in cases], expected_revision=expected_revision,
diagnostic=SCALED_SHEAR if scaled else None,
scope="2D periodic Re100 oblique (2,1) shear wave; one phase and one "
"rectangle; no placement robustness or convergence order")
if not contract_valid:
result.update(status="INVALID_SET", passed=False)
return result
try:
physics = sw.assess(indexed[cases[0]], indexed[cases[1]],
[indexed[cases[2]], indexed[cases[3]]])
except (KeyError, ValueError) as error:
result.update(status="INVALID_SET", passed=False, reason=str(error))
return result
result.update(physics=physics, status=physics["status"],
passed=physics["status"] == "PASS")
try:
result["signed"] = signed_summary([indexed[k] for k in cases])
except (KeyError, TypeError, ValueError) as error:
result["signed_unavailable"] = str(error)
return result
CASE_AGNOSTIC = ("finite_receipts", "same_source_and_harness", "clean_a135252_numerical_source",
"same_recorded_runtime", "same_explicit_device", "same_early_sampling")
# Fixed multi-box presets sharing one contract; only geometry, controls, pins differ.
PRESETS = {
NESTED_SHEAR: dict(
module=nested, physics=nested.assess, snapshot=True,
pins={"shear_nested_sha256": SHEAR_NESTED_SHA256},
finest_control="uniform 4N (U512)",
scope="2D periodic Re100 oblique (2,1) shear wave; fixed three-level/two-region "
"placement; changes placement and refined volume as well as depth, so no "
"isolated-level cost or resolution trend"),
TWIN_SHEAR: dict(
module=twin, physics=sw.assess, snapshot=False,
pins={"shear_nested_sha256": SHEAR_NESTED_SHA256,
"shear_twin_sha256": SHEAR_TWIN_SHA256},
finest_control="uniform 2N (U256)",
scope="2D periodic Re100 oblique (2,1) shear wave; depth-2 twin of the nested "
"preset (same two L0->L1 boxes, no L2). Twin vs nested is the effect of adding "
"this nested refinement; it also changes finest spacing and refined volume, so it "
"does not isolate abstract depth"),
}
def _nested_case_ok(r, n, depth, diagnostic=NESTED_SHEAR):
"""Per-case identity of a fixed multi-box preset (scales, builds, patch, work, pins)."""
preset = PRESETS[diagnostic]
geometry = preset["module"]
runtime = r.get("runtime_source", {})
package = Path(runtime.get("checkout", "")) / "src/CelerisLab"
cells = [n * n] if depth == 1 else geometry.LEAVES
hops = [16 * n * 2**level for level in range(depth)]
work = r.get("workload", {})
snapshot = r.get("final_snapshot")
expected_patch = [] if depth == 1 else geometry.REFINEMENTS
return {
"complete_times": r.get("completed") is True and tg.at_steps(
r.get("rows", []), sample_steps(n)[0], n),
"scales": (r.get("U") == sw.U and tg.finite(r.get("viscosity"))
and abs(r["viscosity"] - sw.U * n / 100) < 1e-15
and r.get("tau_by_level") == [0.5 + 3 * sw.U * n / 100 * 2**l
for l in range(depth)]
and r.get("mach") == sw.U * math.sqrt(3)),
"actual_builds": tg.valid_builds(r, n, depth),
"reference": (r.get("numerical_revision") == tg.NUMERICAL_REVISION
and r.get("reference_harness") == tg.REFERENCE_HARNESS),
"runtime_source": bool(runtime.get("checkout")) and all(
runtime.get(k) == str(package / tail) for k, tail in
(("module", "__init__.py"), ("compiler", "cuda/compiler_v2.py"),
("kernel_root", "lbm/kernels"))),
"patch": json.dumps(r.get("refinements")) == json.dumps(expected_patch),
"workload": (work.get("active_cells_by_level") == cells and work.get("level_hops") == hops
and work.get("root_intervals") == 16 * n
and work.get("active_collision_updates_by_level") == [
c * h for c, h in zip(cells, hops)]
and work.get("active_collision_updates") == sum(
c * h for c, h in zip(cells, hops))),
"shear_metadata": metadata_ok(r, n),
"nested_module_pin": all((r.get("source") or {}).get(key) == pin
for key, pin in preset["pins"].items()),
# Optional, nested projected depth-3 only; bounded and hashed when present.
"snapshot": snapshot is None or (
preset["snapshot"] and depth == nested.DEPTH and r.get("explosion") == "projected_linear"
and snapshot.get("kept") is True and isinstance(snapshot.get("sha256"), str)
and len(snapshot["sha256"]) == 64
and 0 < snapshot.get("bytes", 0) <= nested.SNAPSHOT_MAX_BYTES
and snapshot.get("leaf_counts") == nested.LEAVES
and snapshot.get("step") == sample_steps(n)[0][-1]),
}
def assess_nested(records, expected_revision=None, diagnostic=NESTED_SHEAR):
"""Fixed four-case preset contract: nested_shear_n128 (U128, U512, constant/projected
R128 depth 3) or twin_shear_n128 (U128, U256, constant/projected R128 depth 2)."""
preset = PRESETS[diagnostic]
geometry = preset["module"]
if not isinstance(records, list) or not all(isinstance(r, dict) for r in records):
raise ValueError("records must be a list of receipt objects")
base_checks = tg.assess(records, expected_revision)["checks"]
checks = {name: base_checks.get(name, False) for name in CASE_AGNOSTIC}
indexed, duplicates = {}, []
for r in records:
key = tuple(r.get(k) for k in ("re_L", "n", "depth", "explosion"))
(duplicates.append(key) if key in indexed else indexed.__setitem__(key, r))
checks.update(complete_matrix=all(k in indexed for k in geometry.CASES),
no_extra_cases=set(indexed) <= set(geometry.CASES),
no_duplicate_cases=not duplicates,
shear_diagnostic_identity=all(r.get("diagnostic") == diagnostic
for r in records))
for key in geometry.CASES:
_, n, depth, scheme = key
label = f"{n}/{depth}/{scheme}"
case = _nested_case_ok(indexed[key], n, depth, diagnostic) if key in indexed else {}
for name in ("complete_times", "scales", "actual_builds", "reference", "runtime_source",
"patch", "workload", "shear_metadata", "nested_module_pin", "snapshot"):
checks[f"{label} {name}"] = bool(case.get(name, False))
contract_valid = all(checks.values())
result = dict(checks=checks, contract_valid=contract_valid, diagnostic=diagnostic,
cases=[list(k) for k in geometry.CASES], expected_revision=expected_revision,
finest_control=preset["finest_control"], scope=preset["scope"])
if not contract_valid:
result.update(status="INVALID_SET", passed=False)
return result
try:
cases = geometry.CASES
physics = preset["physics"](indexed[cases[0]], indexed[cases[1]],
[indexed[cases[2]], indexed[cases[3]]])
except (KeyError, ValueError) as error:
result.update(status="INVALID_SET", passed=False, reason=str(error))
return result
result.update(physics=physics, status=physics["status"], passed=physics["status"] == "PASS")
try:
# Role "uniform_fine" is the preset's finest control (4N nested, 2N twin).
result["signed"] = signed_summary([indexed[k] for k in geometry.CASES])
except (KeyError, TypeError, ValueError) as error:
result["signed_unavailable"] = str(error)
return result
def _signed(record):
"""Raw signed quantities of one valid record (None when the fit is unavailable)."""
s = sw.summarize(record)
bias = None if not s["rate_fit_valid"] else s["rate"] / s["exact_rate"] - 1
return dict(n=record["n"], depth=record["depth"], explosion=record["explosion"],
rate_fit_valid=s["rate_fit_valid"], signed_rate_bias=bias,
rate_tstar=s["rate_tstar"], exact_rate_tstar=s["exact_rate_tstar"],
interval_rate_errors=s["interval_rate_errors"],
residual_l2=s["residual_l2"], spurious_pressure_scaled=s["spurious_pressure_scaled"],
spurious_pressure_rms=s["spurious_pressure_rms"],
final_amplitude_error=record["rows"][-1]["amplitude_error"])
def signed_summary(records):
"""Signed bias per case and refined excess over each uniform control."""
rows = dict(zip(ROLES, (_signed(r) for r in records)))
def minus(a, b):
return None if a is None or b is None else a - b
for role in ("constant", "projected_linear"):
for control in ("uniform_root", "uniform_fine"):
rows[role][f"excess_over_{control}"] = {
key: minus(rows[role][key], rows[control][key])
for key in ("signed_rate_bias", "residual_l2", "spurious_pressure_rms")}
return rows
def load_reference(paths):
"""The four audited N128 receipts verbatim; never relabelled as fresh controls."""
blobs = [Path(path).read_bytes() for path in paths]
hashes = {hashlib.sha256(blob).hexdigest() for blob in blobs}
if len(blobs) != 4 or hashes != set(REFERENCE_RECEIPTS.values()):
raise ValueError("N128 receipt hashes differ from the audited de5b0cc evidence")
return [json.loads(blob) for blob in blobs]
def resolution_trend(new_result, reference, reference_revision=REFERENCE_REVISION):
"""Descriptive N128 -> N256 signed changes; never an acceptance gate or an order."""
old_result = assess(reference, reference_revision)
if "signed" not in old_result or "signed" not in new_result:
return dict(valid=False, reason="N128 reference or N256 set lacks a valid signed summary")
out = {}
for role in ROLES:
old, new = old_result["signed"][role], new_result["signed"][role]
out[role] = {}
for key in ("signed_rate_bias", "residual_l2", "spurious_pressure_rms",
*(f"excess_over_{c}" for c in ("uniform_root", "uniform_fine"))):
if key not in old:
continue
pairs = old[key].items() if isinstance(old[key], dict) else [(None, old[key])]
for sub, low in pairs:
high = new[key][sub] if sub else new[key]
ok = low is not None and high is not None
out[role][key if sub is None else f"{key}.{sub}"] = dict(
n128=low, n256=high, signed_change=high - low if ok else None,
n256_to_n128=high / low if ok and low else None)
return dict(valid=True, reference_revision=reference_revision,
reference_receipt_sha256=REFERENCE_RECEIPTS,
reference_status=old_result["status"], changes=out,
scope="same-placement two-resolution trend; not a convergence order, "
"placement robustness or redesign decision")
if __name__ == "__main__":
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("files", type=Path, nargs=4)
p.add_argument("--output", type=Path, required=True)
p.add_argument("--expected-revision", required=True)
p.add_argument("--diagnostic", choices=[SCALED_SHEAR, NESTED_SHEAR, TWIN_SHEAR])
p.add_argument("--reference-files", type=Path, nargs=4)
args = p.parse_args()
if args.output.exists():
p.error("Output already exists; choose a new path to retain prior evidence")
if args.diagnostic in PRESETS:
if args.reference_files:
p.error(f"{args.diagnostic} takes no reference files (no resolution trend)")
result = assess_nested([json.loads(path.read_text()) for path in args.files],
args.expected_revision, args.diagnostic)
text = json.dumps(result, indent=2, allow_nan=False) + "\n"
args.output.write_text(text)
print(text, end="")
raise SystemExit(0 if result["passed"] else 1)
if bool(args.diagnostic) != bool(args.reference_files):
p.error("--diagnostic scaled_shear_n256 requires --reference-files (N128 receipts)")
result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision,
scaled=bool(args.diagnostic))
if args.diagnostic:
try:
reference = load_reference(args.reference_files)
except ValueError as exc:
p.error(str(exc))
result["resolution_trend"] = resolution_trend(result, reference)
text = json.dumps(result, indent=2, allow_nan=False) + "\n"
args.output.write_text(text)
print(text, end="")
raise SystemExit(0 if result["passed"] else 1)
+18 -165
View File
@@ -18,37 +18,6 @@ SOURCE_SHA256 = "a5dbfd804befce868067d79a23e5fd47dce7f505b3f9b1bb2145fb57cb5f217
BUILD_CAPTURE_SHA256 = "3dbaa6547cb0a4cb1541054f3b64ad8b6e18cc70527ac6166e56b944e9bd464a"
REFERENCE_HARNESS = "11901a7f77557a901e017a171eeae1758d584fa5"
INITIALIZATION = "physical FP64 equilibrium G at actual cell centers; rho=1+3p"
SHEAR_WAVE_SHA256 = "81d5c8ee60864372887cddf9239580c82fb6750d956038f1e3ee9de61aa4e877"
SCALED_DIAGNOSTIC = "scaled_n256"
SCALED_SHEAR = "scaled_shear_n256" # mutually exclusive with scaled_n256 (one flow each)
NESTED_SHEAR = "nested_shear_n128" # fixed three-level/two-region shear preset
SHEAR_NESTED_SHA256 = "fa6cfe87fb4df8c6d732c637c885276833c81973b950c7724d734f8fc350bcb0"
TWIN_SHEAR = "twin_shear_n128" # depth-2 twin: nested L0->L1 boxes without L2
SHEAR_TWIN_SHA256 = "1cbf37f18e89aea577581631c4565f51e68d22a3739b698816ef64acbf2b2bb4"
SHEAR_PROBES_SHA256 = "53ba8af8ac9c8078c40f84488c7ade4c11ccc91dd1fc275ff806042e9af30166" # signed interface-probe mechanism diagnostic
def flow_functions(flow="taylor_green"):
"""(initial, metrics, limits, initialization); the default is the frozen TG oracle."""
if flow == "taylor_green":
return initial, metrics, LIMITS, INITIALIZATION
if flow == "shear_wave":
import shear_wave
return shear_wave.initial, shear_wave.metrics, shear_wave.LIMITS, shear_wave.INITIALIZATION
raise ValueError(f"unknown flow {flow!r}")
def finite_row(row):
"""Every scalar or list-valued diagnostic is finite (TG rows are all scalars)."""
return all(np.isfinite(np.asarray(value, dtype=float)).all() for value in row.values())
def sample(run, masks, step, n, viscosity, initial_mass, metrics, depth, label="sample"):
arrays = (run.download(),) if depth == 1 else run.download()
row = metrics(arrays, masks, step, n, viscosity, initial_mass)
if not finite_row(row):
raise FloatingPointError(f"nonfinite {label} diagnostics")
return row
def sample_steps(n, early_pressure=False):
@@ -95,7 +64,7 @@ def provenance(repo):
)
def verify_source(repo, revision, flow="taylor_green", probes=False):
def verify_source(repo, revision):
"""Exact clean harness commit, fixed production tree, and reviewed capture helper."""
import oracle
from tests.validation.multigrid import check_projected_pairs as gate
@@ -114,34 +83,6 @@ def verify_source(repo, revision, flow="taylor_green", probes=False):
if result["source_tree_sha256"] != SOURCE_SHA256:
raise RuntimeError("Wrong numerical source fingerprint")
result["build_capture_sha256"] = BUILD_CAPTURE_SHA256
if probes:
import shear_probes
module = harness / "shear_probes.py"
if (Path(shear_probes.__file__).resolve() != module
or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_PROBES_SHA256):
raise RuntimeError("Wrong shear_probes module checkout or content")
result["shear_probes_sha256"] = SHEAR_PROBES_SHA256
if flow == "shear_twin":
import shear_twin
module = harness / "shear_twin.py"
if (Path(shear_twin.__file__).resolve() != module
or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_TWIN_SHA256):
raise RuntimeError("Wrong shear_twin module checkout or content")
result["shear_twin_sha256"] = SHEAR_TWIN_SHA256
if flow in ("shear_nested", "shear_twin"): # the twin reuses the nested geometry
import shear_nested
module = harness / "shear_nested.py"
if (Path(shear_nested.__file__).resolve() != module
or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_NESTED_SHA256):
raise RuntimeError("Wrong shear_nested module checkout or content")
result["shear_nested_sha256"] = SHEAR_NESTED_SHA256
if flow in ("shear_wave", "shear_nested", "shear_twin"):
import shear_wave
module = harness / "shear_wave.py"
if (Path(shear_wave.__file__).resolve() != module
or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_WAVE_SHA256):
raise RuntimeError("Wrong shear_wave module checkout or content")
result["shear_wave_sha256"] = SHEAR_WAVE_SHA256
return result
@@ -204,79 +145,26 @@ def main():
p.add_argument("--run-gpu", action="store_true")
p.add_argument("--dry-run", action="store_true")
p.add_argument("--re", type=int, choices=[100, 1000], required=True)
p.add_argument("--n", type=int, choices=[32, 64, 128, 256, 512], required=True)
p.add_argument("--n", type=int, choices=[32, 64, 128, 256], required=True)
p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True)
p.add_argument("--output", type=Path, required=True)
p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant")
p.add_argument("--diagnostic",
choices=[SCALED_DIAGNOSTIC, SCALED_SHEAR, NESTED_SHEAR, TWIN_SHEAR],
help="explicit four-case N256/fine512 same-placement diagnostic "
"(scaled_n256: Taylor-Green; scaled_shear_n256: shear wave) or the "
"fixed three-level/two-region shear preset (nested_shear_n128) "
"or its depth-2 twin (twin_shear_n128)")
p.add_argument("--probe-trace", action="store_true",
help="signed interface-probe trace (nested U128/U512/projected d3, "
"twin projected d2); mechanism diagnostic, no gates")
p.add_argument("--final-snapshot", action="store_true",
help="nested_shear_n128 projected depth-3 only: final macro snapshot")
p.add_argument("--early-pressure", action="store_true",
help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64")
p.add_argument("--flow", choices=["taylor_green", "shear_wave"], default="taylor_green")
args = p.parse_args()
nested = args.diagnostic == NESTED_SHEAR
if args.flow == "shear_wave" and (args.re != 100 or (args.depth == 3 and not nested)
or args.early_pressure):
p.error("shear_wave requires --re 100, depth 1 or 2 and no --early-pressure")
probe_cases = {(NESTED_SHEAR, 128, 1, "constant"), (NESTED_SHEAR, 512, 1, "constant"),
(NESTED_SHEAR, 128, 3, "projected_linear"),
(TWIN_SHEAR, 128, 2, "projected_linear")}
if args.probe_trace and (args.final_snapshot or (
args.diagnostic, args.n, args.depth, args.explosion) not in probe_cases):
p.error("--probe-trace requires nested U128/U512/projected depth 3 or twin projected "
"depth 2, without --final-snapshot")
if args.final_snapshot and not (nested and args.depth == 3
and args.explosion == "projected_linear"):
p.error("--final-snapshot requires nested_shear_n128 projected depth 3")
initial, metrics, limits, initialization = flow_functions(args.flow)
if args.output.exists():
p.error("Output already exists; choose a new path to retain prior evidence")
if args.explosion != "constant" and args.depth != 2 and not nested:
if args.explosion != "constant" and args.depth != 2:
p.error("projected_linear requires depth 2")
twin = args.diagnostic == TWIN_SHEAR
if nested or twin:
import shear_nested
import shear_twin
preset = shear_nested if nested else shear_twin
if args.flow != "shear_wave":
p.error(f"--diagnostic {args.diagnostic} requires --flow shear_wave")
if (args.re, args.n, args.depth, args.explosion) not in preset.CASES:
p.error(f"{args.diagnostic} requires one of its four Re100 cases")
elif args.diagnostic:
cases = {(256, 1, "constant"), (512, 1, "constant"),
(256, 2, "constant"), (256, 2, "projected_linear")}
flow = "shear_wave" if args.diagnostic == SCALED_SHEAR else "taylor_green"
if args.flow != flow:
p.error(f"--diagnostic {args.diagnostic} requires --flow {flow}")
if args.diagnostic == SCALED_DIAGNOSTIC and (
args.re != 100 or (args.n, args.depth, args.explosion) not in cases
or not args.early_pressure):
p.error("scaled_n256 requires one of its four Re100 cases and --early-pressure")
if args.diagnostic == SCALED_SHEAR and (args.n, args.depth, args.explosion) not in cases:
p.error("scaled_shear_n256 requires one of its four Re100 cases")
elif args.n == 512:
p.error("N512 requires --diagnostic scaled_n256, scaled_shear_n256 or nested_shear_n128")
selected = os.environ.get("CUDA_VISIBLE_DEVICES", "")
if not args.dry_run and (not args.run_gpu or not selected or "," in selected or selected == "-1"):
p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required")
repo = args.repo.resolve()
sys.path[:0] = [str(repo / "src"), str(repo)]
source = verify_source(repo, args.expected_revision,
"shear_nested" if nested else "shear_twin" if twin else args.flow,
probes=args.probe_trace)
source = verify_source(repo, args.expected_revision)
runtime = runtime_source(repo)
if args.dry_run:
print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True,
**({"diagnostic": args.diagnostic} if args.diagnostic else {}))))
print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True)))
return 0
from CelerisLab.cuda.context import CudaContext
from CelerisLab.lbm.blocks import BlockLayout2D
@@ -286,12 +174,7 @@ def main():
from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition
cfg = config(args.n, args.re)
if args.depth == 1:
refinements = []
elif nested or twin:
refinements = [list(level) for level in preset.REFINEMENTS]
else:
refinements = rectangles(args.n, args.depth)
refinements = [] if args.depth == 1 else rectangles(args.n, args.depth)
h = build_hierarchy(cfg, refinements)
np.testing.assert_array_equal(leaf_partition(h), np.ones((args.n, args.n)))
masks = [k == ACTIVE for k in h.kinds]
@@ -314,7 +197,7 @@ def main():
mach=U * np.sqrt(3),
viscosity=cfg.viscosity,
tau_by_level=[0.5 + 3 * c.viscosity for c in h.configs],
initialization=initialization,
initialization=INITIALIZATION,
numerics=dict(lattice_model=cfg.lattice_model, collision=cfg.collision,
store_precision=cfg.store_precision, ddf_shifting=cfg.ddf_shifting,
streaming=cfg.streaming, les_enabled=cfg.les_enabled,
@@ -322,20 +205,13 @@ def main():
sample_phase="postcollision G at root synchronization"),
cuda_visible_devices=selected,
device_id=0,
limits=limits,
limits=LIMITS,
workload=workload(masks, round(TIMES[-1] * args.n / U)),
rows=[],
early_pressure=dict(enabled=args.early_pressure, rows=[],
sampled_max_pressure_l2=None),
completed=False,
)
if args.flow != "taylor_green":
import shear_wave
report.update(flow=args.flow, mode=list(shear_wave.MODE),
polarization=shear_wave.E.tolist(), angle_deg=shear_wave.ANGLE_DEG,
phase=shear_wave.PHI, eps=shear_wave.eps(args.n))
if args.diagnostic:
report["diagnostic"] = args.diagnostic
args.output.parent.mkdir(parents=True, exist_ok=True)
def save():
temporary = args.output.with_suffix(args.output.suffix + ".tmp")
@@ -370,37 +246,17 @@ def main():
]
)
initial_mass = None
probe_steps = set()
if args.probe_trace:
import shear_probes
probe_root = dict(zip(shear_probes.case_steps(args.n), shear_probes.STEPS))
probe_steps = set(probe_root)
probes = shear_probes.select(masks, args.n)
report["probe_trace"] = shear_probes.new_trace(
probes, cfg.viscosity, [args.diagnostic, args.n, args.depth, args.explosion])
for step in sorted(set(main_steps + early_steps) | probe_steps):
for step in sorted(set(main_steps + early_steps)):
run.step(step - run.step_count)
if step in probe_steps:
trace = report["probe_trace"]
trace["steps"].append(probe_root[step]) # root-physical step
trace["values"].append(shear_probes.sample(
(run.download(),) if args.depth == 1 else run.download(), probes))
if step not in main_steps and step not in early_steps:
continue # probe-only step: no qualification metrics
row = sample(run, masks, step, args.n, cfg.viscosity, initial_mass,
metrics, args.depth)
arrays = (run.download(),) if args.depth == 1 else run.download()
row = metrics(arrays, masks, step, args.n, cfg.viscosity, initial_mass)
if not all(np.isfinite(value) for value in row.values()):
raise FloatingPointError("nonfinite sample diagnostics")
row["step"] = step
if initial_mass is None:
initial_mass = row["mass"]
if step in main_steps:
report["rows"].append(row)
if args.final_snapshot and step == main_steps[-1]:
# Optional macro-only snapshot; scalar receipt stays authoritative.
report["final_snapshot"] = shear_nested.write_snapshot(
args.output.with_name(args.output.stem + ".final_macro.npz"),
run.download(), masks,
shear_nested.snapshot_meta(step, cfg.viscosity, args.explosion,
source.get("revision")))
if step in early_steps:
trace = report["early_pressure"]
trace["rows"].append({key: row[key] for key in ("step", "tstar", "pressure_l2")})
@@ -422,8 +278,10 @@ def main():
# Exercise real kernels/halos/transfers on every level before READY.
# This is the first measured trajectory step, not discarded warmup.
run.step(1)
sample(run, masks, 1, args.n, cfg.viscosity, initial_mass,
metrics, args.depth, "READY")
probe = (run.download(),) if args.depth == 1 else run.download()
probe_metrics = metrics(probe, masks, 1, args.n, cfg.viscosity, initial_mass)
if not all(np.isfinite(value) for value in probe_metrics.values()):
raise FloatingPointError("nonfinite READY diagnostics")
report["ready_after_step"] = 1
save()
print(
@@ -439,16 +297,11 @@ def main():
),
flush=True,
)
if args.probe_trace:
try:
report["probe_trace"]["bytes"] = shear_probes.validate(report["probe_trace"])
except ValueError as error: # keep the scalar receipt; drop an invalid trace
report["probe_trace"] = dict(error=str(error))
report.update(
completed=True,
seconds=time.monotonic() - started,
screening_pass=all(
row[key] <= limit for row in report["rows"] for key, limit in limits.items()
row[key] <= limit for row in report["rows"] for key, limit in LIMITS.items()
),
)
save()
@@ -1,174 +0,0 @@
"""Fixed three-level/two-region shear preset (opt-in nested_shear_n128).
Four fresh cases: U128, U512 (finest control = 4N), constant R128 depth 3,
projected R128 depth 3. Same wave, phase, Re100/U=0.03, initialization, five
sample times, eps(root N), limits and floors as shear_wave. This changes
placement and refined volume as well as depth: no isolated-level cost or
resolution-trend label. shear_wave.py is reused unchanged (historical receipt
identities stay valid); only the depth/finest-control contract is generalized
here.
"""
import hashlib
import json
from pathlib import Path
import numpy as np
import shear_wave as sw
from oracle import CX, CY, U, centers
ROOT_N = 128
DEPTH = 3
FINEST_FACTOR = 2 ** (DEPTH - 1) # finest-spacing uniform control is 4N
# Parent-frame (x0, y0, x1, y1) rectangles; level-wide, zero-origin unions.
REFINEMENTS = [
[(16, 16, 60, 112), (68, 24, 112, 120)], # L0 -> L1, root coordinates
[(44, 44, 108, 212), (148, 60, 212, 228)], # L1 -> L2, L1 coordinates
]
CASES = ((100, ROOT_N, 1, "constant"), (100, FINEST_FACTOR * ROOT_N, 1, "constant"),
(100, ROOT_N, DEPTH, "constant"), (100, ROOT_N, DEPTH, "projected_linear"))
ROOT_STEPS = 16 * ROOT_N
LEAVES = [7936, 12288, 86016]
HOPS = [ROOT_STEPS * 2**level for level in range(DEPTH)] # [2048, 4096, 8192]
WORK = sum(c * h for c, h in zip(LEAVES, HOPS)) # 771227648 per refined case
NU = [U * ROOT_N / sw.RE * 2**level for level in range(DEPTH)] # [.0384, .0768, .1536]
TAU = [0.5 + 3 * nu for nu in NU] # [.6152, .7304, .9608]
SNAPSHOT_MAX_BYTES = 8 * 1024 * 1024
SNAPSHOT_KEYS = ("level", "i", "j", "rho", "ux", "uy", "meta")
def leaf_masks():
"""Independent numpy leaf masks of the fixed preset (root, L1, L2 level-wide grids)."""
masks, covered = [], None
for level in range(DEPTH):
size = ROOT_N * 2**level
inside = np.ones((size, size), bool) if level == 0 else covered
if level < DEPTH - 1:
child = np.zeros((size, size), bool)
for x0, y0, x1, y1 in REFINEMENTS[level]:
child[y0:y1, x0:x1] = True
masks.append(inside & ~child)
covered = np.kron(child, np.ones((2, 2), bool))
else:
masks.append(inside)
return masks
def assess(base_control, finest_control, refined):
"""shear_wave.assess generalized to depth 3 with a 4N finest control.
Same summaries, screens, budgets and outcome semantics; only the
(depth, finest-control) shape contract differs.
"""
if base_control is None or finest_control is None or not refined:
return dict(status="INVALID_CONTROL", reason="missing control or refined run")
n = base_control["n"]
if (base_control["depth"], finest_control["depth"], finest_control["n"]) != (
1, 1, FINEST_FACTOR * n):
raise ValueError("controls must be uniform N and uniform 4N")
if any(r["depth"] != DEPTH or r["n"] != n for r in refined):
raise ValueError("refined runs must be depth 3 at root N")
if len({r.get("explosion") for r in refined}) != len(refined):
raise ValueError("duplicate refined explosion scheme")
sources = {repr(r.get("source")) for r in [base_control, finest_control, *refined]}
base, fine = sw._usable(base_control), sw._usable(finest_control)
if base is None or fine is None or len(sources) != 1:
return dict(status="INVALID_CONTROL", reason="incomplete/invalid control or mixed source")
controls_pass = base["own_screens_pass"] and fine["own_screens_pass"]
report = dict(base_control=base, finest_control=fine, refined={},
controls_pass=controls_pass)
statuses = []
for r in refined:
try:
s = sw.summarize(r)
except (KeyError, TypeError, ValueError) as error:
report["refined"][r["explosion"]] = dict(valid=False, reason=str(error))
statuses.append("INVALID_RUN")
continue
s["valid"] = True
s["budget"] = {
key: (None if base[key] is None else max(sw.BASELINE_FACTOR * base[key], floor))
for key, floor in sw.FLOORS.items()
}
s["budget_pass"] = all(
s[key] is not None and s["budget"][key] is not None and s[key] <= s["budget"][key]
for key in sw.FLOORS)
s["control_ratios"] = {
name: {key: (s[key] / c[key] if s[key] is not None and c[key] else None)
for key in (*sw.FLOORS, "spurious_pressure_rms")}
for name, c in (("root", base), ("finest", fine))
}
s["controlled_pass"] = controls_pass and s["own_screens_pass"] and s["budget_pass"]
report["refined"][r["explosion"]] = s
statuses.append("PASS" if s["controlled_pass"] else "FAIL")
if "INVALID_RUN" in statuses:
report["status"] = "INVALID_RUN"
elif not controls_pass:
report["status"] = "CONTROL_SCREEN_FAIL"
else:
report["status"] = "PASS" if all(v == "PASS" for v in statuses) else "FAIL"
return report
def write_snapshot(path, arrays, masks, meta):
"""Final macro snapshot: packed ACTIVE leaves (level, i, j), rho, ux, uy only.
No populations, no restart state, no time series. Returns the receipt block;
a file over SNAPSHOT_MAX_BYTES is removed and reported, never kept.
"""
path = Path(path)
if path.exists():
raise FileExistsError(path)
cols = {k: [] for k in ("level", "i", "j", "rho", "ux", "uy")}
counts = []
for level, (g, mask) in enumerate(zip(arrays, masks)):
if g.shape != (9, *mask.shape):
raise ValueError("population/mask shape mismatch")
values = g[:, mask].astype(float)
rho = values.sum(axis=0)
j, i = np.nonzero(mask) # row = y index, column = x index
cols["level"].append(np.full(rho.size, level, np.uint8))
cols["i"].append(i.astype(np.uint16))
cols["j"].append(j.astype(np.uint16))
cols["rho"].append(rho)
cols["ux"].append(CX @ values / rho)
cols["uy"].append(CY @ values / rho)
counts.append(int(rho.size))
packed = {k: np.concatenate(v) for k, v in cols.items()}
if not all(np.isfinite(packed[k]).all() for k in ("rho", "ux", "uy")):
raise FloatingPointError("nonfinite snapshot macros")
meta = dict(meta, leaf_counts=counts, cell_center="h*(index+0.5), h=2**-level",
fields="rho, ux, uy from FP64 physical populations at ACTIVE leaves")
with path.open("xb") as handle:
np.savez_compressed(handle, meta=np.array(json.dumps(meta, sort_keys=True)), **packed)
size = path.stat().st_size
block = dict(file=path.name, bytes=size, leaf_counts=counts, max_bytes=SNAPSHOT_MAX_BYTES,
step=meta.get("step"), tstar=meta.get("tstar"))
if size > SNAPSHOT_MAX_BYTES:
path.unlink()
return dict(block, kept=False, sha256=None, reason="exceeds snapshot byte cap")
block.update(kept=True, sha256=hashlib.sha256(path.read_bytes()).hexdigest())
return block
def read_snapshot(path):
"""Validate a snapshot: exact keys, no populations, consistent packed arrays."""
with np.load(path, allow_pickle=False) as data:
if tuple(sorted(data.files)) != tuple(sorted(SNAPSHOT_KEYS)):
raise ValueError("unexpected snapshot contents")
out = {k: data[k] for k in data.files}
meta = json.loads(str(out.pop("meta")))
n = len(out["rho"])
if any(len(v) != n for v in out.values()) or sum(meta["leaf_counts"]) != n:
raise ValueError("inconsistent snapshot arrays")
return meta, out
def snapshot_meta(step, viscosity, explosion, source_revision):
"""Exact time/phase/config identity stored inside the snapshot."""
return dict(step=step, tstar=step * U / ROOT_N, phase=sw.PHI, mode=list(sw.MODE),
polarization=sw.E.tolist(), amplitude=U, re_L=sw.RE, n=ROOT_N, depth=DEPTH,
explosion=explosion, viscosity=viscosity, tau_by_level=TAU,
refinements=REFINEMENTS, source_revision=source_revision,
origin="level-wide zero-origin grids", time="root synchronization, postcollision")
@@ -1,123 +0,0 @@
"""Signed interface-probe traces for the nested/twin shear presets (mechanism diagnostic).
No pass gates. Native-validated 24-target layout through box 1, in root-128
physical coordinates, crossing L0->L1 and (nested only) L1->L2 away from corners:
horizontal line y = 56 across the vertical faces x = 16 and x = 22;
vertical line x = 38 across the horizontal faces y = 16 and y = 22;
normal coordinates NORMAL_COORDS on both (nested: 3 L0 + 5 L1 + 4 L2 each).
Each target maps to the nearest ACTIVE leaf center (tie rule: distance, then
finer level, then smaller j, then smaller i). Grids differ between cases
(U128, U512, nested, twin), so actual centers, levels and displacements are
stored and analytic fields are evaluated at the actual centers.
Samples are POST-collision G at root synchronization only (root-step units);
no substep content is observed, and grid-scale alternation can alias.
"""
import math
import numpy as np
import shear_wave as sw
from oracle import CX, CY, U
ROOT_N = 128
NORMAL_COORDS = (12.5, 14.5, 15.5, 16.25, 17.25, 19.25, 20.75, 21.75,
22.125, 23.125, 24.125, 26.125)
TRANSECTS = {"horizontal_y56": dict(axis=0, other=56.0), # x varies, y fixed
"vertical_x38": dict(axis=1, other=38.0)} # y varies, x fixed
STEPS = tuple(range(0, 129)) + tuple(range(1920, 2049)) # root steps, both inclusive
TRACE_MAX_BYTES = 600_000
def points():
out = []
for name, t in TRANSECTS.items():
for along in NORMAL_COORDS:
xy = (along, t["other"]) if t["axis"] == 0 else (t["other"], along)
out.append((name, along, xy))
return out
def case_steps(n):
"""Case lattice steps at the fixed root-physical sample times (U512: 4 per root step)."""
return [s * (n // ROOT_N) for s in STEPS]
def select(masks, n=ROOT_N):
"""Nearest ACTIVE leaf center for each target (deterministic tie rule).
Targets and stored centers are in root-physical units (root cell = 1); a case
with root N = n has cell size h = ROOT_N / (n * 2**level) there (U512: 0.25).
"""
if n % ROOT_N:
raise ValueError("case root N must be a multiple of 128")
probes = []
for name, along, (x, y) in points():
best = None
for level, mask in enumerate(masks):
h = ROOT_N / (n * 2 ** level)
i0, j0 = int(math.floor(x / h)), int(math.floor(y / h))
for j in range(j0 - 2, j0 + 3):
for i in range(i0 - 2, i0 + 3):
if 0 <= j < mask.shape[0] and 0 <= i < mask.shape[1] and mask[j, i]:
cx, cy = h * (i + 0.5), h * (j + 0.5)
key = (math.hypot(cx - x, cy - y), -level, j, i)
if best is None or key < best[0]:
best = (key, level, i, j, cx, cy)
if best is None:
raise ValueError(f"no ACTIVE leaf near target {(x, y)}")
_, level, i, j, cx, cy = best
probes.append(dict(transect=name, normal=along, point=[x, y], level=level, i=i, j=j,
center=[cx, cy], displacement=[cx - x, cy - y]))
if len({(p["level"], p["i"], p["j"]) for p in probes}) != len(probes):
raise ValueError("probe targets map to duplicate cells")
return probes
def sample(arrays, probes):
"""Flat [rho, ux, uy] per probe from FP64 physical populations."""
row = []
for p in probes:
f = np.asarray(arrays[p["level"]][:, p["j"], p["i"]], dtype=float)
rho = float(f.sum())
row += [rho, float(CX @ f) / rho, float(CY @ f) / rho]
return row
def new_trace(probes, case_viscosity, case):
"""Times/coordinates are root-physical (N=128); nu_root = U*128/Re for all cases."""
return dict(schema=1, kind="mechanism diagnostic, no pass gates", case=case,
time_unit="root step (N=128)", sample_phase="postcollision G at root synchronization",
root_n=ROOT_N, viscosity_root=U * ROOT_N / sw.RE, case_viscosity=case_viscosity,
mode=list(sw.MODE), phase=sw.PHI,
amplitude=U, fields=["rho", "ux", "uy"], probes=probes, steps=[], values=[])
def residuals(trace):
"""Signed (drho, dux, duy) per step and probe, analytic field at each actual center."""
out = []
for step, row in zip(trace["steps"], trace["values"]):
res = []
for n, p in enumerate(trace["probes"]):
x, y = p["center"]
ux, uy, _ = sw.exact(x, y, step, trace["root_n"], trace["viscosity_root"])
rho, vx, vy = row[3 * n: 3 * n + 3]
res.append((rho - 1.0, vx - float(ux), vy - float(uy)))
out.append(res)
return out
def validate(trace):
"""Exact schedule, finite values, fixed probe count and byte cap."""
import json
if trace.get("steps") != list(STEPS) or len(trace["values"]) != len(STEPS):
raise ValueError("trace steps differ from the fixed schedule")
if len(trace["probes"]) != len(NORMAL_COORDS) * len(TRANSECTS):
raise ValueError("unexpected probe count")
width = 3 * len(trace["probes"])
if any(len(r) != width or not all(math.isfinite(v) for v in r) for r in trace["values"]):
raise ValueError("malformed or nonfinite trace row")
size = len(json.dumps(trace, allow_nan=False).encode())
if size > TRACE_MAX_BYTES:
raise ValueError(f"trace exceeds {TRACE_MAX_BYTES} bytes ({size})")
return size
@@ -1,36 +0,0 @@
"""Depth-2 twin of the nested shear preset (opt-in twin_shear_n128).
Same two L0->L1 boxes as shear_nested, no L2. Four fresh cases: U128, U256
(finest control = 2N), constant R128 depth 2, projected R128 depth 2; same
wave, phase, Re100/U=0.03, initialization, times, eps(root N), limits, floors.
Twin vs nested measures the effect of adding this nested L2 refinement.
Removing L2 also changes the finest spacing and refined volume, so the
contrast does not isolate abstract depth. Physics is the unchanged depth-2,
2N-control shear_wave.assess; shear_nested.py stays byte-identical.
"""
import numpy as np
import shear_nested as nested
ROOT_N = nested.ROOT_N
DEPTH = 2
FINEST_FACTOR = 2
REFINEMENTS = [nested.REFINEMENTS[0]] # L0 -> L1 only, root coordinates
CASES = ((100, ROOT_N, 1, "constant"), (100, FINEST_FACTOR * ROOT_N, 1, "constant"),
(100, ROOT_N, DEPTH, "constant"), (100, ROOT_N, DEPTH, "projected_linear"))
ROOT_STEPS = 16 * ROOT_N
LEAVES = [7936, 33792]
HOPS = [ROOT_STEPS * 2**level for level in range(DEPTH)] # [2048, 4096]
WORK = sum(c * h for c, h in zip(LEAVES, HOPS)) # 154664960 per refined case
NU = nested.NU[:DEPTH]
TAU = nested.TAU[:DEPTH]
def leaf_masks():
"""Independent numpy leaf masks (root complement, full L1 boxes)."""
child = np.zeros((ROOT_N, ROOT_N), bool)
for x0, y0, x1, y1 in REFINEMENTS[0]:
child[y0:y1, x0:x1] = True
return [~child, np.kron(child, np.ones((2, 2), bool))]
-312
View File
@@ -1,312 +0,0 @@
"""CPU-only oblique shear-wave reference and composite-leaf diagnostics.
Convention: integer mode MODE=(2,1) on the L x L periodic root domain, so the
wavevector k = (2*pi/L)*(2,1) points 26.565 deg from +x (mode (1,2) would be
63.435 deg). Velocity is polarized along E = (-1,2)/sqrt(5), perpendicular to k.
The frozen Taylor-Green oracle is reused for constants and geometry only.
"""
import math
import numpy as np
from oracle import CX, CY, W, U, TIMES, centers
MODE = (2, 1)
PHI = 0.0
E = np.array([-MODE[1], MODE[0]], dtype=float) / math.hypot(*MODE)
ANGLE_DEG = math.degrees(math.atan2(MODE[1], MODE[0]))
RE = 100 # Re = U L / nu, the same nu as Taylor-Green Re100 at equal N
FIT_FROM_TSTAR = 0.12 # late-time samples only; reduces startup sensitivity
INITIALIZATION = (
"physical FP64 equilibrium G at actual cell centers; rho=1 (exact p const)"
)
# Per-row screens. phase_drift_abs, residual_l2 and the run-level rate screen
# use eps(n) and are preregistered diagnostic choices, not error bounds.
LIMITS = dict(velocity_l2=0.05, spurious_pressure_scaled=0.02, mass_relative=1e-5)
# The comparative pressure budget max(2*control, 0.02) is never below the 0.02
# absolute ceiling, so it is always redundant; raw control ratios are kept.
FLOORS = dict(rate_error=0.002, residual_l2=0.002, spurious_pressure_scaled=0.02)
BASELINE_FACTOR = 2.0
def wavevector(length):
return 2 * np.pi / length * np.array(MODE, dtype=float)
def decay_rate(length, viscosity):
"""Exact amplitude decay rate per root step: nu |k|^2."""
k = wavevector(length)
return viscosity * float(k @ k)
def eps(n):
"""(|k| h_root)^2 with h_root = 1: dimensionless, coefficient one."""
k = wavevector(n)
return float(k @ k)
def exact(x, y, t, length, viscosity, amplitude=U):
k = wavevector(length)
a = amplitude * np.exp(-decay_rate(length, viscosity) * t)
s = np.sin(k[0] * x + k[1] * y + PHI)
p = np.zeros_like(s)
return a * E[0] * s, a * E[1] * s, p # kinematic pressure, constant
def equilibrium(rho, ux, uy):
# Same second-order D2Q9 equilibrium as oracle.initial.
cu = CX[:, None, None] * ux + CY[:, None, None] * uy
return (
W[:, None, None]
* rho
* (1 + 3 * cu + 4.5 * cu * cu - 1.5 * (ux * ux + uy * uy))
)
def initial(shape, level, length, viscosity):
x, y = centers(shape, level)
ux, uy, p = exact(x, y, 0.0, length, viscosity)
return equilibrium(1 + 3 * p, ux, uy)
def _leaves(arrays, masks):
accum = []
for level, (g, mask) in enumerate(zip(arrays, masks)):
if g.shape != (9, *mask.shape):
raise ValueError("population/mask shape mismatch")
values = g[:, mask].astype(float)
if not np.isfinite(values).all():
raise FloatingPointError("nonfinite leaf populations")
rho = values.sum(axis=0)
if (rho <= 0).any():
raise FloatingPointError("nonpositive leaf density")
x, y = centers(mask.shape, level)
area = np.full(rho.size, 0.25**level)
lev = np.full(rho.size, level)
accum.append((rho, CX @ values / rho, CY @ values / rho, x[mask], y[mask], area, lev))
return [np.concatenate(v) for v in zip(*accum)]
def metrics(arrays, masks, t, length, viscosity, initial_mass=None):
rho, ux, uy, x, y, w, lev = _leaves(arrays, masks)
volume = w.sum()
if not np.isclose(volume, length**2, rtol=0, atol=1e-10):
raise ValueError("leaf partition does not cover domain once")
k = wavevector(length)
theta = k[0] * x + k[1] * y + PHI
s, c = np.sin(theta), np.cos(theta)
v = ux * E[0] + uy * E[1]
# Weighted least squares on the actual composite quadrature (no
# continuous-orthogonality assumption).
gram = np.array([[w @ (s * s), w @ (s * c)], [w @ (s * c), w @ (c * c)]])
a_hat, b_hat = np.linalg.solve(gram, [w @ (v * s), w @ (v * c)])
ex, ey, _ = exact(x, y, t, length, viscosity)
exact_norm = w @ (ex * ex + ey * ey)
fit = a_hat * s + b_hat * c
longitudinal = (ux * k[0] + uy * k[1]) / math.sqrt(k @ k)
amplitude = U * math.exp(-decay_rate(length, viscosity) * t)
scale = amplitude * amplitude / 4 # continuum reference scale, not the TG norm
modal = math.hypot(a_hat, b_hat)
p = (rho - 1) / 3
p = p - w @ p / volume # one global gauge
mass = float(w @ rho)
return dict(
tstar=t * U / length,
mass=mass,
mass_relative=(
0.0 if initial_mass is None else abs(mass - initial_mass) / initial_mass
),
a_hat=float(a_hat),
b_hat=float(b_hat),
amplitude_exact=amplitude,
amplitude_ratio=float(a_hat / amplitude),
modal_amplitude=modal,
amplitude_error=modal / amplitude - 1,
phase_drift=float(math.atan2(b_hat, a_hat)),
phase_drift_abs=abs(float(math.atan2(b_hat, a_hat))),
velocity_l2=float(
np.sqrt(w @ ((ux - ex) ** 2 + (uy - ey) ** 2) / exact_norm)
),
residual_l2=float(
np.sqrt(w @ ((ux - fit * E[0]) ** 2 + (uy - fit * E[1]) ** 2) / exact_norm)
),
longitudinal_l2=float(np.sqrt(w @ (longitudinal**2) / exact_norm)),
spurious_pressure_rms=float(np.sqrt(w @ (p * p) / volume)),
spurious_pressure_scaled=float(np.sqrt(w @ (p * p) / volume) / scale),
spurious_pressure_max_scaled=float(np.max(np.abs(p)) / scale),
spurious_pressure_scaled_by_level=[
float(np.sqrt(w[lev == i] @ (p[lev == i] ** 2) / w[lev == i].sum()) / scale)
for i in range(len(masks))
],
rho_min=float(rho.min()),
rho_max=float(rho.max()),
max_speed=float(np.max(np.hypot(ux, uy))),
leaf_cells=int(len(rho)),
)
def fit_rate(rows, length, viscosity):
"""Free-intercept least squares of ln R, R = hypot(a_hat, b_hat), late samples.
The intercept absorbs only a constant amplitude prefactor once a single
exponential dominates; per-interval rates and the fit residual are kept so
a lingering transient stays visible. Rates are per root step of the record
(uniform 2N uses its own nu and doubled step counts).
"""
late = [r for r in rows if r["tstar"] >= FIT_FROM_TSTAR - 1e-12]
amp = np.array([math.hypot(r["a_hat"], r["b_hat"]) for r in late])
if len(late) < 3 or not np.isfinite(amp).all():
raise ValueError("need >= 3 late samples with finite R")
exact_rate = decay_rate(length, viscosity)
to_tstar = length / U # root steps per unit t*
if not (amp > 0).all():
# Finite modal extinction: a physical failure, not malformed data. No
# clipping and no invented rate; fit quantities are unavailable.
return dict(rate_fit_valid=False, rate=None, exact_rate=exact_rate,
rate_tstar=None, exact_rate_tstar=exact_rate * to_tstar,
rate_error=None, interval_rate_errors=None, log_fit_residual_max=None)
t = np.array([r["step"] for r in late], dtype=float)
(slope, intercept) = np.polyfit(t, np.log(amp), 1)
residual = np.log(amp) - (slope * t + intercept)
return dict(
rate_fit_valid=True, rate=float(-slope), exact_rate=exact_rate,
rate_tstar=float(-slope * to_tstar),
exact_rate_tstar=exact_rate * to_tstar, # = 5 (2 pi)^2 / Re
rate_error=abs(-slope / exact_rate - 1),
interval_rate_errors=[
float(-np.log(amp[i + 1] / amp[i]) / (t[i + 1] - t[i]) / exact_rate - 1)
for i in range(len(t) - 1)
],
log_fit_residual_max=float(np.max(np.abs(residual))),
)
NONNEGATIVE = ("tstar", "mass", "mass_relative", "amplitude_exact", "modal_amplitude",
"phase_drift_abs", "velocity_l2", "residual_l2", "longitudinal_l2",
"spurious_pressure_rms", "spurious_pressure_scaled",
"spurious_pressure_max_scaled", "rho_min", "rho_max", "max_speed")
SIGNED = ("a_hat", "b_hat", "amplitude_ratio", "amplitude_error", "phase_drift")
def _number(value):
return (isinstance(value, (int, float)) and not isinstance(value, bool)
and math.isfinite(value))
def validate_rows(record):
"""Exact schedule, finite numbers (incl. lists) and nonnegative error fields.
Signed quantities (amplitudes, amplitude/phase errors) keep their sign.
Raises ValueError so callers can report an invalid run before any screen.
"""
n, rows = record["n"], record["rows"]
steps = [round(tstar * n / U) for tstar in TIMES]
if not isinstance(rows, list) or len(rows) != len(steps):
raise ValueError("rows do not match the scheduled samples")
for row, step in zip(rows, steps):
if type(row.get("step")) is not int or row["step"] != step:
raise ValueError(f"unscheduled or duplicate step {row.get('step')!r}")
if not _number(row.get("tstar")) or abs(row["tstar"] - step * U / n) > 3e-17:
raise ValueError(f"wrong sample time at step {step}")
if not all(_number(row.get(key)) for key in (*NONNEGATIVE, *SIGNED)):
raise ValueError(f"missing or nonfinite diagnostic at step {step}")
if any(row[key] < 0 for key in NONNEGATIVE) or row["rho_min"] <= 0:
raise ValueError(f"negative error quantity or density at step {step}")
levels = row.get("spurious_pressure_scaled_by_level")
if (not isinstance(levels, list) or len(levels) != record["depth"]
or not all(_number(v) and v >= 0 for v in levels)):
raise ValueError(f"invalid per-level pressure at step {step}")
def summarize(record):
validate_rows(record)
rows = record["rows"]
n = record["n"]
out = fit_rate(rows, n, record["viscosity"])
out.update(
eps=eps(n),
complete=record["completed"] is True,
residual_l2=max(r["residual_l2"] for r in rows),
phase_drift_abs=max(r["phase_drift_abs"] for r in rows),
spurious_pressure_scaled=max(r["spurious_pressure_scaled"] for r in rows),
spurious_pressure_rms=max(r["spurious_pressure_rms"] for r in rows),
amplitude_error_abs=max(abs(r["amplitude_error"]) for r in rows),
velocity_l2=max(r["velocity_l2"] for r in rows),
mass_relative=max(r["mass_relative"] for r in rows),
)
if out["rate_fit_valid"] and not all(
_number(v) for v in (out["rate"], out["rate_error"], *out["interval_rate_errors"],
out["log_fit_residual_max"])):
raise ValueError("nonfinite rate fit")
out["own_screens_pass"] = out["complete"] and out["rate_fit_valid"] and all(
out[key] <= out["eps"] for key in ("rate_error", "phase_drift_abs", "residual_l2")
) and all(out[key] <= limit for key, limit in LIMITS.items())
return out
def _usable(record):
try:
out = summarize(record)
except (KeyError, TypeError, ValueError):
return None
return out if out["complete"] else None
def assess(base_control, fine_control, refined):
"""Uniform root-spacing and finest-spacing controls versus refined runs.
INVALID_CONTROL: missing, incomplete, invalid-row or mixed-source controls.
INVALID_RUN: a refined record has invalid rows (schedule, NaN, sign).
CONTROL_SCREEN_FAIL: complete valid controls that fail their own screens;
refined own-screen results are kept, but no controlled comparison passes.
PASS/FAIL: every refined run's controlled_pass (own screens and budgets).
"""
if base_control is None or fine_control is None or not refined:
return dict(status="INVALID_CONTROL", reason="missing control or refined run")
n = base_control["n"]
if (base_control["depth"], fine_control["depth"], fine_control["n"]) != (1, 1, 2 * n):
raise ValueError("controls must be uniform N and uniform 2N")
if any(r["depth"] != 2 or r["n"] != n for r in refined):
raise ValueError("refined runs must be depth 2 at root N")
if len({r.get("explosion") for r in refined}) != len(refined):
raise ValueError("duplicate refined explosion scheme")
sources = {repr(r.get("source")) for r in [base_control, fine_control, *refined]}
base, fine = _usable(base_control), _usable(fine_control)
if base is None or fine is None or len(sources) != 1:
return dict(status="INVALID_CONTROL",
reason="incomplete/invalid control or mixed source")
controls_pass = base["own_screens_pass"] and fine["own_screens_pass"]
report = dict(base_control=base, fine_control=fine, refined={},
controls_pass=controls_pass)
statuses = []
for r in refined:
try:
s = summarize(r)
except (KeyError, TypeError, ValueError) as error:
report["refined"][r["explosion"]] = dict(valid=False, reason=str(error))
statuses.append("INVALID_RUN")
continue
s["valid"] = True
# rate_error is None when a modal fit is unavailable; never a pass.
s["budget"] = {
key: (None if base[key] is None else max(BASELINE_FACTOR * base[key], floor))
for key, floor in FLOORS.items()
}
s["budget_pass"] = all(
s[key] is not None and s["budget"][key] is not None and s[key] <= s["budget"][key]
for key in FLOORS)
s["control_ratios"] = {
name: {key: (s[key] / c[key] if s[key] is not None and c[key] else None)
for key in (*FLOORS, "spurious_pressure_rms")}
for name, c in (("root", base), ("finest", fine))
}
s["controlled_pass"] = controls_pass and s["own_screens_pass"] and s["budget_pass"]
report["refined"][r["explosion"]] = s
statuses.append("PASS" if s["controlled_pass"] else "FAIL")
if "INVALID_RUN" in statuses:
report["status"] = "INVALID_RUN"
elif not controls_pass:
report["status"] = "CONTROL_SCREEN_FAIL"
else:
report["status"] = "PASS" if all(v == "PASS" for v in statuses) else "FAIL"
return report
@@ -17,7 +17,6 @@ from unittest.mock import patch
import numpy as np
import run_case
import assess_projected_pairs as assessor
from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION,
SOURCE_SHA256, assess)
from oracle import LIMITS, TIMES, U, rectangles
@@ -32,7 +31,7 @@ RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0
def source():
return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256,
harness_sha256={"run_case.py": "a" * 64, "oracle.py": "c" * 64},
harness_sha256={"run_case.py": "a" * 64},
build_capture_sha256=run_case.BUILD_CAPTURE_SHA256)
@@ -43,14 +42,12 @@ def build_metadata(size):
arch="sm_70", flags=["--use_fast_math"], ptx_sha256="b" * 64)
def records(early=False, diagnostic=None):
def records(early=False):
from CelerisLab.lbm.hierarchy import build_hierarchy
from CelerisLab.lbm.refinement import ACTIVE
out = []
for re, n, depth, scheme in CASES:
if diagnostic:
n *= 2
cfg = run_case.config(n, re)
refinements = [] if depth == 1 else rectangles(n, depth)
h = build_hierarchy(cfg, refinements)
@@ -69,8 +66,6 @@ def records(early=False, diagnostic=None):
early_pressure=dict(enabled=early, rows=[
dict(step=s, tstar=s*U/n, pressure_l2=.01 if s not in main_steps else 0.0) for s in early_steps],
sampled_max_pressure_l2=.01 if early else None)))
if diagnostic:
out[-1]["diagnostic"] = diagnostic
return out
@@ -256,7 +251,7 @@ class ProjectedPairsTests(unittest.TestCase):
def test_pinned_build_sources_match_host_generated_inputs(self):
from CelerisLab.cuda import compiler_v2
for size in (128, 256, 512):
for size in (128, 256):
cfg = replace(run_case.config(size, 100), nx=10, ny=10)
cfg.validate()
root = compiler_v2.generate_config(cfg)
@@ -268,111 +263,6 @@ class ProjectedPairsTests(unittest.TestCase):
finally:
compiler_v2.cleanup_build_dir(root)
def test_scaled_geometry_work_and_frozen_gates(self):
data = records(True, run_case.SCALED_DIAGNOSTIC)
self.assertEqual(sum(r["workload"]["active_collision_updates"] for r in data), 4714397696)
self.assertEqual([r["viscosity"] for r in data], [.0768, .1536, .0768, .0768])
self.assertEqual(data[-1]["tau_by_level"], [.7303999999999999, .9608])
for r in data:
self.assertEqual([row["tstar"] for row in r["rows"]], TIMES)
self.assertEqual(len(r["early_pressure"]["rows"]), 65)
if r["depth"] == 2:
self.assertEqual(r["refinements"], [[(32, 32, 224, 192)]])
self.assertEqual(r["workload"]["active_cells_by_level"], [34816, 122880])
self.assertEqual(r["workload"]["level_hops"], [4096, 8192])
result = assess(data, MOCK_REVISION, diagnostic=run_case.SCALED_DIAGNOSTIC)
self.assertTrue(result["passed"])
self.assertEqual(result["floors"], dict(velocity_l2=.002, pressure_l2=.02, energy_relative=.004))
self.assertTrue(all(row["floor_dominates"] for row in result["comparisons"]))
data[-1]["rows"][-1]["velocity_l2"] = .00200001
self.assertFalse(assess(data, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
self.assertFalse(assess(data)["passed"])
data[-1]["rows"][-1]["velocity_l2"] = 0.0
for mutate in (lambda r: r.pop("diagnostic"), lambda r: r.update(diagnostic="wrong"),
lambda r: r["source"].update(revision="wrong"),
lambda r: r["early_pressure"].update(enabled=False),
lambda r: r.update(refinements=[[(32, 32, 224, 200)]])):
bad = deepcopy(data)
mutate(bad[-1])
self.assertFalse(assess(bad, MOCK_REVISION, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
def test_raw_resolution_changes_and_reference_identity(self):
old, new = records(True), records(True, run_case.SCALED_DIAGNOSTIC)
for r in old:
r["source"]["revision"] = assessor.REFERENCE_REVISION
old[-2]["rows"][-1]["velocity_l2"] = .004 # historical constant failure remains usable
old[-1]["rows"][-1]["velocity_l2"] = .001
new[-1]["rows"][-1]["velocity_l2"] = .0015 # floor pass, but worse raw error
result = assessor.resolution_trend(new, old, MOCK_REVISION)
self.assertTrue(result["valid"])
self.assertTrue(result["trend_interpretation_ready"])
self.assertFalse(result["reference_scheme_results"]["constant"]["passed"])
self.assertEqual((len(result["per_time"]), len(result["worst_post_initial"]),
len(result["uniform256_bridge"])), (80, 16, 20))
row = next(r for r in result["per_time"] if r["role"] == "projected_linear"
and r["metric"] == "velocity_l2" and r["tstar"] == .48)
self.assertEqual((row["signed_change"], row["new_to_old"], row["change_percent"]), (.0005, 1.5, 50.))
self.assertTrue(assess(new, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
self.assertIsNone(result["per_time"][0]["new_to_old"]) # zero denominator is explicit
self.assertEqual(result["per_time"][0]["ratio_unavailable_reason"], "zero historical error")
new[-1]["rows"][-1]["velocity_l2"] = .06 # retain raw errors even when absolute screen fails
self.assertTrue(assessor.resolution_trend(new, old, MOCK_REVISION)["valid"])
self.assertFalse(assess(new, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
new[-1]["rows"][-1]["velocity_l2"] = .0015
for sample in ("rows", "early_pressure"):
bad = deepcopy(new)
if sample == "rows":
bad[0]["rows"][-1]["velocity_l2"] = .0001
else:
bad[0]["early_pressure"]["rows"][1]["pressure_l2"] = .009
result = assessor.resolution_trend(bad, old, MOCK_REVISION)
self.assertTrue(result["valid"])
self.assertFalse(result["trend_interpretation_ready"])
for mutate in (lambda r: r["source"].update(revision=MOCK_REVISION),
lambda r: r["source"].update(source_tree_sha256="wrong"),
lambda r: r["source"]["harness_sha256"].update(**{"oracle.py": "wrong"}),
lambda r: r.update(diagnostic=run_case.SCALED_DIAGNOSTIC),
lambda r: r.update(refinements=[]), lambda r: r["rows"].pop()):
bad = deepcopy(old)
mutate(bad[-1])
self.assertFalse(assessor.resolution_trend(new, bad, MOCK_REVISION)["valid"])
with tempfile.TemporaryDirectory() as tmp:
paths = [Path(tmp) / f"{i}.json" for i in range(4)]
for path, receipt in zip(paths, old):
path.write_text(json.dumps(receipt))
with self.assertRaisesRegex(ValueError, "receipt hashes"):
assessor.load_reference(paths)
witnesses = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in paths}
with patch.object(assessor, "REFERENCE_RECEIPTS", witnesses):
self.assertEqual(assessor.load_reference(paths), json.loads(json.dumps(old)))
with self.assertRaises(ValueError):
assessor.load_reference(paths[:-1] + paths[:1])
paths[-1].write_text(paths[-1].read_text() + " ")
with self.assertRaises(ValueError):
assessor.load_reference(paths)
def test_scaled_runner_requires_explicit_bounded_diagnostic(self):
with tempfile.TemporaryDirectory() as tmp:
base = ["run_case.py", "--repo", str(REPO), "--expected-revision", MOCK_REVISION,
"--output", str(Path(tmp) / "case.json"), "--dry-run"]
diagnostic = ["--diagnostic", run_case.SCALED_DIAGNOSTIC, "--early-pressure"]
for n, depth, re, extra, valid in (
(512, 1, 100, [], False), (512, 1, 100, diagnostic, True),
(256, 2, 100, diagnostic, True), (128, 2, 100, diagnostic, False),
(256, 3, 100, diagnostic, False), (256, 2, 1000, diagnostic, False),
(256, 2, 100, diagnostic[:2], False), (512, 2, 100, diagnostic, False)):
argv = base + ["--n", str(n), "--depth", str(depth), "--re", str(re)] + extra
with self.subTest(argv=argv), patch.object(sys, "argv", argv), \
patch.object(run_case, "verify_source", return_value=source()) as verify, \
redirect_stdout(io.StringIO()), patch("sys.stderr", io.StringIO()):
if valid:
self.assertEqual(run_case.main(), 0)
else:
with self.assertRaises(SystemExit) as caught:
run_case.main()
self.assertEqual(caught.exception.code, 2)
verify.assert_not_called()
def test_source_preflight_requires_clean_exact_revision_and_tree(self):
import oracle
from tests.validation.multigrid import check_projected_pairs as gate
@@ -429,7 +319,7 @@ class ProjectedPairsTests(unittest.TestCase):
patch.object(run_case, "runtime_identity", side_effect=AssertionError("CUDA reached")), \
redirect_stdout(stdout):
self.assertEqual(run_case.main(), 0)
verify.assert_called_once_with(REPO, MOCK_REVISION, "taylor_green")
verify.assert_called_once_with(REPO, MOCK_REVISION)
self.assertTrue(json.loads(stdout.getvalue())["dry_run"])
self.assertFalse(output.exists())
@@ -459,15 +349,9 @@ class ProjectedPairsTests(unittest.TestCase):
def test_mocked_runner_default_early_and_failure(self):
# Replace the GPU context and steppers before import. No CUDA module or flow is used.
cases = [(early, depth, fail, 128, None, "projected_linear" if early else "constant")
for early, depth, fail in
((False, 1, None), (False, 2, None), (True, 2, None),
for early, depth, fail in ((False, 1, None), (False, 2, None), (True, 2, None),
(True, 2, "resource"), (True, 2, "nonfinite"),
(True, 2, "probe_nonfinite"))]
cases += [(True, depth, None, n, run_case.SCALED_DIAGNOSTIC, scheme)
for n, depth, scheme in ((256, 1, "constant"), (512, 1, "constant"),
(256, 2, "constant"), (256, 2, "projected_linear"))]
for early, depth, fail, n, diagnostic, scheme in cases:
(True, 2, "probe_nonfinite")):
calls, selectors = [], []
class FakeRun:
@@ -511,14 +395,11 @@ class ProjectedPairsTests(unittest.TestCase):
with tempfile.TemporaryDirectory() as tmp:
output = Path(tmp) / "case.json"
argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", str(n),
argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
"--depth", str(depth), "--output", str(output),
"--expected-revision", MOCK_REVISION, "--run-gpu"]
argv += ["--explosion", scheme]
if early:
argv += ["--early-pressure"]
if diagnostic:
argv += ["--diagnostic", diagnostic]
argv += ["--early-pressure", "--explosion", "projected_linear"]
with patch.dict(sys.modules, modules), patch.object(sys, "argv", argv), \
patch.dict("os.environ", CUDA_VISIBLE_DEVICES="mock"), \
patch.object(run_case, "verify_source", return_value=source()) as verify, \
@@ -531,17 +412,16 @@ class ProjectedPairsTests(unittest.TestCase):
else:
self.assertEqual(run_case.main(), 0)
receipt = json.loads(output.read_text())
verify.assert_called_once_with(REPO, MOCK_REVISION, "taylor_green")
self.assertEqual(selectors, [scheme] if depth == 2 else [])
verify.assert_called_once_with(REPO, MOCK_REVISION)
self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else [])
self.assertEqual(receipt["source"], source())
self.assertEqual(receipt["runtime_identity"], RUNTIME)
self.assertEqual(receipt["completed"], not fail)
self.assertEqual(receipt.get("diagnostic"), diagnostic)
self.assertEqual(calls[:2], [0, 1]) # READY probe is part of this trajectory
if not fail:
self.assertEqual([r["tstar"] for r in receipt["rows"]], TIMES)
self.assertEqual(len(receipt["early_pressure"]["rows"]), 65 if early else 0)
self.assertEqual(calls[-1], 16*n)
self.assertEqual(calls[-1], 2048)
else:
self.assertEqual(len(receipt["rows"]), 1)
self.assertEqual(len(receipt["early_pressure"]["rows"]),
File diff suppressed because it is too large Load Diff
+3 -2
View File
@@ -173,8 +173,9 @@ def time_refined(a, h, sched):
cfg, pair, body, host = r.host_geometry(a.source, h, "MRT")
context = stack.enter_context(CudaContext(0))
backend = stack.enter_context(FixedPatchStepper(
cfg, r.BOUNDS, context=context, coarse_boundary="channel", base_block=r.BASE_BLOCK,
fine_block=r.FINE_BLOCK, fine_objects=(body,), observe_raw_exchange=True))
cfg, context=context, coarse_boundary="channel", refinements=[r.BOUNDS],
level_blocks=[r.BASE_BLOCK, r.FINE_BLOCK], level_objects=[(), (body,)],
observe_raw_exchange=True))
masks = [channel_masks(pair.coarse_shape)[0] & ~pair.covered, # as run_refined_dfg
(pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)]
fresh = (r.initial_state(pair.coarse_shape, (0.0, 0.0), 1.0, root=True),
+2 -2
View File
@@ -130,8 +130,8 @@ def main():
cfg, pair, body, host = r.host_geometry(a.source, h, "MRT")
context = stack.enter_context(CudaContext(0))
backend = stack.enter_context(FixedPatchStepper(
cfg, r.BOUNDS, context=context, coarse_boundary="channel", base_block=r.BASE_BLOCK,
fine_block=r.FINE_BLOCK, fine_objects=(body,), observe_raw_exchange=True,
cfg, context=context, coarse_boundary="channel", refinements=[r.BOUNDS],
level_blocks=[r.BASE_BLOCK, r.FINE_BLOCK], level_objects=[(), (body,)], observe_raw_exchange=True,
observe_window_exchange=True))
masks = [channel_masks(pair.coarse_shape)[0] & ~pair.covered,
(pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)]