scpn_fusion.io – Data Interop

The IO subpackage provides data-interoperability adapters for the IMAS (Integrated Modelling & Analysis Suite) data exchange standard used in the fusion community.

IMAS Connector

Facade API for IMAS/IDS adapter modules.

This module intentionally stays as a stable import surface while implementation is decomposed into focused submodules:

  • imas_connector_common: validation/coercion primitives

  • imas_connector_digital_twin: summary/state IDS mappings

  • imas_connector_equilibrium: GEQDSK <-> IMAS equilibrium

  • imas_connector_transport: core_profiles/summary/core_transport

  • imas_connector_storage: JSON I/O helpers

  • imas_connector_omas: OMAS bridge

  • omas_free_boundary_inputs: strict PF/magnetics acquisition contract

scpn_fusion.io.imas_connector.validate_ids_payload(payload)[source]

Validate a complete IDS payload shape and units.

Ensures required top-level fields, nested structures, and numeric coercion constraints are satisfied. This function performs strict checks for time-slice ordering at the millisecond level and required equilibrium / performance keys.

Return type:

None

Parameters:

payload (Mapping[str, Any])

scpn_fusion.io.imas_connector.digital_twin_summary_to_ids(summary, *, machine='ITER', shot=0, run=0)[source]

Map an internal digital-twin summary into an IDS-like payload.

Returns a shallow canonicalised mapping that includes the equilibrium and performance fields required by validate_ids_payload().

Return type:

dict[str, Any]

Parameters:
scpn_fusion.io.imas_connector.digital_twin_state_to_ids(state, *, machine='ITER', shot=0, run=0)[source]

Map a detailed digital-twin state + profiles into an IDS-like payload.

Accepts full internal state records, including optional 1-D profile fields, and delegates to digital_twin_summary_to_ids() before attaching the validated equilibrium.profiles_1d payload.

Return type:

dict[str, Any]

Parameters:
scpn_fusion.io.imas_connector.ids_to_digital_twin_summary(payload)[source]

Convert an IDS-like payload into internal digital-twin summary shape.

Return type:

dict[str, Any]

Parameters:

payload (Mapping[str, Any])

scpn_fusion.io.imas_connector.ids_to_digital_twin_state(payload)[source]

Convert IDS payload into detailed digital-twin state with optional profiles.

Return type:

dict[str, Any]

Parameters:

payload (Mapping[str, Any])

scpn_fusion.io.imas_connector.digital_twin_history_to_ids(history, *, machine='ITER', shot=0, run=0)[source]

Convert a local digital-twin history into IDS payload sequence form.

Return type:

list[dict[str, Any]]

Parameters:
scpn_fusion.io.imas_connector.digital_twin_history_to_ids_pulse(history, *, machine='ITER', shot=0, run=0)[source]

Convert digital-twin history into a single IDS pulse payload.

Return type:

dict[str, Any]

Parameters:
scpn_fusion.io.imas_connector.ids_to_digital_twin_history(payloads)[source]

Convert a sequence of IDS payloads back to digital-twin snapshots.

Return type:

list[dict[str, Any]]

Parameters:

payloads (Sequence[Mapping[str, Any]])

scpn_fusion.io.imas_connector.ids_pulse_to_digital_twin_history(pulse)[source]

Convert IDS pulse payload into a sequence of digital-twin snapshots.

Return type:

list[dict[str, Any]]

Parameters:

pulse (Mapping[str, Any])

scpn_fusion.io.imas_connector.validate_ids_payload_sequence(payloads)[source]

Validate a strictly monotonic sequence of IDS payloads.

The function enforces: - required payload schema fields - shared machine / shot / run identity across the sequence - strictly increasing time index and time value

Return type:

None

Parameters:

payloads (Sequence[Mapping[str, Any]])

scpn_fusion.io.imas_connector.validate_ids_pulse_payload(pulse)[source]

Validate IDS pulse payload integrity and consistency constraints.

Return type:

None

Parameters:

pulse (Mapping[str, Any])

scpn_fusion.io.imas_connector.geqdsk_to_imas_equilibrium(eq, *, time_s=0.0, shot=0, run=0)[source]

Convert a GEqdsk equilibrium to an IMAS Data Dictionary equilibrium IDS.

Return type:

dict[str, Any]

Parameters:
scpn_fusion.io.imas_connector.imas_equilibrium_to_geqdsk(ids)[source]

Convert an IMAS Data Dictionary equilibrium IDS back to a GEqdsk.

Return type:

GEqdsk

Parameters:

ids (Mapping[str, Any])

scpn_fusion.io.imas_connector.state_to_imas_core_profiles(state, *, time_s=0.0)[source]

Convert a plasma state dict to an IMAS core_profiles IDS.

Return type:

dict[str, Any]

Parameters:
scpn_fusion.io.imas_connector.state_to_imas_summary(state)[source]

Convert a performance/state dict to an IMAS summary IDS.

Return type:

dict[str, Any]

Parameters:

state (Mapping[str, Any])

scpn_fusion.io.imas_connector.state_to_imas_core_transport(state, *, time_s=0.0)[source]

Convert a plasma state dict to an IMAS core_transport IDS.

Return type:

dict[str, Any]

Parameters:
scpn_fusion.io.imas_connector.imas_core_transport_to_state(ids)[source]

Convert an IMAS core_transport IDS back to a state dict.

Return type:

dict[str, Any]

Parameters:

ids (Mapping[str, Any])

scpn_fusion.io.imas_connector.write_ids(ids_dict, path)[source]

Write an IDS dict to a JSON file with schema validation.

Return type:

None

