BT_SIGN_VEST_TO_IMAS = <vaft.machine_mapping.conventions.SignTransformation>

calibrate_vest_signal(data: Any, calibration: Mapping[str, Any]) -> np.ndarray

Apply one named calibration to a raw VEST waveform.

DEFAULT_CONSTRAINT_UNCERTAINTIES = <builtins.dict>

DEFAULT_CONSTRAINT_UNCERTAINTY_VECTOR = (0.0001, 0.0001, 0.05, 0.03, 0.01, 0.1, 0.01, 0.1, 0.01)

IMAS_DISCHARGE_SIGNS = <vaft.machine_mapping.conventions.DischargeSignContract>

IP_SIGN_VEST_TO_IMAS = <vaft.machine_mapping.conventions.SignTransformation>

resolve_vest_diagnostic(shot: int, diagnostic: str, *, info_file: str | None = None, with_provenance: bool = False) -> dict[str, Any] | tuple[dict[str, Any], dict[str, Any]]

Return the effective, validated canonical configuration for one diagnostic.

barometry(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> None

barometry_from_raw_database(ods: object, shot: int, tstart: float, tend: float, dt: float, options: dict | None = None) -> None

Also exported as vaft.machine_mapping.barometry_from_raw_database.

vfit_barometry_dynamic(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> None

vfit_barometry_static(ods: object) -> None

camera_visible(ods: Any, shot: int, *, data_root: str | Path | None = None, frame_dir: str | Path | None = None, header_path: str | Path | None = None, near_black_threshold: float | None = None, near_black_percentage: float = 0.98, buffer_frames: int = 2, dark_margin: float = 20.0, plasma_end_s: float | None = None, channel_name: str = 'Fast Camera') -> None

Populate ods with VEST FAST-camera raw frames for shot.

Also exported as vaft.machine_mapping.camera_visible.camera_visible_from_raw_database, vaft.machine_mapping.camera_visible_from_raw_database.

camera_visible_from_frame_dir(shot: int, *, consistency_check: bool = True, **kwargs: Any)

Create and return an OMAS ODS filled with one VEST FAST-camera shot.

Also exported as vaft.machine_mapping.camera_visible_from_frame_dir.

Raised when no non-dark frame can be found in a shot’s frame sequence.

Also exported as vaft.machine_mapping.CameraFrameSelectionError.

CameraHeaderInfo(start_time_ms: float, end_time_ms: float, total_frames: int, exposure_time_s: float | None, exposure_source: str = 'header ShutterSpeed line 25') -> None

Parsed contents of a {shot}_bmp.txt camera header.

DARK_LEVEL_CEILING = 60.0

DARK_LEVEL_FRAME_QUANTILE = 0.25

DARK_LEVEL_PERCENTILE = 1.0

DEFAULT_BUFFER_FRAMES = 2

DEFAULT_DARK_MARGIN = 20.0

DEFAULT_NEAR_BLACK_PERCENTAGE = 0.98

DEFAULT_NEAR_BLACK_THRESHOLD = 35

estimate_dark_level(frames: Sequence[np.ndarray | None], *, percentile: float = 1.0) -> float

Return a shot’s sensor dark level (see DARK_LEVEL_FRAME_QUANTILE).

Also exported as vaft.machine_mapping.estimate_dark_level.

find_valid_frame_interval(frames: Sequence[np.ndarray | None], *, buffer_frames: int = 2, threshold: float | None = None, percentage: float = 0.98, dark_margin: float = 20.0) -> tuple[int, int]

Return only (onset, end) of select_valid_frames.

Also exported as vaft.machine_mapping.find_valid_frame_interval.

FIXED_FRAME_SELECTION_RULE = 'fixed-v1'

FRAME_SELECTION_RULE = 'background-relative-v3'

frame_time_ms(frame_index: int, total_frames: int, start_time_ms: float, end_time_ms: float) -> float

Linear-interpolate a frame’s time, reusing the donor bmp_arranger.py formula.

Also exported as vaft.machine_mapping.frame_time_ms.

FrameSelection(onset: int, end: int, total_frames: int, threshold: float, percentage: float, buffer_frames: int, rule: str, first_retained: int, last_retained: int, dark_level: float | None = None, peak_threshold: float | None = None, reference_frames: int | None = None, reference_sigma: float | None = None, plasma_end_frame: int | None = None) -> None

The outcome of frame selection on one shot’s frame sequence.

is_near_black(image: np.ndarray, *, threshold: float = 35, percentage: float = 0.98) -> bool

Return whether image is near-black, reusing the donor is_near_black formula.

Also exported as vaft.machine_mapping.is_near_black.

narrow_image_storage(ods: Any) -> None

Re-state every camera_visible image_raw frame as int32.

Also exported as vaft.machine_mapping.narrow_image_storage.

parse_frame_selection(comment: str) -> dict[str, Any] | None

Read the frame selection back out of a camera_visible IDS comment.

Also exported as vaft.machine_mapping.parse_frame_selection.

save_camera_visible_ods(output_path: str | Path, shot: int, **kwargs: Any)

Build and save a camera_visible ODS. The file extension selects OMAS format.

Also exported as vaft.machine_mapping.save_camera_visible_ods.

select_valid_frames(frames: Sequence[np.ndarray | None], *, buffer_frames: int = 2, threshold: float | None = None, percentage: float = 0.98, dark_margin: float = 20.0, rule: str = 'background-relative-v3', plasma_end_frame: int | None = None) -> FrameSelection

Find the [onset, end] interval of plasma-lit frames, with padding.

Also exported as vaft.machine_mapping.select_valid_frames.

vfit_camera_visible_dynamic(ods: Any, *, images: Sequence[np.ndarray], times_s: Sequence[float]) -> None

Fill dynamic camera_visible frame data: reindexed frame[:] array.

vfit_camera_visible_static(ods: Any, *, lines_n: int, columns_n: int, exposure_time_s: float | None = None, channel_name: str = 'Fast Camera', source: str | None = None, comment_extra: str = '') -> None

Fill static camera_visible IDS metadata: channel/detector geometry-free facts.

charge_exchange(ods: Any, shotnumber: int, options: str = 'ces', data_root: str | Path | None = None, mat_file: str | Path | None = None) -> None

read_charge_exchange_ces_mat(ods: Any, shotnumber: int | None = None, mat_file: str | Path | None = None, data_root: str | Path | None = None, line: str = 'C3+', default_time_s: float = 0.3) -> None

read_doppler_ids_mat(ods: Any, shotnumber: int | None = None, mat_file: str | Path | None = None, data_root: str | Path | None = None, line: str = 'C3+') -> None

read_doppler_profile(ods: Any, shotnumber: int, line: str = 'C3+', data_root: str | Path | None = None) -> None

Also exported as vaft.machine_mapping.read_doppler_profile.

read_doppler_single(ods: Any, shotnumber: int, data_root: str | Path | None = None) -> None

Also exported as vaft.machine_mapping.read_doppler_single.

vfit_charge_exchange(ods: Any, shotnumber: int, options: str = 'ces', data_root: str | Path | None = None, mat_file: str | Path | None = None) -> None

vfit_ion_doppler_spectroscophy(ods: Any, shotnumber: int, options: str = 'ces', data_root: str | Path | None = None, mat_file: str | Path | None = None) -> None

CLASSICAL_MODEL_DESCRIPTION = 'Classical collisional transport: Braginskii perpendicular conductive heat flux (VAFT private identifier; the data di...

CLASSICAL_MODEL_INDEX = -1

CLASSICAL_MODEL_NAME = 'classical'

classical_parameters(text: Any) -> Optional[dict[str, Any]]

Parse a classical entry’s code.parameters.

core_transport_from_classical(ods: ODS, records: Iterable[dict], profile: Any, *, time: float, state_identity: str, efit_lineage: Optional[str] = None) -> dict[str, Any]

Write classical heat fluxes and diffusivities into core_transport.

FLUX_MULTIPLIER = 0.0

formulation_sha256(model: Any) -> str

SHA-256 of a record’s model mapping, keys sorted: the formulation identity.

apply_coil_excitation(ods: ODS, excitations: Sequence[CoilExcitation], time_s: Sequence[float] | float = 0.0) -> None

Overlay run-specific per-sector currents onto the mapped geometry.

coils_non_axisymmetric(ods: ODS, source: str | None = None, options: Optional[Mapping[str, Any]] = None) -> None

Populate static VEST 3D coil geometry.

sector_identifier(coil_set: CoilSet3D, sector: int) -> str

The stable identifier of one sector coil, e.g. VEST_3D_MID_01.

coil_set_from_dat(path: str | Path, name: str, *, identifier: str | None = None, sector_angles_deg: Sequence[float] | None = None, description: str = '', provenance: str = '') -> CoilSet3D

Read a GPEC .dat file (header GPEC_COIL_DAT_HEADER) as a CoilSet3D.

Also exported as vaft.machine_mapping.coil_geometry_3d.coil_set_from_dat (deprecated).

coil_set_from_xyz_loops(name: str, loops: Sequence[np.ndarray], *, turns: float, identifier: str | None = None, sector_angles_deg: Sequence[float] | None = None, description: str = '', provenance: str = '', dat_path: str | Path | None = None) -> CoilSet3D

Build a CoilSet3D from Cartesian loops of any machine.

Also exported as vaft.machine_mapping.coil_geometry_3d.coil_set_from_xyz_loops (deprecated).

CoilExcitation(coil_set: str, currents_a: tuple[float, ...]) -> None

Run-specific excitation of one coil set: amperes per turn per sector.

Also exported as vaft.machine_mapping.coil_geometry_3d.CoilExcitation (deprecated).

CoilFilament(points_xyz: np.ndarray) -> None

One physical winding path (one toroidal sector).

Also exported as vaft.machine_mapping.coil_geometry_3d.CoilFilament (deprecated).

CoilSet3D(name: str, identifier: str, turns: float, filaments: tuple[CoilFilament, ...], sector_angles_deg: tuple[float, ...], dat_path: Path | None, description: str, provenance: str) -> None

A canonical coil set: parsed geometry plus its static metadata.

Also exported as vaft.machine_mapping.coil_geometry_3d.CoilSet3D (deprecated).

CoilSetSpec(identifier: str, dat_file: str, description: str, z_range: tuple[float, float], sector_angles_deg: tuple[float, ...] = (0.0, 60.0, 120.0, 180.0, 240.0, 300.0), provenance: str = 'GPEC-format coil description; geometry and 20-turn interpretation reviewed with 3D coil developer Gwang-geun Seo.') -> None

Static metadata for one canonical VEST 3D coil set.

Also exported as vaft.machine_mapping.coil_geometry_3d.CoilSetSpec (deprecated).

GPEC_COIL_DAT_HEADER = ('ncoil', 'nsec', 'npts', 'nw')

Also exported as vaft.machine_mapping.coil_geometry_3d.GPEC_COIL_DAT_HEADER (deprecated).

load_vest_3d_coil_config(data_root: str | Path | None = None, *, coil_sets: Sequence[str] | None = None) -> Vest3DCoilConfig

Load and validate the canonical VEST 3D coil configuration.

Also exported as vaft.machine_mapping.coil_geometry_3d.load_vest_3d_coil_config (deprecated).

parse_gpec_coil_dat(path: str | Path) -> tuple[int, int, int, float, np.ndarray]

Parse a GPEC coil .dat file.

Also exported as vaft.machine_mapping.coil_geometry_3d.parse_gpec_coil_dat (deprecated).

Vest3DCoilConfig(coil_sets: Mapping[str, CoilSet3D]) -> None

The full canonical VEST 3D coil configuration.

Also exported as vaft.machine_mapping.coil_geometry_3d.Vest3DCoilConfig (deprecated).

VEST_3D_COIL_SETS = <builtins.dict>

Also exported as vaft.machine_mapping.coil_geometry_3d.VEST_3D_COIL_SETS (deprecated).

BT_SIGN_VEST_TO_IMAS = <vaft.machine_mapping.conventions.SignTransformation>

clock_angle_to_toroidal_angle(clock_angle_deg: float) -> float

A VEST clock angle in degrees to an IMAS toroidal angle in radians.

cylindrical_to_cartesian(r: Any, phi: Any, z: Any) -> tuple[np.ndarray, np.ndarray, np.ndarray]

Machine (R, phi, Z) to right-handed Cartesian (x, y, z), all in metres.

DischargeSignContract(ip: int, b0: int) -> None

The signs a normal VEST discharge must carry in the IMAS convention.

Also exported as vaft.machine_mapping.DischargeSignContract.

equilibrium_orientation_is_resolved() -> bool

True only when both transformations rest on an oriented reference.

Also exported as vaft.machine_mapping.equilibrium_orientation_is_resolved.

expected_q_sign(cocos_index: int, contract: DischargeSignContract | None = None) -> int

The sign q must carry in cocos_index under the discharge contract.

Also exported as vaft.machine_mapping.expected_q_sign.

gpec_coil_directions(contract: DischargeSignContract) -> dict[str, str]

The coil.in direction words a machine’s sign contract implies.

GPEC_DIRECTION_WORDS = <builtins.dict>

IMAS_DISCHARGE_SIGNS = <vaft.machine_mapping.conventions.DischargeSignContract>

IP_SIGN_VEST_TO_IMAS = <vaft.machine_mapping.conventions.SignTransformation>

port_toroidal_angle(clock_hour: float) -> float

The IMAS toroidal angle of a port, in radians.

sign_transformations() -> tuple[SignTransformation, ...]

Both transformations, in the order they appear in the COCOS product.

Also exported as vaft.machine_mapping.sign_transformations.

SignTransformation(quantity: str, sign: int, confirmed: bool, vest_native_source: str, evidence: str, needed_to_confirm: str) -> None

One VEST-native to IMAS sign transformation and its standing.

Also exported as vaft.machine_mapping.SignTransformation.

to_imas_discharge_signs(ip_vest, bt_vest)

Carry a VEST-native (ip, bt) discharge pair into the IMAS convention.

Also exported as vaft.machine_mapping.to_imas_discharge_signs.

vest_clock_angle(clock_hour: float) -> float

The VEST clock angle of a port, in degrees, clockwise-positive.

VEST_GPEC_COIL_DIRECTIONS = <builtins.dict>

VEST_PORT_CLOCK_DEGREES = 30.0

VEST_PORT_CLOCK_SENSE = 'clockwise-positive viewed from above'

VEST_PORT_CLOCK_TO_PHI_SIGN = -1

VEST_PORT_CLOCK_ZERO = "12 o'clock (large main door)"

classify_ti_record(record: Any) -> str

Which of TI_RECORD_KINDS an ion temperature_fit.parameters record is.

CoreProfilesPolicy(shot: Optional[int], coordinate: str, ti_te_ratio: float, ti_te_ratio_sigma: float, ti_te_ratio_status: str, impurity_species: tuple[str, ...], impurity_fractions: Mapping[str, float], impurity_status: Mapping[str, str], provenance: Mapping[str, Any], source: str = 'vest.yaml:diagnostics.core_profiles', impurity_model: Optional[Mapping[str, Any]] = None) -> None

The resolved VEST kinetic-profile policy for one shot.

Also exported as vaft.machine_mapping.CoreProfilesPolicy.

INFERRED_TI_ORIGIN = 'inferred'

inferred_ti_text(method: str = 'equilibrium_pressure_partition') -> str

The record a writer stores beside an ion temperature it inferred, not measured.

policy_for_ods(ods, shot: Optional[int] = None, *, info_file: str | None = None) -> CoreProfilesPolicy

The policy for this ODS: shot if given, else the ODS’s own, else the base revision.

POLICY_STATUSES = ('assumed', 'measured', 'inferred')

ti_record_fields(record: Any) -> dict[str, str]

The key=value (or key: value) fields of a provenance record, lower-cased keys.

TI_RECORD_KINDS = ('measured', 'assumed', 'inferred', 'unknown')

vest_core_profiles_policy(shot: Optional[int], *, info_file: str | None = None) -> CoreProfilesPolicy

Resolve the VEST kinetic-profile policy for shot from vest.yaml.

Also exported as vaft.machine_mapping.vest_core_profiles_policy.

vest_impurity_model(shot: Optional[int], *, info_file: str | None = None) -> Optional[Mapping[str, Any]]

The VEST impurity-composition preset for shot, or None for an era without one.

core_sources_from_nubeam(ods: ODS, result: Any, *, time: float = 0.0, time_index: int = 0, rho: Optional[Sequence[float]] = None, b0: Optional[float] = None) -> dict[str, Any]

Write a NUBEAM result into core_sources as an NBI source term.

NBI_SOURCE_INDEX = 2

dataset_description(ods: Any, source: int, options: dict | None = None) -> None

Compatibility wrapper that maps legacy calls onto the canonical builder.

dataset_description_from_raw_database(ods: Any, shot: int, options: dict | None = None) -> None

Build dataset_description and, when live SQL access is available, fill pulse_time_begin from the authoritative shot table.

Also exported as vaft.machine_mapping.dataset_description_from_raw_database.

vfit_dataset_description(ods: Any, shot: int, run: int, machine: str = 'VEST', pulse_type: str = 'pulse', user: str | None = None, description: str | None = None, pulse_datetime: datetime | None = None) -> None

Populate canonical dataset metadata for a VEST pulse/run pair.

DIAGNOSTIC_FAULTS_KEY = 'diagnostic_faults'

FAULT_KINDS = frozenset({'calibration_unverified', 'no_signal', 'railed'})

known_diagnostic_faults(shot: int, *, info_file: str | None = None) -> list[dict[str, Any]]

The recorded diagnostic faults in force on shot, in file order.

distributions_from_nubeam(ods: ODS, result: Any, *, time: float = 0.0, time_index: int = 0, rho: Optional[Sequence[float]] = None, species: Optional[Sequence[str]] = None) -> dict[str, Any]

Write a NUBEAM result into distributions as the fast-ion population.

NBI_PROCESS_INDEX = 1

ec_launchers(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> None

Map the VEST 6 kW 2.45 GHz ECH power and launch condition into ec_launchers.

ec_launchers_dynamic(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> None

ec_launchers_from_raw_database(ods: object, shot: int, tstart: float, tend: float, dt: float, options: dict | None = None) -> None

Also exported as vaft.machine_mapping.ec_launchers_from_raw_database.

ec_launchers_geometry(ods: object, geometry: Mapping[str, Any], time: Any) -> None

Write one beam’s fixed launch condition on time into ec_launchers.beam.0.

ec_launchers_static(ods: object) -> None

resolve_ec_launcher_geometry(shot: int, beam: str = 'ech_6kw', *, info_file: str | None = None) -> dict[str, Any] | None

The launch condition of one EC beam for shot, or None outside its era.

steering_angles_from_direction(direction: Any) -> tuple[float, float]

IMAS (steering_angle_pol, steering_angle_tor) of a propagation direction.

vest_ec_power(shot: int, *, raw_source: raw_db.RawSource | None = None, time: Any | None = None) -> dict[str, Any]

Calibrated, masked 6 kW ECH forward and reflected power for one shot.

Also exported as vaft.machine_mapping.vest_ec_power.

EdgeQEstimatePolicy(scaling: str, configuration: str, default_shape: Mapping[str, float], status: Mapping[str, str], provenance: Mapping[str, str], source: str = 'vest.yaml:edge_q_estimate') -> None

The stated scaling and stand-in shape for an edge-q estimate.

vest_edge_q_estimate_policy(*, info_file: str | None = None) -> EdgeQEstimatePolicy

The scaling and stand-in shape VEST’s edge-q estimates use.

detect_energised_circuits(pf_active: Any, policy: EFITCoilsetPolicy, *, window: tuple[float, float]) -> EnergisationDecision

Decide which circuits carried current, from the record when there is one.

EFITCoilsetPolicy(group_names: tuple[str, ...], source_circuit: Mapping[str, str], segment_edges: Mapping[str, tuple[float, ...]], ties: tuple[tuple[str, ...], ...], declared_energised: tuple[str, ...], energisation_threshold: float, coil_current_error_floor: float, status: Mapping[str, str], provenance: Mapping[str, str], source: str = 'vest.yaml:efit_coilset') -> None

The resolved coilset: the groups, their circuits, and the series ties.

EnergisationDecision(energised: tuple[str, ...], peak_abs_current: Mapping[str, float], declared: tuple[str, ...], threshold: float, window: tuple[float, float], measured: bool) -> None

Which circuits an experiment actually energised, and how that was known.

vest_efit_coilset_policy(*, info_file: str | None = None) -> EFITCoilsetPolicy

The coilset EFIT’s table and k-file describe.

efund_geometry_from_static(ods: Any, *, manifest: Mapping[str, Any] | None = None, coilset: Any | None = None) -> EFUNDGeometry

Project a static VEST ODS onto EFUND’s description.

efund_probe_angle_deg(poloidal_angle_rad: float) -> float

EFUND amp2 (degrees) for a DD poloidal_angle (radians).

EFUNDGeometry(fcoil_r: np.ndarray, fcoil_z: np.ndarray, fcoil_w: np.ndarray, fcoil_h: np.ndarray, fcoil_a: np.ndarray, fcoil_a2: np.ndarray, fcoil_group: np.ndarray, fcoil_turns: np.ndarray, group_names: tuple[str, ...], group_turns: np.ndarray, vessel_r: np.ndarray, vessel_z: np.ndarray, vessel_w: np.ndarray, vessel_h: np.ndarray, vessel_a: np.ndarray, vessel_a2: np.ndarray, vessel_group: np.ndarray, vessel_resistance: np.ndarray, vessel_names: tuple[str, ...], loop_r: np.ndarray, loop_z: np.ndarray, loop_names: tuple[str, ...], probe_r: np.ndarray, probe_z: np.ndarray, probe_angle_deg: np.ndarray, probe_length: np.ndarray, probe_names: tuple[str, ...], machine: dict[str, Any] = <factory>) -> None

EFUND’s rectangle description of one machine era.

equilibrium_probe_count(ods: Any) -> int

The B-pol probes EFIT’s geometry represents.

rectangle_from_outline(r: Sequence[float], z: Sequence[float]) -> tuple[float, float, float, float]

(centre_r, centre_z, width, height) of an axis-aligned 4-point outline.

vest_acceptance_envelope(ods: Any, *, nw: int = 129, rleft: float = 0.05, rright: float = 1.2, resolved_cells: int = 5, base: Any | None = None, virial_checks: bool = False) -> Any

Derive EFIT’s &incheck geometric bounds from the machine itself.

calculate_em_coupling_from_raw_database(ods: Any, shot: int | dict | None = None, options: dict | None = None) -> None

Populate coupling for a shot through the raw-mapping entry point.

Also exported as vaft.machine_mapping.calculate_em_coupling_from_raw_database.

COUPLING_PROVENANCE_KEY = 'provenance'

The ODS’s PF coils or passive loops are not the ones the packaged coupling was computed for.

DEFAULT_REFERENCE_ODS = <pathlib.PosixPath>

DEFAULT_STATIC_GEOMETRY = <pathlib.PosixPath>

DEFAULT_VERSIONED_COUPLING = <pathlib.PosixPath>

em_coupling(ods: Any, source: str | Path | None = None, options: dict | None = None, *, shot: int | None = None) -> None

Populate canonical, PF- and wall-versioned VEST coupling data.

load_versioned_coupling_provenance(path: str | Path = PosixPath('/tmp/vaft-docs-_y04_kb0/src/development/vaft/data/geometry/VEST_em_coupling_pf_versions.npz')) -> dict | None

The provenance record stored in the packaged coupling asset, or None.

PASSIVE_COUPLING_ASYMMETRY_REJECT = 0.1

PASSIVE_COUPLING_ASYMMETRY_ROUNDOFF = 1e-12

PASSIVE_COUPLING_ASYMMETRY_WARN = 1e-06

equilibrium(ods: ODS, source: Sequence[str | Path], options: Optional[dict[str, Any]] = None) -> None

Merge a sequence of GEQDSK files into ods.equilibrium.

EquilibriumRegimePolicy(phase_rule: str, flat_fraction: float, slope_fraction: float, vacuum_current_amperes: float, topology_labels: Mapping[str, str], status: Mapping[str, str], provenance: Mapping[str, str], current_edges: tuple[float, ...] = (), rate_edges: tuple[float, ...] = (), rate_labels: tuple[str, ...] = (), version: int = 1, source: str = 'vest.yaml:equilibrium_regime') -> None

The stated rule for putting a slice in a cohort.

vest_equilibrium_regime_policy(*, info_file: str | None = None) -> EquilibriumRegimePolicy

The regime rule VEST reconstructions are classified by.

b_field_lines_from_flare_connection(ods: ODS, product, *, time: float, boundary=None, code_name: str = 'FLARE', code_repository: str = '') -> int

Map one FLARE connection-length product at a fixed angle into the IDS.

divertor_incident_fractions_from_flare_footprints(ods: ODS, proxies: Mapping[str, Any], *, time: float, divertor: int = 0, divertor_name: str = '', code_name: str = 'FLARE', code_repository: str = '') -> dict[str, float]

Integrate one proxy per target, and write their shares.

RECTANGULAR_GRID_TYPE = 1

write_b_field_lines(ods: ODS, *, grid_r, grid_z, starting_r, starting_z, lengths, open_fraction: Optional[float], time: float, phi_deg: float, provenance: Mapping[str, Any] | None = None, code_name: str = 'FLARE', code_repository: str = '') -> int

Write one traced (R, z) plane into plasma_initiation.b_field_lines.

write_divertor_incident_fractions(ods: ODS, *, fractions: Mapping[str, float], time: float, divertor: int = 0, divertor_name: str = '', provenance: str = '', code_name: str = 'FLARE', code_repository: str = '') -> None

Write one divertor’s per-target incident fractions.

gpec_ideal(ods: ODS, source: str, options: Optional[dict] = None) -> dict[int, dict[str, Any]]

Map one ideal-GPEC run directory into mhd_linear.

ANOMALOUS_MODEL_INDEX = 6

CGYRO_MODEL_DESCRIPTION = 'Turbulent transport, from a nonlinear local CGYRO gyrokinetic simulation'

conversion_factors(local: Any, outputs: Any) -> Optional[dict[str, float]]

The CGYRO -> DD rescalings for one run, or None when they cannot be formed.

core_transport_from_cgyro(ods, surfaces, profile, *, time=0.0, time_index=0, state_identity=None, realization_id=None, configuration=None, provenance=None)

Map one radial realization transactionally, preserving scientific identity.

FREQUENCY_SIGN_CONVENTION = 'ion_diamagnetic_negative'

geometric_angle(theta: Any, local: Any) -> np.ndarray

Miller angle (extended) -> the DD’s geometric poloidal angle, clockwise.

gyrokinetics_local_from_cgyro(ods: ODS, local: Any, outputs: Any, *, provenance: Optional[Mapping[str, Any]] = None, time: Optional[float] = None, flux_window: Optional[tuple[float, float]] = None, locality: Optional[Mapping[str, Any]] = None) -> dict[str, Any]

Write one CGYRO run into ods['gyrokinetics_local'].

gyrokinetics_local_from_tglf(ods: ODS, local: Any, outputs: Any, *, parameters: Optional[Mapping[str, Any]] = None, provenance: Optional[Mapping[str, Any]] = None, time: Optional[float] = None) -> dict[str, Any]

Write one TGLF surface into ods['gyrokinetics_local'], per TGLF_MAPPING_AUDIT.

IDS = 'gyrokinetics_local'

MAPPING_AUDIT = <builtins.dict>

merge_linear_scan(odss: Iterable[ODS], *, rtol: float = 1e-09) -> ODS

Combine single-k_y linear runs of one local problem into one IDS.

TGLF_MAPPING_AUDIT = <builtins.dict>

hard_x_rays(ods: Any, shot: int, *, data_root: str | Path | None = None, flux_file: str | Path | None = None, time_offset: float | None = None, time_reference: str = 'auto', trigger_settings_path: str | Path | None = None) -> None

Map one archived VEST HXR Eflux record into hard_x_rays.

hard_x_rays_from_flux_csv(shot: int, *, consistency_check: bool = True, **kwargs: Any)

Create a shot-level ODS holding one HXR Eflux record.

Also exported as vaft.machine_mapping.hard_x_rays_from_flux_csv.

HXR_BIN_WIDTH_S = 0.001

HXR_DAQ_LABEL = '17592'

hxr_file_candidates(shot: int, variant: str = 'Eflux', data_root: str | Path | None = None, daq_label: str = '17592') -> list[Path]

Candidate locations of one archived HXR CSV.

HXR_FLUX_ENERGIES_KEV = (40.0, 60.0, 80.0, 110.0, 140.0)

HXR_MODEL_ENERGIES_KEV = (40.0, 60.0, 80.0, 110.0, 140.0, 210.0)

HXR_SCINTILLATOR_THICKNESS_M = (1e-05, 0.0001, 0.0002, 0.0003)

HXR_UNFOLDING_MODEL = <builtins.dict>

HXRTimeAlignment(offset_seconds: float, source: str, detail: str) -> None

Resolved relation between the trigger-relative HXR axis and the shot clock.

load_hxr_flux_csv(path: str | Path) -> np.ndarray

Read one Eflux CSV as a (bins, energies) array.

Also exported as vaft.machine_mapping.load_hxr_flux_csv.

load_hxr_raw_csv(path: str | Path, *, sample_rate: float = 976562.5) -> dict[str, np.ndarray]

Read a legacy HXR raw CSV (one row per scintillator channel).

Also exported as vaft.machine_mapping.load_hxr_raw_csv.

resolve_hxr_time_alignment(shot: int, *, trigger_settings_path: str | Path | None = None) -> HXRTimeAlignment

Resolve the machine-time offset of an HXR record.

Also exported as vaft.machine_mapping.resolve_hxr_time_alignment.

HALL_PROBE_TYPE_INDEX = 3

impa(ods: Any, shot: int, tstart: float = 0.26, tend: float = 0.36, dt: float = 4e-05, *, raw_source: raw_db.RawSource | None = None, reference_shot: int | None = None) -> dict[str, Any]

Map one shot’s calibrated IMPA measurement into the magnetics IDS.

impa_expected_fields(shot: int, config: Mapping[str, Any] | None = None) -> list[int]

Return the raw Hall field codes this shot’s era actually wires.

impa_from_raw_database(ods: Any, shot: int, tstart: float = 0.26, tend: float = 0.36, dt: float = 4e-05, options: dict | None = None) -> dict[str, Any]

Raw-database entry point matching the other VEST diagnostic mappers.

Also exported as vaft.machine_mapping.impa_from_raw_database.

IMPA_IDENTIFIER_PREFIX = 'impa:'

impa_probe_indices(ods: Any, node: str | None = None) -> list[int]

Return the probe indices holding IMPA channels, located by identifier.

Also exported as vaft.machine_mapping.impa_probe_indices.

impa_probe_node(ods: Any) -> str | None

Return the primary magnetics node holding this ODS’s IMPA channels.

IMPA_PROBE_NODES = ('magnetics.b_field_tor_probe', 'magnetics.b_field_pol_probe')

load_impa_inputs(shot: int, config: Mapping[str, Any], *, raw_source: raw_db.RawSource | None = None) -> dict[str, Any]

Load the raw IMPA channels plus the same-shot TF, Ip and PF context.

process_impa_shot(shot: int, *, raw_source: raw_db.RawSource | None = None, config: Mapping[str, Any] | None = None, reference_shot: int | None = None) -> tuple[ImpaResult, dict[str, Any]]

Resolve configuration, load one shot and run the IMPA pipeline.

resolve_impa_config(shot: int, info_file: str | None = None) -> dict[str, Any]

Return the IMPA block for shot, merging shot overrides over defaults.

interferometer(ods: Any, shot: int, *, data_root: str | Path | None = None, mat_file_94ghz: str | Path | None = None, mat_file_282ghz: str | Path | None = None, compute_line_average: bool = False, info_file: str | None = None) -> None

Populate ods with both interferometer systems in one heterogeneous-time IDS.

interferometer_282ghz(ods: Any, shot: int, *, data_root: str | Path | None = None, mat_file: str | Path | None = None, compute_line_average: bool = False, info_file: str | None = None) -> None

Populate ods with the 282 GHz vertical interferometer (own occurrence).

Also exported as vaft.machine_mapping.interferometer_282ghz.

interferometer_94ghz(ods: Any, shot: int, *, data_root: str | Path | None = None, mat_file: str | Path | None = None, compute_line_average: bool = False, info_file: str | None = None) -> None

Populate ods with the 94 GHz horizontal interferometer (own occurrence).

Also exported as vaft.machine_mapping.interferometer_94ghz.

vfit_interferometer_282ghz_dynamic(ods: Any, shot: int, *, data_root: str | Path | None = None, mat_file: str | Path | None = None, compute_line_average: bool = False, info_file: str | None = None) -> None

vfit_interferometer_282ghz_static(ods: Any, *, info_file: str | None = None) -> None

vfit_interferometer_94ghz_dynamic(ods: Any, shot: int, *, data_root: str | Path | None = None, mat_file: str | Path | None = None, compute_line_average: bool = False, info_file: str | None = None) -> None

vfit_interferometer_94ghz_static(ods: Any, *, info_file: str | None = None) -> None

apply_langmuir_probe_measured_positions(ods: object, shot: int, *, csv_path: str | Path | None = None) -> None

Update embedded.{0,1}.position.r from the measured-position CSV.

Also exported as vaft.machine_mapping.apply_langmuir_probe_measured_positions.

langmuir_probes(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None, mid_r: float | None = None, upper_r: float | None = None, position_csv_path: str | Path | None = None, report: dict[str, Any] | None = None) -> None

langmuir_probes_from_raw_database(ods: object, shot: int, tstart: float, tend: float, dt: float, options: dict | None = None) -> None

Also exported as vaft.machine_mapping.langmuir_probes_from_raw_database.

Raised when the VEST langmuir_probes configuration cannot resolve a shot.

The shot falls in a documented gap between bias/tip-geometry eras.

resolve_langmuir_probe_config(assembly_key: str, shot: int, info_file: str | None = None) -> dict[str, Any]

Return the langmuir_probes.<assembly_key> block, with shot overrides merged.

vfit_langmuir_probes_dynamic(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None, mid_r: float | None = None, upper_r: float | None = None, report: dict[str, Any] | None = None) -> None

Map both triple-probe assemblies onto langmuir_probes.embedded.

vfit_langmuir_probes_static(ods: object) -> None

b_field_pol_probe_from_raw_database(ods: object, shot: int, *, tstart: float = 0.26, tend: float = 0.36, dt: float = 4e-05, processing_config: VestMagneticsProcessingConfig | None = None, raw_source: raw_db.RawSource | None = None) -> None

Map calibrated and raw poloidal-field probe signals and metadata.

Also exported as vaft.machine_mapping.b_field_pol_probe_from_raw_database.

detect_plasma_window(shot: int, ip_time: np.ndarray, ip: np.ndarray, raw_source: raw_db.RawSource | None = None, *, policy: PlasmaTimingPolicy | None = None) -> PlasmaWindowChoice

The plasma window for raw-side consumers, from the shared timing policy.

Also exported as vaft.machine_mapping.detect_plasma_window.

diamagnetic_baseline_window(method_name: str | None) -> tuple[float, float] | None

The plasma window a stored diamagnetic flux was baselined over, read back from its method_name.

diamagnetic_field_for_shot(shot: int) -> int | None

The raw field carrying the diamagnetic (TF Rogowski) signal, or None.

diamagnetic_flux_rogowski_coil_from_raw_database(ods: object, shot: int, *, tstart: float = 0.26, tend: float = 0.36, dt: float = 4e-05, processing_config: VestMagneticsProcessingConfig | None = None, raw_source: raw_db.RawSource | None = None) -> None

Map the diamagnetic Rogowski sensor and flux, without plasma current.

Also exported as vaft.machine_mapping.diamagnetic_flux_rogowski_coil_from_raw_database.

diamagnetic_saturation_report(shot: int, *, raw_source: raw_db.RawSource | None = None, plasma_start: float | None = None, plasma_end: float | None = None) -> dict[str, Any]

Report acquisition-limit saturation on the diamagnetic Rogowski channel.

Also exported as vaft.machine_mapping.diamagnetic_saturation_report.

equilibrium_magnetics_processing_config(shot: int) -> VestMagneticsProcessingConfig

Build the processing config for shot from its resolved vest.yaml era.

fluctuation_mirnov_channel_definitions(shot: int = 0) -> tuple[dict[str, Any], ...]

Return ordered VEST outboard fluctuation-Mirnov channel metadata for provenance.

fluctuation_mirnov_gain_by_identifier() -> dict[str, float]

Per-channel voltage gains for raw-Mirnov probes, keyed by ODS identifier.

fluctuation_mirnov_probe_indices(ods: object, *, shot: int = 0) -> dict[str, int]

Map fluctuation-Mirnov identifiers to their b_field_pol_probe indices.

flux_loop_from_raw_database(ods: object, shot: int, *, tstart: float = 0.26, tend: float = 0.36, dt: float = 4e-05, processing_config: VestMagneticsProcessingConfig | None = None, raw_source: raw_db.RawSource | None = None) -> None

Map calibrated flux-loop signals and metadata from the VEST archive.

Also exported as vaft.machine_mapping.flux_loop_from_raw_database.

ip_rogowski_coil_from_raw_database(ods: object, shot: int, *, tstart: float = 0.26, tend: float = 0.36, dt: float = 4e-05, processing_config: VestMagneticsProcessingConfig | None = None, raw_source: raw_db.RawSource | None = None) -> None

Map the plasma-current Rogowski sensor and its derived plasma current.

Also exported as vaft.machine_mapping.ip_rogowski_coil_from_raw_database.

known_magnetics_faults(shot: int) -> dict[tuple[str, int], str]

{(kind, index): reason} for channels vest.yaml records as broken on shot.

Also exported as vaft.machine_mapping.known_magnetics_faults.

LIMITER_SHUNT_BASELINE_WINDOW = (0.0, 0.2)

LIMITER_SHUNT_CHANNELS = <builtins.tuple>

LIMITER_SHUNT_RESISTANCE = 0.1

magnetics(ods: object, shot: int, tstart: float = 0.26, tend: float = 0.36, dt: float = 4e-05, processing_config: VestMagneticsProcessingConfig | None = None, *, raw_source: raw_db.RawSource | None = None) -> None

Canonical machine_mapping entry point for the magnetics IDS.

magnetics_from_raw_database(ods: object, shot: int, tstart: float, tend: float, dt: float, options: dict | None = None) -> None

Also exported as vaft.machine_mapping.magnetics_from_raw_database.

magnetics_wiring_for_shot(shot: int) -> MagneticsWiring

Resolve the shot’s probe wiring from vest.yaml (issue #956).

Also exported as vaft.machine_mapping.magnetics_wiring_for_shot.

MagneticsWiring(layout: str, channels: tuple[Mapping[str, Any], ...], overrides: tuple[Mapping[str, Any], ...], provenance: Mapping[str, Any]) -> None

Which raw field and calibration feed each equilibrium probe for a shot.

PlasmaWindowChoice(start: float, end: float, source: str, flags: tuple[str, ...] = (), evidence: Mapping[str, Any] = <factory>) -> None

The plasma window a raw-side consumer uses, and where it came from.

Also exported as vaft.machine_mapping.PlasmaWindowChoice.

require_supported_magnetics_geometry(shot: int | None) -> None

Raise if shot requires a magnetics layout VAFT has no record of.

select_fluctuation_mirnov_channels(*, role: str | None = 'fluctuation', toroidal_angle_deg: float | Sequence[float] | None = None, sub_array: str | None = None, z_range: tuple[float, float] | None = None, shot: int = 0) -> tuple[dict[str, Any], ...]

Select fluctuation-Mirnov channels by role and geometry, in canonical order.

toroidal_array_for_shot(shot: int) -> dict[str, Any]

Which toroidal Mirnov array shot has, and what it can resolve.

toroidal_mirnov_reference_channels(shot: int = 0) -> tuple[dict[str, Any], ...]

The phase-reference channels published for shot.

UNSUPPORTED_MAGNETICS_GEOMETRY_SHOTS = <builtins.dict>

Raised when a shot needs a magnetics geometry this repository lacks.

vest_diamagnetic_flux(shot: int, plasma_start: float, plasma_end: float, *, raw_source: raw_db.RawSource | None = None) -> tuple[np.ndarray, np.ndarray]

Compute the corrected diamagnetic flux waveform.

Also exported as vaft.machine_mapping.vest_diamagnetic_flux.

vest_diamagnetic_flux_detailed(shot: int, plasma_start: float, plasma_end: float, *, raw_source: raw_db.RawSource | None = None, with_stages: bool = False) -> tuple[np.ndarray, np.ndarray, dict[str, Any]]

Compute the diamagnetic flux waveform and report input saturation.

Also exported as vaft.machine_mapping.vest_diamagnetic_flux_detailed.

vest_equilibrium_magnetics_channel_definitions(shot: int | None = None) -> tuple[dict[str, Any], ...]

Return ordered VEST equilibrium-magnetics channel metadata for provenance/preflight.

Also exported as vaft.machine_mapping.vest_equilibrium_magnetics_channel_definitions.

vfit_equilibrium_magnetics(shot: int, indices: list[int] | np.ndarray | None = None, processing_config: VestMagneticsProcessingConfig | None = None, *, raw_source: raw_db.RawSource | None = None, allow_missing_channels: bool = False, allow_zero_fallback: bool | None = None) -> tuple[np.ndarray, list[np.ndarray], list[np.ndarray]]

Process magnetic probe and flux-loop data using VAFT process helpers.

vfit_equilibrium_magnetics_detailed(shot: int, indices: list[int] | np.ndarray | None = None, processing_config: VestMagneticsProcessingConfig | None = None, *, raw_source: raw_db.RawSource | None = None, allow_missing_channels: bool = False, allow_zero_fallback: bool | None = None) -> VestEquilibriumMagneticsResult

Process magnetics channels, keeping native flux-loop terminal voltages.

vfit_limiter_shunts_dynamic(ods: object, shot: int, *, raw_source: raw_db.RawSource | None = None) -> None

Map baseline-corrected limiter-monitor voltages at their native timebase.

vfit_magnetics_dynamic(ods: object, shot: int, tstart: float, tend: float, dt: float, processing_config: VestMagneticsProcessingConfig | None = None, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> dict[str, str]

Populate dynamic magnetics nodes from required raw waveforms.

vfit_magnetics_for_shot(ods: object, shot: int, tstart: float, tend: float, dt: float, processing_config: VestMagneticsProcessingConfig | None = None, *, raw_source: raw_db.RawSource | None = None) -> None

Populate canonical static and dynamic magnetics nodes for one shot.

vfit_magnetics_static(ods: object, shot: int = 0) -> None

Populate static magnetics metadata from YAML geometry assets.

vfit_mirnov_raw_dynamic(ods: object, shot: int, *, raw_source: raw_db.RawSource | None = None, tstart: float | None = None, tend: float | None = None) -> None

Populate raw Mirnov voltage traces at their native acquisition timebase.

vfit_plasma_current(shot: int, ref: int = -1, *, raw_source: raw_db.RawSource | None = None) -> tuple[np.ndarray, np.ndarray]

Return the processed plasma-current waveform.

claim_ids(ods: ODS, ids: str, owner: str) -> None

Refuse to write ids when a different mapper already owns it.

DCON_FRAGMENT_VERSION = 2

ensure_toroidal_mode_grid(ods: ODS, time_slice: int, n_tor_grid: Sequence[int]) -> None

Lay out mhd_linear.time_slice[t].toroidal_mode as a dense n_tor grid.

extract_dcon_stability(ods: ODS) -> list[dict[str, Any]]

DCON’s stability results from mhd_linear, one dict per (time slice, n_tor).

extract_rdcon_stability(ods: ODS) -> list[dict[str, Any]]

RDCON/STRIDE stability results from the ODS, one dict per (time slice, solver, n_tor).

initialize_output_flags(ods: ODS, ids: str, count: int) -> None

Extend <ids>.code.output_flag to count slices, defaulting to “not run”.

MAX_RADIAL_POINTS = 256

mhd_linear(ods: ODS, source: str, options: Optional[dict] = None) -> dict[int, dict[str, Any]]

Parse GPEC-suite output under source into mhd_linear/ntms.

NTMS_FRAGMENT_VERSION = 2

ntms_solver_surfaces(ods: ODS) -> list[dict[str, Any]]

Which solver produced each ntms.time_slice[t].mode[i], with its native extras.

RESISTIVE_FRAGMENT_VERSION = 2

nbi(ods: ODS, shot: int = 0, options: Optional[dict] = None) -> dict[str, Any]

Populate the static nbi IDS for VEST.

nbi_run_conditions(ods: ODS, result: Any, *, unit: int = 0) -> dict[str, Any]

Write the beam conditions a NUBEAM case actually ran with.

core_profiles_from_neo(ods: ODS, result: Any, *, time: float = 0.0, time_index: int = 0, b0: Optional[float] = None, conductivity: Any = None) -> dict[str, Any]

Write NEO’s bootstrap current, and its conductivity when given both runs.

core_transport_from_neo(ods: ODS, result: Any, *, time: float = 0.0, time_index: int = 0) -> dict[str, Any]

Write NEO’s particle and energy fluxes into core_transport.

NEOCLASSICAL_MODEL_INDEX = 5

coil_field_code_by_index(shot: int) -> dict[int, int]

Which acquisition channel each PF circuit’s current comes from.

pf_active(ods: object, shot: int, tstart: float, tend: float, dt: float, geometry_root: str | Path | None = None, *, raw_source: raw_db.RawSource | None = None) -> None

pf_active_from_raw_database(ods: object, shot: int, tstart: float, tend: float, dt: float, options: dict | None = None) -> None

Also exported as vaft.machine_mapping.pf_active_from_raw_database.

PF_GEOMETRY_2507_FIRST_SHOT = 45968

pf_geometry_version_for_shot(shot: int | None) -> str

Return the PF geometry version used for shot.

Also exported as vaft.machine_mapping.pf_geometry_version_for_shot.

require_acquired_pf_currents(ods: object, shot: int | None, consumer: str) -> None

Refuse to compute with PF currents when a circuit was not acquired.

resolve_geometry_asset(filename: str, geometry_root: str | Path | None = None) -> Path

Also exported as vaft.machine_mapping.resolve_geometry_asset.

unacquired_pf_coils(ods: object) -> list[str]

Names (or coil.<i>) of the PF circuits whose current is not all finite.

UnacquiredPFCircuitError(shot: int, field: int | str, reason: str, *, signal_name: str | None = None) -> None

A PF circuit the computation needs was not recorded for this shot (#1568).

vfit_pf(shot: int, geometry_root: str | Path | None = None, *, raw_source: raw_db.RawSource | None = None) -> tuple[np.ndarray, list[np.ndarray]]

Every PF circuit’s current; raises if a required circuit was not acquired.

vfit_pf_acquired(shot: int, *, raw_source: raw_db.RawSource | None = None) -> tuple[np.ndarray, list[np.ndarray], dict[int, raw_db.RawSignalUnavailableError]]

PF circuit currents, keeping the circuits that were acquired.

vfit_pf_active_dynamic(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> None

vfit_pf_active_for_shot(ods: object, shot: int, tstart: float, tend: float, dt: float, geometry_root: str | Path | None = None, *, raw_source: raw_db.RawSource | None = None) -> None

vfit_pf_active_static(ods: object, shot: int | None = None, geometry_root: str | Path | None = None) -> None

append_wall_loops(ods: Any, version: str) -> int

Append version’s additional loops to ods["pf_passive.loop"].

DEFAULT_REFERENCE_ODS = <pathlib.PosixPath>

DEFAULT_STATIC_GEOMETRY = <pathlib.PosixPath>

DEFAULT_WALL_2409_ADDITIONS = <pathlib.PosixPath>

load_wall_2409_additions() -> dict[str, Any]

The wall 2409 additions asset: loops, coupling rows and provenance.

pf_passive(ods: ODS, source: str | Path | None = None, options: Optional[dict] = None, *, shot: int | None = None) -> None

Populate canonical VEST passive-loop geometry from a static asset.

WALL_GEOMETRY_2409_FIRST_SHOT = 43017

wall_geometry_version_for_shot(shot: int | None) -> str

Return the passive-wall version for shot: "1512" or "2409".

vest_processing_provenance(shot: int) -> dict[str, Any]

Return the effective processing era for shot, per diagnostic.

Also exported as vaft.machine_mapping.vest_processing_provenance.

archive_workbooks(source_dir: str | Path, filedb: FileDB, *, extra: Iterable[str | Path] = (), dry_run: bool = False) -> dict[str, Any]

Copy every ShotLog file from source_dir (and each extra file) into FileDB.

Also exported as vaft.machine_mapping.pulse_schedule.archive_workbooks.

input_dir(filedb: FileDB) -> Path

load_record(shot: int, data_root: str | Path) -> dict[str, Any]

Read {data_root}/{shot}/metadata/shotlog.json.

Also exported as vaft.machine_mapping.pulse_schedule.load_record.

load_records(filedb: FileDB) -> dict[int, dict[str, Any]]

Every per-shot record under legacy/shotlog, keyed by shot.

output_dir(filedb: FileDB) -> Path

record_path(filedb: FileDB, shot: int) -> Path

Also exported as vaft.machine_mapping.pulse_schedule.record_path.

write_extraction(sessions: Iterable[tuple[SourceWorkbook, dict[str, Any]]], records: dict[int, dict[str, Any]], batch_manifest: dict[str, Any], filedb: FileDB) -> dict[str, int]

Write session documents, per-shot records and the batch manifest.

Also exported as vaft.machine_mapping.pulse_schedule.write_extraction.

ArchivedFile(path: Path, role: str, archive_path: str, note: str | None = None) -> None

A file kept byte for byte but not converted as a record.

classify_source(path: Path) -> tuple[SourceWorkbook | None, str | None]

(workbook, None) for a monthly ShotLog by name, else (None, reason).

convert_directory(input_dir: str | Path, registry: dict[str, Any] | None = None, overrides_path: Path | None = None, extra: Iterable[str | Path] = ()) -> tuple[list[tuple[SourceWorkbook, dict[str, Any]]], dict[str, Any]]

Convert every chosen workbook; return the sessions and a batch manifest.

Also exported as vaft.machine_mapping.pulse_schedule.convert_directory.

discover_sources(input_dir: str | Path, extra: Iterable[str | Path] = (), exclude: Iterable[str | Path] = ()) -> Discovery

Give every file under input_dir (and every extra file) a role.

Also exported as vaft.machine_mapping.pulse_schedule.discover_sources.

Discovery(included: list[SourceWorkbook] = <factory>, archived_only: list[ArchivedFile] = <factory>, fallback: list[SourceWorkbook] = <factory>, excluded: list[dict[str, Any]] = <factory>) -> None

iter_sessions(sources: list[SourceWorkbook], registry: dict[str, Any] | None = None, overrides_path: Path | None = None, manifest: dict[str, Any] | None = None) -> Iterator[tuple[SourceWorkbook, dict[str, Any]]]

Yield (workbook, session) for every sheet that logs a shot.

SourceWorkbook(path: Path, year: int | None, month: int | None, first_shot: int | None, last_shot: int | None, role: str = 'monthly_record', span_limit: int | None = None, archive_path: str = '', content_span: tuple[int, int] | None = None) -> None

One converted ShotLog file: a month’s record, a supplementary log, or a fallback.

summarise(manifest: dict[str, Any]) -> dict[str, int]

extract_card(ws: Any, shot_row: int, end_row: int) -> dict[str, Any] | None

Return the structured content of the card whose shot number is on shot_row.

find_header_row(ws: Any, shot_row: int) -> int | None

The card’s first header row within three rows above the shot.

is_header_row(ws: Any, row: int) -> bool

A card’s first header row, recognised by its section titles.

section_columns(ws: Any, header_row: int) -> dict[str, int]

convert_sheet(source: Path, sheet_name: str, registry: dict[str, Any], overrides_path: Path | None = None, workbook: Any | None = None, source_hash: str | None = None, span: tuple[int, int | None] | None = None) -> dict[str, Any]

Convert sheet_name of source into a session document.

Also exported as vaft.machine_mapping.pulse_schedule.convert_sheet.

derive_session(sheet_name: str, source: Path) -> tuple[str | None, str]

(experiment_date, session_id) from a sheet named YYMMDD or YYYYMMDD.

detect_schema(ws: Any, registry: dict[str, Any]) -> tuple[str | None, dict[str, Any]]

Pick the era whose required headers all appear on the sheet.

load_overrides(path: Path | None) -> tuple[list[dict[str, Any]], str | None]

Read a review-overrides document (YAML; JSON is a subset).

Marker(row: int, kind: str, shots: tuple[int, ...] = (), reference: int | None = None, references: tuple[int, ...] = ()) -> None

marker_shots(value: Any) -> tuple[int, ...]

The shots a column-A cell logs, or () when it logs none.

session_path(output_dir: Path, dataset: dict[str, Any]) -> Path

Also exported as vaft.machine_mapping.pulse_schedule.batch.session_path.

set_path(target: dict[str, Any], path: str, value: Any) -> None

sha256_file(path: Path) -> str

validate_dataset(dataset: dict[str, Any], registry: dict[str, Any]) -> list[str]

EVENT_TYPES = <builtins.dict>

Also exported as vaft.machine_mapping.pulse_schedule.EVENT_TYPES.

LISTENERS = <builtins.dict>

Also exported as vaft.machine_mapping.pulse_schedule.LISTENERS.

map_pulse_schedule(ods: Any, record: dict[str, Any], *, offset_ms: float | None = None) -> int

Write record’s events into ods.pulse_schedule; return how many.

Also exported as vaft.machine_mapping.pulse_schedule.map_pulse_schedule.

pulse_schedule(ods: Any, shot: int, data_root: str | Path | None = None) -> Any

Fill ods.pulse_schedule for shot from its FileDB ShotLog record.

Also exported as vaft.machine_mapping.pulse_schedule.pulse_schedule.

The ShotLog has this shot but logged no usable trigger for it.

Also exported as vaft.machine_mapping.pulse_schedule.PulseScheduleUnavailableError.

build_shot_records(sessions: Iterable[tuple[SourceWorkbook, dict[str, Any]]]) -> dict[int, dict[str, Any]]

Invert converted sessions into {shot: record}.

Also exported as vaft.machine_mapping.pulse_schedule.build_shot_records.

CARRY_FORWARD_SYSTEMS = frozenset({'diagnostic'})

coverage_report(records: dict[int, dict[str, Any]]) -> dict[str, Any]

Which shot numbers the ShotLog accounts for, and where it is silent.

DAQ_OFFSET_MS = -200

Also exported as vaft.machine_mapping.pulse_schedule.DAQ_OFFSET_MS.

effective_timing_by_session(dataset: dict[str, Any]) -> dict[int, dict[str, Any]]

{shot: {"timing": [...], "probe_positions": {...}}} for one session.

RECORD_VERSION = 1

to_daq_ms(value: int | float, offset_ms: int | float = -200) -> int | float

trigger_table(records: dict[int, dict[str, Any]], offset_ms: int | float = -200) -> dict[str, Any]

The packaged diagnostic-trigger-settings.yaml content.

Also exported as vaft.machine_mapping.pulse_schedule.trigger_table.

build_registry(directory: str | Path | None = None) -> dict[str, Any]

Load every *.yaml schema in directory into one registry.

FIELD_KEYS = ('path', 'type', 'required', 'meaning_ko', 'aliases', 'parsing', 'validation')

packaged_registry() -> dict[str, Any]

The registry shipped with this vaft version (cached; do not mutate).

Also exported as vaft.machine_mapping.pulse_schedule.packaged_registry.

REGISTRY_VERSION = 2

SCHEMA_KEYS = ('schema_id', 'version', 'status', 'detection', 'fields')

schema_versions(registry: dict[str, Any]) -> dict[str, str]

{schema_id: version}, recorded with every conversion.

Raised when a packaged or caller-supplied schema is malformed.

is_range(value: Any) -> bool

True for an explicit start-end window rather than a lone number.

normalize_status(strings: list[str]) -> dict[str, Any]

Classify a shot’s outcome from its remark text.

number(value: str | int | float) -> int | float

Parse a number, keeping integers integral so YAML stays readable.

parse_position(value: Any) -> list[int | float] | None

Parse one or two probe locations expressed in metres.

parse_time_window(value: Any) -> tuple[int | float, int | float] | None

Parse a trigger time or an inclusive start-end window, in ms.

parse_value(raw: Any, unit: str | None = None, trigger: bool = False) -> dict[str, Any]

Return a value record: raw text plus whatever can be derived from it.

parse_valve(value: Any) -> dict[str, Any] | None

Parse a gas valve setting into {voltage_V, species}.

raw_field_name(field: int) -> str

The legacy offline name of a DAQ field, or "" when the table has none.

Also exported as vaft.machine_mapping.raw_field_name.

raw_sources(diagnostic: str, shot: int) -> tuple[RawField, ...]

The raw DAQ fields registry entry diagnostic is made from for shot.

Also exported as vaft.machine_mapping.raw_sources.

RawField(field: int, label: str, daq_name: str, raw_unit: str, daq_type: str | None, role: str, channel: str | None, calibration: Mapping[str, Any] = <factory>, config: str | None = None, known_fault: str | None = None) -> None

One archive field a diagnostic is made from, resolved for one shot.

Also exported as vaft.machine_mapping.RawField.

AVAILABILITY_VALUES = frozenset({'If-requested', 'Retired', 'Routine'})

Raised when the diagnostic registry has an invalid schema.

documentation_snapshot(path: str | Path | None = None, provenance: Mapping[str, str] | None = None) -> dict[str, Any]

Return the deterministic, publishable representation of the registry.

export_documentation_snapshot(output: str | Path, path: str | Path | None = None, provenance: Mapping[str, str] | None = None) -> Path

Write a normalized YAML documentation snapshot and return its path.

LIFECYCLE_VALUES = frozenset({'available', 'in_maintenance', 'retired'})

load_diagnostic_registry(path: str | Path | None = None) -> dict[str, dict[str, Any]]

Load and validate the top-level VEST diagnostic registry.

load_port_map(path: str | Path | None = None) -> dict[str, dict[str, Any]]

Load and validate the VEST port table, keyed by port name.

main(argv: list[str] | None = None) -> None

MAPPING_STATUS_VALUES = frozenset({'implemented', 'not_implemented', 'partial'})

port_clock(name: str, path: str | Path | None = None) -> float

The clock position of a named port. Clockwise-positive, not IMAS phi.

port_clock_angle(name: str, path: str | Path | None = None) -> float

The VEST clock angle of a named port, in degrees, clockwise-positive.

PORT_MAP_KEY = 'port_map'

port_phi(name: str, path: str | Path | None = None) -> float

The IMAS toroidal angle of a named port, in radians.

Raised when the VEST port table has an invalid schema.

REGISTRY_KEY = 'diagnostic_registry'

registry_path(path: str | Path | None = None) -> Path

Return the registry-bearing VEST YAML file.

validate_diagnostic_registry(registry: Mapping[str, Mapping[str, Any]], *, subjects: Collection[str] | None = None) -> None

Validate the schema shared by runtime checks and documentation export.

validate_port_map(ports: Mapping[str, Mapping[str, Any]]) -> None

Validate the port table, including name/field agreement.

build_time_axis(samples: int, *, sample_rate: float = 976562.5, time_offset: float = 0.0) -> np.ndarray

Return seconds for digitizer samples from the trigger-relative archive axis.

DEFAULT_ARRAY_SPECS = <builtins.dict>

default_channel_map(daq_label: str | int, n_channels: int) -> list[dict[str, Any]]

Return default VEST SXR digitizer-to-array channel metadata.

DEFAULT_DIGITIZER_ARRAYS = <builtins.dict>

DEFAULT_ENERGY_BAND = (0.0, 20000.0)

DEFAULT_SAMPLE_RATE = 976562.5

DEFAULT_SAMPLE_RATES = <builtins.dict>

DEFAULT_TIME_OFFSET = 0.0

load_digitizer_csv(filepath: str | Path, *, channels_as_rows: bool = True, nrows: int | None = None) -> np.ndarray

Load a VEST SXR digitizer record as a time x channel matrix.

load_sxr_geometry_table(geometry_table: str | Path | None = None) -> dict[tuple[str, int], dict[str, Any]]

Load SXR LOS endpoint geometry keyed by (array, channel).

PACKAGED_GEOMETRY_TABLE = 'geometry/line_of_sight_endpoints.csv'

PACKAGED_TRIGGER_SETTINGS = 'legacy/diagnostic-trigger-settings.yaml'

resolve_sxr_geometry_root(geometry_root: str | Path | None = None) -> Path | None

Resolve a legacy directory containing full SXR LOS grid CSV files.

resolve_sxr_geometry_table(geometry_root: str | Path | None = None) -> Path | None

Resolve a single CSV table with SXR LOS endpoint geometry.

resolve_sxr_time_alignment(shot: int, *, trigger_settings_path: str | Path | None = None, allow_hxr_fallback: bool = True) -> SXRTimeAlignment

Resolve a machine-time offset for a trigger-relative SXR CSV.

resolve_sxr_trigger_settings(trigger_settings_path: str | Path | None = None) -> Path | None

Resolve the packaged or caller-provided diagnostic trigger settings.

save_soft_x_rays_ods(output_path: str | Path, shot: int, daq_label: str | int | None = None, **kwargs: Any)

Build and save one shot-level soft_x_rays ODS in the selected OMAS format.

Also exported as vaft.machine_mapping.save_soft_x_rays_ods.

soft_x_rays(ods: Any, shot: int, daq_label: str | int | None = None, *, data_root: str | Path | None = None, digitizer_file: str | Path | None = None, geometry_root: str | Path | None = None, channel_map: Sequence[Mapping[str, Any]] | None = None, channel_maps: Mapping[str | int, Sequence[Mapping[str, Any]]] | None = None, sample_rate: float | None = None, time_offset: float | None = None, time_reference: str = 'auto', trigger_settings_path: str | Path | None = None, energy_band: tuple[float, float] = (0.0, 20000.0), brightness_scale: float = 1.0, baseline_range: tuple[int | None, int | None] | None = None, polarity: float = 1.0, channels_as_rows: bool = True) -> None

Populate one shot-level SXR IDS from all available digitizer files.

Also exported as vaft.machine_mapping.soft_x_rays.soft_x_rays_from_raw_database, vaft.machine_mapping.soft_x_rays_from_raw_database.

soft_x_rays_from_digitizer_csv(shot: int, daq_label: str | int | None = None, *, consistency_check: bool = True, **kwargs: Any)

Create a shot-level ODS, discovering all SXR DAQs when no label is given.

Also exported as vaft.machine_mapping.soft_x_rays_from_digitizer_csv.

sxr_sightlines(arrays: Sequence[str] | None = None, geometry_table: str | Path | None = None)

The VEST SXR chords as vaft.process.line_of_sight.Sightlines.

SXRDigitizerBlock(array: str, start: int, count: int, filter_material: str | None = None, filter_thickness_m: float | None = None, reverse_spatial_order: bool = False) -> None

One contiguous digitizer block and its physical channel semantics.

SXRTimeAlignment(offset_seconds: float, source: str, detail: str) -> None

Resolved relation between a CSV trigger axis and the shot clock.

vfit_soft_x_rays_dynamic(ods: Any, *, data: np.ndarray, time: np.ndarray, channel_map: Sequence[Mapping[str, Any]], channel_offset: int = 0, set_global_time: bool = True, brightness_scale: float = 1.0, baseline_range: tuple[int | None, int | None] | None = None, polarity: float = 1.0) -> None

Fill dynamic channel brightness traces from a time x channel matrix.

vfit_soft_x_rays_static(ods: Any, *, channel_map: Sequence[Mapping[str, Any]], channel_offset: int = 0, homogeneous_time: bool = True, geometry_root: str | Path | None = None, energy_band: tuple[float, float] = (0.0, 20000.0), phi: float = 0.0, brightness_scale: float = 1.0) -> None

Fill static soft_x_rays IDS metadata and LOS geometry.

CHANNEL_CADENCE_HZ = <builtins.dict>

filterscope_from_raw_database(ods: object, shot: int, tstart: float, tend: float, dt: float, options: dict | None = None) -> None

Also exported as vaft.machine_mapping.filterscope_from_raw_database.

legacy_time_shift_s(shot: int) -> float

The offset the mapper adds to a filterscope record’s own time axis.

SIGNALS = <builtins.list>

spectrometer_uv(ods: object, shot: int, t_start: float, t_end: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> None

Canonical machine_mapping entry point for the spectrometer_uv IDS.

Also exported as vaft.machine_mapping.filterscope.filterscope.

vfit_filterscope(ods: object, shot: int, t_start: float, t_end: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None) -> None

Also exported as vaft.machine_mapping.filterscope.vfit_filterscope.

load_static_ods(path: str | Path) -> ODS

Load plain or gzip-compressed OMAS JSON without temporary files.

summary(ods: ODS, source: str, options: Optional[dict] = None) -> None

Add experiment summary information to the ODS structure.

repair_tf_excursions(time: np.ndarray, current: np.ndarray, repair: dict | None) -> tuple[np.ndarray, list[tuple[float, float]]]

Replace acquisition excursions of the TF current by its slow trend (#1543).

tf(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None) -> None

tf_from_raw_database(ods: object, shot: int, tstart: float, tend: float, dt: float, options: dict | None = None) -> None

Also exported as vaft.machine_mapping.tf_from_raw_database.

vfit_tf_bt_r(shot: int, *, raw_source: raw_db.RawSource | None = None) -> Signal

vfit_tf_current(shot: int, *, raw_source: raw_db.RawSource | None = None) -> Signal

vfit_tf_current_detailed(shot: int, *, raw_source: raw_db.RawSource | None = None) -> tuple[np.ndarray, np.ndarray, list[tuple[float, float]]]

TF current [A] and the intervals whose acquisition excursion was repaired.

vfit_tf_dynamic(ods: object, shot: int, tstart: float, tend: float, dt: float, *, raw_source: raw_db.RawSource | None = None, target_time: np.ndarray | None = None, report: dict | None = None) -> None

Map the TF current and vacuum toroidal field onto the target grid.

vfit_tf_static(ods: object) -> None

thomson_file_shot(filename: str) -> int | None

The shot a Thomson MAT filename belongs to, or None if it is not one.

thomson_scattering(ods: Any, shotnumber: int, data_root: str | Path | None = None, mat_file: str | Path | None = None) -> None

thomson_source_rank(filename: str, shotnumber: int) -> tuple[int, int, int] | None

Rank one Thomson MAT filename for shotnumber, or None if unrelated.

vfit_thomson_scattering_dynamic(ods: Any, shotnumber: int, data_root: str | Path | None = None, mat_file: str | Path | None = None) -> None

vfit_thomson_scattering_static(ods: Any) -> None

ANOMALOUS_MODEL_DESCRIPTION = 'Turbulent transport, from the TGLF quasilinear model'

ANOMALOUS_MODEL_INDEX = 6

ANOMALOUS_MODEL_NAME = 'anomalous'

core_transport_from_tglf(ods, surfaces, profile, *, time=0.0, time_index=0, state_identity=None, realization_id=None, configuration=None, provenance=None)

Map one radial realization transactionally, preserving scientific identity.

FLUX_MULTIPLIER = 0.0

MOMENTUM_SIGN = -1.0

AGREEMENT_CONSISTENT = 'consistent'

Also exported as vaft.omas.plasma_timing.AGREEMENT_CONSISTENT.

AGREEMENT_HALPHA_LEADS_IP_LARGE = 'halpha_leads_ip_large'

Also exported as vaft.omas.plasma_timing.AGREEMENT_HALPHA_LEADS_IP_LARGE.

AGREEMENT_IP_BEFORE_HALPHA = 'ip_before_halpha'

Also exported as vaft.omas.plasma_timing.AGREEMENT_IP_BEFORE_HALPHA.

apply_default_constraint_uncertainties(ods: Any, uncertainty: Sequence[float] | Mapping[str, float] | None = None, *, fl_correct_coeff: Sequence[float] | None = None) -> None

Also exported as vaft.machine_mapping.apply_default_constraint_uncertainties.

apply_magnetics_uncertainties(ods: Any, *, ip_relative_error: float | None = None, diamagnetic_flux_relative_error: float | None = None, bpol_inboard_relative_error: float | None = None, bpol_side_relative_error: float | None = None, bpol_outboard_relative_error: float | None = None, flux_loop_inboard_relative_error: float | None = None, flux_loop_outboard_relative_error: float | None = None, fl_correct_coeff: Sequence[float] | None = None) -> None

Also exported as vaft.machine_mapping.apply_magnetics_uncertainties.

apply_pf_active_current_uncertainties(ods: Any, relative_error: float | None = None) -> None

Also exported as vaft.machine_mapping.apply_pf_active_current_uncertainties.

apply_tf_uncertainties(ods: Any, relative_error: float | None = None) -> None

Also exported as vaft.machine_mapping.apply_tf_uncertainties.

build_window_time_axis(source_time: Any, tstart: float, tend: float, dt: float) -> np.ndarray

Build the half-open target grid for one window, clipped to real coverage.

crop_to_span(time: Any, values: Any, *, baseline_start: float, tstart: float, tend: float, min_baseline_s: float | None = None) -> CroppedRecord

Crop a record to [baseline_start, tend) and split it at tstart.

CroppedRecord(t: np.ndarray, y: np.ndarray, baseline_mask: np.ndarray | None, search: np.ndarray, flags: tuple[str, ...]) -> None

One record cropped to the plasma-analysis span, split into baseline and search.

DEFAULT_CONSTRAINT_UNCERTAINTIES = <builtins.dict>