import hashlib
import inspect
import json
import logging
import os
import platform
import re
import typing as ty
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, datetime
from functools import cached_property
from glob import glob
from pathlib import Path
from typing import Self
import attrs
import requests
import yaml
from dateutil.parser import isoparse
from fileformats.application import Yaml
from fileformats.core import FileSet, from_mime, from_paths, to_mime
from fileformats.core.identification import to_mime_format_name
from fileformats.core.utils import collate_metadata_series
from fileformats.generic import Directory, SetOf
from fileformats.medimage import DicomCollection
from filelock import SoftFileLock
from frametree.core.exceptions import FrameTreeDataMatchError
from frametree.core.frameset import FrameSet
from tqdm import tqdm
from ..exceptions import ImagingSessionParseError, StagingError
from ..helpers.arg_types import (
ON_RESOURCE_CLASH,
AssociatedFiles,
ClashSpec,
IDSpec,
MetadataTable,
OnResourceClash,
PathMetadataRegex,
)
from ..helpers.metadata import Metadata
from .resource import ImagingResource
from .scan import ImagingScan
logger = logging.getLogger("xnat-ingest")
Transform = ty.Callable[[ty.Mapping[str, ty.Any]], ty.Any]
# Sentinel returned by ``IDSpec.get_value_from_matching_spec`` when no scan spec
# applies to a fileset's type, signalling that the scan should take the resource's name
_DERIVED_ID: ty.Any = object()
_DATE_FORMATS = ["%d.%m.%y", "%d.%m.%Y", "%Y-%m-%d", "%Y%m%d", "%m/%d/%y", "%m/%d/%Y"]
_TIME_FORMATS = ["%H.%M.%S", "%H:%M:%S", "%H%M%S"]
def _parse_datetime_to_str(date_str: str, time_str: str | None) -> str:
"""Parse date (and optional time) strings using common formats, return YYYYMMDDHHMMSS or YYYYMMDD."""
parsed_date = None
for fmt in _DATE_FORMATS:
try:
parsed_date = datetime.strptime(date_str, fmt)
break
except ValueError:
continue
if parsed_date is None:
raise ValueError(
f"Cannot parse date '{date_str}' — tried formats: {_DATE_FORMATS}"
)
if time_str:
for fmt in _TIME_FORMATS:
try:
parsed_time = datetime.strptime(time_str, fmt)
return parsed_date.strftime("%Y%m%d") + parsed_time.strftime("%H%M%S")
except ValueError:
continue
raise ValueError(
f"Cannot parse time '{time_str}' — tried formats: {_TIME_FORMATS}"
)
return parsed_date.strftime("%Y%m%d")
def scans_converter(
scans: ty.Union[ty.Sequence[ImagingScan], ty.Dict[str, ImagingScan]],
) -> dict[str, ImagingScan]:
if isinstance(scans, ty.Sequence):
duplicates = [i for i, c in Counter(s.id for s in scans).items() if c > 1]
if duplicates:
raise ValueError(f"Found duplicate scan IDs in list of scans: {duplicates}")
scans = {s.id: s for s in scans}
return scans
def _metadata_diff(
orig: ty.Mapping[str, ty.Any], new: ty.Mapping[str, ty.Any]
) -> dict[str, ty.Any]:
"""Return the fields of `orig` that are missing from or differ in `new`, i.e. the
original values of any metadata fields that were stripped/modified. Used to
reconstruct the reid metadata that `FileSet.deidentify()` implementations no
longer report themselves (see `fileformats.medimage.MedicalImagingData.deidentify`
docstring).
"""
diff: dict[str, ty.Any] = {}
for key, val in orig.items():
try:
new_val = new[key]
except KeyError:
diff[key] = val
continue
if isinstance(val, ty.Mapping) and isinstance(new_val, ty.Mapping):
nested = _metadata_diff(val, new_val)
if nested:
diff[key] = nested
elif val != new_val:
diff[key] = val
return diff
def _expand_collated_metadata(
metadata: dict[str, ty.Any], num_members: int
) -> list[dict[str, ty.Any]]:
"""Reconstruct the per-member metadata dicts from a dict previously produced by
``Metadata.collate`` for ``num_members`` members, so a further member can be
collated in without nesting the already-listed values.
A value is treated as per-member only when it is a list whose length matches
``num_members``; with a single existing member there is nothing to expand.
"""
if num_members <= 1:
return [dict(metadata)]
members: list[dict[str, ty.Any]] = [{} for _ in range(num_members)]
for key, value in metadata.items():
if isinstance(value, list) and len(value) == num_members:
for member, item in zip(members, value):
if item is not None:
member[key] = item
else:
for member in members:
member[key] = value
return members
def _set_content_types(fileset: FileSet) -> tuple[type[FileSet], ...]:
"""The content types to classify ``fileset`` by when folding it into a merged
``SetOf`` resource: the classifiers of an existing ``SetOf`` (from a previous
merge) or the fileset's own type otherwise.
"""
content_types = getattr(type(fileset), "content_types", ())
return tuple(content_types) if content_types else (type(fileset),)
def _type_name_resource_label(type_name: str) -> str:
"""Fallback resource label for a fileset with no ``--resource`` spec: the
mime-like rendering of its type name, e.g. 'vectra-export', 'sqlite3-db',
run through the same ID/label escaping as session/scan IDs
(:attr:`IDSpec.xnat_id_escape_re`) so the '.'/'+' that ``to_mime_format_name``
emits for vendor/classifier type names (``SyngoMi_Vr20b_ListMode`` ->
``syngo-mi.vr20b.list-mode``, ``Png___SetOf`` -> ``png+set-of``) collapse to
'_' while '-' is kept.
"""
return IDSpec.xnat_id_escape_re.sub("_", to_mime_format_name(type_name))
def _glob_to_regex(pattern: str) -> re.Pattern[str]:
r"""Anchored regex for a ``/``-aware glob: ``*`` / ``?`` / ``[...]`` do not cross
``/``, ``**`` (optionally followed by ``/``) matches across directory levels.
Equivalent to ``glob.translate(pattern, recursive=True)`` (py3.13+), spelled out
so 3.11/3.12 work too.
"""
i, n = 0, len(pattern)
out = ["(?s:"]
while i < n:
c = pattern[i]
i += 1
if c == "*":
if i < n and pattern[i] == "*":
i += 1
if i < n and pattern[i] == "/":
i += 1
out.append("(?:[^/]*/)*") # '**/' -> zero or more segments
else:
out.append(".*")
else:
out.append("[^/]*")
elif c == "?":
out.append("[^/]")
elif c == "[":
j = i + 1 if i < n and pattern[i] in "!^" else i
j = j + 1 if j < n and pattern[j] == "]" else j
while j < n and pattern[j] != "]":
j += 1
if j >= n:
out.append(r"\[")
else:
inner = pattern[i:j]
i = j + 1
if inner[:1] in ("!", "^"):
inner = "^" + inner[1:]
out.append("[" + inner + "]")
else:
out.append(re.escape(c))
out.append(r")\Z")
return re.compile("".join(out))
def _drop_excluded_paths(
fspaths: ty.Sequence[Path],
input_dirs: ty.Sequence[Path],
exclude_globs: ty.Sequence[str],
) -> list[Path]:
"""Drop every ``fspath`` whose path *relative to one of ``input_dirs``* matches
one of ``exclude_globs``. Unlike ``allow_unrecognised`` this fires before
classification, so it removes a path even if a ``--datatype`` would claim it
(e.g. a vendor thumbnail that is a valid ``image/png``). Globs use the standard
``glob`` syntax - ``*`` does not cross ``/``, ``**`` does - and match the whole
relative path.
"""
if not exclude_globs:
return list(fspaths)
matchers = [_glob_to_regex(g) for g in exclude_globs]
kept: list[Path] = []
for p in fspaths:
rels: list[str] = []
for base in input_dirs:
try:
rels.append(str(p.relative_to(base)))
except ValueError:
continue
if any(m.match(rel) for m in matchers for rel in rels):
logger.debug("Excluding '%s' (matched --exclude-path)", p)
continue
kept.append(p)
return kept
def _fileset_in_scope(fileset: FileSet, scope: type[FileSet] | ty.Any) -> bool:
"""Whether ``fileset`` falls within a ``ClashSpec`` scope - either it is an
instance of ``scope``, or it is a ``SetOf`` whose every content type is a
subclass of ``scope`` (so a re-merge into an existing ``SetOf[Png, Jpeg]``
still counts as covered by an ``image/png|image/jpeg`` scope).
"""
if isinstance(fileset, scope):
return True
content_types = getattr(type(fileset), "content_types", ())
return bool(content_types) and all(issubclass(ct, scope) for ct in content_types)
def _resolve_clash_policy(
specs: ty.Sequence[ClashSpec],
existing: FileSet,
incoming: FileSet,
where: str,
) -> str:
"""The clash policy for a name collision between ``existing`` and ``incoming``:
the first ``ClashSpec`` whose scope covers *both*. Raises if none does.
"""
for spec in specs:
if _fileset_in_scope(existing, spec.scope) and _fileset_in_scope(
incoming, spec.scope
):
return spec.policy
raise KeyError(
f"Resource-name clash between a {type(existing).__name__} and a "
f"{type(incoming).__name__} {where}, and no --on-resource-clash spec's "
"scope covers both. Add one (e.g. "
f"'--on-resource-clash avoid \"{to_mime(type(existing))}|{to_mime(type(incoming))}\"'), "
"or tighten --scan / --resource so the two don't collide."
)
def _recursive_collect(
root: Path,
datatypes: ty.Sequence[type[FileSet]],
ignore_datatypes: ty.Sequence[type[FileSet]],
) -> ty.Iterator[Path]:
"""Walk ``root`` yielding paths for ``from_paths`` to classify.
A directory that validates as one of ``datatypes`` is yielded whole and *not*
descended into; a directory that validates only as an ``ignore_datatypes``
directory format is skipped whole (not yielded, not descended); any other
directory is descended. Every loose file is yielded, so an unlisted file type
still raises in ``from_paths`` as usual.
``dt.matches()`` runs the datatype's full validation on each directory node,
which for rich directory formats (e.g. a Canfield export) is not free - fine
for the export-sized trees this is meant for.
"""
want = tuple(d for d in datatypes if issubclass(d, Directory))
skip = tuple(d for d in ignore_datatypes if issubclass(d, Directory))
stack: list[Path] = [root]
seen: set[Path] = set()
while stack:
current = stack.pop()
resolved = current.resolve()
if resolved in seen: # guard against symlink loops
continue
seen.add(resolved)
for child in sorted(current.iterdir()):
if not child.is_dir() or any(dt.matches(child) for dt in want):
yield child
elif any(dt.matches(child) for dt in skip):
continue
else:
stack.append(child)
def _deidentify_or_copy_resource(
fileset: FileSet,
resource_name: str,
resource_dest_dir: Path,
contains_phi: bool,
spec: ty.Any,
copy_mode: FileSet.CopyMode,
max_workers: int | None,
transforms: dict[str, Transform] | None = None,
) -> tuple[FileSet, ty.Mapping[str, ty.Any]]:
"""Deidentifies (or, for filesets that don't contain PHI, just copies) a single
resource.
"""
if not contains_phi:
return (
fileset.copy(
resource_dest_dir,
mode=copy_mode,
new_stem=resource_name,
avoid_clashes=True,
),
{},
)
orig_metadata = dict(fileset.metadata)
deid_resource = fileset.deidentify(
resource_dest_dir,
spec=spec,
max_workers=max_workers,
transforms=transforms,
)
reid_mdata = _metadata_diff(orig_metadata, deid_resource.metadata)
return deid_resource, reid_mdata
[docs]
@attrs.define(slots=False)
class ImagingSession:
"""Representation of an imaging session to be uploaded to XNAT, which is a set of scans that
belong together under the same project/subject/session IDs.
Parameters
----------
project_id: str, optional
The project ID of the session
subject_id: str, optional
The subject ID of the session
session_id: str, optional
The session (visit) ID of the session
scans: ty.Dict[str, ImagingScan]
The scans in the session
run_uid: ty.Optional[str]
The run UID of the session, if it exists
"""
uid: str
project_id: str | None = None
subject_id: str | None = None
session_id: str | None = None
scans: ty.Dict[str, ImagingScan] = attrs.field(
factory=dict,
converter=scans_converter,
validator=attrs.validators.instance_of(dict),
)
session_resources: ty.Dict[str, ImagingResource] = attrs.field(factory=dict)
run_uid: ty.Optional[str] = attrs.field(default=None)
metadata: Metadata = attrs.field(eq=False, repr=False, init=False)
METADATA_FNAME = "__METADATA__.yaml"
METADATA_DIR = "__metadata__"
# Directory-name prefix used to flag sessions that have been grouped into scans but
# not yet had project/subject/session IDs assigned to them. Session UIDs (e.g. DICOM
# StudyInstanceUID) commonly contain '.'s, so a distinct prefix is needed to tell
# them apart from assigned "PROJECT.SUBJECT.SESSION" directory names when reloading.
PRE_ASSIGN_PREFIX = "_."
# Metadata key the originating session UID is stashed under when saving, so it can
# be recovered on reload even after the directory has been renamed to PROJECT.SUBJECT.SESSION
UID_METADATA_KEY = "__uid__"
# Metadata key under which each fileset's resolved fileformats type name (e.g.
# 'VectraExport', 'Sqlite3Db') is stashed during grouping, so it can be
# referenced from --session/--scan/--resource specs (e.g. '{__datatype__}')
TYPE_METADATA_KEY = "__datatype__"
def __attrs_post_init__(self) -> None:
for scan in self.scans.values():
scan.session = self
def __getitem__(self, fieldname: str) -> ty.Any:
return self.metadata[fieldname]
@metadata.default
def _metadata_default(self):
return Metadata({}, self)
@property
def name(self) -> str:
if any(i is None for i in (self.project_id, self.subject_id, self.session_id)):
return None
return f"{self.project_id}.{self.subject_id}.{self.session_id}"
@property
def invalid_ids(self) -> bool:
return (
self.project_id.startswith("INVALID")
or self.subject_id.startswith("INVALID")
or self.session_id.startswith("INVALID")
)
@property
def path(self) -> str:
return ":".join([self.project_id, self.subject_id, self.session_id])
@property
def staging_relpath(self) -> list[str]:
if self.name is None:
return [f"{self.PRE_ASSIGN_PREFIX}{self.uid}"]
return [self.name]
@cached_property
def modalities(self) -> str | tuple[str, ...]:
try:
modalities_metadata = self.metadata["Modality"]
except KeyError as e:
e.add_note(f"Available metadata: {list(self.metadata)}")
raise e
if isinstance(modalities_metadata, str):
return modalities_metadata
modalities: set[str] = set()
for modality in modalities_metadata:
if isinstance(modality, str):
modalities.add(modality)
else:
assert isinstance(modality, ty.Iterable)
modalities.update(modality)
return tuple(modalities)
@property
def primary_parents(self) -> set[Path]:
"Return parent directories for all resources in the session"
return set(r.fileset.parent for r in self.primary_resources)
@property
def resources(self) -> ty.List[ImagingResource]:
return list(self.session_resources.values()) + [
r for p in self.scans.values() for r in p.resources.values()
]
@property
def primary_resources(self) -> ty.List[ImagingResource]:
return [
r
for s in self.scans.values()
for r in s.resources.values()
if not s.associated
]
def load_metadata(self):
return Metadata.collate(s.metadata for s in self.scans.values())
def new_empty(self) -> Self:
"""Return a new empty session with the same IDs as the current session"""
return type(self)(
uid=self.uid,
project_id=self.project_id,
subject_id=self.subject_id,
session_id=self.session_id,
run_uid=self.run_uid,
)
def select_resources(
self,
dataset: FrameSet | None,
always_include: ty.Sequence[str | FileSet] = (),
) -> ty.Iterator[ImagingResource]:
"""Returns selected resources that match the columns in the dataset definition
Parameters
----------
dataset : FrameSet
Arcana dataset definition
always_include : sequence[str | FileSet]
mime-types or "mime-like" (see https://arcanaframework.github.io/fileformats/)
of file-format to always include in the upload, regardless of whether they are
specified in the dataset or not
Yields
------
scan_id : str
the ID of the scan should be uploaded to
scan_type : str
the desc/type to assign to the scan
resource_name : str
the name of the resource under the scan to upload it to
scan : FileSet
a fileset to upload
"""
if not dataset and not always_include:
raise ValueError(
"Either 'dataset' or 'always_include' must be specified to select "
f"appropriate resources to upload from {self.name} session"
)
store = ImagingSessionMockStore(self)
uploaded = set()
for mime_like in always_include:
if inspect.isclass(mime_like) and issubclass(mime_like, FileSet):
fileformat = mime_like
elif mime_like == "all":
fileformat = FileSet
else:
fileformat = from_mime(mime_like) # type: ignore[assignment]
if not issubclass(fileformat, FileSet):
raise ValueError(
f"{mime_like!r} does not correspond to a file format ({fileformat})"
)
for resource in self.session_resources.values():
if isinstance(resource.fileset, fileformat):
uploaded.add((None, resource.name))
yield resource
for scan in self.scans.values():
for resource in scan.resources.values():
if isinstance(resource.fileset, fileformat):
uploaded.add((scan.id, resource.name))
yield resource
if dataset is not None:
for column in dataset.columns.values():
try:
entry = column.match_entry(store.row)
except FrameTreeDataMatchError as e:
raise StagingError(
f"Did not find matching entry for {column} column in {dataset} from "
f"{self.name} session"
) from e
else:
scan_id, resource_name = entry.uri
scan = self.scans[scan_id]
if (scan.id, resource_name) in uploaded:
logger.info(
"%s/%s resource is already uploaded as 'always_include' is set to "
"%s and doesn't need to be explicitly specified",
scan.id,
resource_name,
always_include,
)
continue
resource = scan.resources[resource_name]
if not isinstance(resource.fileset, column.datatype):
resource = ImagingResource(
name=resource_name,
fileset=column.datatype(resource.fileset),
scan=scan,
)
uploaded.add((scan.id, resource_name))
yield resource
[docs]
@classmethod
def from_paths(
cls,
files_path: str | Path | ty.Sequence[str | Path],
datatypes: type[FileSet] | ty.Sequence[type[FileSet]],
session_field: ty.Sequence[IDSpec],
scan_field: ty.Sequence[IDSpec] = (),
resource_field: ty.Sequence[IDSpec] = (),
recursive: bool = False,
on_resource_clash: OnResourceClash | ty.Sequence[ClashSpec] = "error",
allow_unrecognised: ty.Sequence[str] | None = None,
exclude_paths: ty.Sequence[str] | None = None,
ignore_datatypes: ty.Sequence[type[FileSet]] | None = None,
path_metadata_regex: ty.Sequence[PathMetadataRegex] = (),
metadata_tables: list[MetadataTable] | None = None,
) -> list[Self]:
"""Loads all imaging sessions from a list of DICOM files
Parameters
----------
files_path : str or Path
Path to a directory containing the resources to load the sessions from, or a
glob string that selects the paths
datatypes : type or list[type]
the fileformats to load from the paths, e.g. DicomSeries or
[DicomSeries, NiftiGz]
session_field: ty.Sequence[IDSpec]
the metadata field that uniquely identifies the session, used to group files
together before project/subject/visit IDs are extracted (e.g. StudyInstanceUID)
scan_field: ty.Sequence[IDSpec]
the value of this field is used to group resources under single scans.
For a fileset whose type is not matched by any spec here (including the
empty default), the scan is named after that fileset's resource.
resource_field: ty.Sequence[IDSpec]
the value of this field is used to identify resources. If empty, the
resource is labelled with the mime-like rendering of the fileset's type
name, e.g. 'vectra-export', 'sqlite3-db'
recursive : bool, optional
recurse into directories passed as file paths. For file datatypes this
flattens the tree; when any ``datatypes`` / ``ignore_datatypes`` entry is
a directory format the walk stops descending into a directory as soon as
it validates as one of them (yielding a ``datatypes`` match whole,
skipping an ``ignore_datatypes`` match whole). ``generic/directory`` and
``generic/file-set`` cannot be used as ``datatypes`` with ``recursive``.
By default False
on_resource_clash : OnResourceClash or Sequence[ClashSpec], optional
how to handle two filesets resolving to the same scan/resource name.
A bare policy string ("error"/"avoid"/"merge"/"overwrite") applies to
any clash. A sequence of ``ClashSpec`` (policy + datatype scope) resolves
each clash with the first spec whose scope covers *both* filesets;
"merge" folds them into a ``SetOf``, "avoid" suffixes, "overwrite"
replaces; a clash no spec covers raises. Default "error".
allow_unrecognised : Sequence[str] or None, optional
regular expressions matched against the *basename* of any input path
that no datatype recognised - matches are skipped instead of raising
``FormatRecognitionError``. Does not affect recognised filesets.
exclude_paths : Sequence[str] or None, optional
glob patterns matched against each input path *relative to its input
directory*, applied before classification so a match is dropped even if
a datatype would claim it (e.g. a vendor thumbnail that is a valid
``image/png``). ``*`` does not cross ``/``, ``**`` does.
ignore_datatypes : ty.Sequence[type[FileSet]] or None, optional
datatypes that are expected in the input but not wanted: recognised
filesets of these types are dropped from the result rather than raising,
and (when ``recursive``) matching directories are skipped without
descending. An input path matching neither ``datatypes`` nor
``ignore_datatypes`` (nor ``allow_unrecognised``/``exclude_paths``)
still raises.
path_metadata_regex : ty.Sequence[PathMetadataRegex], optional
Regular expressions to extract "metadata" values from resource file paths as named groups. The named
groups are used as metadata fields for the resource files, and the extracted values will be used to populate
the corresponding metadata fields to complement the metadata read from the file headers.
metadata_tables : list[MetadataTable] or None, optional
a list of MetadataTable objects that define how to extract metadata from input files (e.g. CSV files and spreadsheets)
and join them with the sessions. If None, no metadata tables will be used.
Returns
-------
list[ImagingSession]
all imaging sessions that are present in list of dicom paths
Raises
------
ImagingSessionParseError
if values extracted from IDs across the DICOM scans are not consistent across
DICOM files within the session
"""
if not isinstance(datatypes, ty.Sequence):
datatypes = [datatypes]
datatypes = list(datatypes)
ignore_datatypes = list(ignore_datatypes or [])
if contradicting := set(datatypes) & set(ignore_datatypes):
raise ValueError(
"The following datatypes were listed for both inclusion (`datatypes`) and exclusion "
f"(`ignore_datatypes`): {list(contradicting)}"
)
# When recursing, a directory format among datatypes/ignore_datatypes makes
# the walk prune-on-match (see _recursive_collect) rather than flatten. The
# bare generic types match every directory / path so they can't be used.
recurse_into_dirs = False
if recursive:
if Directory in datatypes or FileSet in datatypes:
raise ValueError(
"Cannot use `generic/directory` or `generic/file-set` as a `--datatype` "
"with `--recursive` (they match every directory / path at every depth). "
f"Use a specific directory datatype instead (datatypes={datatypes})"
)
recurse_into_dirs = any(
isinstance(d, type) and issubclass(d, Directory)
for d in (*datatypes, *ignore_datatypes)
)
if not recurse_into_dirs:
# file-only recursion: flatten everything, and let a generic
# Directory soak up the bare directory nodes so from_paths doesn't
# choke on them
ignore_datatypes.append(Directory)
if isinstance(files_path, (Path, str)):
files_path = [files_path]
elif not isinstance(files_path, ty.Sequence):
raise TypeError(
"Invalid type of 'files_path', must be a pathlib.Path, str or list of"
)
fspaths: list[Path] = []
input_dirs: list[Path] = []
for fspath in files_path:
logger.debug("Searching for file types in '%s'", str(fspath))
if isinstance(fspath, Path) or "*" not in fspath:
fspath = Path(fspath)
if not fspath.exists():
raise ValueError(
f"Provided file-system path '{fspath}' does not exist"
)
if fspath.is_dir():
input_dirs.append(fspath)
if recurse_into_dirs:
logger.debug(
"Walking '%s' for directory datatypes (prune-on-match)",
str(fspath),
)
fspaths.extend(
_recursive_collect(fspath, datatypes, ignore_datatypes)
)
elif recursive:
logger.debug(
"Recursively searching for all paths '%s' directory",
str(fspath),
)
fspaths.extend(
Path(p) for p in glob(str(fspath) + "/**/*", recursive=True)
)
else:
logger.debug(
"Adding contents of '%s' directory to list", str(fspath)
)
fspaths.extend(Path(fspath).iterdir())
else:
logger.debug(
"Directly appending '%s' to list of files", str(fspath)
)
fspaths.append(fspath)
else:
logger.debug("Searching for file-system paths using glob '%s'", fspath)
fspaths.extend(Path(p) for p in glob(fspath, recursive=True))
fspaths = [fix_long_path(p) for p in fspaths]
if exclude_paths:
fspaths = _drop_excluded_paths(fspaths, input_dirs, exclude_paths)
if nonexistent := [str(p) for p in fspaths if not Path(p).exists()]:
raise ValueError(
"The following paths do not exist:\n"
+ "\n".join(nonexistent[:100])
+ ("\n..." if len(nonexistent) > 100 else "")
)
# Create a UID out of the paths that session was created from and the
# timestamp
crypto = hashlib.sha256()
for fspath in fspaths:
crypto.update(str(fspath.absolute()).encode())
run_uid: str = crypto.hexdigest()[:6] + datetime.strftime(
datetime.now(UTC),
"%Y%m%d%H%M%S",
)
from_paths_kwargs = {}
# Sort loaded series by StudyInstanceUID (imaging session)
logger.info(f"Loading {datatypes} from {files_path}...")
filesets = from_paths(
fspaths,
*(datatypes + ignore_datatypes),
ignore="|".join(allow_unrecognised) if allow_unrecognised else None,
**from_paths_kwargs, # type: ignore[arg-type]
)
if ignore_datatypes:
# drop filesets of an ignored datatype, but never one that also matches
# an explicitly-requested datatype (a specific Directory subclass in
# `datatypes` would otherwise be filtered by the generic `Directory`
# added for file-only recursion)
filesets = [
f
for f in filesets
if any(isinstance(f, d) for d in datatypes)
or not any(isinstance(f, t) for t in ignore_datatypes)
]
if path_metadata_regex:
for fileset in tqdm(
filesets,
"Extracting metadata from file paths...",
):
for path_mdata in path_metadata_regex:
if isinstance(fileset, path_mdata.datatype):
fileset_path = str(getattr(fileset, "fspath", fileset.parent))
match = re.match(path_mdata.regex, fileset_path)
if match is None:
raise ValueError(
f"Could not extract metadata from path '{fileset_path}' "
f"using pattern '{path_mdata.regex}'"
)
fileset.metadata.update(match.groupdict())
# Expose each fileset's resolved type name as a metadata field so it can be
# referenced from --session/--scan/--resource specs (setdefault so an
# explicit path-regex group of the same name still wins)
for fileset in filesets:
fileset.metadata.setdefault(cls.TYPE_METADATA_KEY, fileset.type_name)
MetadataTable.inject_list(metadata_tables, filesets)
sessions: dict[tuple[str, str, str] | str, Self] = {}
for fileset in tqdm(
filesets,
"Sorting resources into XNAT tree structure...",
):
session_uid = IDSpec.get_value_from_matching_spec(fileset, session_field)
# XNAT requires DICOM datasets to have 'DICOM'/'secondary' resource
# labels otherwise some features don't work
resource_derived = False
if isinstance(fileset, DicomCollection):
try:
image_type = fileset.contents[0].metadata["ImageType"]
except (KeyError, IndexError):
resource_label = "DICOM"
else:
resource_label = dicom_image_type_to_resource_label(image_type)
elif not resource_field:
# No --resource spec given: label the resource with the mime-like
# rendering of the fileset's type name, e.g. 'vectra-export'
resource_label = _type_name_resource_label(fileset.type_name)
resource_derived = True
else:
resource_label = IDSpec.get_value_from_matching_spec(
fileset, resource_field
)
# No --scan spec matches this fileset's type (e.g. the datatype-scoped
# 'SeriesNumber' default doesn't apply to a non-DICOM fileset): put the
# resource in a scan of the same name
scan_id = IDSpec.get_value_from_matching_spec(
fileset, scan_field, default=_DERIVED_ID
)
scan_derived = scan_id is _DERIVED_ID
if scan_derived:
scan_id = resource_label
derived_specs = [
spec
for spec, was_derived in (
("--scan", scan_derived),
("--resource", resource_derived),
)
if was_derived
]
clash_hint = (
f"the {' and '.join(derived_specs)} ID(s) for this resource were "
f"auto-derived from its fileset type; pass explicit "
f"{' / '.join(derived_specs)} specifier(s) to control grouping"
if derived_specs
else None
)
try:
session = sessions[session_uid]
except KeyError:
session = cls(
uid=session_uid,
run_uid=run_uid,
)
sessions[session_uid] = session
logger.debug(
"Adding resource '%s' to %s scan in %s session",
resource_label,
scan_id,
session_uid,
)
session.add_resource(
scan_id,
None,
resource_label,
fileset,
on_clash=on_resource_clash,
clash_hint=clash_hint,
)
# Inject metadata from the metadata tables into the sessions, scans, and resources
MetadataTable.inject_list(metadata_tables, list(sessions.values()))
for session in sessions.values():
MetadataTable.inject_list(metadata_tables, list(session.scans.values()))
for scan in session.scans.values():
MetadataTable.inject_list(
metadata_tables, list(scan.resources.values())
)
return list(sessions.values())
[docs]
def assign(
self,
project_field: str,
subject_field: str,
session_field: str,
scan_field: str | None = None,
constant_project_id: str | None = None,
) -> None:
"""Assigns project, subject and session IDs to the session, extracted from its
metadata. Also resolves a description for each scan in the session, if
'scan_field' is provided.
Parameters
----------
project_field : str
metadata field to extract the XNAT project ID from
subject_field : str
metadata field to extract the XNAT subject ID from
session_field: str
metadata field to extract the XNAT session ID from
constant_project_id : str
Override the project ID loaded from the metadata (useful when invoking
manually)
scan_field : str, optional
metadata field to extract a description for each scan from. Scans for which
the field can't be resolved are left without a description (saved with a
trailing-dot '<scan_id>.' directory name)
Notes
-----
If a project/subject/session field can't be resolved from the session's
metadata, a unique 'INVALID_MISSING_<FIELD>_<random>' placeholder is used
instead of raising, so the session can still be saved (see `invalid_ids`) for
manual review/reprocessing rather than being silently dropped.
"""
missing_ids: dict[str, str] = {}
if constant_project_id is None:
self.project_id = IDSpec(project_field).get_value(
self.metadata, missing_ids=missing_ids
)
else:
self.project_id = constant_project_id
self.subject_id = IDSpec(subject_field).get_value(
self.metadata, missing_ids=missing_ids
)
self.session_id = IDSpec(session_field).get_value(
self.metadata, missing_ids=missing_ids
)
if scan_field is not None:
for scan in self.scans.values():
try:
scan.type = IDSpec(scan_field).get_value(
scan.metadata, escape=False, missing_ids=missing_ids
)
except ImagingSessionParseError:
logger.debug(
"Could not resolve a description for scan '%s' from field "
"'%s', using scan ID instead",
scan.id,
scan_field,
)
scan.type = scan.id
return missing_ids
@classmethod
def from_orthanc(
cls,
url: str,
output_dir: Path,
store_dir: Path,
user: str,
password: str,
to_process_label: str | None = None,
processed_label: str = "xnat-sorted",
max_workers: int | None = None,
wait_period: int = 0,
) -> list["ImagingSession"]:
"""Stage DICOM studies from Orthanc directly into output_dir using hardlinks.
Requires orthanc_storage_dir and output_dir to be on the same filesystem.
Parameters
----------
url : str
Base URL of the Orthanc REST API, e.g. 'http://orthanc:8042'
output_dir : Path
Staging directory. Hardlinks land here directly, must be on the same
filesystem as orthanc_storage_dir.
store_dir : Path
Orthanc's StorageDirectory as mounted.
user : str, optional
Orthanc basic auth credentials username
password : str, optional
Orthanc basic auth credentials password
processed_label : str, optional
Label applied after staging to prevent re-processing, by default 'xnat-sorted'.
Remove via the Orthanc UI to re-sort a study.
max_workers : int, optional
the number of threads to use to fetch per-instance attachment info from
Orthanc concurrently. If None, defaults to
`concurrent.futures.ThreadPoolExecutor`'s default.
wait_period : int, optional
Minimum number of seconds since Orthanc last updated a study before it is
staged, by default 0.
Returns
-------
list[ImagingSession]
Staged sessions loaded from output_dir.
"""
auth = (user, password) if user else None
def get_json(path: str) -> ty.Any:
resp = requests.get(f"{url}{path}", auth=auth)
resp.raise_for_status()
return resp.json()
resp = requests.post(
f"{url}/tools/find",
auth=auth,
json={
"Level": "Study",
"Query": {},
"Labels": [processed_label],
"LabelsConstraint": "None",
},
)
resp.raise_for_status()
study_ids = resp.json()
logger.info("Found %d unstaged studies in Orthanc at '%s'", len(study_ids), url)
def _find_studies(labels: list[str], constraint: str) -> set[str]:
body: dict[str, ty.Any] = {"Level": "Study", "Query": {}}
if labels:
body["Labels"] = labels
body["LabelsConstraint"] = constraint
resp = requests.post(f"{url}/tools/find", auth=auth, json=body)
resp.raise_for_status()
return set(resp.json())
if to_process_label:
candidates = _find_studies([to_process_label], "All")
else:
candidates = _find_studies([], "All")
if processed_label:
candidates -= _find_studies([processed_label], "All")
study_ids = sorted(candidates)
logger.info(
"Found %d studies in Orthanc at '%s' (label=%r, skip label=%r)",
len(study_ids),
url,
to_process_label,
processed_label,
)
staged: list[ImagingSession] = []
for study_id in tqdm(study_ids, "Staging studies from Orthanc"):
study = get_json(f"/studies/{study_id}")
if wait_period:
try:
last_update = isoparse(study["LastUpdate"])
except (KeyError, TypeError, ValueError) as e:
raise ValueError(
f"Could not parse LastUpdate for Orthanc study '{study_id}'"
) from e
if last_update.tzinfo is None:
last_update = last_update.replace(tzinfo=UTC)
age = (datetime.now(UTC) - last_update).total_seconds()
if age < wait_period:
logger.info(
"Skipping Orthanc study '%s' because it was updated %.0f "
"seconds ago (wait period: %d seconds)",
study_id,
age,
wait_period,
)
continue
study_tags = {**study["MainDicomTags"], **study["PatientMainDicomTags"]}
session_uid = IDSpec("StudyInstanceUID").get_value(study_tags)
session_dir = output_dir / f"_.{session_uid}"
session_dir.mkdir(parents=True, exist_ok=True)
modalities: set[str] = set()
staged_instance_ids: dict[str, set[str]] = {}
for series_id in study["Series"]:
series = get_json(f"/series/{series_id}")
if modality := series["MainDicomTags"].get("Modality"):
modalities.add(modality)
all_tags = {**study_tags, **series["MainDicomTags"]}
scan_id = IDSpec("SeriesNumber").get_value(all_tags)
scan_type = IDSpec("SeriesDescription").get_value(all_tags)
if "ImageType" in all_tags:
resource_label = dicom_image_type_to_resource_label(
IDSpec("ImageType").get_value(all_tags)
)
else:
resource_label = "DICOM"
resource_dir = session_dir / f"{scan_id}.{scan_type}" / resource_label
resource_dir.mkdir(parents=True, exist_ok=True)
instances = get_json(f"/series/{series_id}/instances")
staged_instance_ids[series_id] = {
instance["ID"] for instance in instances
}
def _link_instance(
instance: ty.Mapping[str, ty.Any],
resource_dir: Path = resource_dir,
series_id: str = series_id,
) -> tuple[str, str]:
instance_id = instance["ID"]
sop_uid = instance["MainDicomTags"].get(
"SOPInstanceUID", instance_id
)
fname = f"{sop_uid}.dcm"
dest_path = resource_dir / fname
attachment = get_json(
f"/instances/{instance_id}/attachments/dicom/info"
)
if attachment["CompressedSize"] != attachment["UncompressedSize"]:
raise ValueError(
f"Instance '{instance_id}' in series '{series_id}' is stored "
"compressed in Orthanc — disable StorageCompression in the "
"Orthanc config to use hardlink sorting."
)
if not dest_path.exists():
uuid = attachment["Uuid"]
src_path = Path(store_dir) / uuid[0:2] / uuid[2:4] / uuid
os.link(src_path, dest_path)
return fname, attachment["UncompressedMD5"]
with ThreadPoolExecutor(max_workers=max_workers) as executor:
linked = executor.map(_link_instance, instances)
checksums: dict[str, str] = dict(linked)
manifest = {"datatype": "medimage/dicom-series", "checksums": checksums}
with open(resource_dir / ImagingResource.MANIFEST_FNAME, "w") as f:
json.dump(manifest, f, indent=4)
metadata_path = session_dir / Metadata.FNAME
if metadata_path.exists():
with open(metadata_path, "r") as f:
existing_tags = json.load(f)
study_tags.update(existing_tags)
if modalities:
study_tags["Modality"] = (
next(iter(modalities)) if len(modalities) == 1 else list(modalities)
)
with open(metadata_path, "w") as f:
json.dump(study_tags, f, indent=4, default=str)
staged_session = cls.load(session_dir)
if processed_label:
current_study = get_json(f"/studies/{study_id}")
current_series_ids = set(current_study["Series"])
if current_study.get("LastUpdate") != study.get(
"LastUpdate"
) or current_series_ids != set(staged_instance_ids):
raise RuntimeError(
f"Orthanc study '{study_id}' changed while it was being staged"
)
for series_id, expected_instance_ids in staged_instance_ids.items():
current_instances = get_json(f"/series/{series_id}/instances")
if {instance["ID"] for instance in current_instances} != (
expected_instance_ids
):
raise RuntimeError(
f"Orthanc study '{study_id}' changed while it was being staged"
)
requests.put(
f"{url}/studies/{study_id}/labels/{processed_label}", auth=auth
).raise_for_status()
logger.info(
"Staged and labelled study '%s' -> '%s'", study_id, session_dir.name
)
staged.append(staged_session)
return staged
[docs]
def deidentify(
self,
dest_dir: Path,
specs: dict[type[FileSet], ty.Any] | None = None,
copy_mode: FileSet.CopyMode = FileSet.CopyMode.hardlink_or_copy,
on_resource_clash: OnResourceClash = "error",
require_matching_spec: bool = True,
max_workers: int | None = None,
transforms: dict[type[FileSet], dict[str, Transform]] | None = None,
) -> tuple[Self, dict[str, ty.Any]]:
"""Creates a new session with deidentified images
Parameters
----------
dest_dir : Path
the directory to save the deidentified files into
specs : dict[type[FileSet], Any], optional
a project-specific specification that defines how to deidentify the different
file types within the imaging session. The keys of the project spec are
the mime-like of the file types (see https://arcanaframework.github.io/fileformats/)
and the values are arbitrary file-format-specific specifications.
copy_mode : FileSet.CopyMode, optional
the mode to use to copy the files that don't need to be deidentified,
by default FileSet.CopyMode.hardlink_or_copy
on_resource_clash : OnResourceClash, optional
when copying a file that doesn't need to be deidentified, if "avoid", if a resource with the same name already exists in the scan, increment the
resource name by appending _1, _2 etc. to the name until a unique name is found, by default "avoid"
if "merge", existing sessions with the same name will be merged.
if "error", an error will be raised if a session with the same name already exists in the staging directory.
if "overwrite", an existing resource with the same name will be overwritten.
require_matching_spec : bool, optional
whether to require a matching specification for each fileset, by default True
max_workers : int, optional
passed through as `max_workers` to each resource's `FileSet.deidentify`, for
formats whose deidentification implementation can parallelise work *within* a
single resource (e.g. the per-file loop for a DICOM series) using threads.
Formats that don't accept/use it just ignore it. Resources themselves are
deidentified/copied sequentially, one at a time, to keep failures easy to
trace back to the resource that caused them.
transforms : dict[type[FileSet], dict[str, Transform]], optional
per-format transforms that compute de-identification replacement values.
Keys are file-format types; values are dicts mapping transform names to
callables that accept a dataset/mapping and return a replacement value.
Passed through as ``variable_builders`` to ``FileSet.deidentify()``.
Returns
-------
ImagingSession
a new session with deidentified images
dict[str, Any]
a mapping containing the original values of metadata fields that
have been removed or modified
"""
if specs is None:
specs = {}
if transforms is None:
transforms = {}
def select_spec(fileset: FileSet) -> ty.Any:
"""Select the appropriate deidentification specification for the
resource based on its file type
"""
matching_specs = {k: v for k, v in specs.items() if isinstance(fileset, k)}
if not matching_specs:
return None
elif len(matching_specs) > 1:
for k in matching_specs:
if all(issubclass(k, other_k) for other_k in matching_specs):
return matching_specs[k]
raise KeyError(
f"Multiple deidentification specifications found for '{to_mime(type(fileset))}'"
f"file types. Please provide a more specific formats to map the specification"
f"specifications: {list(matching_specs)}"
)
return next(iter(matching_specs.values()))
def select_transforms(
fileset: FileSet,
) -> dict[str, Transform] | None:
"""Select the transforms that match this fileset's type."""
matching = {k: v for k, v in transforms.items() if isinstance(fileset, k)}
if not matching:
return None
if len(matching) == 1:
return next(iter(matching.values()))
# Prefer the most specific type
for k in matching:
if all(issubclass(k, other_k) for other_k in matching):
return matching[k]
# Fall back to merging all matching transforms
merged: dict[str, Transform] = {}
for v in matching.values():
merged.update(v)
return merged
# Create a new session to save the deidentified files into
deidentified = self.new_empty()
reid_series = []
for scan in self.scans.values():
for resource_name, resource in scan.resources.items():
resource_dest_dir = dest_dir / scan.id / resource_name
contains_phi = getattr(resource.fileset, "contains_phi", False)
resource_spec = None
resource_transforms = None
if contains_phi:
resource_spec = select_spec(resource.fileset)
resource_transforms = select_transforms(resource.fileset)
if resource_spec is None:
msg = (
"No deidentification specification found for %s fileset in %s/%s resource. "
"Please provide a project specification for %s in the file format hierarchy to "
"deidentify this resource. Returning None and copying the files without "
"deidentification, which may lead to PHI being uploaded to XNAT if the fileset "
"contains PHI. Matching specifications found in project spec: %s"
)
msg_vars = (
type(resource.fileset).__name__,
scan.id,
resource_name,
type(resource.fileset).__name__,
list(specs),
)
if require_matching_spec:
raise KeyError(msg % msg_vars)
else:
logger.warning(msg, *msg_vars)
deid_resource, reid_mdata = _deidentify_or_copy_resource(
resource.fileset,
resource_name,
resource_dest_dir,
contains_phi,
resource_spec,
copy_mode,
max_workers,
transforms=resource_transforms,
)
if reid_mdata is not None:
reid_series.append(reid_mdata)
deidentified.add_resource(
scan.id,
scan.type,
resource_name,
deid_resource,
on_clash=on_resource_clash,
)
# SESSION-LEVEL RESOURCES, which this loop used to drop entirely.
#
# A session can carry resources attached to the SESSION rather than to a
# scan -- a report, a summary, anything added with add_session_resource.
# deidentified starts from new_empty(), which copies the ids and nothing
# else, and the loop above walks self.scans only, so those resources
# never reached the output. They were not de-identified, not copied, and
# nothing said so.
#
# save() has always handled them (included_session_resources), so this
# was a hole in deidentify alone, and it is worse than a plain data loss:
# the per-session completeness gate counts data files on both sides, so a
# session carrying one would come out short, be reported incomplete, and
# correctly refuse to unlink its input -- for ever, on every cycle,
# because the next run drops it again.
#
# They go at the top of dest_dir rather than under a scan id, which is
# where save() puts them and where load() looks for them.
for resource_name, resource in self.session_resources.items():
contains_phi = getattr(resource.fileset, "contains_phi", False)
resource_spec = None
resource_transforms = None
if contains_phi:
resource_spec = select_spec(resource.fileset)
resource_transforms = select_transforms(resource.fileset)
if resource_spec is None:
msg = (
"No deidentification specification found for %s fileset in the "
"session-level %s resource. Please provide a project "
"specification for %s in the file format hierarchy to "
"deidentify this resource. Returning None and copying the files "
"without deidentification, which may lead to PHI being uploaded "
"to XNAT if the fileset contains PHI. Matching specifications "
"found in project spec: %s"
)
msg_vars = (
type(resource.fileset).__name__,
resource_name,
type(resource.fileset).__name__,
list(specs),
)
if require_matching_spec:
raise KeyError(msg % msg_vars)
else:
logger.warning(msg, *msg_vars)
deid_resource, reid_mdata = _deidentify_or_copy_resource(
resource.fileset,
resource_name,
dest_dir / resource_name,
contains_phi,
resource_spec,
copy_mode,
max_workers,
transforms=resource_transforms,
)
if reid_mdata is not None:
reid_series.append(reid_mdata)
deidentified.add_session_resource(resource_name, deid_resource)
return deidentified, collate_metadata_series(reid_series)
[docs]
def associate_files(
self,
patterns: list[AssociatedFiles],
spaces_to_underscores: bool = True,
on_resource_clash: OnResourceClash = "error",
) -> list[FileSet]:
"""Adds files associated with the primary files to the session
Parameters
----------
patterns : list[AssociatedFiles]
list of patterns to associate files with the primary files in the session
spaces_to_underscores : bool, optional
when building associated file globs, convert spaces underscores in fields
extracted from source file metadata, false by default
"""
all_associated = []
for associated_files in patterns:
# substitute string templates int the glob template with values from the
# DICOM metadata to construct a glob pattern to select files associated
# with current session
associated_fspaths: set[Path] = set()
primary_parents = self.primary_parents
if primary_parents:
for parent_dir in primary_parents:
assoc_glob = str(
parent_dir / associated_files.glob.format(**self.metadata)
)
if spaces_to_underscores:
assoc_glob = assoc_glob.replace(" ", "_")
# Select files using the constructed glob pattern
associated_fspaths.update(
Path(p) for p in glob(assoc_glob, recursive=True)
)
elif self.metadata:
assoc_glob = associated_files.glob.format(**self.metadata)
if spaces_to_underscores:
assoc_glob = assoc_glob.replace(" ", "_")
associated_fspaths.update(
Path(p) for p in glob(assoc_glob, recursive=True)
)
logger.info(
"Found %s associated file paths matching '%s'",
len(associated_fspaths),
associated_files.glob,
)
# Identify scan id, type and resource names from deidentified file paths
assoc_re = re.compile(associated_files.identity_pattern)
for fspath in tqdm(associated_fspaths, "sorting files into resources"):
match = assoc_re.match(str(fspath))
if not match:
raise RuntimeError(
f"Regular expression '{associated_files.identity_pattern}' "
f"did not match file path {fspath}"
)
scan_id = match.group("id")
resource_name = match.group("resource")
try:
scan_type = match.group("type")
except IndexError:
scan_type = scan_id
fspaths = from_paths([fspath], associated_files.datatype)
self.add_resource(
scan_id,
scan_type,
resource_name,
fspaths[0],
associated=associated_files,
on_clash=on_resource_clash,
)
all_associated.extend(fspaths)
return all_associated
def add_resource(
self,
scan_id: str,
scan_type: str | None,
resource_name: str,
fileset: FileSet,
associated: AssociatedFiles | None = None,
on_clash: OnResourceClash | ty.Sequence[ClashSpec] = "error",
metadata: dict[str, ty.Any] | None = None,
clash_hint: str | None = None,
) -> None:
"""Adds a resource to the imaging session
Parameters
----------
scan_id : str
the ID of the scan to add the resource to
scan_type : str
short description of the type of the scan
resource_name: str
the name of the resource to add
fileset : FileSet
the fileset to add as the resource
associated : bool, optional
whether the resource is primary or associated to a primary resource
on_clash : OnResourceClash or Sequence[ClashSpec], optional
a bare policy ("error"/"avoid"/"merge"/"overwrite") applied to any
clash, or a sequence of ``ClashSpec`` (policy + datatype scope) where
the clash is resolved by the first spec whose scope covers *both* the
existing and incoming filesets - a clash no spec covers raises.
"avoid" suffixes the name, "merge" folds both into a ``SetOf``,
"overwrite" replaces the existing one, "error" raises. Default "error".
metadata : dict[str, Any], optional
Dictionary containing metadata values to update the resource with.
clash_hint : str, optional
extra context appended to the message when a resource-name clash is
hit (raised for ``on_clash="error"``, logged for ``"avoid"``), e.g. to
note that the clashing IDs were auto-derived because no ``--scan``/
``--resource`` spec was given.
"""
try:
scan = self.scans[scan_id]
except KeyError:
scan = self.scans[scan_id] = ImagingScan(
id=scan_id, type=scan_type, associated=associated, session=self
)
else:
if scan.type != scan_type:
raise ValueError(
f"Non-matching scan types ({scan.type} and {scan_type}) "
f"for scan ID {scan_id}"
)
if associated != scan.associated:
raise ValueError(
f"Non-matching associated files ({scan.associated} and {associated}) "
f"for scan ID {scan_id}"
)
resource = ImagingResource(name=resource_name, fileset=fileset, scan=scan)
if metadata:
resource.metadata.update(metadata)
try:
existing = scan.resources[resource_name]
except KeyError:
pass
else:
if resource.checksums == existing.checksums:
logger.info(
"Not adding resource '%s' to %s scan in %s session as it is identical "
"to a resource that is already present %s",
resource_name,
scan_id,
self.name,
existing,
)
return
if isinstance(on_clash, str):
policy = on_clash
else:
policy = _resolve_clash_policy(
on_clash,
existing.fileset,
fileset,
f"for resource '{resource_name}' in {scan_id} scan of "
f"{self.name} session",
)
if policy == "overwrite":
logger.warning(
"Overwriting existing resource '%s' in %s scan in %s session",
resource_name,
scan_id,
self.name,
)
del scan.resources[resource_name]
elif policy == "merge":
logger.info(
"Merging resource '%s' with existing resource in %s scan in %s session",
resource_name,
scan_id,
self.name,
)
# Combine the members into a single ``SetOf[...]`` resource,
# classified by the union of their content types, and collate their
# metadata the same way a scan collates its resources' (see
# ``ImagingScan.metadata``): fields every member agrees on stay
# scalar, fields that differ (e.g. a per-file 'relpath' from
# ``--path-metadata-regex``) become a list aligned with the merged
# files.
existing_fspaths = list(existing.fileset.fspaths)
content_types = tuple(
dict.fromkeys(
_set_content_types(existing.fileset)
+ _set_content_types(fileset)
)
)
merged_fileset = SetOf[content_types](
[*existing_fspaths, *fileset.fspaths]
)
members = _expand_collated_metadata(
dict(existing.fileset.metadata), len(existing_fspaths)
)
members.append(dict(fileset.metadata))
merged_fileset.metadata.update(Metadata.collate(members))
resource = ImagingResource(
name=resource_name, fileset=merged_fileset, scan=scan
)
if metadata:
resource.metadata.update(metadata)
elif policy == "avoid":
match = re.match(r"^(.*)__(\d+)$", resource_name)
if match:
base_name, num = match.groups()
num = int(num) + 1
else:
base_name = resource_name
num = 2
while resource_name in scan.resources:
resource_name = f"{base_name}__{num}"
num += 1
logger.warning(
"Incremented resource name to '%s' to avoid clash with existing "
"resources%s",
resource_name,
f". {clash_hint}" if clash_hint else "",
)
resource = ImagingResource(
name=resource_name, fileset=fileset, scan=scan
)
elif policy == "error":
raise KeyError(
f"Clash between resource names ('{resource_name}') for {scan_id} scan in "
f"{self.name} session. Pass --on-resource-clash <policy> <scope> "
"(policy one of 'avoid'/'merge'/'overwrite') with a scope covering "
"the clashing datatype(s), or tighten --scan / --resource so they "
"don't collide." + (f" {clash_hint}" if clash_hint else ""),
)
else:
assert False, (
f"Invalid resource-clash policy: {policy} (should be one of {ON_RESOURCE_CLASH})"
)
scan.resources[resource_name] = resource
def add_session_resource(
self,
resource_name: str,
fileset: FileSet,
overwrite: bool = False,
) -> None:
"""Adds a session-level resource
Parameters
----------
resource_name : str
the name of the resource
fileset : FileSet
the fileset to add as the resource
overwrite : bool
whether to overwrite an existing resource with the same name
"""
resource = ImagingResource(name=resource_name, fileset=fileset)
if resource_name in self.session_resources:
existing = self.session_resources[resource_name]
if resource.checksums == existing.checksums:
return
if not overwrite:
raise KeyError(
f"Session resource '{resource_name}' already exists in {self.name}. "
"Use 'overwrite=True' to overwrite."
)
self.session_resources[resource_name] = resource
@classmethod
def from_metadata_yaml(cls, yaml_path: Path) -> Self:
"""Creates a metadata-only session from a __metadata__/ YAML file.
Parameters
----------
yaml_path : Path
path to a YAML file named PROJECT.SUBJECT.SESSION.yaml
Returns
-------
ImagingSession
a session with no scans but with metadata populated
"""
stem = yaml_path.stem
parts = stem.split(".")
if len(parts) != 3:
raise ValueError(
f"Expected metadata YAML filename to have format "
f"PROJECT.SUBJECT.SESSION.yaml, got '{yaml_path.name}'"
)
project_id, subject_id, session_id = parts
with open(yaml_path) as f:
metadata = yaml.safe_load(f)
session = cls(
uid=metadata[cls.UID_METADATA_KEY],
project_id=project_id,
subject_id=subject_id,
session_id=session_id,
)
session.metadata = Metadata(metadata, session)
return session
[docs]
@classmethod
def load(
cls,
session_dir: Path,
require_manifest: bool = True,
check_checksums: bool = True,
) -> Self:
"""Loads a session from a directory. Assumes that the name of the directory is
the name of the session dir and the parent directory is the subject ID and the
grandparent directory is the project ID. The scan information is loaded from a YAML
along with the scan type, resources and fileformats. If the YAML file is not found
or `use_manifest` is set to True, the session is loaded based on the directory
structure.
Parameters
----------
session_dir : Path
the path to the directory where the session is saved
require_manifiest: bool, optional
whether a manifest file is required to load the resources in the session,
if true, resources will only be loaded if the manifest file is found,
if false, resources will be loaded as FileSet types and checksums will not
be checked, by default True
check_checksums: bool, optional
whether to check the checksums of the files in the session, by default True
Returns
-------
ImagingSession
the loaded session
"""
if session_dir.name.startswith(cls.PRE_ASSIGN_PREFIX):
# Session has been grouped into scans but not yet had project/subject/session
# IDs assigned to it
session = cls(uid=session_dir.name[len(cls.PRE_ASSIGN_PREFIX) :])
else:
if "." in session_dir.name:
parts = session_dir.name.split(".")
else:
# Backwards compatibility with old delimiter
parts = session_dir.name.split("-")
if len(parts) == 4:
project_id, subject_id, session_id, run_uid = parts
else:
project_id, subject_id, session_id = parts
run_uid = None
session = cls(
uid=session_dir.name,
project_id=project_id,
subject_id=subject_id,
session_id=session_id,
run_uid=run_uid,
)
for item in session_dir.iterdir():
if not item.is_dir():
continue
if "." in item.name:
# scan directory: <scan_id>.<scan_type>
scan = ImagingScan.load(
item,
require_manifest=require_manifest,
check_checksums=check_checksums,
)
scan.session = session
session.scans[scan.id] = scan
else:
# session resource directory: <resource_name> (no dot)
resource = ImagingResource.load(
item,
require_manifest=require_manifest,
check_checksums=check_checksums,
)
session.session_resources[resource.name] = resource
if (session_dir / Metadata.FNAME).exists():
session.metadata = Metadata.load(session_dir, session)
session.uid = session.metadata.get(cls.UID_METADATA_KEY, None)
return session
def staging_dirname(self, available_projects: list[str] | None = None) -> str:
"""The directory name this session is saved under by :meth:`save`.
Split out of ``save`` so that a caller can work out where the session
WILL land before saving it. ``deidentify_api`` needs that to decide
whether an output already exists, and rebuilding the rule at the call
site would let the two drift: the name is not simply the input
directory's, because it is derived from the assigned ids, gains a
``run_uid`` suffix when one is set and an invalid-project prefix when
the project is unrecognised.
"""
if self.name is None:
# Project/subject/session IDs haven't been assigned yet, so flag the
# directory as not-yet-assigned rather than assuming they're set
return self.staging_relpath[0]
if available_projects is None or self.project_id in available_projects:
project_id = self.project_id
else:
project_id = "INVALID_UNRECOGNISED_" + self.project_id
session_dirname = f"{project_id}.{self.subject_id}.{self.session_id}"
if self.run_uid:
session_dirname += f".{self.run_uid}"
return session_dirname
[docs]
def save(
self,
dest_dir: Path,
available_projects: list[str] | None = None,
copy_mode: FileSet.CopyMode = FileSet.CopyMode.hardlink_or_copy,
collation_map: dict[type[FileSet], FileSet.CopyCollation] | None = None,
conversion_map: (
dict[type[FileSet], tuple[ty.Type[FileSet], dict[str, str]]] | None
) = None,
include: ty.Sequence[type[FileSet]] = (),
) -> tuple[Self, Path]:
r"""Saves the session to a directory. The session will be saved to a directory
with the project, subject and session IDs as subdirectories of this directory,
along with the scans manifest
Parameters
----------
dest_dir : Path
destination directory to save the deidentified files. The session will be saved
to a directory with the project, subject and session IDs as subdirectories of
this directory, along with the scans manifest
available_projects : list[str], optional
list of available project IDs on the XNAT server, if the project ID of the
session is not in this list, it will be prefixed with ``INVALID_UNRECOGNISED_``
to avoid upload errors, by default None
copy_mode : FileSet.CopyMode, optional
the mode to use to copy the files that don't need to be deidentified,
by default FileSet.CopyMode.hardlink_or_copy
include : sequence[type[FileSet]], optional
only save resources matching at least one of these datatypes. An empty
sequence saves all resources.
Returns
-------
ImagingSession
a deidentified session with updated paths
Path
the path to the directory where the session is saved
"""
included_scans = (
[
scan
for scan in self.scans.values()
if any(
resource.matches_datatypes(include)
for resource in scan.resources.values()
)
]
if include
else list(self.scans.values())
)
included_session_resources = [
resource
for resource in self.session_resources.values()
if resource.matches_datatypes(include)
]
if include and not included_scans and not included_session_resources:
raise ValueError(
f"No resources in {self.name or self.uid!r} match the included datatypes"
)
saved = self.new_empty()
session_dir = dest_dir / self.staging_dirname(available_projects)
session_dir.mkdir(parents=True, exist_ok=True)
for scan in tqdm(included_scans, f"Staging sessions to {session_dir}"):
saved_scan = scan.save(
session_dir,
copy_mode=copy_mode,
collation_map=collation_map,
conversion_map=conversion_map,
include=include,
)
saved_scan.session = saved
saved.scans[saved_scan.id] = saved_scan
for resource in included_session_resources:
saved_resource = resource.save(session_dir, copy_mode=copy_mode)
saved.session_resources[saved_resource.name] = saved_resource
logger.debug("Saving session metadata")
self.metadata[self.UID_METADATA_KEY] = self.uid
self.metadata.save(session_dir)
return saved, session_dir
@classmethod
def move_dir(cls, src: Path, dest: Path):
with SoftFileLock(dest.with_suffix(".lock")):
if dest.exists():
logger.info(
"Merging sorted session '%s' into existing directory '%s'",
src.name,
dest,
)
for scan_dir in src.iterdir():
if scan_dir.is_dir():
scan_dir.rename(dest / scan_dir.name)
exist_mdata_path = dest / cls.METADATA_FNAME
new_mdata_path = src / cls.METADATA_FNAME
if new_mdata_path.exists():
if exist_mdata_path.exists():
# Merge metadata files
mdata = Yaml(exist_mdata_path).load()
new_mdata = Yaml(new_mdata_path).load()
for key in set(mdata) & set(new_mdata):
if mdata[key] != new_mdata[key]:
raise ValueError(
f"Conflict in metadata for key '{key}' between existing session at "
f"'{exist_mdata_path}' and new session at '{new_mdata_path}'"
)
mdata.update(new_mdata)
Yaml(exist_mdata_path).save(mdata)
else:
new_mdata_path.rename(exist_mdata_path)
if remaining := list(src.iterdir()):
raise ValueError(
f"Unexpected files/directories {remaining} found in saved session directory '{src}' "
f"after merging with existing session directory '{dest}'"
)
src.rmdir()
else:
src.rename(dest)
MANIFEST_FNAME = "MANIFEST.yaml"
def unlink(self, keep_metadata: bool = False) -> None:
"""Unlink all resources in the session
Parameters
----------
keep_metadata : bool, optional
if True, each resource's directory is removed in its entirety (data
files plus its own manifest/metadata), but the enclosing scan and
session directories — and their own ``__METADATA__.json`` files, which
are always written by :meth:`save` — are left in place. This leaves a
lightweight metadata-only skeleton of the session on disk that can
still be loaded later (e.g. by ``associate`` to work out which scan a
late-arriving file belongs to) without needing to know whether the
session's data has already been cleaned up. Only safe to use on a
staged session directory that this session exclusively owns — never on
a session loaded from a shared source directory (see
:meth:`ImagingResource.unlink`), by default False
"""
for scan in self.scans.values():
for resource in scan.resources.values():
resource.unlink(remove_dir=keep_metadata)
for resource in self.session_resources.values():
resource.unlink(remove_dir=keep_metadata)
def last_modified(self) -> int:
"""Returns the timestamp of the most recently modified file in the session
in nanoseconds
Returns
-------
int
the mtime of the most recently modified file in the session in nanoseconds
"""
return max(
resource.fileset.last_modified
for scan in self.scans.values()
for resource in scan.resources.values()
)
def fix_long_path(p: str | Path) -> Path:
r"""Add \\?\ or \\?\UNC\ prefix on Windows for long paths."""
if platform.system() != "Windows":
return Path(p)
path = Path(p)
path_str = str(path.absolute())
# Already has prefix, don't double-apply
if path_str.startswith("\\\\?\\"):
return path
# UNC path: \\server\share\... -> \\?\UNC\server\share\...
if path_str.startswith("\\\\"):
return Path(f"\\\\?\\UNC\\{path_str[2:]}")
# Local path: C:\... -> \\?\C:\...
return Path(f"\\\\?\\{path_str}")
from .store import ImagingSessionMockStore # noqa: E402
def json_serializer(obj: ty.Any) -> ty.Any:
if isinstance(obj, Path):
return str(obj)
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
def dicom_image_type_to_resource_label(image_type: list[str]) -> str:
"""Maps the image type of a DICOM series to the hard-coded resource names
required by XNAT"""
if image_type[:2] == [
"DERIVED",
"SECONDARY",
]:
resource_label = "secondary"
else:
resource_label = "DICOM" # special case
return resource_label