Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions test/test_impa_mapping.py
Original file line number Diff line number Diff line change
Expand Up @@ -631,3 +631,19 @@ def test_toroidal_hall_channels_declare_a_toroidal_normal_per_the_dd():
# The normal lies in the horizontal plane, which is what poloidal_angle 0
# declares; only the toroidal angle distinguishes grad(phi) from grad(R).
assert IMPA_TOROIDAL_PROBE_POLOIDAL_ANGLE == pytest.approx(0.0)


def test_resolve_gain_fallback_is_canonical_negative():
from vaft.machine_mapping.impa import _resolve_gain

# When no gain is present in calibration dictionary, fallback must be -2.0 / 15.0
assert _resolve_gain({}, 39204) == pytest.approx(-2.0 / 15.0)


def test_window_criteria_matches_vest_yaml_without_max_relative_noise():
from vaft.machine_mapping.impa import _window_criteria

config = resolve_impa_config(39204)
criteria = _window_criteria(config)
assert not hasattr(criteria, "max_relative_noise")
assert "max_relative_noise" not in (config.get("calibration_window") or {})
52 changes: 39 additions & 13 deletions test/test_impa_processing.py
Original file line number Diff line number Diff line change
Expand Up @@ -227,12 +227,12 @@ def test_calibration_applies_gain_and_baseline_in_the_legacy_order():
raw = np.vstack([np.full(time.size, 1.5), 1.5 + np.linspace(0.0, 1.0, time.size)])

calibrated = impa_calibrate_signals(
raw, gain=2.0 / 15.0, cutoff_hz=250.0, sample_rate=SAMPLE_RATE, baseline="first_sample"
raw, gain=-2.0 / 15.0, cutoff_hz=250.0, sample_rate=SAMPLE_RATE, baseline="first_sample"
)

# A constant input is entirely baseline, so it must vanish.
assert np.allclose(calibrated[0], 0.0, atol=1e-9)
assert calibrated[1, -1] == pytest.approx(2.0 / 15.0, rel=5e-2)
assert calibrated[1, -1] == pytest.approx(-2.0 / 15.0, rel=5e-2)


def test_unknown_baseline_is_rejected():
Expand All @@ -253,7 +253,7 @@ def test_process_impa_reports_valid_for_a_well_behaved_synthetic_array():
bz = 3.0e-5 * np.sin(2 * np.pi * 3 * time)
# process_impa filters and gain-calibrates, so hand it volts.
measured = _synthetic_array(time, i_tf, radii, alpha, bz=bz)
raw = measured / (2.0 / 15.0)
raw = measured / (-2.0 / 15.0)

result = process_impa(
time,
Expand Down Expand Up @@ -288,7 +288,7 @@ def test_a_toroidal_array_is_rejected_by_the_poloidal_model():
time = _time()
i_tf = _tf_ramp(time)
radii = 0.4 + np.arange(8) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.ones(8)) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.ones(8)) / (-2.0 / 15.0)

result = process_impa(
time, raw, i_tf, r=radii, config=ImpaProcessingConfig(orientation="poloidal")
Expand All @@ -304,7 +304,7 @@ def test_a_toroidal_array_is_accepted_by_the_toroidal_model():
time = _time()
i_tf = _tf_ramp(time)
radii = 0.4 + np.arange(8) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.ones(8)) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.ones(8)) / (-2.0 / 15.0)

result = process_impa(time, raw, i_tf, r=radii) # toroidal is the default

Expand All @@ -317,7 +317,7 @@ def test_a_poloidal_array_is_rejected_by_the_toroidal_model():
time = _time()
i_tf = _tf_ramp(time)
radii = 0.4 + np.arange(8) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.full(8, 0.05)) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.full(8, 0.05)) / (-2.0 / 15.0)

result = process_impa(time, raw, i_tf, r=radii)

Expand All @@ -330,7 +330,7 @@ def test_a_reference_shot_calibration_can_be_reused():
time = _time()
i_tf = _tf_ramp(time)
radii = 0.42 + np.arange(8) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.ones(8)) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.ones(8)) / (-2.0 / 15.0)
reference = process_impa(time, raw, i_tf)

# A shot with no clean TF interval of its own still calibrates.
Expand All @@ -348,7 +348,7 @@ def test_missing_channels_are_reported_and_never_silently_zero_filled():
time = _time()
i_tf = _tf_ramp(time)
radii = 0.4 + np.arange(8) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.full(8, 0.05)) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.full(8, 0.05)) / (-2.0 / 15.0)
valid = np.ones(8, dtype=bool)
valid[3] = False

