ClaudeandClaude Opus 5.5
6f95c34c0b
protocol.py holds the pure arithmetic (phase/cycle bookkeeping from the right
turning point, 1-degree bins, interval-averaged Morison bases and fit with
added-mass c_i, solid-aware bilinear sampling, settling and common-window
rules). run.py drives the existing Simulation API (set_body + run(1) +
read_force + moving_diagnostics; macroscopic fields and flags only inside the
three phase bins) and writes raw data outside the checkout. analyze.py builds
the cycle ledger, K_case, the common window, Morison fits, Fy sanity and the
cross-resolution waveform difference. CPU tests cover the §6a arithmetic.
Carries the protocol doc from claude/duetsch-protocol 8c1f3c4d.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
102 lines
4.1 KiB
Python
102 lines
4.1 KiB
Python
"""Ledger, settling, common window and Morison fits for Dütsch cases (numpy only).
|
|
|
|
python analyze.py CASE_DIR [CASE_DIR ...] # cases of one comparison
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
|
|
import numpy as np
|
|
|
|
import protocol as p
|
|
|
|
|
|
def load(path):
|
|
with open(os.path.join(path, "case.json")) as fh:
|
|
case = json.load(fh)
|
|
raw = np.load(os.path.join(path, "raw.npz"))
|
|
case.update(force=raw["force"], diag=raw["diag"], samples=raw["samples"], path=path)
|
|
return case
|
|
|
|
|
|
def cycle_slices(case):
|
|
"""{k: slice of step indices m whose interval [m, m+1] lies in paper cycle k}."""
|
|
T, q = case["T"], case["T"] // 4
|
|
return {k: slice(q + (k - 1) * T, q + k * T)
|
|
for k in range(1, case["cycles"] + 1) if q + k * T <= len(case["force"])}
|
|
|
|
|
|
def coarse_wave(case, k, r=1):
|
|
"""Normalized per-step Fx of cycle k, averaged over r native steps per sample."""
|
|
f = case["force"][cycle_slices(case)[k], 0] / case["F_ref"]
|
|
return f.reshape(-1, r).mean(axis=1)
|
|
|
|
|
|
def profiles(case, k):
|
|
s = case["samples"]
|
|
rows = s[s[:, 1] == k]
|
|
return {(ph, xs, x2): (u, v) for _, _, ph, xs, x2, u, v in rows}
|
|
|
|
|
|
def cycle_morison(case, ks):
|
|
T = case["T"]
|
|
idx = np.concatenate([np.arange(len(case["force"]))[cycle_slices(case)[k]] for k in ks])
|
|
b = np.array([p.interval_bases(int(m), T) for m in idx])
|
|
return p.morison_fit(case["force"][idx, 0] / case["F_ref"], b[:, 0], b[:, 1])
|
|
|
|
|
|
def pair_metrics(case, k):
|
|
a, b = coarse_wave(case, k - 1), coarse_wave(case, k)
|
|
pa, pb = profiles(case, k - 1), profiles(case, k)
|
|
common = [key for key in pa if key in pb and not any(map(math.isnan, pa[key] + pb[key]))]
|
|
prof = max((max(abs(pa[c][0] - pb[c][0]), abs(pa[c][1] - pb[c][1])) for c in common),
|
|
default=math.inf) / case["u_max"]
|
|
(cd0, ci0), (cd1, ci1) = cycle_morison(case, [k - 1]), cycle_morison(case, [k])
|
|
m = dict(wave=float(np.sqrt(np.mean((b - a) ** 2))),
|
|
scalar=float(max(abs(b.mean() - a.mean()), abs(np.sqrt(np.mean(b * b)) - np.sqrt(np.mean(a * a))),
|
|
abs(np.abs(b).max() - np.abs(a).max()))),
|
|
profile=float(prof), morison=float(max(abs(cd1 - cd0), abs(ci1 - ci0))),
|
|
profile_points=len(common))
|
|
m["pass"] = all(m[key] <= tol for key, tol in p.SETTLE.items())
|
|
return m
|
|
|
|
|
|
def ledger(case):
|
|
ks = sorted(cycle_slices(case))
|
|
pairs = {k: pair_metrics(case, k) for k in ks if k - 1 > p.DISCARD}
|
|
return pairs, p.settled_start({k: v["pass"] for k, v in pairs.items()})
|
|
|
|
|
|
def main(paths):
|
|
cases = [load(path) for path in paths]
|
|
ledgers = [ledger(c) for c in cases]
|
|
window = p.common_window([k for _, k in ledgers],
|
|
[{k: v["pass"] for k, v in pairs.items()} for pairs, _ in ledgers])
|
|
report = dict(window=window, cases=[])
|
|
for case, (pairs, k_case) in zip(cases, ledgers):
|
|
entry = {key: case[key] for key in ("path", "D", "T", "box", "n", "u_max", "nu", "tau", "A",
|
|
"steps", "cycles", "F_ref")}
|
|
entry.update(duration_T=case["steps"] / case["T"], K_case=k_case, pairs=pairs,
|
|
no_donor=float(case["diag"][:, 13].sum()), eq_only=float(case["diag"][:, 12].sum()))
|
|
if window:
|
|
entry["morison"] = cycle_morison(case, window)
|
|
entry["fy_mean"] = float(np.mean(np.concatenate(
|
|
[case["force"][cycle_slices(case)[k], 1] for k in window])) / case["F_ref"])
|
|
report["cases"].append(entry)
|
|
if window and len(cases) > 1: # cross-resolution waveform against the finest case
|
|
fine = max(cases, key=lambda c: c["D"])
|
|
ref = np.concatenate([coarse_wave(fine, k) for k in window])
|
|
for case, entry in zip(cases, report["cases"]):
|
|
r = fine["T"] // case["T"]
|
|
ref_c = ref.reshape(-1, r).mean(axis=1)
|
|
own = np.concatenate([coarse_wave(case, k) for k in window])
|
|
entry["wave_rms_vs_finest"] = float(np.sqrt(np.mean((own - ref_c) ** 2)))
|
|
print(json.dumps(report, indent=1, default=float))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main(sys.argv[1:])
|