API Reference¶

Library entry points for programmatic use of tokamunch.

Context ¶

class tokamunch.MappingContext(mapper: Any, tokamap: TokamapInterface, device: str, shot: int | None, cli_config: CLIConfig | None = None, concurrency: ConcurrencyConfig = <factory>)¶

Holds all runtime state needed to execute a mapping run.

Can be constructed directly for programmatic (library) use, or via from_config to load settings from a TOML config file for CLI use.

cli_config is required for process-based concurrency: each worker process receives the full CLIConfig and reconstructs its own mapper without needing any config file to be present on disk.

cli_config: CLIConfig | None = None¶

concurrency: ConcurrencyConfig¶

device: str¶

classmethod from_config(config: str, *, device: str | None = None, shot: int | None = None) → MappingContext¶

Build a MappingContext from a munchi TOML config file.

device and shot override the values in the config when provided.

ids_helper(ids_name: str) → IDSHelper¶

mapper: Any¶

shot: int | None¶

tokamap: TokamapInterface¶

Selection ¶

class tokamunch.IdsSelection(ids: str, match: str | None = None, leaves_only: bool = False, mapping_keys: frozenset[str] | None = None)¶: Expand all concrete paths for an IDS, with optional filtering.

class tokamunch.SinglePathSelection(path: str, mapping_keys: frozenset[str] | None = None)¶: Map a single known concrete path.

tokamunch.selection.generate_selected_paths(selection: Selection, ctx: MappingContext) → Iterator[str]¶

IDS schema helpers ¶

class tokamunch.IDSHelper(ids_paths: Iterable[str])¶

reset_expansion_cache() → None¶: Discard all cached array-length results so the next expansion re-queries.

tokamunch.generate_ids_paths(ids_name: str) → Iterator[str]¶

Mapper interface ¶

class tokamunch.TokamapInterface(mapper: MapperProtocol, device: str, *, shot: int | None = None, extra_args: dict[str, Any] | None = None)¶

class tokamunch.MapperProtocol(*args, **kwargs)¶

Protocol for the underlying mapper object (e.g. libtokamap.Mapper).

Typing TokamapInterface.mapper against this protocol allows tests and library users to supply a lightweight fake without installing libtokamap.

tokamunch.DataSource¶: alias of PythonDataSourceProtocol

class tokamunch.DataSourceFactory(*args, **kwargs)¶

Protocol for entry-point data-source factories.

A factory receives plugin-specific configuration args and returns a PythonDataSourceProtocol-compatible object suitable for mapper.register_python_data_source(...).

Configuration ¶

class tokamunch.CLIConfig(mapper: 'MapperConfig', run: 'RunConfig', data_sources: 'list[DataSourceConfig]')¶

class tokamunch.MapperConfig(device: 'str', config: 'str | None' = None, config_params: 'dict[str, Any] | None' = None)¶

class tokamunch.RunConfig(default_shot: 'int | None' = None, concurrency: 'ConcurrencyConfig' = <factory>, log_level: 'str' = 'WARNING', binary_arrays: 'bool' = False, on_imas_error: 'str' = 'fallback-json')¶

class tokamunch.ConcurrencyConfig(mode: 'ConcurrencyMode' = <ConcurrencyMode.SERIAL: 'serial'>, workers: 'int' = 1)¶

class tokamunch.DataSourceConfig(mapper_name: 'str', plugin: 'str', enabled: 'bool' = True, args: 'dict[str, Any]'=<factory>)¶

tokamunch.load_cli_config(path: str | Path) → CLIConfig¶

tokamunch.config.apply_config_overrides(cfg: CLIConfig, overrides: list[str]) → CLIConfig¶

Apply dotted.key=value overrides to a CLIConfig, returning a new instance.

Supported keys:: run.concurrency.mode → ConcurrencyMode run.concurrency.workers → int run.log_level → str (must be a valid log level) run.binary_arrays → bool (“true”/”false”/”1”/”0”) run.on_imas_error → str (“fallback-json” or “raise”) run.default_shot → int mapper.device → str

Since all dataclasses use slots=True, new instances are constructed rather than mutating the originals.

Path utilities ¶

tokamunch.concrete_path_to_schema_path(path: str) → str¶

Convert a concrete runtime path to its schema form.

Replaces every concrete array index [N] with the schema array-struct marker (:).

Example:

"magnetics/flux_loop[0]/position[2]/r"
-> "magnetics/flux_loop(:)/position(:)/r"

tokamunch.concrete_path_to_template(path: str) → str¶

Convert a concrete runtime path to its mapping-template form.

