Michael Hanke
bbd6a602db
Do not change internal layout based on the external archive name. This would impact the checksum.
124 lines
3.5 KiB
Python
124 lines
3.5 KiB
Python
import tarfile
|
|
from datetime import datetime
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
|
|
# this implementation works with pydicom 2x
|
|
from pydicom import (
|
|
dcmread,
|
|
)
|
|
from pydicom.valuerep import (
|
|
DA,
|
|
TM,
|
|
)
|
|
from pydicom.errors import InvalidDicomError
|
|
|
|
|
|
default_date = datetime(1970, 1, 1)
|
|
|
|
|
|
def scan_dir(
|
|
path: Path,
|
|
progress_handler,
|
|
) -> Dict:
|
|
for p in progress_handler(path.rglob('*'), description='Scanning'):
|
|
if p.is_dir():
|
|
continue
|
|
try:
|
|
# determine a reproducible timestamp for this DICOM file
|
|
# based on required attributes (0008, 0020) and (0008, 0030)
|
|
with dcmread(p) as dcm:
|
|
studydate = DA(dcm.StudyDate)
|
|
studytime = TM(dcm.StudyTime)
|
|
timestamp = datetime.combine(studydate, studytime)
|
|
yield (p, timestamp)
|
|
except InvalidDicomError:
|
|
# this is not a DICOM file, report path without timestamp
|
|
yield (p, None)
|
|
|
|
|
|
def write_archive(
|
|
dest_path: Path,
|
|
input_base_dir: Path,
|
|
content: Dict,
|
|
default_timestamp: datetime,
|
|
progress_handler,
|
|
):
|
|
# might have ben done already, but the check is cheap, so do a localized
|
|
# one here
|
|
if dest_path.exists():
|
|
# be safe
|
|
raise ValueError(
|
|
f'output path {dest_path} already exists, refusing to overwrite')
|
|
|
|
dest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# write uncompressed TAR
|
|
with tarfile.open(dest_path, "w") as tar:
|
|
# order of member in archive is significant, sort by path
|
|
for p in progress_handler(sorted(content), 'Composing archive'):
|
|
tinfo = tar.gettarinfo(
|
|
name=p.resolve(),
|
|
)
|
|
# adjust properties to make archive builds reproducible
|
|
tinfo = normalize_tarinfo(
|
|
tinfo,
|
|
p.relative_to(input_base_dir),
|
|
# go with the reported timestamp from DICOM or with default
|
|
content[p] or default_timestamp,
|
|
)
|
|
# ingest into archive
|
|
with p.open('rb') as fp:
|
|
tar.addfile(tinfo, fp)
|
|
|
|
|
|
def normalize_tarinfo(tinfo, member_name, timestamp):
|
|
tinfo.name = str(member_name)
|
|
# be safe
|
|
tinfo.uid = 0
|
|
tinfo.gid = 0
|
|
tinfo.uname = 'root'
|
|
tinfo.gname = 'root'
|
|
tinfo.mtime = timestamp.timestamp()
|
|
if tinfo.isfile():
|
|
# for any regular file normalize the permission
|
|
# leave unexpected extra-ordinary content untouched
|
|
tinfo.mode = int('0o100644', 8)
|
|
return tinfo
|
|
|
|
|
|
def make_dicom_archive(
|
|
src: Path,
|
|
dest: Path,
|
|
progress_handler,
|
|
):
|
|
if dest.exists():
|
|
# be safe
|
|
raise ValueError(
|
|
f'{dest} already exists, refusing to overwrite')
|
|
|
|
# scan input directory, get a mapping of path->timestamp
|
|
content = dict(scan_dir(src, progress_handler))
|
|
# for non-DICOM file the timestamp is `None`, determine "youngest"
|
|
# timestamp and use in such cases
|
|
default_timestamp = (
|
|
sorted(v for v in content.values() if v) or [default_date])[-1]
|
|
|
|
write_archive(
|
|
dest,
|
|
src,
|
|
content,
|
|
default_timestamp,
|
|
progress_handler,
|
|
)
|
|
# be nice (?) an give the generated archive the mtime of the DICOM set
|
|
os.utime(
|
|
dest,
|
|
times=(
|
|
# access time
|
|
datetime.now().timestamp(),
|
|
# modification time
|
|
default_timestamp.timestamp(),
|
|
),
|
|
)
|