export(shot, source=None, *, backend, output=None, overwrite=False, occurrence=0, cache='auto', transport='auto')

Export one shot from an HSDS source as portable local files.

export_summary(df, path, *, mode='replace', key_columns=None, replace_groups=None)

Serialize or upsert an already-generated summary DataFrame.

get_summary_preset(name)

Return the SummaryPreset describing one canonical summary sheet.

load(shot: int | list[int], source: str | None = None, *, directory: str | None = None, representation: Literal['omas', 'imas'] = 'omas', paths: str | list[str] | None = None, occurrence: int | Mapping[str, int] | None = None, imas_version: str | None = None, cache: str = 'auto', transport: Literal['auto', 'canonical', 'h5image'] = 'auto')

Materialize an OMAS ODS or native IMAS IDS from remote HSDS storage.

open(shot: int, *, source: str | None = None, directory: str | None = None, representation: Literal['omas', 'imas'] = 'omas', paths: str | list[str] | None = None, occurrence: int | Mapping[str, int] | None = None, imas_version: str | None = None)

Open a read-only lazy OMAS ODS or native IMAS IDS adapter over HSDS.

save(data, shot: int, *, source: str | None = None, target: str | None = None, directory: str | None = None, representation: Literal['omas', 'imas'] | None = None, occurrence: int | Mapping[str, int] | None = None, imas_version: str | None = None, derived_cache: Literal['auto', 'none', 'imas-images', 'omas', 'both'] = 'auto', finalize_master: Callable[[Path], None] | None = None)

Write an OMAS ODS or native IDS to remote HSDS storage.

summary(shot_range=None, *, preset='equilibrium_global', source=None, directory=None)

Return a canonical preset summary for a range or all available shots.

compose(shot: int, sources: Sequence[str] = ('main', 'impa'), *, paths: Iterable[str] | None = None, occurrence: int | None = None) -> tuple[Any, dict[str, Any]]

Load shot from several sources and append the optional channels.

Also exported as vaft.database.compose.

compose_stage_products(*, diagnostics: str | Path, eddy: str | Path, eddy_manifest: str | Path | None = None, strict: bool = True) -> tuple[Any, dict[str, Any]]

Union a diagnostics product with the eddy product computed from it.

eddy_no_output_reason(eddy_manifest: str | Path | None) -> str | None

The recorded reason when eddy_manifest says no_output, else None.

EddyNoOutputError(path: Path, reason: str) -> None

The eddy stage finished no_output: there are no passive currents to compose.

impa_channels(ods: Any, node: str) -> list[int]

Return the indices of node holding IMPA channels, by identifier.

Two stage products that cannot be composed into one shot.

CORRECTIVE_LINEAGE = ('thomson', 'ces', 'core_profiles', 'composition', 'ti_inferred', 'resistive_zeff', 'electron_efit', 'kinetic_efit')

DEFAULT_STAGES = ('thomson', 'ces', 'core_profiles', 'composition', 'electron_efit', 'kinetic_efit')

scoped(scope: frozenset[str] | None) -> tuple[str, ...]

The lineage’s stages inside scope, in lineage order (all of them for None).

stage_scope(value: Iterable[str] | str | None) -> frozenset[str] | None

The stages a run builds and judges, from the configuration’s stages key.

UPSTREAM = <builtins.dict>

container_path(csv_path: str | Path) -> Path

Where the container for digitizer_{daq}_{shot}.csv lives.

A CSV could not be packed without loss; the CSV was left untouched.

fsync_path(path: str | Path) -> None

Force a file (or directory) to stable storage before anything depends on it.

GZIP_LEVEL = 4

load_digitizer_hdf5(path: str | Path, *, nrows: int | None = None) -> np.ndarray

Return the (channels, samples) array a container holds.

pack_digitizer_csv(csv_path: str | Path, target: str | Path | None = None, *, shot: int | None = None, daq_label: str | None = None, overwrite: bool = False) -> PackResult

Write the lossless container for one digitizer CSV and verify it.

PackResult(source: Path, target: Path, shape: tuple[int, int], source_size: int, target_size: int, source_sha256: str) -> None

read_attributes(path: str | Path) -> dict[str, Any]

rebuild_csv_bytes(data: np.ndarray, line_terminator: str = '\r\n') -> bytes

Regenerate the CSV text sample_v3.py wrote for data.

restore_csv(path: str | Path, target: str | Path | None = None) -> Path

Write the original CSV back out of a verified container.

SCHEMA = 'vaft.soft_x_rays.digitizer'

SCHEMA_VERSION = 1

SUFFIX = '.h5'

verify_container(path: str | Path, *, expected_sha256: str | None = None) -> str

Regenerate the source CSV from a container and check its sha256.

audit_filedb_grammar(root: str | os.PathLike[str]) -> GrammarRelocationReport

Plan a canonical root’s move onto the lineage-bearing grammar, changing nothing.

audit_legacy_filedb(legacy_root: str | os.PathLike[str], *, target_root: str | os.PathLike[str] | None = None, expected_products: Mapping[str, str] | None = None) -> LegacyAuditReport

Inventory and propose mappings without modifying either FileDB root.

The reconstruction lineage an equilibrium came from.

FileDB(root: str | os.PathLike[str]) -> None

Resolve canonical OMAS-first FileDB paths without filesystem writes.

Raised when the FileDB storage root cannot be configured.

Base error for invalid FileDB configuration or path requests.

Raised when a path request is outside the canonical grammar.

Legacy stability code names.

GrammarRelocation(source: str, target: str, rule: str) -> None

One subtree that moves, and the rule that says where.

GrammarRelocationReport(root: str, applied: bool, relocations: tuple[GrammarRelocation, ...], already_canonical: tuple[str, ...], collisions: tuple[str, ...], unrecognized: tuple[str, ...], unattributable: tuple[str, ...] = ()) -> None

