"""
Dataset download and verification helpers.
Slicer-Heart-CT, KCL-Heart-Model, CHOP-Valve4D, Chest-CT, TCIA-4DLung, and
Duke-Heart-4DLabelmaps are downloaded automatically. Other datasets require manual download, and
the verification helpers check the file layouts used by the repository
tutorials, experiments, and tests.
"""
from __future__ import annotations
import logging
import shutil
import tarfile
import tempfile
import urllib.request
import zipfile
from pathlib import Path
from typing import ClassVar, Optional, Union
import itk
import numpy as np
from .convert_image_4d_to_3d import ConvertImage4DTo3D
_DOWNLOAD_TIMEOUT_SECONDS = 60.0
_logger = logging.getLogger(__name__)
[docs]
class DownloadData:
"""Download and verify optional MONAI Physio example datasets."""
SLICER_HEART_CT_URL = (
"https://github.com/Project-MONAI/monai-physio/releases/download/"
"2026.07.1/TruncalValve_4DCT.seq.nrrd"
)
SLICER_HEART_CT_FILENAME = "TruncalValve_4DCT.seq.nrrd"
SLICER_HEART_CT_SLICE_BASENAME = "slice"
[docs]
@staticmethod
def DownloadSlicerHeartCTData(dirname: Union[str, Path]) -> Path:
"""Download the Slicer-Heart-CT 4-D CT sample into ``dirname``.
Also splits the downloaded 4-D sequence into per-frame
``slice_???.mha`` volumes in the same directory (via
``ConvertImage4DTo3D``), skipping the split if those files already
exist.
Args:
dirname: Directory where ``TruncalValve_4DCT.seq.nrrd`` and the
converted ``slice_???.mha`` files should live.
Returns:
Path to the downloaded or already-cached ``.seq.nrrd`` file.
"""
data_dir = Path(dirname)
data_dir.mkdir(parents=True, exist_ok=True)
data_file = data_dir / DownloadData.SLICER_HEART_CT_FILENAME
if not (data_file.exists() and data_file.stat().st_size > 0):
DownloadData._DownloadFile(DownloadData.SLICER_HEART_CT_URL, data_file)
slice_basename = DownloadData.SLICER_HEART_CT_SLICE_BASENAME
if not any(data_dir.glob(f"{slice_basename}_???.mha")):
# Convert into a temp directory first, then atomically move each
# finished frame into data_dir, so a failed or interrupted
# conversion leaves no partial slice_???.mha files behind and
# concurrent callers never observe a half-written frame.
with tempfile.TemporaryDirectory(dir=str(data_dir)) as tmp_dir_name:
tmp_dir = Path(tmp_dir_name)
conv = ConvertImage4DTo3D()
conv.load_image_4d(str(data_file))
conv.save_3d_images(tmp_dir, slice_basename)
for tmp_slice_file in sorted(tmp_dir.glob(f"{slice_basename}_???.mha")):
tmp_slice_file.replace(data_dir / tmp_slice_file.name)
_logger.info(
"Converted %s to %s_???.mha frames",
DownloadData.SLICER_HEART_CT_FILENAME,
slice_basename,
)
return data_file
@staticmethod
def _DownloadFile(url: str, target_file: Path) -> None:
"""Stream-download ``url`` and atomically replace ``target_file``.
Streams to a unique temp file in the target's directory with an
explicit timeout, then atomically replaces the target on success. The
temp name is unique so concurrent callers do not clobber each other.
Avoids partial files on interrupt and the indefinite hang that
urlretrieve has without a timeout.
"""
tmp_handle = tempfile.NamedTemporaryFile( # noqa: SIM115 - closed by the with below
dir=str(target_file.parent),
prefix=f".{target_file.name}.",
suffix=".tmp",
delete=False,
)
tmp_file = Path(tmp_handle.name)
try:
with (
urllib.request.urlopen(
url, timeout=_DOWNLOAD_TIMEOUT_SECONDS
) as response,
tmp_handle as out,
):
shutil.copyfileobj(response, out)
if tmp_file.stat().st_size == 0:
raise RuntimeError(f"Downloaded file is empty: {url}")
tmp_file.replace(target_file)
_logger.info("Downloaded %s", target_file.name)
except BaseException:
tmp_handle.close()
if tmp_file.exists():
tmp_file.unlink()
raise
[docs]
@staticmethod
def VerifySlicerHeartCTData(dirname: Union[str, Path]) -> bool:
"""Return True when Slicer-Heart-CT has the expected 4-D CT file."""
return (Path(dirname) / DownloadData.SLICER_HEART_CT_FILENAME).is_file()
KCL_HEART_MODEL_MESH_COUNT = 20
KCL_HEART_MODEL_INDIVIDUAL_URL_TEMPLATE = (
"https://zenodo.org/records/4590294/files/{index:02d}.tar.gz?download=1"
)
KCL_HEART_MODEL_AVERAGE_URL = (
"https://zenodo.org/records/4593739/files/average.tar.gz?download=1"
)
[docs]
@staticmethod
def DownloadKCLHeartModelData(dirname: Union[str, Path]) -> Path:
"""Download the KCL-Heart-Model dataset into ``dirname``.
Downloads and extracts the 20 individual four-chamber heart meshes
from https://zenodo.org/records/4590294 into
``dirname/input_meshes/01.vtk`` through ``20.vtk``, and the average
heart mesh from https://zenodo.org/records/4593739 into
``dirname/average_mesh.vtk``. Already-extracted files are reused.
Args:
dirname: Directory where the KCL-Heart-Model dataset should live.
Returns:
Path to ``dirname``.
"""
data_dir = Path(dirname)
input_meshes_dir = data_dir / "input_meshes"
input_meshes_dir.mkdir(parents=True, exist_ok=True)
for index in range(1, DownloadData.KCL_HEART_MODEL_MESH_COUNT + 1):
target_file = input_meshes_dir / f"{index:02d}.vtk"
if target_file.exists() and target_file.stat().st_size > 0:
continue
url = DownloadData.KCL_HEART_MODEL_INDIVIDUAL_URL_TEMPLATE.format(
index=index
)
DownloadData._DownloadAndExtractTarMember(
url, member_name=f"{index:02d}.vtk", target_file=target_file
)
_logger.info(
"Downloaded %02d.vtk (%d/%d)",
index,
index,
DownloadData.KCL_HEART_MODEL_MESH_COUNT,
)
average_file = data_dir / "average_mesh.vtk"
if not (average_file.exists() and average_file.stat().st_size > 0):
DownloadData._DownloadAndExtractTarMember(
DownloadData.KCL_HEART_MODEL_AVERAGE_URL,
member_name="average.vtk",
target_file=average_file,
)
_logger.info("Downloaded average_mesh.vtk")
return data_dir
@staticmethod
def _DownloadAndExtractTarMember(
url: str, member_name: str, target_file: Path
) -> None:
"""Download a ``.tar.gz`` archive and extract one member to ``target_file``."""
target_file.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(dir=str(target_file.parent)) as tmp_dir_name:
tmp_dir = Path(tmp_dir_name)
archive_file = tmp_dir / "archive.tar.gz"
with (
urllib.request.urlopen(
url, timeout=_DOWNLOAD_TIMEOUT_SECONDS
) as response,
open(archive_file, "wb") as out,
):
shutil.copyfileobj(response, out)
if archive_file.stat().st_size == 0:
raise RuntimeError(f"Downloaded archive is empty: {url}")
with tarfile.open(archive_file) as tar:
try:
tar.extract(member_name, path=tmp_dir, filter="data")
except TypeError:
# Python < 3.12 (without the PEP 706 backport) does not
# accept the ``filter`` keyword argument.
tar.extract(member_name, path=tmp_dir)
extracted_file = tmp_dir / member_name
if not extracted_file.is_file():
raise RuntimeError(
f"Expected member {member_name!r} not found in archive: {url}"
)
extracted_file.replace(target_file)
CHOP_VALVE4D_RELEASE_URL = (
"https://github.com/Project-MONAI/monai-physio/releases/download/2026.07.1/"
)
# subdirectory name -> release asset filename. Alterra and TPV25 are
# each >1 GB (per-frame valve mesh time series); CT is the smaller
# image/segmentation bundle. See data/CHOP-Valve4D/README.md.
CHOP_VALVE4D_ASSETS: ClassVar[dict[str, str]] = {
"Alterra": "CHOP-Valve4D-Alterra.zip",
"CT": "CHOP-Valve4D-CT.zip",
"TPV25": "CHOP-Valve4D-TPV25.zip",
}
[docs]
@staticmethod
def DownloadCHOPValve4DData(dirname: Union[str, Path]) -> Path:
"""Download the CHOP-Valve4D convenience release into ``dirname``.
Downloads the three zip archives attached to the MONAI Physio
2026.07.1 GitHub release
(https://github.com/Project-MONAI/monai-physio/releases/tag/2026.07.1)
and extracts each into its matching subdirectory: ``Alterra/`` and
``TPV25/`` (valve mesh time series, >1 GB each) and ``CT/`` (source
CT volume and Simpleware segmentation). A subdirectory that already
has its expected files (see ``_CHOPValve4DSubdirIsPopulated``) is
left alone, so re-running resumes an interrupted download without
re-fetching a subdirectory a partial extraction left behind. See
``data/CHOP-Valve4D/README.md`` for what this converted data
contains and how it relates to the original FEBio source model.
Args:
dirname: Directory where the CHOP-Valve4D dataset should live.
Returns:
Path to ``dirname``.
"""
data_dir = Path(dirname)
for (
subdir_name,
asset_name,
) in DownloadData.CHOP_VALVE4D_ASSETS.items():
target_dir = data_dir / subdir_name
if DownloadData._CHOPValve4DSubdirIsPopulated(target_dir):
continue
url = DownloadData.CHOP_VALVE4D_RELEASE_URL + asset_name
DownloadData._DownloadAndExtractZip(url, target_dir)
_logger.info("Downloaded %s (%s)", subdir_name, asset_name)
return data_dir
@staticmethod
def _CHOPValve4DSubdirIsPopulated(
target_dir: Path,
) -> bool:
"""Return True when ``target_dir`` already has subdir_name's expected files.
Used to decide whether a subdirectory can be skipped on a re-run.
Checking for *any* file is not enough - an interrupted extraction
can leave a partially-populated directory that would then be
skipped forever - so this checks for the specific files each
subdirectory is expected to contain once fully extracted.
"""
if not target_dir.is_dir():
return False
subdir_name = target_dir.name
if subdir_name == "CT":
has_ct_volume = any(
(target_dir / filename).is_file()
for filename in ("RVOT28-Dias.nii.gz", "RVOT28-Dias.mha")
)
has_simpleware_parts = (target_dir / "Simpleware" / "parts").is_dir()
return has_ct_volume or has_simpleware_parts
# Alterra and TPV25 are valve mesh time series.
return any(target_dir.glob("*.vtk"))
@staticmethod
def _DownloadAndExtractZip(
url: str, target_dir: Path, extract_dir: Optional[Path] = None
) -> None:
"""Stream-download a ``.zip`` archive and extract it into ``extract_dir``.
``extract_dir`` defaults to ``target_dir.parent``, which is correct
when the archive's top-level entry is ``target_dir.name`` itself; pass
``extract_dir=target_dir`` when the archive's entries already sit
directly inside ``target_dir``.
"""
target_dir.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(dir=str(target_dir.parent)) as tmp_dir_name:
archive_file = Path(tmp_dir_name) / "archive.zip"
with (
urllib.request.urlopen(
url, timeout=_DOWNLOAD_TIMEOUT_SECONDS
) as response,
open(archive_file, "wb") as out,
):
shutil.copyfileobj(response, out, length=1024 * 1024)
if archive_file.stat().st_size == 0:
raise RuntimeError(f"Downloaded archive is empty: {url}")
with zipfile.ZipFile(archive_file) as archive:
archive.extractall(extract_dir or target_dir.parent)
[docs]
@staticmethod
def VerifyCHOPValve4DData(dirname: Union[str, Path]) -> bool:
"""Return True when CHOP-Valve4D files referenced by the repo exist.
Accepted layouts are the CT volume used by Simpleware/model-to-patient
experiments and the valve time-series folders used by VTK-to-USD
experiments.
"""
data_dir = Path(dirname)
has_ct = DownloadData._CHOPValve4DSubdirIsPopulated(data_dir / "CT")
has_alterra = DownloadData._CHOPValve4DSubdirIsPopulated(data_dir / "Alterra")
has_tpv25 = DownloadData._CHOPValve4DSubdirIsPopulated(data_dir / "TPV25")
return has_ct or (has_alterra and has_tpv25)
CHEST_CT_URL = (
"https://github.com/Project-MONAI/monai-physio/releases/download/2026.07.1/"
"Chest-CT.mha"
)
CHEST_CT_FILENAME = "Chest-CT.mha"
[docs]
@staticmethod
def DownloadChestCTData(dirname: Union[str, Path]) -> Path:
"""Download the Chest-CT sample volume into ``dirname``.
Fetches ``Chest-CT.mha`` - an ungated 3-D chest CT - from the
MONAI Physio 2026.07.1 GitHub release. An existing non-empty file is
reused, so re-running resumes an interrupted download. See
``data/Chest-CT/README.md`` for the data source and required citation.
Args:
dirname: Directory where ``Chest-CT.mha`` should live.
Returns:
Path to the downloaded or already-cached ``Chest-CT.mha`` file.
"""
data_dir = Path(dirname)
data_dir.mkdir(parents=True, exist_ok=True)
data_file = data_dir / DownloadData.CHEST_CT_FILENAME
if not (data_file.exists() and data_file.stat().st_size > 0):
DownloadData._DownloadFile(DownloadData.CHEST_CT_URL, data_file)
return data_file
[docs]
@staticmethod
def VerifyChestCTData(dirname: Union[str, Path]) -> bool:
"""Return True when Chest-CT has its expected CT volume."""
return (Path(dirname) / DownloadData.CHEST_CT_FILENAME).is_file()
TCIA_4D_LUNG_RELEASE_URL = (
"https://github.com/Project-MONAI/monai-physio/releases/download/2026.07.1/"
)
TCIA_4D_LUNG_ASSETS: ClassVar[list[str]] = [
"TCIA-4DLung-Part1.zip",
"TCIA-4DLung-Part2.zip",
]
[docs]
@staticmethod
def DownloadTCIA4DLungData(dirname: Union[str, Path]) -> Path:
"""Download the TCIA-4DLung converted-subset release into ``dirname``.
Fetches the two zip archives attached to the MONAI Physio 2026.07.1
GitHub release and extracts them directly into ``dirname``, giving
the per-case ``<case>/<case>_g0??.nii.gz`` phase volumes the lung
tutorials read. This is a tutorial subset of the full TCIA 4D-Lung
collection; see ``data/TCIA-4DLung/README.md`` for how to obtain the
full collection manually. Already-populated data is left alone, so
re-running is a no-op once the subset is present.
Args:
dirname: Directory where the TCIA-4DLung dataset should live.
Returns:
Path to ``dirname``.
"""
data_dir = Path(dirname)
if DownloadData.VerifyTCIA4DLungData(data_dir):
return data_dir
for asset_name in DownloadData.TCIA_4D_LUNG_ASSETS:
url = DownloadData.TCIA_4D_LUNG_RELEASE_URL + asset_name
DownloadData._DownloadAndExtractZip(url, data_dir, extract_dir=data_dir)
_logger.info("Downloaded %s", asset_name)
return data_dir
[docs]
@staticmethod
def VerifyTCIA4DLungData(dirname: Union[str, Path]) -> bool:
"""Return True when TCIA-4DLung has at least one case's phase volumes."""
return any(Path(dirname).glob("*_HM10395/*_g0??.nii.gz"))
DUKE_HEART_4D_LABELMAPS_URL = (
"https://github.com/Project-MONAI/monai-physio/releases/download/2026.07.1/"
"Duke-Heart-4DLabelmaps.zip"
)
[docs]
@staticmethod
def DownloadDukeHeart4DLabelmapsData(dirname: Union[str, Path]) -> Path:
"""Download the Duke-Heart-4DLabelmaps release into ``dirname``.
Fetches ``Duke-Heart-4DLabelmaps.zip`` from the MONAI Physio 2026.07.1
GitHub release and extracts it directly into ``dirname``, giving the
per-case ``pm00*/`` directories of ``*_labelmap.nii.gz`` and
``*_landmark.mrk.json`` files. The labelmaps were generated with the
ASCardio module of Simpleware (``SegmentHeartSimplewareTrimmedBranches``);
models trained on them should only be applied to labelmaps generated
the same way. See ``data/Duke-Heart-4DLabelmaps/README.md``.
Already-populated data is left alone, so re-running is a no-op once
the data is present.
Args:
dirname: Directory where the Duke-Heart-4DLabelmaps dataset should
live.
Returns:
Path to ``dirname``.
"""
data_dir = Path(dirname)
if DownloadData.VerifyDukeHeart4DLabelmapsData(data_dir):
return data_dir
DownloadData._DownloadAndExtractZip(
DownloadData.DUKE_HEART_4D_LABELMAPS_URL, data_dir, extract_dir=data_dir
)
_logger.info("Downloaded Duke-Heart-4DLabelmaps.zip")
return data_dir
[docs]
@staticmethod
def VerifyDukeHeart4DLabelmapsData(dirname: Union[str, Path]) -> bool:
"""Return True when Duke-Heart-4DLabelmaps has at least one labelmap."""
return any(Path(dirname).glob("pm*/*_labelmap.nii.gz"))
@staticmethod
def _MetaImageHeaderHasBackingData(mhd_file: Path) -> bool:
"""Return True when a MetaImage ``.mhd`` header's pixel data exists.
Committed ``.mhd`` headers are tiny text files (a few hundred
bytes) that ship with the repository as documentation of the
expected DirLab-4DCT layout; their presence alone does not mean
the (large, gitignored) raw pixel data has been downloaded. This
checks that the file the header's ``ElementDataFile`` line points
to actually exists, resolved relative to the header's directory.
"""
try:
lines = mhd_file.read_text(errors="ignore").splitlines()
except OSError:
return False
for line in lines:
key, sep, value = line.partition("=")
if sep and key.strip() == "ElementDataFile":
data_file = value.strip()
if not data_file or data_file.upper() == "LOCAL":
# Pixel data is embedded in this same file.
return True
return (mhd_file.parent / data_file).is_file()
return False
[docs]
@staticmethod
def VerifyDirLab4DCTData(dirname: Union[str, Path]) -> bool:
"""Return True when a supported DirLab-4DCT case layout exists."""
data_dir = Path(dirname)
case1_dir = data_dir / "Case1"
has_case_dir_layout = case1_dir.is_dir() and any(case1_dir.glob("*.mha"))
has_case_dir_layout = has_case_dir_layout or (
case1_dir.is_dir()
and any(
DownloadData._MetaImageHeaderHasBackingData(mhd_file)
for mhd_file in case1_dir.glob("*.mhd")
)
)
has_pack_layout = any(data_dir.glob("Case1Pack_T*.mha")) or any(
DownloadData._MetaImageHeaderHasBackingData(mhd_file)
for mhd_file in data_dir.glob("Case1Pack_T*.mhd")
)
return has_case_dir_layout or has_pack_layout
DIRLAB_4DCT_HU_OFFSET = 1024
DIRLAB_4DCT_HU_CLIP_RANGE = (-1024, 1024)
[docs]
@staticmethod
def FixDirLab4DCTData(
dirname: Union[str, Path],
output_dirname: Optional[Union[str, Path]] = None,
) -> list[Path]:
"""Convert DirLab-4DCT raw intensities to clipped Hounsfield units.
DirLab-4DCT's raw ``.img``/``.mhd`` volumes store intensities offset
by +1024 from Hounsfield units. This finds every ``.mhd`` header
under ``dirname`` (recursively) whose raw data has been downloaded,
subtracts 1024, clips the result to ``DIRLAB_4DCT_HU_CLIP_RANGE``,
and writes the corrected volume with the same base filename but a
``.mha`` extension, compressed.
Args:
dirname: Directory to search recursively for ``.mhd`` headers.
output_dirname: Directory to write the ``.mha`` files into.
Defaults to writing each result next to its source ``.mhd``
header.
Returns:
Paths to the written ``.mha`` files.
"""
output_dir = Path(output_dirname) if output_dirname is not None else None
if output_dir is not None:
output_dir.mkdir(parents=True, exist_ok=True)
output_files = []
for mhd_file in sorted(Path(dirname).rglob("*.mhd")):
if not DownloadData._MetaImageHeaderHasBackingData(mhd_file):
continue
image = itk.imread(str(mhd_file))
image_arr = itk.array_from_image(image) - DownloadData.DIRLAB_4DCT_HU_OFFSET
image_arr = np.clip(image_arr, *DownloadData.DIRLAB_4DCT_HU_CLIP_RANGE)
fixed_image = itk.image_from_array(image_arr)
fixed_image.CopyInformation(image)
target_dir = output_dir if output_dir is not None else mhd_file.parent
output_file = target_dir / mhd_file.with_suffix(".mha").name
itk.imwrite(fixed_image, str(output_file), compression=True)
output_files.append(output_file)
_logger.info("Fixed %s -> %s", mhd_file.name, output_file.name)
return output_files
PHYSICSNEMO_MGN_LUNG_MOTION_URL = (
"https://github.com/Project-MONAI/monai-physio/releases/download/2026.07.1/"
"physicsnemo_mgn_lung_motion.zip"
)
PHYSICSNEMO_MGN_LUNG_MOTION_DIRNAME = "physicsnemo_mgn_lung_motion"
[docs]
@staticmethod
def DownloadPhysicsNeMoMGNLungMotionData(dirname: Union[str, Path]) -> Path:
"""Download the pretrained lung-motion PhysicsNeMo MGN checkpoint.
Fetches ``physicsnemo_mgn_lung_motion.zip`` from the MONAI Physio
2026.07.1 GitHub release and extracts it into ``dirname``, giving
the ``physicsnemo_mgn_lung_motion/mgn_stage_model.pt`` checkpoint
Tutorial 10 and later lung tutorials read. Already-extracted data is
left alone, so re-running is a no-op once it is present.
Args:
dirname: Directory where ``physicsnemo_mgn_lung_motion/`` should
live, typically ``tutorials/network_weights``.
Returns:
Path to ``dirname/physicsnemo_mgn_lung_motion``.
"""
data_dir = Path(dirname)
target_dir = data_dir / DownloadData.PHYSICSNEMO_MGN_LUNG_MOTION_DIRNAME
if not DownloadData.VerifyPhysicsNeMoMGNLungMotionData(data_dir):
DownloadData._DownloadAndExtractZip(
DownloadData.PHYSICSNEMO_MGN_LUNG_MOTION_URL, target_dir
)
_logger.info(
"Downloaded %s", DownloadData.PHYSICSNEMO_MGN_LUNG_MOTION_DIRNAME
)
return target_dir
[docs]
@staticmethod
def VerifyPhysicsNeMoMGNLungMotionData(dirname: Union[str, Path]) -> bool:
"""Return True when the lung-motion PhysicsNeMo MGN checkpoint exists."""
return (
Path(dirname)
/ DownloadData.PHYSICSNEMO_MGN_LUNG_MOTION_DIRNAME
/ "mgn_stage_model.pt"
).is_file()
[docs]
@staticmethod
def VerifyKCLHeartModelData(dirname: Union[str, Path]) -> bool:
"""Return True when KCL-Heart-Model has its expected mesh inputs."""
data_dir = Path(dirname)
input_meshes_dir = data_dir / "input_meshes"
return (data_dir / "average_mesh.vtk").is_file() and any(
input_meshes_dir.glob("*.vtk")
)