API

Public

The following functions mirror the command line tools, and can be used to evoke the same modular behaviour via a Python workflow.

xnat_ingest.api.group(input_paths: list[str], output_dir: Path, datatypes: list[FileSet], session: list[IDSpec], scan: list[IDSpec], resource: list[IDSpec], path_metadata_regex: Sequence[PathMetadataRegex] = (), unlink_source: str | None = None, raise_errors: bool = False, copy_mode: CopyMode = CopyMode.hardlink_or_copy, collation_map: dict[Type[FileSet], CopyCollation] | None = None, wait_period: int = 0, avoid_clashes: bool = True, recursive: bool = False) → list[str]

Groups the input files into sessions/scans/resources and stages them into the staging directory. Project/subject/visit IDs and scan descriptions are not assigned at this point, see the 'assign' function for that.

Parameters:

• input_paths (list[str]) -- List of paths to search for input files. Can be local paths or S3 paths.

• output_dir (Path) -- Path to the staging directory where the grouped sessions will be saved. This should be a local path.

• datatypes (list[MimeType]) -- List of datatypes to look for in the input files. Only files with these datatypes will be considered for staging.

• session (list[IDSpec] | None) -- List of field specifications to use for extracting the session UIDs from the input files to group them into separate sessions

• scan (list[IDSpec]) -- List of field specifications to use for extracting the scan IDs from the input files to group them into scans

• resource (list[IDSpec]) -- List of field specifications to use for extracting the resource IDs from the input files to group them into resources

• path_metadata_regex (ty.Sequence[PathMetadataRegex]) -- 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.