Replaces every concrete array index [N] with the template placeholder [#], enabling lookup against mapping-file keys.

Example:

"magnetics/flux_loop[0]/position[2]/r"
-> "magnetics/flux_loop[#]/position[#]/r"

tokamunch.parse_concrete_path(path: str) → Iterator[IDSNode]¶

tokamunch.parse_schema_path(path: str, interner: SegmentInterner | None = None) → Iterator[IDSNode]¶

tokamunch.render_concrete_path(nodes: Iterable[IDSNode]) → str¶

tokamunch.render_schema_path(nodes: Iterable[IDSNode]) → str¶

tokamunch.render_array_length_query_path(nodes: Iterable[IDSNode]) → str¶

tokamunch.normalise_schema_segment(raw: str) → str¶

Mapping execution ¶

tokamunch.mapping.collect_mapped_values(ctx: MappingContext, selection: Selection, *, verbose_errors: bool, progress_callback: _ProgressCallback | None = None, on_paths_ready: Callable[[int], None] | None = None, profile: ProfileData | None = None, dry_run: bool = False, limit: int | None = None) → tuple[list[MappingRecord], MappingSummary]¶

tokamunch.mapping.map_path(ctx: MappingContext, ids_path: str) → Any¶

class tokamunch.mapping.MappingRecord(ids_path: 'str', value: 'Any | None' = None, error: 'Exception | None' = None, suppressed: 'bool' = False)¶

class tokamunch.mapping.MappingSummary(total_paths: 'int' = 0, mapped: 'int' = 0, returned_none: 'int' = 0, suppressed_errors: 'int' = 0, unexpected_errors: 'int' = 0, elapsed_s: 'float' = 0.0)¶

IDS writers ¶

tokamunch.set_ids_value(ids_obj: Any, segments: Iterable[IDSNode], value: Any, *, skip_root_segment: bool = True) → None¶

tokamunch.resize_and_set_ids_value(ids_obj: Any, segments: Iterable[IDSNode], value: Any, array_sizes: dict[str, int], *, write_context: WriteContext | None = None, skip_root_segment: bool = True) → None¶

tokamunch.ensure_ids_arrays_resized(ids_obj: Any, segments: Iterable[IDSNode], array_sizes: dict[str, int], *, write_context: WriteContext | None = None, skip_root_segment: bool = True) → None¶

tokamunch.resolve_ids_parent(ids_obj: Any, segments: Iterable[IDSNode], *, skip_root_segment: bool = True) → tuple[Any, IDSNode]¶

tokamunch.resolve_ids_segments(ids_obj: Any, segments: Iterable[IDSNode], *, skip_root_segment: bool = True) → Any¶

class tokamunch.write_ids.IdsWriteError(ids_name: str, records: list[Any], cause: Exception)¶: Captures a per-IDS write failure together with the records that couldn’t be written.

Format conversion ¶

Functions for moving data between munchi JSON output, imas-python IMAS files (.h5, .nc), and in-memory imas IDS objects.

tokamunch.read_json_records(path: Path) → list[MappingRecord]¶

Read a munchi JSON output file and return the entries as MappingRecord list.

Binary-encoded ndarrays ({"__ndarray__": ..., "dtype": ..., "shape": ...}) are decoded back to numpy arrays automatically.

tokamunch.records_to_ids_objects(records: list[MappingRecord]) → dict[str, Any]¶

Convert a list of MappingRecords into a dict of imas IDS objects.

Returns a {ids_name: ids_obj} mapping. Each IDS object is populated with the supplied records via setattr traversal, with array-struct nodes resized as needed.

tokamunch.read_ids_records(ids_obj: Any, ids_name: str) → list[MappingRecord]¶

Extract all non-empty leaf values from a loaded imas IDS object.

Uses the IMAS data-dictionary schema to enumerate candidate paths and reads each field value directly from ids_obj — no mapper required. Empty/unset fields (zero-length arrays, None) are silently skipped.

tokamunch.read_imas_records(path: Path, ids_names: list[str]) → list[MappingRecord]¶: Read IDS records from an imas HDF5 or NetCDF file.

tokamunch.convert_file(input_path: Path, output_path: Path, *, ids_names: list[str] | None = None, force: bool = False, binary_arrays: bool = False, on_imas_error: str = 'fallback-json') → list[IdsWriteError]¶

Convert data between supported file formats.

Supported input formats:

.json — munchi JSON output
.h5 / .nc — imas-python IMAS files (requires ids_names)

Supported output formats:

.json — munchi JSON output
.h5 / .nc — imas-python IMAS files

Diff and comparison ¶

tokamunch.diff.diff_records(records_a: list[MappingRecord], records_b: list[MappingRecord]) → list[DiffEntry]¶: Compare two sets of mapping records.

tokamunch.diff.diff_files(path_a: Path, path_b: Path, *, ids_names: list[str] | None = None) → list[DiffEntry]¶

Load two files and return diff entries.

Supports .json (munchi JSON output) and IMAS files (.h5, .nc). For IMAS files, ids_names must be supplied.

tokamunch.diff.render_diff(entries: list[DiffEntry], label_a: str, label_b: str, *, show_unchanged: bool = False) → str¶: Render a human-readable diff.

class tokamunch.diff.DiffEntry(path: str, value_a: Any, value_b: Any, status: str)¶

A single entry in a mapping diff.

status: str¶: One of "added", "removed", "changed", "unchanged".

value_a: Any¶: Value from the first file, or None if the path was absent.

value_b: Any¶: Value from the second file, or None if the path was absent.

Checkpointing ¶

class tokamunch.checkpoint.Checkpoint(output_path: str, completed_paths: list[str] = <factory>, results: dict[str, ~typing.Any]=<factory>)¶

Persistent state for a partially-completed mapping run.

completed_paths: list[str]¶: Concrete paths whose mapping has already been stored.

output_path: str¶: The target output file this checkpoint corresponds to.

results: dict[str, Any]¶: path → JSON-safe value for each completed path.

tokamunch.checkpoint.save_checkpoint(path: Path, cp: Checkpoint) → None¶

Write cp to path atomically (write to a .tmp file then rename).

The parent directory is created if it does not already exist.

tokamunch.checkpoint.load_checkpoint(path: Path) → Checkpoint | None¶

Load a checkpoint from path.

Returns None if the file does not exist. Raises ValueError if the file exists but cannot be parsed as a valid checkpoint.

tokamunch.checkpoint.apply_checkpoint(paths: list[str], cp: Checkpoint) → tuple[list[str], list[MappingRecord]]¶: Split paths into remaining work and already-completed records.

Shell completions ¶

tokamunch.completions.generate_bash_completion(ids_names: list[str]) → str¶: Generate a bash completion script for munchi.

tokamunch.completions.generate_zsh_completion(ids_names: list[str]) → str¶: Generate a zsh completion script for munchi.

tokamunch.completions.generate_fish_completion(ids_names: list[str]) → str¶: Generate a fish shell completion script for munchi.

tokamunch.completions.get_ids_names() → list[str]¶

Return sorted list of known IDS names from imas_data_dictionary.

Returns an empty list if the data dictionary is not available.

Mapping annotations ¶

tokamunch.templates.is_comment_stub(value: Any) → bool¶

Return True if value is a documented stub with no mapping expression.

A comment stub is a dict that has a "comment" key and whose remaining keys are limited to the allowed metadata set ("units", "source"). Any other key (e.g. an expression) means this entry carries real data and is therefore not a stub.

tokamunch.templates.merge_mapping_stubs(ids_name: str, existing_path: Path, *, leaves_only: bool = False) → dict[str, Any]¶

Merge an existing mapping file with blank stubs for any missing IDS paths.

All existing entries are preserved with their original values. Any template path present in the IDS schema but absent from the file is added as {"comment": ""}.

Existing entries appear first (in their original file order), followed by new stubs sorted alphabetically.

tokamunch.templates.build_blank_mapping_template(ids_name: str, *, leaves_only: bool) → dict[str, dict[str, Any]]¶

Profiling ¶

See also: Profiling and performance for usage examples.

class tokamunch.profiling.ProfileData(phases: 'PhaseTimings' = <factory>, mapper_map: 'CallStats' = <factory>, array_length: 'CallStats' = <factory>)

class tokamunch.profiling.PhaseTimings(expansion_s: 'float' = 0.0, mapping_s: 'float' = 0.0, output_s: 'float' = 0.0)

class tokamunch.profiling.CallStats(count: int = 0, total_s: float = 0.0, min_s: float = inf, max_s: float = 0.0): Thread-safe accumulator for per-call timing statistics.

tokamunch.profiling.render_profile_report(data: ProfileData, total_elapsed_s: float) → str

Trie ¶

tokamunch.build_ids_path_trie(ids_paths: Iterable[str]) → TrieNode¶

tokamunch.generate_schema_paths_from_trie(root: TrieNode, *, leaves_only: bool = False) → Iterator[str]¶

tokamunch.expand_ids_path_trie(root: TrieNode, get_length_callback: Callable[[str], int], *, context: ExpansionContext | None = None, leaves_only: bool = False) → Iterator[str]¶

tokamunch.expand_ids_path_trie_segments(root: TrieNode, get_length_callback: Callable[[str], int], *, context: ExpansionContext | None = None, leaves_only: bool = False) → Iterator[tuple[IDSNode, ...]]¶