Skip to main content

functions

Shared helpers for the scan_metadata runtime.

The centrepiece is normalize_to_scan_metadata: the single mapping from a _process_file record to the scan_metadata normalized field set. It is vendor-agnostic by construction — private-eye records already carry the normalized snake_case keys, while DICOM/Zeiss records carry only raw DICOM tag-name keys, and every private-eye record also carries the DICOM alias keys (see private_eye_base_source.py:1220-1238). So the mapper reads the normalized key first and falls back to the DICOM alias (computing dimensions_mm_* from Pixel Spacing × Rows/Columns × frames when only the raw tags are present). Fields with no source on either path stay None.

Cache/datasource helpers (open_background_cache, _resolve_datasource_class, _hash_file, make_dataset_identifier) are re-exported from the file_metadata runtime so the two runtimes cannot drift.

Module​

Functions​

_resolve_datasource_class​

def _resolve_datasource_class(datasource_type: str) ‑> type | None:

Resolve a datasource type string to its class, or None on failure.

Follows the same lookup path as _create_datasource_from_config in bitfount.runners.config_utils:

  1. Try DataSourceType(datasource_type).name to convert the enum value (e.g. "bitfount.CSVSource") to its class name (e.g. "CSVSource"). Falls back to the raw string for plugin datasources not in the enum.
  2. getattr(importlib.import_module("bitfount.data"), cls_name) to obtain the class object.

Arguments

  • datasource_type: The datasource type string from the pod config, e.g. "bitfount.CSVSource" or a plugin class name.

Returns The resolved class, or None if the class cannot be imported.

accepted_extensions​

def accepted_extensions(datasource: BaseSource) ‑> set[str] | None:

Return datasource's accepted file extensions (lowercased), or None.

Read from the datasource's FileSystemFilter (datasource.filter). Keeps selection from handing a consumer files the datasource would never load (e.g. a stray .txt sharing the root). None means no extension restriction, so every row under the root is considered.

Probed rather than typed, for the same reason as datasource_root: a source reaching here always has a filter in practice, and a missing one reads as "no restriction" — which keeps files in rather than dropping them.

Arguments

  • datasource: The datasource being selected for.

Returns Lowercased extensions, or None for no restriction.

get_file_metadata_inventory​

def get_file_metadata_inventory(    cache: CacheProtocol,) ‑> CollectedFileMetadataQueryResult:

Return the whole pod inventory (every file row, datasource-agnostic).

scan_metadata reuses this inventory (paths + content hashes) instead of walking the datasource's directory itself — file_metadata_runtime already paid the (potentially very expensive) recursive walk and persisted the result, so scan_metadata is a cheap indexed read chained after it. An empty result means file_metadata has not populated yet; the caller does nothing (scan_metadata never walks).

The inventory is now datasource-agnostic (keyed by file_path, not task_hash), so this returns every file in the pod. Selection of the files relevant to a given datasource lives with the datasource — the caller filters by the datasource's own root path + accepted_extensions (decision #12), rather than by a stored owner.

instantiate_ophthalmology_datasource​

def instantiate_ophthalmology_datasource(    datasource_type: str, path: str, datasource_kwargs: dict[str, Any] | None = None,) ‑> Optional[Any]:

Construct an ophthalmology datasource for path, or None.

Resolves datasource_type to its class and instantiates it with path.

A HeidelbergE2ESource config is instead instantiated as its file-intrinsic sibling, HeidelbergE2EAllSeriesSource: scan_metadata is file-intrinsic, so every series in a file is extracted regardless of laterality/series_protocol, and that selection is applied later at read time. The laterality/series_protocol ctor-kwargs the trigger may still forward are therefore not threaded into extraction — a file yields the same scan rows no matter which config initiated the run. Other ophthalmology sources (DICOM, Zeiss) are already file-intrinsic and are constructed unchanged.