What a canonical root needs before it resolves under the new grammar.

LEGACY_DCON_PRODUCT = 'dcon'

LEGACY_FAMILY = 'magnetic'

LEGACY_REFINEMENT = 'chease'

LegacyAuditEntry(source: str, proposed_target: str | None, status: str, reason: str | None = None, size: int | None = None, sha256: str | None = None) -> None

LegacyAuditReport(legacy_root: str, target_root: str, entries: tuple[LegacyAuditEntry, ...], collisions: tuple[LegacyCollision, ...], duplicates: tuple[LegacyDuplicate, ...], symlinks: tuple[str, ...], missing_products: tuple[LegacyMissingProduct, ...]) -> None

LegacyCollision(proposed_target: str, sources: tuple[str, ...], existing_target: bool = False, existing_target_kind: str | None = None) -> None

LegacyDuplicate(sha256: str, size: int, sources: tuple[str, ...]) -> None

LegacyMissingProduct(shot: int, product: str, expected_source: str) -> None

LegacyResolution(path: Path, exists: bool, read_only: bool = True, layout: str = 'legacy-shot-first') -> None

An explicitly read-only path into the former shot-first layout.

The equilibrium actually handed to the stability solver.

relocate_filedb_grammar(root: str | os.PathLike[str], *, apply: bool = False) -> GrammarRelocationReport

Move a canonical root onto the lineage-bearing grammar.

One stability calculation.

stage_lineage(stage: str, *, family: str, refinement: str | None = None, product: str | None = None) -> dict[str, str]

The lineage arguments an OMAS stage’s path takes, and only those.

canonical_info(directory: str, shot: int, filename: str) -> dict[str, str | None]

derived_filename(filename: str) -> str

H5IMAGE_CHUNK_BYTES = 4194304

H5IMAGE_SCHEMA_VERSION = 1

A requested derived HDF5 image is missing, stale, or corrupt.

is_derived_filename(filename: str) -> bool

materialize_image(directory: str, shot: int, filename: str, target: Path, *, source_info: dict[str, str | None]) -> dict[str, Any]

Restore and validate one derived image into an ordinary HDF5 file.

publish_image(local_path: Path, directory: str, shot: int, *, imas_version: str | None, source_info: dict[str, str | None] | None = None) -> dict[str, Any]

Publish one local canonical HDF5 file as a compressed opaque image.

read_manifest(domain) -> dict[str, Any] | None

active_path(cwd: Path | None = None) -> Path

The file h5pyd will actually read: ./.hscfg shadows ~/.hscfg.

configured_keys(path: Path) -> tuple[str, ...]

Names of the h5pyd keys path sets to a non-empty value. Never values.

default_path() -> Path

~/.hscfg, the file vaft hsds configure writes unless told otherwise.

HSCFG_KEYS = ('hs_endpoint', 'hs_username', 'hs_password', 'hs_api_key')

insecure_permissions(path: Path) -> bool

Whether path is group- or world-accessible. Always False on Windows.

read_values(path: Path) -> dict[str, str]

Every key = value pair in path; the last occurrence wins, as in h5pyd.

render_update(original: str, updates: Mapping[str, str]) -> str

original with each key in updates set, everything else untouched.

SECRET_KEYS = frozenset({'hs_api_key', 'hs_password'})

update_hscfg(path: Path, updates: Mapping[str, str]) -> None

Set updates in the .hscfg at path, creating it 0600 if needed.

validate_value(key: str, value: str) -> None

Refuse a value h5pyd would read back differently from what was typed.

write_private(path: Path, text: str) -> None

Atomically replace path with text, readable by the owner only.

load(shot: int, ids_name: Union[str, list[str]], source: Optional[str] = None, occurrence: int | dict[str, int] = 0, dd_version: Optional[str] = None, local_dir: Optional[str] = None, cache: str | Path = 'auto', transport: Literal['auto', 'canonical', 'h5image'] = 'auto', *, directory: Optional[str] = None) -> Union[object, dict[str, object]]

Load IDS object(s) from HSDS as native IMAS objects.

save(ids: imas.ids_toplevel.IDSToplevel, shot: int, source: Optional[str] = None, dd_version: Optional[str] = None, *, directory: Optional[str] = None, derived_cache: Literal['auto', 'none', 'imas-images', 'omas', 'both'] = 'auto') -> Optional[str]

Save one native IMAS IDS to a remote HSDS namespace.

active_recorder() -> IORecorder | None

The innermost recorder, or None when nothing is recording.

IORecorder(run_id: str, requests: list[dict[str, Any]] = <factory>, subprocesses: list[dict[str, Any]] = <factory>, started_wall: float = <factory>, ended_wall: float | None = None) -> None

The requests and subprocess transfers seen while it was active.

record_io(run_id: str | None = None) -> Iterator[IORecorder]

Record HSDS I/O inside the block; patches requests.Session.send only while active.

RUN_ID_HEADER = 'X-VAFT-Run-ID'

decode_hdf5_value(value: Any) -> Any

Restore conventional Python/NumPy values from IMAS HDF5 storage.

discover_hsds_ids(h5pyd_module: Any, directory: str, shot: int) -> tuple[str, ...]

List canonical IDS image domains for a shot.

normalize_ids(ids: str | Iterable[str] | None) -> tuple[str, ...] | None

Normalize an optional IDS restriction while preserving input order.

HSDSIMASHandle(shot: int, source: str | None = None, *, directory: str | None = None, ids: str | Iterable[str] | None = None, imas_version: str, h5pyd_module: Any = None) -> None

Read-only context manager returning native lazy IDSToplevel objects.

