552 lines
18 KiB
Python
552 lines
18 KiB
Python
# /// script
|
|
# requires-python = ">=3.12"
|
|
# dependencies = [
|
|
# "datalad",
|
|
# "datalad-next",
|
|
# "datalad-container",
|
|
# "datalad-core",
|
|
# "pydicom",
|
|
# "trr379-rdmtools @ git+https://hub.trr379.de/q02/rdmtools@main",
|
|
# ]
|
|
# ///
|
|
|
|
# Known issues
|
|
#
|
|
# - Conversion is reproducible, but it will still change the dataset due to
|
|
# https://github.com/nipy/heudiconv/issues/852
|
|
|
|
from datalad_core import repo
|
|
from datalad_core.config import ConfigItem
|
|
from datalad_core.runners import (
|
|
CommandError,
|
|
call_git,
|
|
call_git_oneline,
|
|
call_git_success,
|
|
)
|
|
from pathlib import (
|
|
Path,
|
|
PurePosixPath,
|
|
)
|
|
import rich_click as click
|
|
from rich.progress import (
|
|
Progress,
|
|
track,
|
|
)
|
|
from shutil import copyfile
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
|
|
from trr379_rdmtools.dicom_archive import make_dicom_archive
|
|
from trr379_rdmtools.utils import (
|
|
get_remote_dataset_state,
|
|
has_staged_changes,
|
|
post_failure,
|
|
post_success,
|
|
post_noop_result,
|
|
shallow_clone_submodule,
|
|
)
|
|
|
|
|
|
hub_url = 'https://hub.trr379.de'
|
|
dicom_org = 'q01-heidelberg-mrisessions'
|
|
all_dicoms_ds_url = 'https://hub.trr379.de/q01-heidelberg/mri-dicoms'
|
|
bids_ds_url = 'https://hub.trr379.de/q01-heidelberg/q01-bids-dataset'
|
|
envvar_prefix = 'TRR_'
|
|
|
|
default_dataset_content = {
|
|
'.datalad/.gitattributes': 'config annex.largefiles=nothing\n',
|
|
'.gitattributes':
|
|
'* annex.backend=MD5E\n**/.git* annex.largefiles=nothing\n',
|
|
}
|
|
|
|
|
|
@click.group(
|
|
'uk-heidelberg',
|
|
epilog='See https://hub.trr379.de/q02/rdmtools for a changelog, and (open) issues.'
|
|
)
|
|
def cli():
|
|
"""\
|
|
Connect MRI data workflows at the UK Heidelberg with the TRR379 RDM system.
|
|
|
|
Use of this tool requires no explicit use of DataLad, nor the dedicated
|
|
maintenance of any DataLad dataset. All commands run idempotent.
|
|
|
|
BIDS conversion is fully reproducible.
|
|
The brain imaging facility only needs to store the original DICOMs.
|
|
The BIDS-converted outputs are tracked with DataLad, but provided in a plain directory.
|
|
The content of this directory can be reproduced from the original DICOM tarballs as needed.
|
|
|
|
All commands are designed on run in temporary directories (can be RAM-disks). Placement
|
|
of these temporary directories is controlled by standard environment variable (e.g., `TMPDIR`).
|
|
"""
|
|
|
|
|
|
def get_archive_dest_rpath(
|
|
dicoms: Path,
|
|
sub_id: str,
|
|
ses_id: str,
|
|
):
|
|
return f'{sub_id}-{ses_id}.tar'
|
|
|
|
|
|
def get_session_dataset_url(sub_id: str, ses_id: str) -> str:
|
|
return f'{hub_url}/{dicom_org}/{sub_id}-{ses_id}.git'
|
|
|
|
|
|
def get_session_dataset_rpath(sub_id: str, ses_id: str) -> str:
|
|
"""Within the DICOM sessions dataset"""
|
|
return f'sessions/{sub_id}/{ses_id}'
|
|
|
|
|
|
def create_dataset(path, private_annex=False) -> repo.Worktree:
|
|
wt = repo.Worktree.init_at(path)
|
|
if private_annex:
|
|
wt.config.sources['git-local'].add(
|
|
'annex.private',
|
|
ConfigItem('true', coercer=bool),
|
|
)
|
|
wt.init_annex()
|
|
wt.config.sources['datalad-branch'].add(
|
|
'datalad.dataset.id',
|
|
ConfigItem(str(uuid.uuid4())),
|
|
)
|
|
for rpath, content in default_dataset_content.items():
|
|
(wt.path / PurePosixPath(rpath)).write_text(content)
|
|
call_git(['add', '.'], cwd=wt.path)
|
|
call_git(['commit', '-q', '-m', 'new datalad dataset'], cwd=wt.path)
|
|
return wt
|
|
|
|
|
|
param_cfg = {
|
|
'dicoms': {
|
|
'type': click.Path(
|
|
exists=True, file_okay=False, dir_okay=True,
|
|
readable=True, path_type=Path,
|
|
),
|
|
'help': 'Path to a directory with DICOMs for a single scanning session',
|
|
},
|
|
'sub-id': {
|
|
'help': 'Pseudonymized TRR379 Q01 subject identifier corresponding to the scan',
|
|
},
|
|
'ses-id': {
|
|
'help': 'TRR379 session identifier for this subject',
|
|
},
|
|
('-m', '--message'): {
|
|
'help': 'Commit message WHY a registration was updated',
|
|
},
|
|
'--container-cache-dir': {
|
|
'envvar': f'{envvar_prefix}CONTAINERCACHE_DIR', 'show_envvar': True,
|
|
'type': click.Path(
|
|
exists=True, file_okay=False, dir_okay=True, path_type=Path,
|
|
),
|
|
'help': 'Directory for locally caching container image files '
|
|
'to speed up processing',
|
|
},
|
|
}
|
|
|
|
|
|
@cli.command(
|
|
'create-session-dataset',
|
|
help="Create and deposit a new DataLad dataset for a single scanning session",
|
|
)
|
|
@click.argument('dicoms', **param_cfg['dicoms'])
|
|
@click.argument('sub-id', **param_cfg['sub-id'])
|
|
@click.argument('ses-id', **param_cfg['ses-id'])
|
|
def create_session_dataset(dicoms, sub_id, ses_id):
|
|
try:
|
|
with tempfile.TemporaryDirectory() as wdir:
|
|
gitsha, deposit_url, dsid = _do_session_dataset(
|
|
Path(wdir),
|
|
dicoms=dicoms,
|
|
sub_id=sub_id,
|
|
ses_id=ses_id,
|
|
)
|
|
except ValueError as e:
|
|
post_failure(str(e))
|
|
sys.exit(1)
|
|
|
|
post_success(
|
|
f'DICOM session DataLad dataset deposited at {deposit_url}',
|
|
)
|
|
|
|
|
|
def _do_session_dataset(wdir: Path, dicoms: Path, sub_id: str, ses_id: str):
|
|
"""
|
|
"""
|
|
# TODO check at the very start whether we already have the target dataset
|
|
# on the hub
|
|
# check if the
|
|
|
|
deposit_url = get_session_dataset_url(sub_id, ses_id)
|
|
|
|
# try clone as a check for conflicts. later this might be an approach
|
|
# to implement updates.
|
|
if call_git_success(
|
|
['ls-remote', deposit_url],
|
|
capture_output=True,
|
|
):
|
|
# TODO: proper error
|
|
msg = 'There is already a DICOM session dataset for ' \
|
|
f'{sub_id}-{ses_id} at {deposit_url}. If the subject/session ' \
|
|
'identifier is correct, this dataset needs to be updated ' \
|
|
'manually, if needed. DO NOT DELETE this existing repository, ' \
|
|
'because it would invalidate existing metadata.'
|
|
raise ValueError(msg)
|
|
wt = create_dataset(wdir, private_annex=True)
|
|
archive_dest_path = wt.path / get_archive_dest_rpath(dicoms, sub_id, ses_id)
|
|
make_dicom_archive(dicoms, archive_dest_path, track)
|
|
call_git(['annex', 'add', archive_dest_path], cwd=wt.path)
|
|
call_git(['commit', '-m', 'ingest DICOM session archive'], cwd=wt.path)
|
|
call_git(['remote', 'add', 'origin', deposit_url], cwd=wt.path)
|
|
call_git(['push', '--all'], cwd=wt.path)
|
|
gitsha = call_git_oneline(['rev-parse', 'HEAD'], cwd=wt.path)
|
|
return gitsha, deposit_url, wt.config.get('datalad.dataset.id').value
|
|
|
|
|
|
@cli.command(
|
|
'register-session-dataset',
|
|
help="Register a single-session dataset in the all-scan-sessions "
|
|
"DataLad dataset",
|
|
)
|
|
@click.argument('sub-id', **param_cfg['sub-id'])
|
|
@click.argument('ses-id', **param_cfg['ses-id'])
|
|
@click.option('-m', '--message', **param_cfg[('-m', '--message')])
|
|
def register_session_dataset(sub_id, ses_id, message: str | None = None):
|
|
deposit_url = get_session_dataset_url(sub_id, ses_id)
|
|
dsid, gitsha = get_remote_dataset_state(deposit_url)
|
|
|
|
with tempfile.TemporaryDirectory() as wdir:
|
|
gitsha, updated = _do_dicoms_dataset(
|
|
Path(wdir),
|
|
sub_id=sub_id,
|
|
ses_id=ses_id,
|
|
gitsha=gitsha,
|
|
deposit_url=deposit_url,
|
|
dsid=dsid,
|
|
message=message,
|
|
)
|
|
if updated:
|
|
post_success(
|
|
f'Pushed updated DICOM sessions DataLad dataset',
|
|
)
|
|
else:
|
|
post_noop_result('Registration found up-to-date, no changes made')
|
|
|
|
|
|
def _do_dicoms_dataset(
|
|
wdir: Path,
|
|
sub_id: str,
|
|
ses_id: str,
|
|
gitsha: str,
|
|
deposit_url: str,
|
|
dsid: str,
|
|
message: str | None,
|
|
) -> tuple[str, bool]:
|
|
call_git(['clone', '-q', '--depth', '1', all_dicoms_ds_url, wdir])
|
|
subm_rpath = get_session_dataset_rpath(sub_id, ses_id)
|
|
# subproject commit
|
|
call_git(
|
|
['update-index', '--add', '--replace', '--cacheinfo', '160000',
|
|
gitsha, subm_rpath],
|
|
cwd=wdir,
|
|
)
|
|
for k, v in (
|
|
('path', subm_rpath),
|
|
('url', deposit_url),
|
|
('datalad-id', dsid),
|
|
):
|
|
call_git(
|
|
# with newer Git use
|
|
# ['config', 'set', '--file', '.gitmodules',
|
|
['config', '--file', '.gitmodules',
|
|
f'submodule.{subm_rpath}.{k}', v],
|
|
cwd=wdir,
|
|
)
|
|
call_git(['add', '.gitmodules'], cwd=wdir)
|
|
updated = False
|
|
if has_staged_changes(wdir):
|
|
call_git(['commit', '-m',
|
|
message or f'add DICOM {sub_id}-{ses_id} session dataset'],
|
|
cwd=wdir)
|
|
call_git(['push'], cwd=wdir)
|
|
updated = True
|
|
gitsha = call_git_oneline(['rev-parse', 'HEAD'], cwd=wdir)
|
|
return gitsha, updated
|
|
|
|
|
|
@cli.command(
|
|
'update-bids-sourcedata',
|
|
help="Update the BIDS dataset to the latest DICOM source data",
|
|
)
|
|
def update_bids_sourcedata():
|
|
with tempfile.TemporaryDirectory() as wdir:
|
|
updated = _do_update_bids_sourcedata(Path(wdir))
|
|
|
|
if updated:
|
|
post_success(
|
|
'Pushed updated BIDS DataLad dataset',
|
|
)
|
|
else:
|
|
post_noop_result(
|
|
'DICOM dataset registration found up-to-date, no changes made'
|
|
)
|
|
|
|
|
|
def _do_update_bids_sourcedata(wdir: Path, gitsha: str | None = None) -> bool:
|
|
if gitsha is None:
|
|
_, gitsha = get_remote_dataset_state(all_dicoms_ds_url)
|
|
|
|
call_git(['clone', '-q', '--depth', '1', bids_ds_url, wdir])
|
|
call_git(
|
|
['update-index', '--add', '--replace', '--cacheinfo', '160000',
|
|
gitsha, 'sourcedata/dicoms'],
|
|
cwd=wdir,
|
|
)
|
|
if has_staged_changes(wdir):
|
|
call_git(['commit', '-m', 'update DICOM source dataset'], cwd=wdir)
|
|
call_git(['push'], cwd=wdir)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
@cli.command(
|
|
'bids-conversion',
|
|
help="""\
|
|
Convert MRI data from a single DICOM session to BIDS format,
|
|
and incorporate it into the Q01 BIDS DataLad dataset.
|
|
|
|
Pseudonymized TRR379 Q01 subject and session identifiers have to be used.
|
|
These identifiers must match those used to register the corresponding
|
|
single-session DICOM dataset.
|
|
""",
|
|
)
|
|
@click.argument('dicoms', **param_cfg['dicoms'])
|
|
@click.argument('sub-id', **param_cfg['sub-id'])
|
|
@click.argument('ses-id', **param_cfg['ses-id'])
|
|
@click.option('-m', '--message', **param_cfg[('-m', '--message')])
|
|
@click.option('--container-cache-dir', **param_cfg['--container-cache-dir'])
|
|
def bids_conversion(
|
|
dicoms: Path,
|
|
sub_id: str,
|
|
ses_id: str,
|
|
message: str | None = None,
|
|
container_cache_dir: Path | None = None,
|
|
):
|
|
with tempfile.TemporaryDirectory() as wdir:
|
|
_do_bids_dataset(
|
|
Path(wdir),
|
|
dicoms,
|
|
sub_id,
|
|
ses_id,
|
|
gitsha=None,
|
|
container_cache_dir=container_cache_dir,
|
|
)
|
|
|
|
|
|
def _init_container_from_cache(
|
|
wdir: Path,
|
|
container_cache_dir: Path,
|
|
):
|
|
container_ds_rpath = 'code/heudiconv'
|
|
shallow_clone_submodule(wdir, container_ds_rpath)
|
|
container_ds_path = wdir / container_ds_rpath
|
|
# we need to fetch the git-annex branch to avoid it being created
|
|
# after the previous shallow clone
|
|
call_git(['fetch', '-q', 'origin', 'git-annex:git-annex', '--depth', '1'],
|
|
cwd=container_ds_path)
|
|
call_git(['annex', 'init'], cwd=container_ds_path)
|
|
|
|
container_img_rpath = call_git_oneline(
|
|
# with newer Git use
|
|
# ['config', 'get', '--file', '.datalad/config',
|
|
['config', '--file', '.datalad/config',
|
|
'datalad.containers.apptainer.image'],
|
|
cwd=container_ds_path,
|
|
)
|
|
img_key = call_git_oneline(
|
|
['annex', 'lookupkey', container_img_rpath],
|
|
cwd=container_ds_path,
|
|
)
|
|
cached_img = container_cache_dir / img_key
|
|
if cached_img.exists():
|
|
with Progress() as progress:
|
|
progress.add_task("Injecting cached container image", total=None)
|
|
copyfile(cached_img, container_ds_path / '.img_key')
|
|
call_git(
|
|
['annex', 'reinject', '.img_key', container_img_rpath],
|
|
cwd=container_ds_path,
|
|
)
|
|
|
|
|
|
def _cache_container(
|
|
wdir: Path,
|
|
container_cache_dir: Path,
|
|
):
|
|
container_img_rpath = call_git_oneline(
|
|
# with newer Git use
|
|
# ['config', 'get', '--file', '.datalad/config',
|
|
['config', '--file', '.datalad/config',
|
|
'datalad.containers.apptainer.image'],
|
|
cwd=wdir,
|
|
)
|
|
img_key = call_git_oneline(
|
|
['annex', 'lookupkey', container_img_rpath], cwd=wdir,
|
|
)
|
|
key_rpath = PurePosixPath(call_git_oneline(
|
|
['annex', 'contentlocation', img_key], cwd=wdir,
|
|
))
|
|
if not (container_cache_dir / key_rpath.name).exists():
|
|
with Progress() as progress:
|
|
progress.add_task("Cache container image", total=None)
|
|
copyfile(wdir / key_rpath, container_cache_dir / key_rpath.name)
|
|
|
|
|
|
def _init_annex(repodir: Path, *, private):
|
|
if private:
|
|
# with newer Git use
|
|
# call_git(['config', 'set', 'annex.private', 'true'], cwd=repodir)
|
|
call_git(['config', 'annex.private', 'true'], cwd=repodir)
|
|
# we need to fetch the git-annex branch to avoid it being created
|
|
# after the previous shallow clone
|
|
try:
|
|
call_git(
|
|
['fetch', '-q', 'origin', 'git-annex:git-annex', '--depth', '1'],
|
|
cwd=repodir,
|
|
)
|
|
except CommandError:
|
|
# we allow for failure, maybe there is none, and we will spawn it here
|
|
pass
|
|
call_git(['annex', 'init'], cwd=repodir)
|
|
|
|
|
|
def _do_bids_dataset(
|
|
wdir: Path,
|
|
dicoms: Path,
|
|
sub_id: str,
|
|
ses_id: str,
|
|
gitsha: str | None = None,
|
|
container_cache_dir: Path | None = None,
|
|
):
|
|
_do_update_bids_sourcedata(Path(wdir), gitsha=gitsha)
|
|
# now we prep the two repos that take writes. we want their
|
|
# local annexes to be private (we have ephemeral clones), and we do not
|
|
# want to fetch all history for speed, but nevertheless need to avoid
|
|
# creating detached git-annex branches
|
|
shallow_clone_submodule(wdir, '.heudiconv')
|
|
# we need a branch to push update
|
|
call_git(['switch', 'main'], cwd=wdir / '.heudiconv')
|
|
for repodir in (wdir, wdir / '.heudiconv'):
|
|
_init_annex(repodir, private=True)
|
|
|
|
dicom_src_rpath = 'sourcedata/dicoms'
|
|
# grab the relevant source dataset to inject the DICOM archive key we know
|
|
# we have, and avoid the GB download
|
|
# not using datalad-get, it cannot handle shallow clones
|
|
shallow_clone_submodule(wdir, dicom_src_rpath)
|
|
|
|
dicom_session_rpath = get_session_dataset_rpath(sub_id, ses_id)
|
|
shallow_clone_submodule(wdir / dicom_src_rpath, dicom_session_rpath)
|
|
|
|
session_dicom_ds_path = wdir / dicom_src_rpath / dicom_session_rpath
|
|
call_git(['annex', 'init'], cwd=session_dicom_ds_path)
|
|
# 'reinject' will move the file, hence we need a copy to maintain
|
|
# the input in its pristine location
|
|
with Progress() as progress:
|
|
progress.add_task("Injecting DICOM session archive", total=None)
|
|
make_dicom_archive(dicoms, session_dicom_ds_path / '.archive_key', track)
|
|
call_git(
|
|
['annex', 'reinject', '.archive_key',
|
|
get_archive_dest_rpath(dicoms, sub_id, ses_id)],
|
|
cwd=session_dicom_ds_path,
|
|
)
|
|
if container_cache_dir:
|
|
_init_container_from_cache(wdir, container_cache_dir)
|
|
else:
|
|
shallow_clone_submodule(wdir, 'code/heudiconv')
|
|
subm_path = wdir / 'code' / 'heudiconv'
|
|
# we need to fetch the git-annex branch to avoid it being created
|
|
# after the previous shallow clone
|
|
call_git(['fetch', '-q', 'origin', 'git-annex:git-annex', '--depth', '1'],
|
|
cwd=subm_path)
|
|
call_git(['annex', 'init'], cwd=subm_path)
|
|
|
|
pre_gitsha = call_git_oneline(['rev-parse', 'HEAD'], cwd=wdir)
|
|
# this is the only thing for which we need legacy datalad
|
|
import datalad.api as dl
|
|
with Progress() as progress:
|
|
progress.add_task("BIDS conversion", total=None)
|
|
dl.containers_run(
|
|
dataset=dl.Dataset(wdir),
|
|
message=f"Convert {sub_id}-{ses_id} DICOM data",
|
|
container_name='code/heudiconv/apptainer',
|
|
outputs=[
|
|
# the rest is taken care of by --overwrite
|
|
f"sub-{sub_id}/ses-{ses_id}/sub-{sub_id}_ses-{ses_id}_scans.tsv",
|
|
f".heudiconv/sub-{sub_id}/ses-{ses_id}",
|
|
],
|
|
inputs=[
|
|
'sourcedata/dicoms/code/heuristic-q01.py',
|
|
f'{dicom_src_rpath}/{dicom_session_rpath}',
|
|
],
|
|
cmd='--bids notop --overwrite --minmeta '
|
|
'-o . '
|
|
"-f '{inputs[0]}' "
|
|
f'-s "{sub_id}" -ss "{ses_id}" '
|
|
f'--files {dicom_src_rpath}/{dicom_session_rpath}',
|
|
)
|
|
# only as the very last step push the new state.
|
|
# if there was any error before this point, we can simply rerun
|
|
# everything -- there is no local state
|
|
for repodir in (wdir, wdir / '.heudiconv'):
|
|
call_git(['annex', 'copy', '-t', 'origin', '.'], cwd=repodir)
|
|
call_git(['push'], cwd=repodir)
|
|
# update the cache
|
|
if container_cache_dir:
|
|
_cache_container(wdir / 'code' / 'heudiconv', container_cache_dir)
|
|
if pre_gitsha == call_git_oneline(['rev-parse', 'HEAD'], cwd=wdir):
|
|
post_noop_result('No changes to tracked content.')
|
|
else:
|
|
post_success('Conversion successfull.')
|
|
|
|
|
|
@cli.command(
|
|
'ingest-new-session',
|
|
help="""\
|
|
Convenience command to perform all processing steps until (and incl. BIDS-conversion) for a new scan.
|
|
|
|
This command is only for processing new data. In case of updates for previously registered data,
|
|
the corresponding individual commands must be used instead (`create-session-dataset`,
|
|
`register-session-dataset`, `update-bids-sourcedata`, `bids-conversion`).
|
|
""",
|
|
)
|
|
@click.argument('dicoms', **param_cfg['dicoms'])
|
|
@click.argument('sub-id', **param_cfg['sub-id'])
|
|
@click.argument('ses-id', **param_cfg['ses-id'])
|
|
@click.option('--container-cache-dir', **param_cfg['--container-cache-dir'])
|
|
@click.pass_context
|
|
def ingest_new_session(
|
|
ctx,
|
|
dicoms: Path,
|
|
sub_id: str,
|
|
ses_id: str,
|
|
container_cache_dir: Path | None = None,
|
|
) -> None:
|
|
ctx.invoke(
|
|
create_session_dataset,
|
|
dicoms=dicoms, sub_id=sub_id, ses_id=ses_id,
|
|
)
|
|
ctx.invoke(
|
|
register_session_dataset,
|
|
sub_id=sub_id, ses_id=ses_id,
|
|
)
|
|
ctx.invoke(
|
|
bids_conversion,
|
|
dicoms=dicoms, sub_id=sub_id, ses_id=ses_id,
|
|
container_cache_dir=container_cache_dir,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|