Expand Down Expand Up @@ -485,7 +485,7 @@ def test_an_inactive_bz_sensor_is_flagged_rather_than_trusted():
time = _time()
i_tf = _tf_ramp(time)
radii = 0.45 + np.arange(2) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.ones(2)) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.ones(2)) / (-2.0 / 15.0)
dead = np.random.default_rng(1).normal(0.0, 1e-5, (2, time.size))

result = process_impa(time, raw, i_tf, r=radii, b_z_raw=dead)
Expand All @@ -505,7 +505,7 @@ def test_incident_angle_is_recovered_from_the_rigid_pitch():
incident = 30.0
projected = 0.05 * np.cos(np.radians(incident))
radii = 0.45 + np.arange(8) * projected
raw = _synthetic_array(time, i_tf, radii, np.ones(8), noise=0.0) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.ones(8), noise=0.0) / (-2.0 / 15.0)

result = process_impa(time, raw, i_tf, pitch=0.05, fit_pitch=True)

Expand All @@ -518,7 +518,7 @@ def test_a_radial_insertion_reports_no_incident_angle():
time = _time()
i_tf = _tf_ramp(time)
radii = 0.45 + np.arange(8) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.ones(8), noise=0.0) / (2.0 / 15.0)
raw = _synthetic_array(time, i_tf, radii, np.ones(8), noise=0.0) / (-2.0 / 15.0)

result = process_impa(time, raw, i_tf, pitch=0.05, fit_pitch=True)

Expand Down Expand Up @@ -559,11 +559,37 @@ def test_process_impa_rejects_a_reference_with_a_different_channel_count():
time = _time()
i_tf = _tf_ramp(time)
reference_radii = 0.4 + np.arange(7) * 0.05
reference_raw = _synthetic_array(time, i_tf, reference_radii, np.ones(7)) / (2.0 / 15.0)
reference_raw = _synthetic_array(time, i_tf, reference_radii, np.ones(7)) / (-2.0 / 15.0)
reference = process_impa(time, reference_raw, i_tf, r=reference_radii)

target_radii = 0.4 + np.arange(8) * 0.05
target_raw = _synthetic_array(time, i_tf, target_radii, np.ones(8)) / (2.0 / 15.0)
target_raw = _synthetic_array(time, i_tf, target_radii, np.ones(8)) / (-2.0 / 15.0)

with pytest.raises(ValueError, match="7 channels but this shot has 8"):
process_impa(time, target_raw, i_tf, reference=reference)


def test_impa_code_defaults_match_canonical_values():
"""Verify code defaults agree with hardware specs and vest.yaml (#624, #625)."""
config = ImpaProcessingConfig()
assert config.gain == pytest.approx(-2.0 / 15.0)

# TfWindowCriteria no longer defines unread max_relative_noise
criteria = TfWindowCriteria()
assert not hasattr(criteria, "max_relative_noise")


def test_process_impa_defaults_agree_with_vest_yaml():
"""Verify process_impa defaults bz_radial_offset=0.01 and max_normalized_rmse=0.15."""
time = _time()
i_tf = _tf_ramp(time)
radii = 0.4 + np.arange(8) * 0.05
raw = _synthetic_array(time, i_tf, radii, np.ones(8)) / (-2.0 / 15.0)
bz_raw = np.zeros((8, time.size))

# Without passing explicit config or bz_radial_offset, the defaults apply
result = process_impa(time, raw, i_tf, r=radii, b_z_raw=bz_raw)

assert result.bz_r is not None
np.testing.assert_allclose(result.bz_r, radii + 0.01)
assert result.provenance["gain"] == pytest.approx(-2.0 / 15.0)
7 changes: 3 additions & 4 deletions vaft/machine_mapping/impa.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ def _channel_specs(config: Mapping[str, Any], shot: int) -> list[dict[str, Any]]

def _resolve_gain(calibration: Mapping[str, Any], shot: int) -> float:
"""Return the Hall gain for ``shot``, honouring verified shot-era ranges."""
gain = float(calibration.get("gain", 2.0 / 15.0))
gain = float(calibration.get("gain", -2.0 / 15.0))
for era in calibration.get("shot_era_overrides") or ():
if int(era.get("min_shot", 0)) <= int(shot) <= int(era.get("max_shot", 0)):
return float(era.get("gain", gain))
Expand Down Expand Up @@ -202,7 +202,6 @@ def _window_criteria(config: Mapping[str, Any]) -> TfWindowCriteria:
pf_current_max=float(window.get("pf_current_max", defaults.pf_current_max)),
min_duration=float(window.get("min_duration", defaults.min_duration)),
tf_dynamic_range_min=float(window.get("tf_dynamic_range_min", defaults.tf_dynamic_range_min)),
max_relative_noise=float(window.get("max_relative_noise", defaults.max_relative_noise)),
smoothing_samples=int(window.get("smoothing_samples", defaults.smoothing_samples)),
)