Raised when an uncached native IDS leaf is read after close().

open_imas(shot: int, source: str | None = None, *, directory: str | None = None, ids: str | Iterable[str] | None = None, imas_version: str) -> HSDSIMASHandle

HSDSODS(*, store: HSDSStore | None = None, dynamic: HSDSStore | None = None, imas_version: str | None = None, consistency_check: bool = True, **kwargs: Any) -> None

An omas.ODS whose missing leaves are fetched from HSDS.

HSDSStore(shot: int, source: str | None = None, *, directory: str | None = None, ids: str | Iterable[str] | None = None, h5pyd_module: Any = None) -> None

OMAS dynamic backend for occurrence-zero IMAS HDF5 HSDS domains.

Raised when an uncached leaf is requested after closing a lazy ODS.

open_ods(shot: int, source: str | None = None, *, directory: str | None = None, ids: str | list[str] | None = None, imas_version: str | None = None, consistency_check: bool = True) -> HSDSODS

Open an occurrence-zero ODS that fetches only requested HSDS leaves.

audit_master_link(shot: int, *, source: str | None = None, repair: bool = False) -> dict[str, Any]

Does the shot’s master.h5 link every IDS file stored beside it (#913)?

audit_master_links(shots: Iterable[int], *, source: str | None = None, repair: bool = False) -> list[dict[str, Any]]

Run audit_master_link over many shots.

What one published shot still carries, and whether it can be removed.

Raised when a published shot cannot be stripped safely.

Raised when the rewrite failed after it had started touching the source.

inspect_impa_residue(ods: Any) -> ImpaResidue

Report the IMPA channels in ods and whether they form a tail block.

MASTER_ABSENT = 'absent'

MASTER_COMPLETE = 'complete'

MASTER_LINKS_MISSING = 'links_missing'

MASTER_MISSING = 'no_master'

MASTER_NAN_ONLY_UNLINKED = 'nan_only_unlinked'

MASTER_REPAIRED = 'repaired'

MASTER_STUBS_UNLINKED = 'stubs_unlinked'

MASTER_UNREADABLE = 'unreadable'

strip_impa_from_shots(shots: Iterable[int], *, source: str | None = None, apply: bool = False) -> list[dict[str, Any]]

Run strip_impa_from_source over many shots, recording refusals.

strip_impa_from_source(shot: int, *, source: str | None = None, apply: bool = False) -> dict[str, Any]

Remove IMPA channels from one published shot’s baseline magnetics.

exist_ts_file(*args, **kwargs)

Return processed Thomson-scattering shots through the current helper.

load(shot, source: Optional[str] = None, **kwargs)

Deprecated compatibility alias for load_ods.

Deprecated: Deprecated compatibility alias for ``load_ods``.

load_ods(shot: Union[int, list[int]], source: Optional[str] = None, *, directory: Optional[str] = None, occurrence: Optional[dict] = None, paths: Optional[list] = None, time: Optional[float] = None, imas_version: Optional[str] = None, skip_uncertainties: bool = False, consistency_check: bool = True, verbose: bool = False, path: Optional[Union[str, Path]] = None, local_dir: Optional[Union[str, Path]] = None, cache: Literal['auto', 'off'] | str | Path = 'auto', transport: Literal['auto', 'canonical', 'h5image'] = 'auto') -> Union[omas.ODS, list[omas.ODS]]

Load ODS data by reading IMAS images from {source}/{shot} and converting via OMAS.

MASTER_FILENAME = 'master.h5'

save_ods(ods: omas.ODS, shot: int, filename: Optional[str] = None, env: str = 'server', *, source: Optional[str] = None, directory: Optional[str] = None, path: Optional[Union[str, Path]] = None, occurrence: Optional[dict] = None, user: Optional[str] = None, machine: Optional[str] = None, run: Optional[int] = None, imas_version: Optional[str] = None, verbose: bool = True, derived_cache: Literal['auto', 'none', 'imas-images', 'omas', 'both'] = 'auto', finalize_master: Optional[Callable[[Path], None]] = None) -> Optional[str]

Save ODS data by converting it to IMAS HDF5 images.

upload_order(filenames: Iterable[str]) -> list[str]

Order a shot’s files so the master is uploaded last.

available_plots(shot: Any = None, source: str | None = None, *, query: str | None = None, detail: bool = False, available_only: bool | None = None, **filters: Any)

What can be plotted from a database shot, without downloading it.

compose(composition: Any, shot: Any, source: str | None = None, *, occurrence: Any = None, backend: str | None = None, show: bool = False, label: Any = 'shot', **presentation: Any) -> Any

Draw a vaft.plot.FigureComposition for database shots (issue #1467).

dd(name: str) -> tuple

The Data Dictionary paths plot name reads; nothing is opened.

extract(name: str, shot: Any, source: str | None = None, *, lazy: bool = True, occurrence: Any = None, label: Any = 'shot', **options: Any) -> Any

Open what plot name needs of shot and return its view model, undrawn.

plot_diagnostics_time_interactive(shot: Any, source: str | None = None, *, occurrence: Any = None, **options: Any) -> Any

The diagnostics overview with live controls, from the database (issue #482).

plot_equilibrium_interactive(shot: Any, source: str | None = None, *, occurrence: Any = None, **options: Any) -> Any

Explore one shot’s equilibrium slices, from the database.

render(name: str, shot: Any, source: str | None = None, *, lazy: bool = True, occurrence: Any = None, ax: Any = None, show: bool = False, label: Any = 'shot', **options: Any) -> tuple[Any, Any]

Open what plot name needs of shot in source and render it.

render_to_file(name: str, shot: Any, path: Any, source: str | None = None, *, lazy: bool = True, occurrence: Any = None, label: Any = 'shot', **options: Any) -> Any

Render plot name for shot and write it to path; returns path.

stored_ids(shot: Any, source: str | None = None) -> tuple[str, ...]

The IDS shot stores in source, listed without reading any data.

audit_product_containers(root: str | Path, *, stages: Sequence[str] | None = None, verify_shape: bool = False) -> ProductMigrationReport

Classify every stage product under root. Reads only; writes nothing.

LEGACY_SUFFIX = '.json'

migrate_product_containers(root: str | Path, *, stages: Sequence[str] | None = None, apply: bool = False) -> ProductMigrationReport

Bring every stage product under root onto its declared shape.

MIGRATIONS = <builtins.tuple>

ProductMigration(stage: str, directory: str, source: str, target: str, verify: str) -> None

One product’s move, as the plan describes it.

A migration that refuses to run rather than guess.

ProductMigrationReport(root: str, applied: bool, pending: tuple[ProductMigration, ...] = (), migrated: tuple[str, ...] = (), settled: tuple[str, ...] = (), conflicting: tuple[str, ...] = (), orphan_temporaries: tuple[str, ...] = (), unrecognized: tuple[str, ...] = (), deletion_blocked: tuple[str, ...] = (), failures: tuple[str, ...] = (), manifest_repairs: tuple[ProductMigration, ...] = ()) -> None

What a canonical root needs before its products match their declarations.

PROJECTIONS = <builtins.dict>

Recontainment(stage: str, from_suffix: str, to_suffix: str, keep_ids: tuple[str, ...] | None) -> None

One stage’s move from what is on disk to what it now declares.

sweep_superseded_products(report: ProductMigrationReport, *, apply: bool = False) -> dict[str, Any]

Delete the originals the migration superseded. Irreversible.

mhd_linear_run_coverage_model(manifest: Mapping[str, Any])

Build the DCON/RDCON/STRIDE run-coverage view model (issue #173 phase 1).

Also exported as vaft.validation.mhd_linear_run_coverage_model (deprecated).

raw_acquisition_qa_model(payload: Mapping[str, Any], *, required_fields: Sequence[int] = (), shot: int | None = None, max_signal_panels: int = 4)

Build the compact raw-acquisition QA view model for one shot.

Also exported as vaft.validation.raw_acquisition_qa_model (deprecated).

render_stage_plots(stage: str, source: Any, output_dir: str | os.PathLike[str], **context: Any) -> dict[str, Any]

Render and persist stage’s validation plots into output_dir.

Also exported as vaft.validation.render_stage_plots (deprecated).

stage_plot_filenames(stage: str, *, required_only: bool = False) -> tuple[str, ...]

File names stage writes into its plot/ directory.

Also exported as vaft.validation.stage_plot_filenames (deprecated).

STAGE_VALIDATION_PLOTS = <builtins.dict>

stages() -> tuple[str, ...]

Every stage with a declared validation-plot set, sorted.

Also exported as vaft.validation.stages (deprecated).

validation_plots(stage: str) -> tuple[ValidationPlot, ...]

The declared plots for stage.

Also exported as vaft.validation.validation_plots (deprecated).

ValidationPlot(plot: str, filename: str = '', required: bool = True, kind: str = 'ods', options: Mapping[str, Any] = <factory>) -> None

One figure a stage owes after its data product is generated.

Also exported as vaft.validation.ValidationPlot (deprecated).

ChainLink(stage: str, upstream_stage: str, consumed_sha256: str, produced_by_upstream_sha256: str, status: str, detail: str = '') -> None

One consumed-produced pair, and whether the two agree.

ChainReport(shot: int | None, links: tuple[ChainLink, ...]) -> None

Whether a product’s recorded ancestry holds together.

NOT_RECORDED = ''

verify_chain(*, chease_manifest: dict[str, Any], stability_input_sha256: str, efit_gfile_sha256: str | None = None, shot: int | None = None) -> ChainReport

Check a stability product’s recorded ancestry against its upstreams.

compare_db_and_dumped_raw_signals_for_shot(shot: int, output_path: str = None, max_retries: int = 3, daq_type: int = 0, slow_dt_threshold: float = 5e-06) -> bool

Compare and plot original signals from database with signals loaded from JSON file

CONFIG_FILE = '/home/runner/.vest/database_raw_info.yaml'

configuration() -> Tuple[str, str, str, str]

Load database configuration.

date_from_shot(shot: int) -> tuple

Returns (date_str, datetime_obj) for the given shot number from the shot table.

dump_all_raw_signals_for_shot(shot: int, output_path: str = None, max_retries: int = 3, daq_type: int = 0, slow_dt_threshold: float = 5e-06, plot_opt: bool = False, sample_opt: bool | RawSource = False, fields: Optional[List[int]] = None) -> bool

Store shot data as JSON GZIP file (.json.gz) with the following steps: 1. Retrieve list of field codes 2. Load (time, data1D) using load_raw 3. Classify as fast/slow based on sampling interval (time[1]-time[0]) 4. Create shot_data = { “shot”: shot, “fields”: {fcode: {type, data}} } 5. Save as gzip compressed JSON (.json.gz) 6. If plot_opt is True, display and save signals as subplots

EXCLUDED_FIELD_CODES = set({110, 111, 112, 113})

FAST_DT = 4e-06

FAST_SPAN_S = 0.1

get_all_field_codes_for_shot(shot: int, max_retries: int = 3)

Returns all field codes used in the given shot.

init_pool() -> None

Initialize the global MySQLConnectionPool.

KEY_FILE = '/home/runner/.vest/encryption_key.key'

last_shot() -> int

Returns the maximum shotCode from the shot table.

list_shots(*, shot_min: int | None = None, shot_max: int | None = None, start_date: str | date | None = None, end_date: str | date | None = None) -> list[tuple[int, datetime]]

Return (shotNumber, recordDateTime) pairs from the SQL shot table.

load(shot: int, field: int, max_retries: int = 3)

Legacy alias kept for existing machine_mapping wrappers.

load_or_generate_key() -> bytes

Load or generate an encryption key.

load_raw(shot: int, fields: Optional[Union[int, List[int]]] = None, max_retries: int = 3, daq_type: Optional[int] = None, sample_opt: bool | RawSource = False) -> Optional[Tuple[np.ndarray, np.ndarray]]

High-level data loader for the VEST database.

MAX_RETRIES = 3

Fields with different sampling cadences were batch-loaded together.

name(field: int) -> tuple

Retrieves the shotDataFieldName and shotDataFieldRemark from the ‘shotDataField’ table for a given field code.

plot(shots, fields, semilogy_opt: bool = False, norm_opt: bool = False, xlims=None, *, ax=None, show: bool = True)

Plots raw waveforms for the 3 standard scenarios:

POOL_SIZE = 4

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

Raised when a required VEST raw waveform is absent or unusable.

RawSource = <typing.Union>

require_signal(loaded: Optional[Tuple[np.ndarray, np.ndarray]], *, shot: int, field: int | str, signal_name: str | None = None, min_samples: int = 2) -> Tuple[np.ndarray, np.ndarray]

Validate and return a required single-channel raw waveform.

SecureConfigManager()

setup_raw_db() -> None

Initialize database configuration.

shots_from_date(date_str: str) -> list

Returns a list of shotNumbers for the given date (YYYY-MM-DD).

SLOW_DT = 4e-05

SLOW_DT_THRESHOLD = 5e-06

SLOW_SPAN_S = 1.0

sql_loading_available() -> bool

Return whether SQL-backed waveform loading is available in this env.

SQL_TABLE_PATH = <pathlib.PosixPath>

upgrade_archive_timebase(payload: dict) -> dict

Add self-describing t0/dt entries to a legacy raw archive.

V2_LAST_SHOT = 42190

vest_check_table(mydb: Any, shot: int) -> int

Return the waveform table generation used by a shot.

vest_connection_pool(pool_size: int = 4) -> None

Compatibility wrapper for donor-style SQL pool initialization.

vest_date(shot: int) -> Optional[str]

Return shot date as YYYY-MM-DD.

vest_load(shot: int, field: int, max_retries: int = 3, sample_opt: bool | RawSource = False) -> Optional[Tuple[np.ndarray, np.ndarray]]

Compatibility wrapper for donor-style single-signal loading.

vest_load_by_name(shot: int, name: str, sample_opt: bool | RawSource = False) -> Optional[Tuple[np.ndarray, np.ndarray]]

Load a waveform by signal name using the shipped lookup table.

Also exported as vaft.database.raw.vest_loadn.

vest_load_shot_waveform_2(mydb: Any, shot: int, field: int) -> Tuple[np.ndarray, np.ndarray]

Compatibility wrapper for donor-style waveform table 2 access.

Also exported as vaft.database.raw.vest_load_shotWaveform_2.

vest_load_shot_waveform_3(mydb: Any, shot: int, field: int) -> Tuple[np.ndarray, np.ndarray]

Compatibility wrapper for donor-style waveform table 3 access.

Also exported as vaft.database.raw.vest_load_shotWaveform_3.

vest_shots(date: str) -> list[int]

Return shot list for a date string.

vestdb_is_data_exist(shot_code: int, shot_data_field_code: int) -> bool

Return whether a signal exists for a shot in raw storage.

waveform_generation_for_shot(conn: Any, shot: int) -> Optional[int]

Return which waveform table holds shot, or None if neither does.

is_reusable(record: ReplicationRecord | None, *, product_sha256: str, require_validation: bool) -> bool

Whether a recorded replication still stands for the current product.

merge_remote_master(source: str, shot: int, previous_master: Path | None) -> tuple[str, ...]

Fold a pre-write master’s links back into the one the write just left.

NOTHING_TO_REPLICATE_STATUSES = frozenset({'no_output'})

Raised when the local stage product is not finished enough to send.

read_record(path: Path) -> ReplicationRecord | None

Return a previously written record, or None when there is none.

REPLICABLE_STATUSES = frozenset({'completed', 'partial', 'success'})

replicate_stage(stage: str | OMASStage, shot: int, *, filedb: FileDB, family: str = 'magnetic', refinement: str = 'chease', product: str | None = None, attempts: int = 3, retry_delay: float = 5.0, validate: bool = True, force: bool = False) -> ReplicationRecord

Replicate one completed canonical FileDB stage product into HSDS.

REPLICATION_SCHEMA_VERSION = 1

Raised when a stage product cannot be replicated.

ReplicationRecord(stage: str, shot: int, source: str, remote_uri: str, ids: tuple[str, ...], occurrence: int, product_sha256: str, state: str, attempts: int, started_at: str, completed_at: str | None = None, round_trip: dict[str, Any] | None = None, error: str | None = None, schema_version: int = 1, vaft_version: str = <factory>) -> None

What happened when one stage product was sent to a named HSDS source.

Raised when the replica does not match what was sent.

Raised when a stage has no destination or no wired rule.

write_record(record: ReplicationRecord, path: Path) -> Path

copy_refinement(shot: int, *, source: str = 'chease-mhd-stability', apply: bool = False) -> dict[str, Any]

Copy one shot’s CHEASE refinement into main/chease, verified.

delete_retired_source(report: RetirementReport, *, apply: bool = False) -> dict[str, Any]

Delete the combined source, only once its report says nothing is lost.

plan_retirement(shots: Iterable[int], *, products: Iterable[str] = ('dcon-peeling', 'dcon-kink', 'rdcon', 'stride'), source: str = 'chease-mhd-stability') -> RetirementReport

Say what each shot still needs before the combined source can go.

REFINEMENT_DESTINATION = 'main/chease'

REFINEMENT_IDS = ('equilibrium',)

Raised when a source cannot be retired safely.

RetirementReport(source: str, applied: bool, shots: tuple[ShotRetirement, ...], unexamined: tuple[int, ...] = (), enumeration_error: str | None = None) -> None

Whether the combined source still holds anything unique.

RETIRING_SOURCE = 'chease-mhd-stability'

ShotRetirement(shot: int, refinement: str, stability: str, detail: str) -> None

What stands between one shot and the old source being deletable.

STABILITY_IDS = ('mhd_linear', 'ntms')

ATLAS_LINEAGE_FAMILIES = <builtins.dict>

CATALOG = <builtins.dict>

children(name: str, *, environment: Mapping[str, str] | None = None) -> tuple[HSDSSource, ...]

The sources that hang directly beneath name, in catalog order.

DEFAULT_SOURCE = 'main'

describe(name: str, *, environment: Mapping[str, str] | None = None) -> HSDSSource

Return the catalog entry for one resolved namespace name.

EXTRA_SOURCES_VARIABLE = 'VAFT_HSDS_EXTRA_SOURCES'

family_for_atlas_lineage(lineage: str) -> str

The equilibrium family one routine-TGLF lineage name is filed under.

FAMILY_LINEAGE_NAMES = <builtins.dict>

FAMILY_SOURCES = <builtins.dict>

HSDSSource(name: str, purpose: str, writable: bool = True, sparse: bool = False, parent: str | None = None, projects_to: str | None = None, transitional_until: str | None = None) -> None

One named HSDS namespace and the lineage it stores.

Base error for an unusable HSDS source name.

is_writable(name: str, *, environment: Mapping[str, str] | None = None) -> bool

Return whether VAFT may publish into name.

known_sources(*, environment: Mapping[str, str] | None = None) -> tuple[HSDSSource, ...]

Return every source a call may name right now, catalog order first.

LEGACY_SOURCE = 'public'

MissingSourceError(source: str, detail: str = '') -> None

Raised when a named source has no folder on the HSDS deployment.

OccurrenceMigration(source: str, ids: str, name: str, occurrence: int | None = None, allocated_by: str = 'the #1129 VEST deployment catalog (not assigned yet)') -> None

Where one transitional source’s IDS goes in the #1128/#1129 model.

parse_tglf_configuration(name: Any) -> tuple[int, str]

Return (sat_rule, field_model) for a configuration name, or raise.

Raised when a write is requested against a read-only source.

replicable_stages(*, produced_by: str | None = None) -> tuple[str, ...]

Return the stages with a destination and a wired replication rule.

replication_for_stage(stage: Any, *, family: str = 'magnetic', refinement: str = 'chease', product: str | None = None) -> StageReplication

Return the replication contract for one canonical FileDB OMAS stage.

resolve(source: str | None = None, *, directory: str | None = None, target: str | None = None, default: str | None = 'main', writable: bool = False, label: str = 'source', stacklevel: int = 3, environment: Mapping[str, str] | None = None) -> str

Resolve one namespace from source and its deprecated aliases.

source_for_stage(stage: Any, *, family: str = 'magnetic', refinement: str = 'chease', product: str | None = None) -> str

Return the HSDS source a canonical FileDB OMAS stage is replicated into.

STAGE_REPLICATION = <builtins.dict>

STAGE_SOURCE = <builtins.dict>

StageReplication(source: str | None, ids: tuple[str, ...] = (), occurrence: int = 0, note: str = '', deferred_to: str | None = None, optional: bool = False, produced_by: str = 'routine') -> None

How one canonical FileDB OMAS stage reaches a remote backend.

tglf_configuration(sat_rule: int, field_model: str) -> str

tglf-sat<n>-<field>: the name of one TGLF configuration (#1482).

tglf_configurations() -> tuple[str, ...]

Every TGLF configuration name, ordered by SAT rule then field model.

TGLF_FIELD_MODELS = ('es', 'em-bper', 'em-bper-bpar')

TGLF_SAT_RULES = (0, 1, 2, 3)

TRANSITIONAL_UNTIL = '#1128/#1129'

turbulent_transport_migration() -> tuple[OccurrenceMigration, ...]

Every transitional TGLF source with the occurrence name it migrates to.

turbulent_transport_occurrence_name(family: str, configuration: str) -> str

The future ids_properties.name of one TGLF configuration’s IDS.

turbulent_transport_source(family: str, configuration: str) -> str

The transitional HSDS source of one TGLF configuration on one family.

Raised when a name is not in the catalog and was not opted into.

create_partial_master(source_master: Path, destination_master: Path, requested_ids: Iterable[str]) -> tuple[str, ...]

Copy source_master and retain only requested IDS external links.

external_h5_links(master_path: Path) -> list[str]

Return filenames referenced by top-level external links in master.

HSDSDomainCache(directory: str, shot: int, cache: str | Path = 'auto') -> None

Persistent validated cache for HSDS HDF5 images of one shot.

merge_master_links(current_master: Path, previous_master: Path | None, *, present_files: Iterable[str]) -> tuple[str, ...]

Fold a previous master’s IDS links back into a freshly written one.

requested_ids_from_paths(paths: Iterable[Any] | None) -> tuple[str, ...] | None

Infer requested top-level IDS names from OMAS/IMAS paths.

stage_imas_shot(directory: str, shot: int, staging_dir: Path, *, requested_ids: Iterable[str] | None, cache: str | Path = 'auto', transport: Literal['auto', 'canonical', 'h5image'] = 'auto') -> dict[str, Any]

Materialize a complete or IDS-selective local IMAS HDF5 shot.

HSDSCommandError(command: str, exit_code: int, detail: str = '') -> None

Raised when an HSDS CLI process exits unsuccessfully.

Raised when a required h5pyd CLI executable is not on PATH.

Base error raised when an HSDS command cannot complete.

Raised when an uploaded domain differs from its local HDF5 source.

run_hsget(remote_uri: str, out_path: str | Path) -> Path

Materialize one remote HSDS domain as a local HDF5 file.

run_hsload(local_path: str | Path, remote_uri: str) -> str

Upload one local HDF5 file as an HSDS domain.

verify_uploaded_image(local_path: str | Path, remote_uri: str, *, h5pyd_module: object | None = None) -> None

Compare critical IMAS metadata after hsload transcoding.

CP_REGISTRY_GROUP = 'cp_shots'

CX_REGISTRY_GROUP = 'cx_shots'

ensure_imas_hdf5_userblock(path: Union[str, Path], entry_dir: Union[str, Path]) -> None

Repair an IMAS HDF5 image downloaded from HSDS so IMAS-Core can open it.

exist_shot(source: Optional[str] = None, shot: Optional[int] = None, data_filter: Optional[str] = None, sort: int = -1, *, username: Optional[str] = None, strict: bool = False) -> Union[List[str], bool, pd.DataFrame, None]

Return a list of shot names or processed-diagnostic data from HSDS.

is_connect() -> bool

Check if the user is connected to the HSDS server.

processed_registry_uri(source: Optional[str] = None, *, writable: bool = False) -> str

Return the legacy corrective-pipeline registry domain for one named source.

read_legacy_processed_registry(group: str) -> dict

Return the legacy public registry entries for one diagnostic group.

require_source_exists(source: str) -> None

Fail with the administrator fix when a source folder is not there yet.

TS_REGISTRY_GROUP = 'shots'

Classification(label: str, reason: str, applicable_stages: tuple[str, ...] | None) -> None

Also exported as vaft.database.worker.Classification.

Classifier(*args, **kwargs)

Also exported as vaft.database.worker.Classifier.

DAQ_MISSING = 'daq_missing'

load_classifier(spec: str | None, pipeline_config: Mapping[str, Any]) -> Classifier

Resolve module:callable to a classifier, or the raw-field default.

RAW_INCOMPLETE = 'raw_incomplete'

RAW_ONLY = ('raw',)

RawFieldClassifier(required_fields: Sequence[int] = (1, 12, 25, 59, 109))

Decide from the SQL field inventory alone.

Also exported as vaft.database.worker.RawFieldClassifier.

ShotObservation(shot: int, record_datetime: datetime | str | None, field_codes: frozenset[int]) -> None

What SQL says about one settled shot.

Also exported as vaft.database.worker.ShotObservation.

UNCLASSIFIED = 'unclassified'

CONFIG_ENVIRONMENT_VARIABLE = 'VAFT_WORKER_CONFIG'

load_pipeline_config(config: WorkerConfig) -> dict[str, Any]

The configuration Snakemake will actually run with.

load_worker_config(path: str | os.PathLike[str] | None = None) -> WorkerConfig

Read the worker configuration from path or $VAFT_WORKER_CONFIG.

Also exported as vaft.database.worker.load_worker_config.

worker_config_from_mapping(data: Mapping[str, Any], *, base_dir: Path) -> WorkerConfig

Build a WorkerConfig from parsed YAML.

WorkerConfig(state_db: Path, log_dir: Path, workflow_dir: Path, pipeline_config: Path, first_shot: int, poll_interval: float, quiet_seconds: float, cores: int, snakemake_cmd: tuple[str, ...], extra_args: tuple[str, ...] = (), max_shots_per_run: int = 20, max_attempts: int = 3, recheck_seconds: float = 259200.0, max_reprocess: int = 3, run_timeout: float | None = None, env: Mapping[str, str] = <factory>, classifier: str | None = None, record_shot_class: bool = False, stages: tuple[str, ...] | None = None, min_free_gb: float | None = None, resume_free_gb: float | None = None, disk_paths: tuple[Path, ...] = ()) -> None

What one deployment of the worker runs, and where it keeps its state.

Also exported as vaft.database.worker.WorkerConfig.

Raised when the worker configuration cannot be used.

Also exported as vaft.database.worker.WorkerConfigError.

CONFIG_ENVIRONMENT_VARIABLE = 'VAFT_CORRECTIVE_WORKER_CONFIG'

corrective_config_from_mapping(data: Mapping[str, Any], *, base_dir: Path) -> CorrectiveWorkerConfig

Build a CorrectiveWorkerConfig from parsed YAML; relative paths resolve against base_dir.

CorrectiveCycleReport(recovered: list[int] = <factory>, scanned: int = 0, new: list[int] = <factory>, changed: list[int] = <factory>, run_id: int | None = None, run_shots: list[int] = <factory>, busy: bool = False, held: bool = False, scan_error: str | None = None, outcomes: dict[int, str] = <factory>, gave_up: list[int] = <factory>) -> None

CorrectiveHarvester(pipeline_config: Mapping[str, Any], *, paths_workflow_dir: Path, environment: Mapping[str, str] | None = None)

Judge each shot by the manifests of the stages in scope, never by an exit code.

CorrectivePlan(run_id: int, full_shots: tuple[int, ...], file_targets: tuple[str, ...] = (), force: bool = False) -> None

A RunPlan that may force the root rules (a changed or failed shot is not up to date).

CorrectiveRunner(config: WorkerConfig, pipeline_config: Mapping[str, Any])

The routine worker’s runner, plus --forcerun on the two root rules when the plan asks.

CorrectiveState(path: str | os.PathLike[str], *, read_only: bool = False)

The corrective worker’s SQLite file: one row per shot with an upload, plus runs and events.

CorrectiveWorker(config: CorrectiveWorkerConfig, *, state: CorrectiveState | None = None, scanner: Any = None, runner: Any = None, harvester: CorrectiveHarvester | None = None, pipeline_config: Mapping[str, Any] | None = None, clock: Callable[[], datetime] = <built-in method now of type object>)

Watch the uploads, run pipeline 2, harvest. Collaborators are injectable for tests.

CorrectiveWorkerConfig(state_db: Path, log_dir: Path, workflow_dir: Path, pipeline_config: Path, paths_workflow_dir: Path, first_shot: int, poll_interval: float, quiet_seconds: float, cores: int, snakemake_cmd: tuple[str, ...], last_shot: int | None = None, extra_args: tuple[str, ...] = (), max_shots_per_run: int = 20, max_attempts: int = 3, run_timeout: float | None = None, env: Mapping[str, str] = <factory>, stages: tuple[str, ...] | None = None, hold_file: Path | None = None) -> None

What one deployment of the corrective worker runs, and where it keeps its state.

load_corrective_config(path: str | os.PathLike[str] | None = None) -> CorrectiveWorkerConfig

Read the configuration from path or $VAFT_CORRECTIVE_WORKER_CONFIG.

load_corrective_pipeline_config(config: CorrectiveWorkerConfig) -> dict[str, Any]

The configuration Snakemake will run with: the workflow’s own, overlaid by the deployment’s.

read_corrective_state(path: str | os.PathLike[str]) -> Iterator[CorrectiveState]

Open the state file read-only, for monitoring.

SHOT_STATES = ('pending', 'running', 'completed', 'unavailable', 'failed', 'gave_up', 'excluded')

ShotOutcome(state: str, reason: str, stages: tuple[StageStatus, ...]) -> None

UploadScanner(data_root: str | os.PathLike[str])

List the Thomson and CES uploads under data_root and group them by shot.

UploadSignature(files: tuple[tuple[str, int, int], ...]) -> None

The files uploaded for one shot, as (relative name, size, mtime_ns).

MATCH_QUIET_SECONDS = 30.0

require_sql_credentials() -> None

Refuse to start without provisioned SQL credentials.

ShotSource(*args, **kwargs)

SqlShotSource() -> None

The live VEST SQL database.

upload_finished(*, field_codes: frozenset[int], reference: frozenset[int] | None, quiet_seconds: float | None, quiet_limit: float) -> str | None

Why a shot’s upload counts as finished, or None while it may continue.

live_snakemake_in(workflow_dir: Path) -> bool | None

Is any Snakemake process working in workflow_dir right now?

RunPlan(run_id: int, full_shots: tuple[int, ...], file_targets: tuple[str, ...] = ()) -> None

RunResult(exit_code: int | None, log_path: str, busy: bool = False, timed_out: bool = False, interrupted: bool = False) -> None

SnakemakeRunner(config: WorkerConfig, pipeline_config: Mapping[str, Any])

Builds and runs the Snakemake command for a RunPlan.

CycleReport(recovered: list[int] = <factory>, detected: list[int] = <factory>, queued: list[int] = <factory>, excluded: list[int] = <factory>, run_id: int | None = None, run_shots: list[int] = <factory>, busy: bool = False, outcomes: dict[int, str] = <factory>, gave_up: list[int] = <factory>, reprocessed: list[int] = <factory>, disk_paused: str | None = None) -> None

Also exported as vaft.database.worker.CycleReport.

PipelineWorker(config: WorkerConfig, *, state: S.WorkerState | None = None, source: ShotSource | None = None, classifier: Classifier | None = None, runner: Runner | None = None, harvester: PipelineHarvester | None = None, pipeline_config: Mapping[str, Any] | None = None, clock: Callable[[], datetime] = <built-in method now of type object>, disk_free: Callable[[str], int] = <function PipelineWorker.<lambda>>)

Poll SQL, classify, run and harvest. Collaborators are injectable for tests.

Also exported as vaft.database.worker.PipelineWorker.

COMPLETED = 'completed'

DETECTED = 'detected'

EXCLUDED = 'excluded'

FAILED = 'failed'

GAVE_UP = 'gave_up'

PARTIAL = 'partial'

QUEUED = 'queued'

read_worker_state(path: str | os.PathLike[str]) -> WorkerState

Open a worker’s state read-only – the monitoring entry point (#1347).

Also exported as vaft.database.worker.read_worker_state.

RUNNING = 'running'

SHOT_STATES = ('detected', 'queued', 'running', 'completed', 'partial', 'excluded', 'failed', 'gave_up')

Also exported as vaft.database.worker.SHOT_STATES.

WorkerState(path: str | os.PathLike[str], *, readonly: bool = False)

The worker’s SQLite store. Every mutation is one transaction.

Also exported as vaft.database.worker.WorkerState.

load_pipeline_paths_module(workflow_dir: Path) -> ModuleType

Import the workflow’s paths.py from the configured checkout.

MANIFEST = 'manifest'

MISSING = 'missing'

OK_STATUSES = frozenset({'completed', 'present', 'success', 'validated'})

PipelineHarvester(pipeline_config: Mapping[str, Any], *, workflow_dir: Path, environment: Mapping[str, str] | None = None)

Evaluate each shot against the routine pipeline’s declared products.

REPLICATION = 'replication'

ShotOutcome(state: str, reason: str, stages: tuple[StageStatus, ...]) -> None

StageStatus(stage: str, product: str | None, kind: str, status: str, path: str) -> None

StageTarget(stage: str, product: str | None, kind: str, path: str) -> None