Precondition: the caller has already confirmed the type is an ophthalmology source via is_ophthalmology_datasource (the flow's skip gate); this function therefore does not re-check. It returns None only when the class cannot be resolved, which the flow treats as a config error.

No trigger forwards ctor kwargs today, and that is deliberate rather than an oversight: HeidelbergE2ESource is the only ophthalmology source whose constructor accepts laterality/series_protocol, and it is precisely the type replaced here — so a forwarded value could never reach a constructor. The parameter is kept because it is part of the scan_metadata_runtime deployment's schema and would be the route for a future source that genuinely needs a kwarg the type string cannot convey.

Arguments

  • datasource_type: The datasource type string from the pod config.
  • path: The directory the datasource should walk.
  • datasource_kwargs: Extra ctor kwargs from the trigger. Ignored for HeidelbergE2ESource (the all-series variant supersedes them); splatted into other sources' constructors.

Returns The constructed datasource instance, or None if the class cannot be resolved.

is_ophthalmology_datasource​

def is_ophthalmology_datasource(datasource_type: str) ‑> bool:

Return True when datasource_type resolves to an ophthalmology source.

The scan_metadata runtime only handles ophthalmology datasources (the only ones that produce per-scan header metadata via _process_file(skip_non_tabular_data=True)); every other datasource type is skipped by the flow.

Arguments

  • datasource_type: The datasource type string from the pod config.

Returns True when the resolved class is an _OphthalmologySource subclass.

make_dataset_identifier​

def make_dataset_identifier(pod_name: str, datasource_name: str) ‑> str:

Return the "{pod_name}/{datasource_name}" dataset identifier.

This is the same slash-joined key generate_prefect_task_hash hashes, and the value stored in FileMetadata.dataset_identifier.

normalize_to_scan_metadata​

def normalize_to_scan_metadata(    data: dict[str, Any],) ‑> NormalizedScanFields:

Map one _process_file record to the scan_metadata field set.

Vendor-agnostic: prefers the normalized snake_case keys emitted by the private-eye pipeline, and falls back to the raw DICOM tag-name keys emitted by the DICOM/Zeiss pipeline (computing dimensions_mm_* from the pixel spacing and matrix size when only the raw tags are present). Any field with no source on either path is left None.

The returned NormalizedScanFields holds only the normalized, per-scan vendor columns; the collection task adds the PK (file_path, series_index) and the control/provenance columns.

Arguments

  • data: One record dict as returned by datasource._process_file(skip_non_tabular_data=True).

Returns The normalized per-scan field values.

open_background_cache​

def open_background_cache(    pod_name: str, *, pod_key_path: Path | None = None, create_dir: bool = False,) ‑> CacheProtocol:

Open the per-pod background cache DB.

Single source of truth for resolving the per-pod background cache path (PODS_CACHE_ROOT/<pod_name>/BACKGROUND_CACHE_FILENAME) and opening it, shared by the DATASET_PROJECT_LINKED gRPC pod handler and the run_background_task CLI so the two entry points cannot drift.

Arguments

  • pod_name: Name of the pod whose cache DB to open.
  • pod_key_path: Path to the pod's pod_rsa.pem (typically pod.pod_key_path). When provided, it is bound into the cache-encryption layer so encrypted columns can derive their key from this pod's RSA private key. Optional so test/CLI flows that don't need encryption can still call with just pod_name.
  • create_dir: When True the parent directory is created if missing — the live pod handler path, where the DB is created on first write. When False the cache file must already exist or FileNotFoundError is raised — the CLI path, which requires the pod to have been started at least once to populate file metadata.

Returns A CacheProtocol instance for the pod's background cache.

Raises

  • FileNotFoundError: When create_dir is False and the cache file does not exist.

select_for_datasource​

def select_for_datasource(    records: Iterable[FileMetadataRecord],    datasource: FileSystemIterableSource,    root: Path,) ‑> Iterator[FileMetadataRecord]:

Filter a pod-wide inventory down to datasource's own files.

Takes the root as an argument rather than deriving it: a caller has to look it up anyway to know whether path selection applies at all (datasource_root returns None for a SQL source, which keeps the task_hash scope), so re-deriving it here would resolve the path twice and turn a case the caller already handled into an exception it has to know not to trigger.

Arguments

  • records: The whole inventory, e.g. from file_metadata.v1.store.get_file_metadata_inventory. Streamed, so a large inventory is not materialised.
  • datasource: The datasource being selected for, read for its extension filter.
  • root: Its resolved walk root, from datasource_root.