• unlink_source (str | None) -- If "all" or "keep-metadata", the input files will be unlinked one by one after staging (both behave the same here, since the source isn't a directory tree that xnat-ingest owns). If None, the input files will be left in place.

• raise_errors (bool) -- If True, any errors encountered during staging will raise an exception. If False, errors will be logged and the staging process will continue for the remaining sessions.

• copy_mode (FileSet.CopyMode) -- The copy mode to use when saving the sessions. This determines whether files are copied, moved or symlinked when saving the sessions to the staging directory.

• wait_period (int) -- If provided, this is the number of seconds that must have passed since the last modification time of the session before it will be staged. This can be used to avoid staging sessions that are still being modified or created.

• avoid_clashes (bool) -- If True, if a session with the same name already exists in the staging directory, a suffix will be added to the session name to avoid overwriting the existing session. If False, existing sessions with the same name will be overwritten.

• recursive (bool) -- If True, the input paths will be searched recursively for files to stage. If False, only the files directly within the input paths will be considered for staging.

xnat_ingest.api.group_orthanc(url: str, store_dir: Path, output_dir: Path, user: str, password: str, to_process_label: str | None = None, processed_label: str | None = None, unlink_source: str | None = None, raise_errors: bool = False, copy_mode: CopyMode = CopyMode.hardlink_or_copy) → list[str]

Groups the input files into sessions and stages them into the staging directory.

Parameters:

• url (str) -- Orthanc server to retrieve the DICOM resources from.

• output_dir (Path) -- Path to the staging directory where the grouped sessions will be saved. This should be

• user (str) -- Orthanc user to login with

• password (str) -- Orthanc password to login with

• processed_label (str | None) -- The label applied to the sessions in Orthanc by this script to signify that they have already been processed.

• to_process_label (str | None) -- The label externally applied to sessions in Orthanc to signify that should be processed. If None, all sessions will be processed.

• session_id (list[FieldSpec] | None) -- List of field specifications to use for extracting the session ID from the input files. If None, the session ID will be generated from the subject and visit IDs.

• scan_id (list[FieldSpec]) -- List of field specifications to use for extracting the scan ID from the input files.

• unlink_source (str | None) -- If "all" or "keep-metadata", the source studies in Orthanc will be unlinked after staging. Not yet implemented. If None, the source studies will be left in place.

• raise_errors (bool) -- If True, any errors encountered during staging will raise an exception. If False, errors will be logged and the staging process will continue for the remaining sessions.

• copy_mode (FileSet.CopyMode) -- The copy mode to use when saving the sessions. This determines whether files are copied, moved or symlinked when saving the sessions to the staging directory.

• wait_period (int) -- If provided, this is the number of seconds that must have passed since the last modification time of the session before it will be staged. This can be used to avoid staging sessions that are still being modified or created.

xnat_ingest.api.assign(input_dir: Path, output_dir: Path, project_field: str, subject_field: str, session_field: str, scan_field: str | None = None, project_id: str | None = None, copy_mode: CopyMode = CopyMode.hardlink_or_copy, unlink_source: str | None = None, raise_errors: bool = False) → list[str]

Sorts the input files into sessions and stages them into the staging directory.

Parameters:

• input_dir (Path) -- List of paths to search for input files. Can be local paths or S3 paths.

• output_dir (Path) -- Path to the staging directory where the sorted sessions will be saved. This should be a local path.

• project_field (str) -- Field name to use for extracting the project ID from the input files.

• subject_field (str) -- Field name to use for extracting the subject ID from the input files.

• session_field (str) -- Field name to use for extracting the session ID from the input files.

• scan_field (str | None) -- Field name to use for extracting a description for each scan. Scans for which the field can't be resolved are left without a description.

• project_id (str | None) -- If provided, this project ID will be used for all sessions instead of extracting it from the input files. Useful for instruments that upload to a single project.

• copy_mode (FileSet.CopyMode) -- The copy mode to use when saving the sessions. This determines whether files are copied, moved or symlinked when saving the sessions to the staging directory.

• unlink_source (str | None) -- If "all", the grouped session directory is removed in its entirety after assignment. If "keep-metadata", the resource data is removed but the session/scan-level metadata is left behind as a lightweight skeleton. If None, the grouped session directory is left in place.

• raise_errors (bool) -- If True, any errors encountered during staging will raise an exception. If False, errors will be logged and the staging process will continue for the remaining sessions.

xnat_ingest.api.deidentify(input_dir: Path, output_dir: Path, spec_dir: Path, reid_dir: Path, avoid_clashes: bool = False, raise_errors: bool = False, copy_mode: CopyMode = CopyMode.copy, require_manifest: bool = True, unlink_source: str | None = None, reid_encrypt_key: bytes | None = None) → list[str]

xnat_ingest.api.upload(input_dir: str, xnat_repo: Xnat, always_include: Sequence[str | FileSet] = (), store_credentials: StoreCredentials | None = None, require_manifest: bool = True, methods: Sequence[UploadMethod] = (), wait_period: int = 0, num_files_per_batch: int = 0, check_checksums: bool = True, s3_cache_dir: Path | None = None, raise_errors: bool = False, dry_run: bool = False) → list[str]

Upload sorted sessions in the given staging directory to XNAT

Parameters:

• input_dir (Path) -- The directory containing the sessions to upload. Each session should be in a separate subdirectory.

• xnat_repo (Xnat) -- The XNAT repository to upload to

• always_include (Sequence[str]) -- A sequence of scan types or file paths to always include in the upload regardless of whether they are explicitly specified in the frameset definition

• raise_errors (bool) -- Whether to raise errors that occur during upload or to log them and continue with the next session

• store_credentials (StoreCredentials) -- Whether to store credentials for accessing staging directories that require authentication (e.g. S3)

• require_manifest (bool) -- Whether to require a manifest file in each session directory that specifies the resources to upload and their checksums

• methods (Sequence[UploadMethod]) -- The upload method to use for each datatype (e.g. 'tgz_file' or 'directory')

• wait_period (int) -- The minimum age in seconds of session directories to upload (only applicable for local staging directories)

• num_files_per_batch (int) -- The number of files to upload in each batch when uploading resources with the 'directory' method (if 0, all files will be uploaded in a single batch)

• check_checksums (bool) -- Whether to check checksums of uploaded resources against the checksums specified in the manifest file and the checksums of the files in the staged resources (if available) to verify that they were

• dry_run (bool) -- Whether to list the sessions that would be uploaded instead of actually uploading them

xnat_ingest.api.check_upload(input_dir: str, server: str, user: str, password: str, store_credentials: ~xnat_ingest.helpers.arg_types.StoreCredentials, always_include: ~typing.Sequence[str | ~typing.Type[~fileformats.core.fileset.FileSet]] = (<class 'fileformats.medimage.dicom.DicomSeries'>, ), temp_dir: ~pathlib.Path | None = None, verify_ssl: bool = True, use_curl_jsession: bool = False, disable_progress: bool = False) → None

Checks the staged sessions against the XNAT server to check for any issues before upload.

This is intended to be used as a pre-upload check to identify any issues with the staged sessions before attempting to upload them to XNAT. It checks that the project, session, scan and resource exist on XNAT and that the checksums of the files in the resource match those on XNAT. If any issues are found, they are logged and a summary of the issues is logged at the end. If the --raise-errors flag is set, then any issues will raise an exception instead of being logged.

Parameters:

• input_dir (str) -- The directory containing the staged sessions to check.

• server (str) -- The URL of the XNAT server to check against.

• user (str) -- The username to use when connecting to the XNAT server.

• password (str) -- The password to use when connecting to the XNAT server.

• loggers (list[LoggerConfig]) -- The logger configurations to use for logging the output of the function.

• always_include (list[str]) -- A list of MIME types to always include in the check, even if they aren't defined in the frameset on XNAT. This can be used to include additional file formats that aren't defined in the frameset, or to include all file formats by using "all".

xnat_ingest.api.associate(input_dir: Path, output_dir: Path, datatype: Type[FileSet], glob: str, identity_pattern: str, spaces_to_underscores: bool = False, avoid_clashes: bool = False, raise_errors: bool = False, require_manifest: bool = True, copy_mode: CopyMode = CopyMode.copy, unlink_source: str | None = None) → list[str]

Model classes

These classes represent an imaging session as it moves through the pipeline stages above: a session contains scans, each of which contains one or more resources (e.g. the original DICOM files, a derived NIfTI conversion).

class xnat_ingest.model.session.ImagingSession(uid: str, project_id: str | None = None, subject_id: str | None = None, session_id: str | None = None, scans: Sequence[ImagingScan] | Dict[str, ImagingScan] = NOTHING, session_resources: Dict[str, ImagingResource] = NOTHING, run_uid: str | None = None)

Bases: object

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

assign(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)

Raises:

ImagingSessionParseError -- if none of the candidate fields for a given ID resolve to a value in the session's metadata

associate_files(patterns: List[AssociatedFiles], spaces_to_underscores: bool = True, avoid_clashes: bool = False) → 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

deidentify(dest_dir: Path, specs: dict[type[FileSet], Any] = None, copy_mode: CopyMode = CopyMode.hardlink_or_copy, avoid_clashes: bool = False, require_matching_spec: bool = True) → tuple[Self, dict[str, 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

• avoid_clashes (bool, optional) -- when copying a file that doesn't need to be deidentified, 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 False

• require_matching_spec (bool, optional) -- whether to require a matching specification for each fileset, by default True

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

classmethod from_paths(files_path: str | Path | Sequence[str | Path], datatypes: Type[FileSet] | Sequence[Type[FileSet]], session_field: list[IDSpec], scan_field: list[IDSpec], resource_field: list[IDSpec], recursive: bool = False, avoid_clashes: bool = True, path_metadata_regex: Sequence[PathMetadataRegex] = ()) → 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 (list[IdField]) -- 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 (list[IdField]) -- the value of this field is used to group resources under single scans.

• resource_field (list[IdField]) -- the value of this field is to resources

• recursive (bool, optional) -- recurse into directories passed as file paths (i.e. by appending **/* and running a glob), by default False

• avoid_clashes (bool, optional) -- 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 False

• 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.

Returns:

all imaging sessions that are present in list of dicom paths

Return type:

list[ImagingSession]

Raises:

ImagingSessionParseError -- if values extracted from IDs across the DICOM scans are not consistent across DICOM files within the session

classmethod load(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:

the loaded session

Return type:

ImagingSession

metadata: Metadata

project_id: str | None

run_uid: str | None

save(dest_dir: Path, available_projects: List[str] | None = None, copy_mode: CopyMode = CopyMode.hardlink_or_copy, collation_map: dict[Type[FileSet], CopyCollation] | None = None) → tuple[Self, Path]

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

Returns:

• ImagingSession -- a deidentified session with updated paths

• Path -- the path to the directory where the session is saved

scans: Dict[str, ImagingScan]

session_id: str | None

session_resources: Dict[str, ImagingResource]

subject_id: str | None

uid: str

class xnat_ingest.model.scan.ImagingScan(id: str, type: str | None = None, resources: dict[str, ImagingResource | FileSet] = NOTHING, associated: AssociatedFiles | None = None, session: ImagingSession = None)

Bases: object

Representation of a scan to be uploaded to XNAT

Parameters:

• id (str) -- the ID of the scan on XNAT

• type (str | None) -- the scan type/description

• resources (dict[str, ImagingResource]) -- the list of the resources associated with the scan

• associated (AssociatedFiles, optional) -- Secondary files that are associated with the given imaging scan (but may not have the necessary metadata to identify the scan itself)

associated: AssociatedFiles | None

id: str

classmethod load(scan_dir: Path, require_manifest: bool = True, check_checksums: bool = True) → Self

metadata: Metadata

resources: Dict[str, ImagingResource]

save(dest_dir: Path, copy_mode: CopyMode = CopyMode.hardlink_or_copy, collation_map: dict[Type[FileSet], CopyCollation] | None = None) → Self

type: str

class xnat_ingest.model.resource.ImagingResource(name: str, fileset: FileSet, checksums: dict[str, str] = NOTHING, scan: ImagingScan = None)

Bases: object

Representation of a resource to be uploaded to XNAT, which is a set of files associated with a scan. There can be multiple resources associated with a scan, e.g. the original DICOM files, a BIDS-converted version of the scan, and a brain mask generated from the scan.

Parameters:

• name (str) -- The name of the resource

• fileset (FileSet) -- The set of files associated with the resource

• checksums (dict[str, str]) -- The checksums of the files in the resource

• scan (ImagingScan) -- The scan that the resource is associated with

checksums: dict[str, str]

fileset: FileSet

classmethod load(resource_dir: Path, require_manifest: bool = True, check_checksums: bool = True) → Self

Load a resource from a directory, reading the manifest file if it exists. If the manifest file doesn't exist and 'require_manifest' is True then an exception is raised, if it is False, then a generic FileSet object is loaded from the files that were found

metadata: Metadata

name: str

save(dest_dir: Path, copy_mode: CopyMode = CopyMode.copy, collation_map: dict[Type[FileSet], CopyCollation] | None = None, calculate_checksums: bool = True, overwrite: bool | None = None) → Self

Save the resource to a directory

Parameters:

• dest_dir (Path) -- The directory to save the resource

• copy_mode (FileSet.CopyMode) -- The method to copy the files

• calculate_checksums (bool) -- Whether to calculate the checksums of the files

• overwrite (bool) -- Whether to overwrite the resource if it already exists, if None then the files are overwritten if they are newer than the ones saved, otherwise a warning is issued, if False an exception will be raised, if True then the resource is saved regardless of the files being newer

Returns:

The saved resource

Return type:

ImagingResource

Raises:

FileExistsError -- If the resource already exists and overwrite is False or None and the files are not newer

scan: ImagingScan