Parameters:
scpn_fusion.io.imas_connector.read_ids(path)[source]

Read an IDS JSON file and validate minimal schema.

Return type:

dict[str, Any]

Parameters:

path (str | Path)

scpn_fusion.io.imas_connector.ids_to_omas_core_profiles(ids_dict)[source]

Convert an IMAS core_profiles IDS dict to an OMAS ODS.

Return type:

Any

Parameters:

ids_dict (Mapping[str, Any])

scpn_fusion.io.imas_connector.ids_to_omas_equilibrium(ids_dict)[source]

Convert an IMAS equilibrium IDS dict to an OMAS ODS.

Return type:

Any

Parameters:

ids_dict (Mapping[str, Any])

scpn_fusion.io.imas_connector.omas_core_profiles_to_ids(ods)[source]

Convert OMAS ODS core_profiles data back to an IDS dict.

Return type:

dict[str, Any]

Parameters:

ods (Any)

scpn_fusion.io.imas_connector.omas_equilibrium_to_ids(ods)[source]

Convert OMAS ODS equilibrium data back to an IDS dict.

Return type:

dict[str, Any]

Parameters:

ods (Any)

class scpn_fusion.io.imas_connector.FluxLoopInput(identifier, r_m, z_m, flux_wb)[source]

Bases: object

Flux-loop geometry and poloidal-flux history.

Parameters:
identifier: str
r_m: tuple[float, ...]
z_m: tuple[float, ...]
flux_wb: TimeSeriesSI
class scpn_fusion.io.imas_connector.OmasFreeBoundaryInputs(schema, cocos, imas_version, provenance, time_alignment, pf_coils, bpol_probes, flux_loops, ingestion_blockers, ingestion_ready, tier0_claim_blockers, tier0_claim_admission_ready, payload_sha256)[source]

Bases: object

Validated OMAS channels with distinct ingestion and Tier-0 states.

Parameters:
schema: str
cocos: int
imas_version: str | None
provenance: OmasSourceProvenance | None
time_alignment: Literal['native_unaligned', 'exact_common_axis']
pf_coils: tuple[PfCoilInput, ...]
bpol_probes: tuple[PoloidalFieldProbeInput, ...]
flux_loops: tuple[FluxLoopInput, ...]
ingestion_blockers: tuple[str, ...]
ingestion_ready: bool
tier0_claim_blockers: tuple[str, ...]
tier0_claim_admission_ready: Literal[False]
payload_sha256: str
class scpn_fusion.io.imas_connector.OmasSourceProvenance(machine, shot_id, run_id, source_uri, source_sha256, license_id)[source]

Bases: object

External binding that an ODS alone cannot prove reliably.

Parameters:

  • machine (str)

  • shot_id (int)

  • run_id (int)

  • source_uri (str)

  • source_sha256 (str)

  • license_id (str)

machine: str
shot_id: int
run_id: int
source_uri: str
source_sha256: str
license_id: str
class scpn_fusion.io.imas_connector.PfCoilInput(identifier, elements, current_a)[source]

Bases: object

PF-coil current history and matching signed geometry.

Parameters:
identifier: str
elements: tuple[PfElementGeometry, ...]
current_a: TimeSeriesSI
class scpn_fusion.io.imas_connector.PfElementGeometry(identifier, turns_with_sign, geometry_type, r_m, z_m, width_m, height_m)[source]

Bases: object

One signed PF element with an explicit IMAS geometry representation.

Parameters:
identifier: str
turns_with_sign: float
geometry_type: Literal[1, 2]
r_m: tuple[float, ...]
z_m: tuple[float, ...]
width_m: float | None
height_m: float | None
class scpn_fusion.io.imas_connector.PoloidalFieldProbeInput(identifier, r_m, z_m, poloidal_angle_rad, length_m, field_t)[source]

Bases: object

Poloidal-field probe position, orientation, and field history.

Parameters:
identifier: str
r_m: float
z_m: float
poloidal_angle_rad: float
length_m: float
field_t: TimeSeriesSI
class scpn_fusion.io.imas_connector.TimeSeriesSI(time_s, values, error_lower, error_upper, validity)[source]

Bases: object

One finite, strictly ordered SI-unit time series.

Parameters:
time_s: tuple[float, ...]
values: tuple[float, ...]
error_lower: tuple[float, ...] | None
error_upper: tuple[float, ...] | None
validity: int | None
scpn_fusion.io.imas_connector.extract_omas_free_boundary_inputs(ods, *, provenance=None, cocos=None, time_alignment='native_unaligned', require_ingestion_ready=True)[source]

Extract strict SI-unit PF and magnetics channels from an OMAS ODS.

Parameters:

  • ods (ODSLike) – OMAS ODS (or compatible dotted-path mapping). IMAS schema units are preserved: seconds, amperes, metres, radians, tesla, and webers.

  • provenance (OmasSourceProvenance | None) – Immutable source binding supplied by the acquisition layer.

  • cocos (int | None) – Canonical COCOS index. When omitted, ods.cocos is used.

  • time_alignment (Literal['native_unaligned', 'exact_common_axis']) – Declare whether every channel was acquired on one exact time axis. The adapter verifies an exact_common_axis declaration byte-for-byte.

  • require_ingestion_ready (bool) – Raise when any provenance, uncertainty, validity, channel, or alignment gate is missing. Set false only for explicit development inspection.

Returns:

Immutable extracted inputs, ingestion blockers, readiness state, and digest. Full Tier-0 claim admission remains explicitly false.

Return type:

OmasFreeBoundaryInputs

Raises:

ValueError – If the ODS is structurally malformed or strict ingestion is blocked.