FAILURE_MODES = ('accepted', 'rejected', 'collapsed', 'boundary_failure', 'axis_failure', 'solver_failure', 'unknown')

PHASES = ('vacuum', 'ramp_up', 'flat', 'ramp_down', 'unknown')

STAGE_METRICS = <builtins.dict>

STAGE_PRECONDITIONS = <builtins.dict>

STAGE_VALIDATION_PLOTS = <builtins.dict>

Deprecated: vaft.validation.STAGE_VALIDATION_PLOTS moved to vaft.database.production_qa (issue #338): it is stage-QA artifact policy, not scientific validation. Import it from there.

APPLICABILITY_STATUSES = ('SUPPORTED', 'OUTSIDE', 'UNASSESSED', 'NOT_APPLICABLE')

ApproximationContract(name: str, physical_model: str, assumptions: tuple[OrderingAssumption, ...], references: tuple[str, ...] = (), limitations: tuple[str, ...] = (), applies_to: Mapping[str, tuple] = <factory>) -> None

The orderings one model or approximation relies on, with where they come from.

as_evidence(result: ContractEvaluation, *, key: str | None = None)

Place a contract result on the applicability credibility axis.

AssumptionResult(quantity: str, value: float, margin: float, status: str, reason: str = '') -> None

One assumption evaluated on one state.

ContractEvaluation(contract: str, status: str, assumptions: tuple[AssumptionResult, ...], limiting: str | None, reason: str = '') -> None

A contract evaluated on one state: every assumption, then the composition.

evaluate_contract(contract: ApproximationContract, state: Any) -> ContractEvaluation

Evaluate every assumption of contract on one state.

evaluate_population(contract: ApproximationContract, table: Any, *, keep: Sequence[str] = ()) -> Any

Evaluate contract on every row of a table; one output row per input row.

ordering_margin(value: Any, ordering: str, threshold: float = 1.0) -> float

Decades by which value satisfies an ordering against threshold.

OrderingAssumption(quantity: str, ordering: str, scale: str, scope: str = 'global', threshold: float = 1.0, threshold_source: str = '', meaning: str = '') -> None

One asymptotic assumption a model makes: quantity is small or large.

ORDERINGS = ('small', 'large')

SCOPES = ('global', 'local', 'profile', 'edge', 'time_history', 'event', 'mode')

successive_discrepancy(values: Iterable[Any], *, reference: Any = None) -> list[float]

Relative change of an observable along a model or representation hierarchy.

summarize_population(evaluated: Any, contract: ApproximationContract) -> dict[str, Any]

Population-level applicability of one contract from evaluate_population output.

ChannelDecision(kind: str, index: int, name: str, state: np.ndarray, weight_factor: np.ndarray, value: np.ndarray | None = None, uncertainty: np.ndarray | None = None, provenance: str = 'measured', reasons: tuple[str, ...] = ()) -> None

One channel’s decision on the consumer’s time grid.

ChannelDecisions(times: np.ndarray, entries: Mapping[tuple[str, int], ChannelDecision] = <factory>) -> None

Every submitted channel’s decision on one time grid.

decision_from_codes(kind: str, index: int, name: str, codes: np.ndarray | None, n_slices: int, *, min_validity: int, suspect_code: int, suspect_weight_factor: float | None = None, reasons: Iterable[str] = ()) -> ChannelDecision

A decision from per-slice validity codes.

MISSING = 3

missing_decision(kind: str, index: int, name: str, n_slices: int) -> ChannelDecision

PROVENANCE_MEASURED = 'measured'

REASON_CONDEMNED = 'condemned_whole_record'

REASON_MANUAL_LIST = 'manual_exclusion_list'

REASON_NO_SIGNAL = 'no_signal'

REASON_PROJECTED_VALIDITY = 'projected_validity'

RECOVERED = 4

REJECTED = 2

rejected_decision(kind: str, index: int, name: str, n_slices: int, *, reason: str | Iterable[str]) -> ChannelDecision

state_name(code: int) -> str

STATE_NAMES = ('usable', 'suspect', 'rejected', 'missing', 'recovered')

SUSPECT = 1

USABLE = 0

usable_decision(kind: str, index: int, name: str, n_slices: int, *, reasons: Iterable[str] = ()) -> ChannelDecision

weight_factor_for(state: np.ndarray, *, suspect_weight_factor: float = 1.0) -> np.ndarray

The deterministic state -> weight-factor mapping.

AXES = ('source_evidence', 'transformation', 'inference', 'applicability', 'numerical', 'independent_validation')

CATEGORY_AXES = <builtins.mappingproxy>

CHECK_AXES = <builtins.mappingproxy>

compose(evidence: Iterable[Evidence]) -> dict[str, ValidationStatus]

One status per axis that has evidence, never one status overall.

COST_CLASSES = ('cheap', 'moderate', 'expensive')

CRITERIA_AXES = <builtins.mappingproxy>

DIAMAGNETIC_CHECKS = ('physical_validity.diamagnetic_flux', 'independent_validation.diamagnetic_energy', 'independent_validation.virial_me...

Evidence(axis: str, key: str, status: ValidationStatus, metrics: Mapping[str, Any] = <factory>, reasons: tuple[str, ...] = (), cost: str = 'cheap', role: str | None = None) -> None

One assessment’s conclusion, placed on one credibility axis.

evidence_from_efit_criteria(evaluation: Mapping[str, Any], *, used_for_inference: Iterable[str] = ()) -> tuple[Evidence, ...]

Pilot A: the #891/#1331 study criteria (version 2) on the credibility axes.

evidence_from_equilibrium_report(report: Mapping[str, Any], *, used_for_inference: Iterable[str] = ()) -> tuple[Evidence, ...]

Pilot A: vaft.validation.validate_equilibrium on the credibility axes.

EVIDENCE_ROLES = ('used_for_inference', 'independent_validation', 'available_not_used', 'rejected', 'assumed', 'inferred')

condemned_channels(source: Any, *, nbprobe: int) -> set[int]

Legacy-style combined zero-based indexes of channels condemned outright.

decide_efit_channels(source: Any, times: Any, *, nbprobe: int | None = None, manual_rejections: Iterable[int] = (), min_validity: int = 0, suspect_weight_factor: float | None = None, suspects: Mapping[tuple[str, int], str] | None = None, kinds: Iterable[str] = ('b_field_pol_probe', 'flux_loop')) -> ChannelDecisions

Decide every EFIT-facing magnetics channel at every slice of times.

EFIT_KINDS = ('b_field_pol_probe', 'flux_loop')

efit_probe_count(source: Any) -> int

The number of B-pol probes EFIT’s geometry represents.

EFIT_QUANTITY = <builtins.dict>

split_manual_rejections(manual: Iterable[int], *, nbprobe: int) -> dict[tuple[str, int], int]

Combined one-based indexes -> (kind, index).

aggregate_status(statuses: Iterable[ValidationStatus | str]) -> ValidationStatus

Summarize several statuses without collapsing the undecided ones.

EQUILIBRIUM_CATEGORIES = ('verification', 'diagnostic_fit', 'physical_validity', 'independent_validation')

status_summary(report: Mapping[str, Any]) -> dict[str, ValidationStatus]

One ValidationStatus per category present in report.

thomson_pressure_samples(equilibrium: Any, time_slice: int, diagnostics: Any, *, tolerance: float | None = None) -> dict[str, Any]

The Thomson channels of one slice, placed on its flux surfaces.

validate_equilibrium(equilibrium: Any, *, diagnostics: Any | None = None, kinetic_profiles: Any | None = None, checks: Iterable[str] = ('verification', 'diagnostic_fit', 'physical_validity', 'independent_validation'), time_slice: int | Sequence[int] | None = None) -> dict[str, Any]

The unified equilibrium validation report (#72).

Also exported as vaft.validation.validate_equilibrium.

validate_independent(equilibrium: Any, *, time_slice: int, kinetic_profiles: Any | None = None, diagnostics: Any | None = None) -> dict[str, dict[str, Any]]

Agreement with measurements the reconstruction never used, one slice (#72 §4).

validate_magnetic_fit(equilibrium: Any, *, time_slice: int) -> dict[str, dict[str, Any]]

Constraint-fit assessment for one slice, per check (#72 §2, #69).

validate_physical(equilibrium: Any, *, time_slice: int, diagnostics: Any | None = None) -> dict[str, dict[str, Any]]

Force-balance and plausibility checks for one slice (#72 §3).

verify_continuity(equilibrium: Any, *, time_slice: int | Sequence[int] | None = None) -> dict[str, Any]

Time-slice continuity across the selected slices (#72 §1).

verify_convention(equilibrium: Any, *, time_slice: int) -> dict[str, Any]

COCOS sign relations for one slice, via vaft.process.cocos.validate_cocos.

verify_convergence(equilibrium: Any, *, time_slice: int) -> dict[str, Any]

The solver’s own convergence evidence for one slice (#70).

verify_structure(equilibrium: Any, *, time_slice: int) -> dict[str, Any]

Numerical and structural checks for one slice (#72 §1).

cohort_rows(outcomes: Iterable[SliceOutcome]) -> list[dict[str, Any]]

One flat record per slice, every column of _COLUMNS.

CohortJoin(outcomes: tuple[SliceOutcome, ...], unmatched_log_times: tuple[int, ...]) -> None

The joined slices, and every way the join was incomplete.

Also exported as vaft.validation.CohortJoin.

FAILURE_MODES = ('accepted', 'rejected', 'collapsed', 'boundary_failure', 'axis_failure', 'solver_failure', 'unknown')

join_regime_and_termination(ods: Any, slices: Sequence[Mapping[str, Any]], *, policy=None, slice_statuses: Sequence[Mapping[str, Any]] | None = None, shot: int | None = None, source: str = 'unknown', **evidence: Any) -> CohortJoin

Put each reconstruction slice’s regime beside how EFIT ended it.

Also exported as vaft.validation.join_regime_and_termination.

representative_cases(outcomes: Iterable[SliceOutcome], *, group_by: Sequence[str] = ('phase',), per_cohort: int = 1) -> dict[str, dict[str, list[dict[str, Any]]]]

Deterministic example slices per cohort: successes and failures.

SliceOutcome(time: float, regime: RegimeLabels, termination: Mapping[str, Any] | None, mode: str, quality: str = 'unknown', shot: int | None = None, source: str = 'unknown') -> None

One slice: the regime it was in, and what became of it.

Also exported as vaft.validation.SliceOutcome.

summarize_by_cohort(outcomes: Iterable[SliceOutcome], *, group_by: Sequence[str] = ('phase',)) -> dict[str, Any]

Outcome counts per cohort, plus what each cohort’s failures blame.

Also exported as vaft.validation.summarize_by_cohort.

write_cohort_table(rows: Sequence[Mapping[str, Any]], path: str | Path, *, policy=None) -> Path

Write the joined table as CSV or Parquet (by suffix), with its rule.

ADMISSIBILITY_RULES = <builtins.tuple>

COHORTS = ('good', 'admissible', 'unreconstructible')

CONFINEMENT_RULES = ('finite', 'ip_min', 'dwdt_fraction', 'ip_change_per_tau')

CRITERIA_PATH = <pathlib.PosixPath>

CROSSWALK = <builtins.tuple>

efit_evidence_columns(ods: Any, time_slice: int, *, diagnostics: Any = None, kinetic_profiles: Any = None) -> dict[str, Any]

The vaft.omas.efit_quality and generic-validation columns of one equilibrium slice.

Also exported as vaft.validation.efit_evidence_columns.

equilibrium_quality_confinement_funnel(confinement_table, *, quality_column: str = 'efit_quality', rules: Sequence[str] = ('finite', 'ip_min', 'dwdt_fraction', 'ip_change_per_tau')) -> dict[str, Any]

Equilibrium quality × confinement acceptance (#1644 §9), per quality cohort.

equilibrium_quality_constraint_points(ods: Any, time_slice: int) -> list[dict[str, Any]]

Every constraint channel of one slice: measured, reconstructed and normalised residual.

equilibrium_quality_crosswalk(table) -> list[dict[str, Any]]

Generic validation status × study verdict, per paired concept, where both exist.

Also exported as vaft.validation.equilibrium_quality_crosswalk.

equilibrium_quality_failure_census(table) -> dict[str, Any]

The rule × cohort matrix (status fractions) and a failure Pareto per cohort.

Also exported as vaft.validation.equilibrium_quality_failure_census.

equilibrium_quality_summary(table) -> dict[str, Any]

Slice counts per cohort, and Thomson consistency within each.

Also exported as vaft.validation.equilibrium_quality_summary.

equilibrium_quality_table(records: Iterable[Mapping[str, Any]], *, criteria: ModuleType | None = None, evidence: Callable[[Mapping[str, Any]], Mapping[str, Any] | None] | None = None)

One row per (shot, time, setting): raw metrics, verdicts and the study classification.

Also exported as vaft.validation.equilibrium_quality_table.

FIT_QUALITY_RULES = ('rule_convergence', 'rule_pressure_nonnegative', 'rule_beta_p_positive', 'rule_w_positive', 'rule_q95', 'rule_ip_rat...

load_study_criteria(path: str | Path | None = None) -> ModuleType

The #891/#1331 study criteria module: the shipped one, or a caller’s file loaded from path.

MEASUREMENT_RULES = <builtins.tuple>

read_equilibrium_quality_table(source: Any) -> Any

A cohort table written with DataFrame.to_csv read back as equilibrium_quality_table made it.

REPRESENTATIVE_CASES = ('good_consistent', 'good_inconsistent', 'admissible_only', 'unreconstructible')

ROUTINE_SETTING = 'routine'

RULE_COLUMNS = ('rule_convergence', 'rule_pressure_nonnegative', 'rule_beta_p_positive', 'rule_w_positive', 'rule_q95', 'rule_ip_rat...

select_representative_cases(table) -> list[dict[str, Any]]

Deterministic representative slices for the four #1644 cases, with why each was chosen.

slice_cohorts(table) -> Any

One row per slice: its label and, for the census, the row of its best setting.

STATUSES = ('pass', 'fail', 'indeterminate', 'not_available')

TEXT_COLUMNS = ('good_settings', 'admissible_settings', 'consistent_settings', 'inconsistent_settings', 'failure_reasons')

TYPICALITY_METRICS = ('probe_reduced_chi2', 'loop_reduced_chi2', 'gs_residual', 'virial_log_ratio')

classify_equilibrium_regime(ods: Any, *, time_index: int, policy: EquilibriumRegimePolicy | None = None, **evidence: Any) -> RegimeLabels

Label one slice, by classifying the discharge and taking that slice.

Also exported as vaft.validation.classify_equilibrium_regime.

classify_equilibrium_regimes(ods: Any, *, policy: EquilibriumRegimePolicy | None = None, era: str | None = None, passive_current: tuple[Sequence[float], Sequence[float]] | None = None, events: Sequence[tuple[float, float, str]] | None = None, limloc: Mapping[int, str] | None = None) -> tuple[RegimeLabels, ...]

Label every slice of one reconstruction.

Also exported as vaft.validation.classify_equilibrium_regimes.

PHASES = ('vacuum', 'ramp_up', 'flat', 'ramp_down', 'unknown')

RegimeLabels(time: float, phase: str, topology: str, current: float | None, current_rate: float | None, fraction_of_peak: float | None, limloc: str | None, rule: str, reasons: tuple[str, ...] = (), current_bin: str = 'unknown', magnitude_rate: float | None = None, rate_bin: str = 'unknown', era: str = 'unknown', event: str = 'unknown', eddy_current: float | None = None, eddy_fraction: float | None = None, version: int = 1) -> None

One slice’s cohort membership, and the evidence that put it there.

Also exported as vaft.validation.RegimeLabels.

assess_flux_loops(qualities: Iterable[Any], residual_rows: Sequence[Mapping[str, Any]] | None, *, window: tuple[float, float] | None, nbprobe: int, time: np.ndarray | None = None, projected: Mapping[int, ValidityRecord] | None = None, field_codes: Sequence[int | None] | None = None, policy: FluxLoopPolicy | None = None) -> tuple[FluxLoopAssessment, ...]

Combine intrinsic quality and model agreement into one record per loop.

flux_loop_evidence(ods: Any, *, policy: FluxLoopPolicy | None = None, window: tuple[float, float] | None = None, benchmark: bool = True, nbprobe: int | None = None) -> dict[str, Any]

Assemble the flux-loop evidence for one processed ODS.

FluxLoopAssessment(index: int, name: str, field_code: int | None, combined_index_one_based: int, state: str, reasons: tuple[str, ...], source_validity: Mapping[str, Any], model_agreement: Mapping[str, Any] | None, window: tuple[float, float] | None, policy: Mapping[str, Any] = <factory>) -> None

One loop’s evidence and the state the policy derives from it.

FluxLoopPolicy(min_valid_fraction_in_window: float = 0.0, max_normalized_residual: float | None = None, min_correlation: float | None = None, min_wall_authority_to_score: float = 0.1, reject_on_model_disagreement: bool = False) -> None

The thresholds that turn evidence into a state. All of them are policy (#253 §7) and every record repeats the values it was judged under.

manual_exclusion_index(index: int, nbprobe: int) -> int

The one-based combined index the routine broken list uses for a loop.

NOT_AVAILABLE = 'not_available'

REJECT_FOR_EFIT = 'reject_for_efit'

STATES = ('usable', 'suspect', 'reject_for_efit', 'not_available')

SUSPECT = 'suspect'

USABLE = 'usable'

is_condemned_channel(source: Any, base: str, *, min_validity: int = 0) -> bool

Whether base was assessed and has no usable sample anywhere.

is_usable_channel(source: Any, base: str, *, times: np.ndarray | None = None, min_validity: int = 0, default: bool = True) -> bool

Whether base has any usable sample at all, or any of times.

read_convergence_result(source: Any, *, time_slice: int) -> dict[str, Any] | None

equilibrium.time_slice[i].convergence.result – a solver’s own verdict.

read_output_flag(source: Any, ids: str = 'equilibrium', *, time_slice: int | None = None) -> Any

<ids>.code.output_flag – an external code’s execution status.

read_validity(source: Any, base: str) -> int | None

<base>.validity, or None when the node carries none.

read_validity_record(source: Any, base: str) -> ValidityRecord

The validity nodes of base, extracted and uninterpreted.

read_validity_timed(source: Any, base: str) -> np.ndarray | None

<base>.validity_timed as an integer array, or None when absent.

resolve_signal_time(source: Any, base: str) -> np.ndarray | None

The time coordinate of the signal node at base.

resolve_signal_waveform(source: Any, base: str) -> tuple[np.ndarray, np.ndarray] | None

(time, data) of the signal node at base, or None.

signal_label(source: Any, base: str, fallback: str) -> str

The channel’s name, else its identifier, else fallback.

signal_matches_root_time(source: Any, base: str) -> bool

Whether base holds a waveform the length of its IDS root time.

valid_fraction(source: Any, base: str, *, times: np.ndarray | None = None, window: np.ndarray | None = None, min_validity: int = 0, default: bool = True) -> float

Fraction of the samples in window that base declares usable.

validity_codes(source: Any, base: str, *, times: np.ndarray | None = None) -> np.ndarray | None

Validity codes for the node at base, per sample.

validity_mask(source: Any, base: str, *, times: np.ndarray | None = None, min_validity: int = 0, default: bool = True) -> np.ndarray

Which samples of base a consumer accepting min_validity may use.

write_validity(source: Any, base: str, timed: Iterable[int], *, scalar: int | None = None) -> int

Project a per-sample assessment into base’s native validity nodes.

assemble_kinetic_population(state: Any, profiles: Any, virial: Any = None, *, manifest: Optional[Mapping[str, Any]] = None, schemas: Optional[Mapping[str, Mapping[str, Any]]] = None, strict: bool = False) -> KineticPopulation

Join the atlas state, profile and virial tables under the population contract.

EVIDENCE_STATUSES = ('available', 'unavailable', 'failed')

ISSUE_COLUMNS = ('scope', 'shot', 'time_efit_s', 'efit_lineage', 'check', 'status', 'detail')

KineticPopulation(population: Any, profiles: Any, issues: Any, provenance: Mapping[str, Any] = <factory>) -> None

The checked population: the joined state table, its profiles, and every finding.

population_evidence_summary(population: KineticPopulation, *, by: Sequence[str] = ('efit_lineage',))

Counts and quartiles of every evidence quantity, by lineage (or by) and validation role.

The atlas tables break the population contract (duplicate keys, missing columns, or a strict failure).

pressure_evidence(population: KineticPopulation)

One row per (state, pressure ratio): value, support, validation role and status.

PRESSURE_RATIOS = <builtins.dict>

read_kinetic_population(atlas: str | Path, *, virial: Any = None, strict: bool = False) -> KineticPopulation

Read an atlas directory (v1/state.csv, profiles.csv, MANIFEST.json, schema/) and assemble it.

STATE_KEY = ('shot', 'time_efit_s', 'efit_lineage')

validation_role(lineage: str) -> str

The validation role of an EFIT lineage; raises for an unknown one.

VALIDATION_ROLES = <builtins.dict>

virial_evidence(population: KineticPopulation)

One row per (state, virial quantity): value, family, the study verdict, validation role and status.

VIRIAL_QUANTITIES = <builtins.dict>

CONTRACT_VERSION = '1'

core_profiles_electron_pressure(profiles: Any, *, time_s: float, tolerance_s: float | None = None) -> dict[str, Any]

p_e = e n_e T_e of the core_profiles slice at time_s, on its rho_tor_norm.

core_profiles_grid_on_slice(coordinate: Mapping[str, Any], profile: Mapping[str, Any], *, tolerance: float = 0.02) -> dict[str, Any]

Place a core_profiles grid on the equilibrium slice it was written on.

default_tolerance(times: np.ndarray) -> float

Half the median cadence, at least 1 ms – the rule vaft.validation uses.

GRID_RHO_TOLERANCE = 0.02

infer_ti_pressure_partition(*args: Any, **kwargs: Any) -> dict[str, Any]

Moved to vaft.process.kinetic_closure.infer_ti_pressure_partition (#1842).

integrated_pressure_ratio(equilibrium: Any, index: int, rho_tor_norm: np.ndarray, p_e: np.ndarray, *, psi_norm_span: tuple[float, float] | None = None) -> dict[str, Any]

int p_EFIT dV / int p_e dV over a flux span of one slice.

ion_composition(z_eff: float = 2.0, impurity: str = 'C') -> dict[str, Any]

Ion species and densities per electron from a Z_eff closure.

LINEAGES = ('magnetics', 'electron_kinetic')

match_thomson_pressure(equilibrium: Any, thomson: Any, *, time_s: float, slice_tolerance_s: float | None = None, ts_tolerance_s: float | None = None) -> dict[str, Any]

Thomson electron pressure against the EFIT pressure of the slice at time_s.

pressure_sigma(pressure: np.ndarray, neighbours: Sequence[np.ndarray] = (), *, floor: float = 0.17) -> tuple[np.ndarray, str]

The EFIT pressure uncertainty: ensemble spread with a model-form floor.

QUALITIES = ('good', 'admissible')

rho_tor_norm_of(ods: Any, index: int) -> dict[str, Any]

The slice’s (psi_norm, rho_tor_norm) profile pair, or why there is none.

slice_at_time(ods: Any, time_s: float, *, ids: str = 'equilibrium', tolerance_s: float | None = None) -> tuple[int, float]

The ids slice nearest time_s, as (index, signed offset).

state_time(time_s: float) -> float

A slice time as it appears in the state key: seconds, to 1e-4 s.

array_contradiction_fractions_of(heights: Sequence[float], waveforms: np.ndarray, *, floor: float, sigma: float, min_scored: float, min_witnesses: int = 3, scale: str = 'witness') -> tuple[np.ndarray, np.ndarray]

Per-probe fraction of scorable samples over sigma, and the mask of those samples.

array_contradiction_scores(heights: Sequence[float], waveforms: np.ndarray, *, floor: float = 0.0, min_witnesses: int = 3) -> np.ndarray

Robust-z of each probe’s departure from its two nearest neighbours, per sample.

array_departure_ratios(heights: Sequence[float], waveforms: np.ndarray, *, floor: float = 0.0) -> np.ndarray

Each probe’s departure from its neighbours as a multiple of the array’s own amplitude.

attribute_array_contradictions(heights: Sequence[float], waveforms: np.ndarray, *, floor: float, sigma: float, fraction: float, min_members: int, min_scored: float, min_witnesses: int = 3, scale: str = 'witness') -> list[dict[str, Any]]

Which probes of one array contradict it, attributed by leave-one-out.

channel_node(kind: str, index: int, quantity: str) -> str

The IDS node one channel’s processed quantity lives at.

ChannelQuality(kind: str, index: int, name: str, quantity: str, unit: str, status: ValidationStatus, validity: int, validity_timed: np.ndarray, valid_fraction: float, metrics: Mapping[str, float] = <factory>, events: tuple[QualityEvent, ...] = (), reason: str = '') -> None

The signal-quality assessment of one magnetic channel.

implausible_magnitude(amplitude: float, ceiling: float | None) -> bool

Whether a record’s amplitude exceeds an absolute physical ceiling.

magnetics_quality_metrics(source: Any, report: Iterable[ChannelQuality], *, config: MagneticsQualityConfig | None = None) -> dict[str, Any]

The diagnostics-stage magnetics block for a stage manifest (#189 req 4).

MagneticsQualityConfig(baseline_fraction: float = 0.2, min_constant_run: int = 16, significance_floor: float = 0.01, spike_sigma: float = 8.0, max_spike_samples: int = 3, offset_jump_sigma: float = 10.0, coherence_window: float = 0.0005, coherent_jump_fraction: float = 0.3, max_plausible_amplitude: Mapping[str, float] = <factory>, population_peak_factor: float | None = 4.0, array_departure_factor: float | None = 1.0, array_contradiction_fraction: float = 0.3, min_array_members: int = 7, array_min_scored: float = 0.5, array_amplitude_floor: float = 0.01) -> None

Detector thresholds, carried into the report as provenance.

population_peak_outliers(amplitudes: Mapping[Any, float], factor: float | None) -> set[Any]

The members whose amplitude dwarfs the family’s median amplitude.

population_peak_ratios(amplitudes: Mapping[Any, float]) -> dict[Any, float]

Each member’s amplitude as a multiple of its family’s median amplitude.

project_validity(source: Any, report: Iterable[ChannelQuality]) -> dict[str, int]

Write a report’s per-sample assessment into the native validity nodes.

QualityEvent(reason: str, start: float, end: float, samples: int, validity: int) -> None

One contiguous stretch of a waveform that a detector objected to.

QUANTITY_BY_KIND = <builtins.dict>

recorded_faults_for(source: Any) -> dict[tuple[str, int], str]

The channels recorded as broken for the pulse source describes.

unusable_channels_at(source: Any, times: np.ndarray | Sequence[float], *, min_validity: int = 0, kinds: Iterable[str] = ('b_field_pol_probe', 'flux_loop')) -> dict[tuple[str, int], np.ndarray]

Which channels a consumer must not use at each instant in times.

validate_magnetics_signals(source: Any, *, config: MagneticsQualityConfig | None = None, kinds: Iterable[str] = ('b_field_pol_probe', 'flux_loop'), known_faults: Mapping[tuple[str, int], str] | None | object = 'from_pulse') -> tuple[ChannelQuality, ...]

Assess every equilibrium magnetics channel’s processed waveform.

build_ml_report(model, evaluation, dataset, split, criteria, *, protocol)

Bind numerical evaluation to declared identities and criteria.

MetricCriterion(metric: str, axis: str, unit: str, provider: str, source: str, revision: str, lower: float | None = None, upper: float | None = None) -> None

Versioned owner-defined bounds; units/provider/source remain explicit.

ml_report_evidence(report, model, evaluation, dataset, split, criteria, *, protocol, expected_report_sha256)

Verify a report against trusted inputs and emit existing per-axis Evidence.

CATEGORIES = ('verification', 'source_validity', 'diagnostic_fit', 'physical_validity', 'independent_validation')

Also exported as vaft.validation.CATEGORIES.

SEVERITIES = ('error', 'warning')

ValidationIssue(severity: str, code: str, field: str, message: str) -> None

One specific objection to a datum or a result.

Also exported as vaft.data.ValidationIssue, vaft.data.equilibrium.ValidationIssue.

ValidationReport(issues: tuple[ValidationIssue, ...] = ()) -> None

The findings from one assessment, in the order they were made.

Also exported as vaft.data.ValidationReport, vaft.data.equilibrium.ValidationReport.

What one assessment concluded.

Also exported as vaft.validation.ValidationStatus.

adapter_mode(key: str) -> AdapterMode

The mode "<adapter>:<mode>"; KeyError lists the known ones.

ADAPTER_MODES = <vaft.validation.model_contracts._Modes>

AdapterMode(adapter: str, mode: str, family: str, formalism: Any, audit: str, note: str = '', transport_coupled: bool = False) -> None

A scientifically distinct mode of one code adapter, with its formalism.

advisory_ordering_check(formalism, state: Mapping[str, Any], points: Any = None, *, key: Optional[str] = None, transport_coupled: bool = False) -> Tuple[Tuple[AdvisoryResult, ...], Tuple[str, ...]]

Evaluate the contracts a run’s formalism consumes on that run’s input.

AdvisoryResult(use: ContractUse, evaluation: ContractEvaluation, counts: Mapping[str, int], point: Optional[int], evidence: Any) -> None

One contract evaluated on a run’s input.

contracts_for(formalism, *, transport_coupled: bool = False) -> Tuple[Tuple[ContractUse, ...], Tuple[str, ...]]

The contracts a formalism consumes, and why any part of it maps to none.

ContractUse(contract: str, role: str) -> None

One contract a calculation consumes, and the role it plays there (ROLES).

ordering_provenance(formalism, state: Optional[Mapping[str, Any]] = None, points: Any = None, *, transport_coupled: bool = False, notes: Sequence[str] = ()) -> Optional[Dict[str, Any]]

The advisory ordering check of one run, as a JSON-ready provenance record (#1823).

ROLES = ('governing', 'outer_solution', 'layer_solution', 'fluid_part', 'kinetic_response', 'closure', 'inherited', 'transpor...

safe_ordering_provenance(formalism, state: Optional[Mapping[str, Any]] = None, points: Any = None, **kwargs) -> Optional[Dict[str, Any]]

ordering_provenance that never raises: an advisory record must not cost a run its result.

bootstrap_models(ods: Any, *, models: Sequence[str] = ('sauter', 'redl'), include_stored: bool = True, time_slice: Optional[int] = None, solver_charge: Optional[float] = None, solver_ion_fraction: Optional[float] = None, **options: Any) -> dict[str, Any]

Evaluate each analytic model on ods, alongside whatever it already stores.

input_readiness(ods: Any, *, time: Optional[float] = None, time_index: Optional[int] = None, rho_max: Optional[float] = 0.95, z_eff: Optional[float] = None, ion_index: int = 0, minimum_points: int = 8) -> dict[str, Any]

Whether this ODS is a kinetic state NEO can be run on, and what is missing.

METRICS = ('integrated_relative_difference', 'rms_over_peak', 'peak_relative_difference')

model_agreement(comparison: Mapping[str, Any], tolerances: Mapping[str, float]) -> dict[str, Any]

Apply the caller’s tolerances to a comparison, and echo them back.

model_comparison(models: Mapping[str, Any], *, reference: str = 'neo') -> dict[str, Any]

Measure how far each series sits from reference. Numbers only, no status.

REQUIREMENTS = ('convertible', 'ion_temperature', 'effective_charge', 'radial_extent')

contract(name: str) -> ApproximationContract

The registered contract name; KeyError lists the known ones.

CONTRACTS = <builtins.dict>

GROUPS = ('foundational', 'global', 'profile', 'perturbation', 'layer', 'time_history')

margin_label(margin: float) -> str

The MARGIN_LABELS band of an ordering margin in decades; "unassessed" for NaN.

MARGIN_LABELS = <builtins.tuple>

ORDERING_QUANTITIES = <builtins.dict>

ORDERING_SPACES = <builtins.dict>

ordering_threshold(quantity: str) -> float

The threshold every contract that uses quantity orders it against (order unity unless sourced).

OrderingQuantity(definition: str, scale: str, scope: str, group: str, kernels: Tuple[str, ...]) -> None

What a state column means: its definition, scale, scope, group and the kernels that compute it.

OrderingSpace(name: str, x: str, y: str, title: str, x_reading: Tuple[str, str], y_reading: Tuple[str, str]) -> None

Two ordering quantities read together, and what each side of each means.

required_ordering(quantity: str) -> str

"small" or "large": the side every contract that uses quantity requires.

SYMBOLS = <builtins.dict>

GPEC_COIL_FIELD = <vaft.validation.perturbation_phase.PhaseAuditCriteria>

phase_convention_verdict(audit, *, criteria: PhaseAuditCriteria) -> PhaseConventionVerdict

Decide whether a phase audit settled which reconstruction a file belongs to.

PhaseAuditCriteria(separation_ratio: float, relative_norm: float, name: str = '') -> None

What a toroidal phase audit has to achieve before it settles anything.

PhaseConventionVerdict(status: ValidationStatus, convention: str | None, reason: str) -> None

What one phase audit settled, if anything.

assess_plasma_rogowski(time: np.ndarray, current: np.ndarray, config: PlasmaCurrentQualityConfig | None = None) -> PlasmaCurrentQuality

Judge a calibrated plasma-current Rogowski record over its full span.

PlasmaCurrentQuality(validity: int, reason: str, metrics: dict[str, float] = <factory>) -> None

PlasmaCurrentQualityConfig(quiet_windows: tuple[tuple[float, float], ...] = ((0.0, 0.2), (0.45, 1.0)), reference_window: tuple[float, float] = (0.2, 0.24), max_off_discharge_p2p: float = 120000.0, min_quiet_samples: int = 1000) -> None

Where the record must be quiet, and how quiet.

CHECKS = <builtins.dict>

checks_in(category: str) -> tuple[CheckSpec, ...]

Every registered check of one category, in report order.

CheckSpec(key: str, category: str, unit: str, provider: str, method: str, measure: str = 'rule', tolerance: tuple[float, float] | None = None) -> None

What is stable about one check, independent of any particular result.

describe(key: str) -> CheckSpec

The registry entry for key ("<category>.<check>").

MEASURES = <builtins.dict>

as_evidence(result: Mapping[str, Any], *, key: str, axis: str | None = None, cost: str = 'moderate')

A graded comparison as vaft.validation.credibility.Evidence.

compare_jacobians(reference: Jacobian, candidate: Jacobian, *, output_scale: Sequence[float] | None = None, tolerance: tuple[float, float] | None = None, allow_mixed: bool = False) -> dict[str, Any]

One derivative validated against another of different provenance.

compare_linear_to_monte_carlo(linear_covariance: np.ndarray, sampled: Mapping[str, Any], *, outputs: Sequence[str] | None = None, tolerance: tuple[float, float] | None = None) -> dict[str, Any]

The linear J Sigma J^T against a sampled covariance (#1642 s16).

Jacobian(provenance: str, kind: str, perturbation: str, inputs: tuple[str, ...], outputs: tuple[str, ...], matrix: np.ndarray | None = None, jvp: Callable[[np.ndarray], np.ndarray] | None = None, source: str = '') -> None

A derivative with its provenance: an explicit matrix or a matrix-free product.

JACOBIAN_KINDS = ('forward', 'observation', 'governing_operator')

MIN_SCAN_MEMBERS = 3

PERTURBATIONS = ('physical', 'numerical')

PROVENANCES = ('native', 'finite_difference', 'autodiff', 'analytic')

scan_evidence(report: Mapping[str, Any], *, key: str = 'efit_sensitivity.model_form') -> Any

A targeted-scan ensemble report as model-form evidence on the inference axis.

Q_AT_PSIEDGE_ATOL = 0.0001

SELF_CONSISTENCY_RTOL = 0.2

STABILITY_CHECKS = <builtins.dict>

SURFACE_IDENTITY_ATOL = 1e-06

SURFACE_MATCH_PSI_ATOL = 0.001

validate_stability(*, dcon: Any = None, dcon_check: Any = None, matching: Any = None, matching_check: Any = None) -> dict[str, Any]

Verify one stability case; returns a JSON-serializable report.

STAGE_METRICS = <builtins.dict>

Also exported as vaft.database.production_qa.STAGE_METRICS.

STAGE_PRECONDITIONS = <builtins.dict>

Also exported as vaft.database.production_qa.STAGE_PRECONDITIONS.

aggregate_benchmark(cases: Iterable[Mapping[str, Any]]) -> dict[str, Any]

Residual structure across cases, channels, excitations and machine eras.

ARRAY_CONTRADICTION_FLOOR = 0.01

ARRAY_CONTRADICTION_FRACTION = 0.05

array_contradiction_fractions(channels: Iterable[Any], window: tuple[float, float], *, sigma: float = 20.0, min_members: int = 7, exclude: Iterable[str] = ()) -> dict[str, float]

Fraction of its scorable samples each probe contradicts its array for.

ARRAY_CONTRADICTION_MIN_SCORED = 0.5

ARRAY_CONTRADICTION_SIGMA = 20.0

array_contradictions(channels: Iterable[Any], window: tuple[float, float], *, sigma: float = 20.0, fraction: float = 0.05, min_members: int = 7) -> list[dict[str, Any]]

Probes that contradict their own array over the plasma-free window.

BENCHMARK_CASE_SCHEMA = 5

benchmark_wall_currents(ods: Any, *, resistance_scale: float = 1.0, calibration: 'WallResistanceCalibration | None' = None, dt_sub: float = 5e-05) -> Any

Re-solve the passive-wall currents driven by measured PF currents alone.

Raised when a shot cannot supply a usable plasma-free benchmark case.

coil_drive_check(ods: Any, window: tuple[float, float]) -> dict[str, Any]

Whether the coils drove the vessel inside window at all.

DEFAULT_HISTORY_TIME_CONSTANTS = 3.0

MIN_ARRAY_MEMBERS = 7

MIN_COIL_DRIVE_FRACTION = 0.1

MIN_WITNESSES = 3

PLASMA_FREE_EVIDENCE_SCHEMA = 4

plasma_free_interval(ods: Any, *, reference_fraction: float = 0.2) -> PlasmaFreeInterval

The interval of a shot over which no plasma contributes, with evidence.

The plasma-free stretch of one shot, with the evidence for it.

run_benchmark_case(ods: Any, *, shot: int | None = None, machine_era: str | None = None, window: tuple[float, float] | None = None, per_family: int | None = None, resistance_scale: float = 1.0, calibration: 'WallResistanceCalibration | None' = None, dt_sub: float = 5e-05, n_tau: float = 3.0, min_samples: int = 2) -> dict[str, Any]

Evaluate one plasma-free case: PF-only wall solve, then measured vs model.

solver_history_check(ods: Any, validation_start: float, *, n_tau: float = 3.0, time_constants: np.ndarray | None = None) -> dict[str, Any]

Whether the solver has forgotten its assumed initial state by validation_start.

wall_time_constants(ods: Any) -> np.ndarray

The passive structure’s L/R decay times, slowest first.

aggregate_validity(timed: Iterable[int]) -> int

The scalar validity summarizing a time-resolved one: worst state wins.

Also exported as vaft.validation.imas.aggregate_validity.

codes_at(record: ValidityRecord, *, times: Any | None = None) -> np.ndarray | None

The record’s validity code per sample.

is_condemned(record: ValidityRecord, *, min_validity: int = 0) -> bool

Assessed, and no usable sample anywhere.

is_usable(record: ValidityRecord, *, times: Any | None = None, min_validity: int = 0, default: bool = True) -> bool

Whether any sample at all – or any of times – reaches the floor.

record_from_mask(scalar: int | None, mask: Any | None) -> ValidityRecord

A record for a consumer that holds a boolean usable-sample mask.

usable_fraction(record: ValidityRecord, *, times: Any | None = None, window: Any | None = None, min_validity: int = 0, default: bool = True) -> float

Fraction of the samples in window the record declares usable.

usable_mask(record: ValidityRecord, *, times: Any | None = None, min_validity: int = 0, default: bool = True) -> np.ndarray

Which samples a consumer accepting min_validity may use.

VALIDITY_CERTIFIED = 1

Also exported as vaft.validation.imas.VALIDITY_CERTIFIED.

VALIDITY_INVALID = -2

Also exported as vaft.validation.imas.VALIDITY_INVALID.

VALIDITY_MEANINGS = <builtins.dict>

Also exported as vaft.validation.imas.VALIDITY_MEANINGS.

VALIDITY_SUSPECT = -1

Also exported as vaft.validation.imas.VALIDITY_SUSPECT.

VALIDITY_VALID = 0

Also exported as vaft.validation.imas.VALIDITY_VALID.

ValidityRecord(scalar: int | None, timed: np.ndarray | None, time: np.ndarray | None) -> None

What a datum’s validity nodes say – extracted, not yet interpreted.

DEFAULT_ORDERS = (1, 2, 4, 8, 16, 19, 32, 64, 128, 256)

DEFAULT_RULES = ('tau', 'drive_gain', 'output_weight', 'uniform', 'moments')

drive_set(system: Mapping[str, Any], kinds: Sequence[str] = ('shot', 'single_coil', 'step')) -> dict[str, np.ndarray]

Source waveforms the reduced wall is tested under, (n_times, n_coils) each.

experimental_comparison(ods: Any, selections: Mapping[str, Any], *, window: tuple[float, float] | None = None, per_family: int | None = None, min_wall_authority: float | None = None, dt_sub: float = 5e-05, **system_options: Any) -> dict[str, Any]

The #190 benchmark for the full wall and for reduced walls, side by side.

find_plasma_free_shots(shots: Iterable[int], *, max_plasma_current: float = 5000.0, min_coil_current: float = 1000.0, raw_source: Any = None, loaders: Mapping[str, Any] | None = None) -> list[dict[str, Any]]

Classify shots by whether they can serve as PF-only benchmark cases.

observation_set(ods: Any, *, n_coils: int | None = None, per_family: int | None = None, window: tuple[float, float] | None = None, ring_shrink: float = 0.85, ring_points: int = 64, grid_shape: tuple[int, int] = (25, 33)) -> dict[str, Any]

Wall-column response matrices at every observation the study scores.

order_convergence(ods_or_system: Any, *, rules: Sequence[str] = ('tau', 'drive_gain', 'output_weight', 'uniform', 'moments'), orders: Sequence[int] = (1, 2, 4, 8, 16, 19, 32, 64, 128, 256), drives: Sequence[str] = ('shot', 'single_coil', 'step'), observation: Mapping[str, Any] | None = None, scores_drive: str = 'shot', dt_sub: float = 5e-05, **system_options: Any) -> list[dict[str, Any]]

Full-versus-reduced response error for every (rule, order, drive).

representation_order(rows: Iterable[Mapping[str, Any]], tolerances: Mapping[str, float], *, drive: str = 'shot') -> dict[str, Any]

The smallest retained order meeting tolerances – per rule, per metric and jointly.

wall_system(ods: Any, *, remap_em_coupling: bool = False, gap_factor: float | None = None, **basis_options: Any) -> dict[str, Any]

The wall matrices, the eigenbasis and the shot’s PF drive, once.

The ODS or the requested study cannot be evaluated as asked.