Module dcm2bids
dcm2bids
Reorganising NIfTI files from dcm2niix into the Brain Imaging Data Structure
View Source
# -*- coding: utf-8 -*-
"""
dcm2bids
--------
Reorganising NIfTI files from dcm2niix into the Brain Imaging Data Structure
"""
from .dcm2bids import Dcm2bids
from .scaffold import scaffold
from .version import __version__
__all__ = ["__version__", "Dcm2bids", "scaffold"]
Sub-modules
- dcm2bids.dcm2bids
- dcm2bids.dcm2niix
- dcm2bids.helper
- dcm2bids.logger
- dcm2bids.scaffold
- dcm2bids.sidecar
- dcm2bids.structure
- dcm2bids.utils
- dcm2bids.version
Variables
__version__
Functions
scaffold
def scaffold(
output_dir_override: Optional[str] = None
)
scaffold entry point
View Source
def scaffold(output_dir_override: Optional[str] = None):
"""scaffold entry point"""
args = _get_arguments()
output_dir_ = output_dir_override if output_dir_override is not None else args.output_dir
for _ in ["code", "derivatives", "sourcedata"]:
os.makedirs(os.path.join(output_dir_, _), exist_ok=True)
for _ in [
"dataset_description.json",
"participants.json",
"participants.tsv",
"README",
]:
dest = os.path.join(output_dir_, _)
with resources.path(__name__, _) as src:
shutil.copyfile(src, dest)
with resources.path(__name__, "CHANGES") as changes_template:
with open(changes_template) as _:
data = _.read().format(datetime.date.today().strftime("%Y-%m-%d"))
write_txt(
os.path.join(output_dir_, "CHANGES"),
data.split("\n")[:-1],
)
Classes
Dcm2bids
class Dcm2bids(
dicom_dir,
participant,
config,
output_dir=PosixPath('/home/sam/Projects/Dcm2Bids'),
session='',
clobber=False,
forceDcm2niix=False,
log_level='WARNING',
**_
)
Object to handle dcm2bids execution steps
Attributes
| Name | Type | Description | Default |
|---|---|---|---|
| dicom_dir | str or list | A list of folder with dicoms to convert | None |
| participant | str | Label of your participant | None |
| config | path | Path to a dcm2bids configuration file | None |
| output_dir | path | Path to the BIDS base folder | None |
| session | str | Optional label of a session | None |
| clobber | boolean | Overwrite file if already in BIDS folder | None |
| forceDcm2niix | boolean | Forces a cleaning of a previous execution of dcm2niix |
None |
| log_level | str | logging level | None |
View Source
class Dcm2bids(object):
""" Object to handle dcm2bids execution steps
Args:
dicom_dir (str or list): A list of folder with dicoms to convert
participant (str): Label of your participant
config (path): Path to a dcm2bids configuration file
output_dir (path): Path to the BIDS base folder
session (str): Optional label of a session
clobber (boolean): Overwrite file if already in BIDS folder
forceDcm2niix (boolean): Forces a cleaning of a previous execution of
dcm2niix
log_level (str): logging level
"""
def __init__(
self,
dicom_dir,
participant,
config,
output_dir=DEFAULT.outputDir,
session=DEFAULT.session,
clobber=DEFAULT.clobber,
forceDcm2niix=DEFAULT.forceDcm2niix,
log_level=DEFAULT.logLevel,
**_
):
self._dicomDirs = []
self.dicomDirs = dicom_dir
self.bidsDir = valid_path(output_dir, type="folder")
self.config = load_json(valid_path(config, type="file"))
self.participant = Participant(participant, session)
self.clobber = clobber
self.forceDcm2niix = forceDcm2niix
self.logLevel = log_level
# logging setup
self.set_logger()
self.logger.info("--- dcm2bids start ---")
self.logger.info("OS:version: %s", platform.platform())
self.logger.info("python:version: %s", sys.version.replace("\n", ""))
self.logger.info("dcm2bids:version: %s", __version__)
self.logger.info("dcm2niix:version: %s", dcm2niix_version())
self.logger.info("participant: %s", self.participant.name)
self.logger.info("session: %s", self.participant.session)
self.logger.info("config: %s", os.path.realpath(config))
self.logger.info("BIDS directory: %s", os.path.realpath(output_dir))
@property
def dicomDirs(self):
"""List of DICOMs directories"""
return self._dicomDirs
@dicomDirs.setter
def dicomDirs(self, value):
dicom_dirs = value if isinstance(value, list) else [value]
valid_dirs = [valid_path(_dir, "folder") for _dir in dicom_dirs]
self._dicomDirs = valid_dirs
def set_logger(self):
""" Set a basic logger"""
logDir = self.bidsDir / DEFAULT.tmpDirName / "log"
logFile = logDir / f"{self.participant.prefix}_{datetime.now().isoformat().replace(':', '')}.log"
logDir.mkdir(parents=True, exist_ok=True)
setup_logging(self.logLevel, logFile)
self.logger = logging.getLogger(__name__)
def run(self):
"""Run dcm2bids"""
dcm2niix = Dcm2niix(
self.dicomDirs,
self.bidsDir,
self.participant,
self.config.get("dcm2niixOptions", DEFAULT.dcm2niixOptions),
)
check_latest()
check_latest("dcm2niix")
dcm2niix.run(self.forceDcm2niix)
sidecars = []
for filename in dcm2niix.sidecarFiles:
sidecars.append(
Sidecar(filename, self.config.get("compKeys", DEFAULT.compKeys))
)
sidecars = sorted(sidecars)
parser = SidecarPairing(
sidecars,
self.config["descriptions"],
self.config.get("searchMethod", DEFAULT.searchMethod),
self.config.get("caseSensitive", DEFAULT.caseSensitive)
)
parser.build_graph()
parser.build_acquisitions(self.participant)
parser.find_runs()
self.logger.info("moving acquisitions into BIDS folder")
intendedForList = [[] for i in range(len(parser.descriptions))]
for acq in parser.acquisitions:
acq.setDstFile()
intendedForList = self.move(acq, intendedForList)
def move(self, acquisition, intendedForList):
"""Move an acquisition to BIDS format"""
for srcFile in glob(acquisition.srcRoot + ".*"):
ext = Path(srcFile).suffixes
ext = [curr_ext for curr_ext in ext if curr_ext in ['.nii','.gz',
'.json',
'.bval','.bvec']]
dstFile = (self.bidsDir / acquisition.dstRoot).with_suffix("".join(ext))
dstFile.parent.mkdir(parents = True, exist_ok = True)
# checking if destination file exists
if dstFile.exists():
self.logger.info("'%s' already exists", dstFile)
if self.clobber:
self.logger.info("Overwriting because of --clobber option")
else:
self.logger.info("Use --clobber option to overwrite")
continue
# it's an anat nifti file and the user using a deface script
if (
self.config.get("defaceTpl")
and acquisition.dataType == "func"
and ".nii" in ext
):
try:
os.remove(dstFile)
except FileNotFoundError:
pass
defaceTpl = self.config.get("defaceTpl")
cmd = [w.replace('srcFile', srcFile) for w in defaceTpl]
cmd = [w.replace('dstFile', dstFile) for w in defaceTpl]
run_shell_command(cmd)
intendedForList[acquisition.indexSidecar].append(acquisition.dstIntendedFor + "".join(ext))
elif ".json" in ext:
data = acquisition.dstSidecarData(self.config["descriptions"],
intendedForList)
save_json(dstFile, data)
os.remove(srcFile)
# just move
else:
os.rename(srcFile, dstFile)
intendedFile = acquisition.dstIntendedFor + ".nii.gz"
if intendedFile not in intendedForList[acquisition.indexSidecar]:
intendedForList[acquisition.indexSidecar].append(intendedFile)
return intendedForList
Instance variables
dicomDirs
List of DICOMs directories
Methods
move
def move(
self,
acquisition,
intendedForList
)
Move an acquisition to BIDS format
View Source
def move(self, acquisition, intendedForList):
"""Move an acquisition to BIDS format"""
for srcFile in glob(acquisition.srcRoot + ".*"):
ext = Path(srcFile).suffixes
ext = [curr_ext for curr_ext in ext if curr_ext in ['.nii','.gz',
'.json',
'.bval','.bvec']]
dstFile = (self.bidsDir / acquisition.dstRoot).with_suffix("".join(ext))
dstFile.parent.mkdir(parents = True, exist_ok = True)
# checking if destination file exists
if dstFile.exists():
self.logger.info("'%s' already exists", dstFile)
if self.clobber:
self.logger.info("Overwriting because of --clobber option")
else:
self.logger.info("Use --clobber option to overwrite")
continue
# it's an anat nifti file and the user using a deface script
if (
self.config.get("defaceTpl")
and acquisition.dataType == "func"
and ".nii" in ext
):
try:
os.remove(dstFile)
except FileNotFoundError:
pass
defaceTpl = self.config.get("defaceTpl")
cmd = [w.replace('srcFile', srcFile) for w in defaceTpl]
cmd = [w.replace('dstFile', dstFile) for w in defaceTpl]
run_shell_command(cmd)
intendedForList[acquisition.indexSidecar].append(acquisition.dstIntendedFor + "".join(ext))
elif ".json" in ext:
data = acquisition.dstSidecarData(self.config["descriptions"],
intendedForList)
save_json(dstFile, data)
os.remove(srcFile)
# just move
else:
os.rename(srcFile, dstFile)
intendedFile = acquisition.dstIntendedFor + ".nii.gz"
if intendedFile not in intendedForList[acquisition.indexSidecar]:
intendedForList[acquisition.indexSidecar].append(intendedFile)
return intendedForList
run
def run(
self
)
Run dcm2bids
View Source
def run(self):
"""Run dcm2bids"""
dcm2niix = Dcm2niix(
self.dicomDirs,
self.bidsDir,
self.participant,
self.config.get("dcm2niixOptions", DEFAULT.dcm2niixOptions),
)
check_latest()
check_latest("dcm2niix")
dcm2niix.run(self.forceDcm2niix)
sidecars = []
for filename in dcm2niix.sidecarFiles:
sidecars.append(
Sidecar(filename, self.config.get("compKeys", DEFAULT.compKeys))
)
sidecars = sorted(sidecars)
parser = SidecarPairing(
sidecars,
self.config["descriptions"],
self.config.get("searchMethod", DEFAULT.searchMethod),
self.config.get("caseSensitive", DEFAULT.caseSensitive)
)
parser.build_graph()
parser.build_acquisitions(self.participant)
parser.find_runs()
self.logger.info("moving acquisitions into BIDS folder")
intendedForList = [[] for i in range(len(parser.descriptions))]
for acq in parser.acquisitions:
acq.setDstFile()
intendedForList = self.move(acq, intendedForList)
set_logger
def set_logger(
self
)
Set a basic logger
View Source
def set_logger(self):
""" Set a basic logger"""
logDir = self.bidsDir / DEFAULT.tmpDirName / "log"
logFile = logDir / f"{self.participant.prefix}_{datetime.now().isoformat().replace(':', '')}.log"
logDir.mkdir(parents=True, exist_ok=True)
setup_logging(self.logLevel, logFile)
self.logger = logging.getLogger(__name__)
Last update:
2023-07-13
Created: 2023-07-13
Created: 2023-07-13