Expand Down Expand Up @@ -439,15 +438,15 @@ def process_impa_shot(
pitch=float(geometry.get("radial_pitch", 0.05)),
r_bounds=(float(min(r_bounds)), float(max(r_bounds))),
r0_initial=float(geometry.get("r0_initial", 0.4)),
max_normalized_rmse=float(quality.get("max_normalized_rmse", 0.1)),
max_normalized_rmse=float(quality.get("max_normalized_rmse", 0.15)),
reference=reference,
b_z_raw=inputs["bz_raw"],
bz_channel_valid=inputs["bz_channel_valid"],
fit_pitch=bool(geometry.get("fit_pitch", False)),
max_crosstalk_angle_deg=float(crosstalk_config.get("max_angle_deg", 30.0)),
min_crosstalk_r_squared=float(crosstalk_config.get("min_r_squared", 0.8)),
bz_gain=config.get("calibration", {}).get("bz_gain"),
bz_radial_offset=float(geometry.get("bz_radial_offset", 0.0)),
bz_radial_offset=float(geometry.get("bz_radial_offset", 0.01)),
)
object.__setattr__(result, "provenance", {**result.provenance, "shot": int(shot)})
return result, inputs
Expand Down
1 change: 0 additions & 1 deletion vaft/machine_mapping/vest.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -644,7 +644,6 @@
pf_current_max: 500.0
min_duration: 0.005
tf_dynamic_range_min: 0.02
max_relative_noise: 0.5
smoothing_samples: 25 # 1 ms: Ip/PF quiet must be sustained
quality:
# Set from the verified 2022-04-23 alignment shots, whose rigid-array
Expand Down
12 changes: 4 additions & 8 deletions vaft/process/impa.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ class ImpaProcessingConfig:
#: as a separate stage because the two legacy routines genuinely differ.
position_lowpass_hz: float = 2_500.0
#: Hall calibration factor [T/V] in the canonical VAFT sign convention.
gain: float = 2.0 / 15.0
gain: float = -2.0 / 15.0
baseline: str = "first_sample"
baseline_samples: int = 2_500
tf_turns: int = 24
Expand Down Expand Up @@ -210,8 +210,6 @@ class TfWindowCriteria:
#: Relative TF spread ``(max-min)/max|I_TF|`` below which the interval is
#: usable but reported as poorly conditioned for a slope fit.
tf_dynamic_range_min: float = 0.02
#: Per-channel sample-to-sample noise relative to the in-window spread.
max_relative_noise: float = 0.5
#: Median-filter width applied to |Ip| and the PF peak before thresholding.
#: "No plasma" is a sustained condition, so isolated noise spikes must not
#: fragment an otherwise clean interval. 25 samples is 1 ms at 25 kHz.
Expand Down Expand Up @@ -616,7 +614,6 @@ def fit_impa_geometry(
i_tf = np.asarray(i_tf, dtype=float)
idx = np.asarray(window.indices, dtype=int)
n_channels = b_measured.shape[0]
offsets = np.arange(n_channels) * float(pitch)
observed = b_measured[:, idx]
lower, upper = float(min(r_bounds)), float(max(r_bounds))

Expand All @@ -627,7 +624,6 @@ def fit_impa_geometry(
if not usable.any():
raise ValueError("No IMPA channel has finite samples inside the calibration window")
fitted = observed[usable]
fitted_offsets = offsets[usable]

# The array is rigid, so the physical spacing is fixed; a probe inserted at
# an angle to the midplane projects it onto a shorter radial step, and the
Expand Down Expand Up @@ -975,7 +971,7 @@ def grade_impa_quality(
coupling: ImpaCouplingFit | None,
*,
expected_channels: int = IMPA_CHANNEL_COUNT,
max_normalized_rmse: float = 0.1,
max_normalized_rmse: float = 0.15,
r_bounds: Sequence[float] = (0.1, 0.9),
pitch_tolerance: float = 0.01,
window_reasons: Sequence[str] = (),
Expand Down Expand Up @@ -1177,15 +1173,15 @@ def process_impa(
pitch: float = 0.05,
r_bounds: Sequence[float] = (0.1, 0.9),
r0_initial: float = 0.4,
max_normalized_rmse: float = 0.1,
max_normalized_rmse: float = 0.15,
reference: ImpaResult | None = None,
b_z_raw: np.ndarray | None = None,
bz_channel_valid: np.ndarray | None = None,
fit_pitch: bool = False,
max_crosstalk_angle_deg: float = 30.0,
min_crosstalk_r_squared: float = 0.8,
bz_gain: float | None = None,
bz_radial_offset: float = 0.0,
bz_radial_offset: float = 0.01,
) -> ImpaResult:
"""Run the full single-shot IMPA pipeline.

Expand Down
Loading