import tempfile, shutil
from fastcore.test import test_eq, test_fail
_tmp = Path(tempfile.mkdtemp())
# Create synthetic NIfTI files
for i in range(3):
tio.ScalarImage(tensor=torch.randn(1, 10, 10, 10)).save(_tmp / f'img_{i}.nii.gz')
tio.LabelMap(tensor=torch.randint(0, 2, (1, 10, 10, 10))).save(_tmp / f'mask_{i}.nii.gz')
# Image-only cache: originals are preserved and outputs live below a version directory.
_df1 = pd.DataFrame({'img': [str(_tmp / f'img_{i}.nii.gz') for i in range(3)]})
_orig_paths1 = _df1['img'].tolist()
_cache_root = _tmp / 'cache'
_result1 = preprocess_dataset(
_df1, img_col='img', output_dir=_cache_root, apply_reorder=False
)
test_eq(_df1['img'].tolist(), _orig_paths1)
test_eq(_result1.processed, 3)
test_eq(_result1.reused, False)
test_eq(_result1.output_dir.parent, _cache_root)
test_eq(all(Path(p).parent == _result1.output_dir / 'images' for p in _df1.img_preprocessed), True)
test_eq(all(Path(p).is_file() for p in _df1.img_preprocessed), True)
_manifest1 = json.loads(_result1.manifest_path.read_text())
test_eq(_manifest1['manifest_schema'], 1)
test_eq(_manifest1['preprocessing_cache_version'], _result1.cache_version)
test_eq(_manifest1['dataset_version'], None)
test_eq(_manifest1['contract']['apply_reorder'], False)
# Identical source signatures and contract reuse the published cache.
_df2 = pd.DataFrame({'img': _orig_paths1})
_result2 = preprocess_dataset(
_df2, img_col='img', output_dir=_cache_root, apply_reorder=False
)
test_eq(_result2.cache_version, _result1.cache_version)
test_eq(_result2.processed, 0)
test_eq(_result2.reused, True)
# A preprocessing-contract change gets a different cache directory.
_df_spacing = pd.DataFrame({'img': _orig_paths1})
_result_spacing = preprocess_dataset(
_df_spacing, img_col='img', output_dir=_cache_root,
target_spacing=[2, 1, 1], apply_reorder=False,
)
assert _result_spacing.cache_version != _result1.cache_version
# A cheap source signature change also invalidates reuse without reading raw contents.
_source0 = Path(_orig_paths1[0])
_stat0 = _source0.stat()
os.utime(_source0, ns=(_stat0.st_atime_ns, _stat0.st_mtime_ns + 1_000_000))
_df_changed = pd.DataFrame({'img': _orig_paths1})
_result_changed = preprocess_dataset(
_df_changed, img_col='img', output_dir=_cache_root, apply_reorder=False
)
assert _result_changed.cache_version not in {
_result1.cache_version, _result_spacing.cache_version
}
# A damaged cache fails closed; an explicit rebuild republishes it.
Path(_df_changed.img_preprocessed.iloc[0]).unlink()
_df_damaged = pd.DataFrame({'img': _orig_paths1})
test_fail(
lambda: preprocess_dataset(
_df_damaged, img_col='img', output_dir=_cache_root, apply_reorder=False
),
contains='no valid matching manifest',
)
_result_rebuilt = preprocess_dataset(
_df_damaged, img_col='img', output_dir=_cache_root,
apply_reorder=False, skip_existing=False,
)
test_eq(_result_rebuilt.cache_version, _result_changed.cache_version)
test_eq(_result_rebuilt.processed, 3)
test_eq(all(Path(p).is_file() for p in _df_damaged.img_preprocessed), True)
# Masks and the existing annotation fingerprint are bound into the manifest.
_df3 = pd.DataFrame({
'img': _orig_paths1,
'mask': [str(_tmp / f'mask_{i}.nii.gz') for i in range(3)],
})
_orig_mask3 = _df3['mask'].tolist()
_result3 = preprocess_dataset(
_df3, img_col='img', mask_col='mask', output_dir=_tmp / 'mask_cache',
apply_reorder=False, dataset_version='annotation-v1',
)
test_eq(_df3['img'].tolist(), _orig_paths1)
test_eq(_df3['mask'].tolist(), _orig_mask3)
test_eq(all(Path(p).is_file() for p in _df3.img_preprocessed), True)
test_eq(all(Path(p).is_file() for p in _df3.mask_preprocessed), True)
test_eq(json.loads(_result3.manifest_path.read_text())['dataset_version'], 'annotation-v1')
# A worker failure removes staging and publishes no cache directory.
_bad_source = _tmp / 'not_a_nifti.nii.gz'
_bad_source.write_text('not a medical image')
_failure_root = _tmp / 'failed_cache'
test_fail(
lambda: preprocess_dataset(
pd.DataFrame({'img': [str(_bad_source)]}),
img_col='img', output_dir=_failure_root, apply_reorder=False,
),
contains='no cache was published',
)
test_eq(list(_failure_root.iterdir()), [])
# Input validation
test_fail(lambda: preprocess_dataset(pd.DataFrame(), img_col='img'), contains='empty')
test_fail(lambda: preprocess_dataset(pd.DataFrame({'x': [1]}), img_col='img'), contains='not found')
_df_dup = pd.DataFrame({'img': [str(_tmp / 'img_0.nii.gz')] * 2})
test_fail(lambda: preprocess_dataset(_df_dup, img_col='img'), contains='Duplicate')
# Failed metadata files are surfaced (not silently dropped from statistics).
_files = [str(_tmp / 'img_0.nii.gz'), str(_tmp / 'img_1.nii.gz'), str(_tmp / 'does_not_exist.nii.gz')]
with warnings.catch_warnings(record=True) as _w:
warnings.simplefilter('always')
_ds = MedDataset(img_list=_files, apply_reorder=False, use_cache=False)
test_eq(_ds.failed_files, [str(_tmp / 'does_not_exist.nii.gz')])
test_eq(len(_ds.df), 2)
test_eq('error' in _ds.df.columns, False)
test_eq(any('failed to load' in str(_x.message) for _x in _w), True)
test_eq(bool(np.isnan(_ds.get_size_statistics()['median']).any()), False)
# All-files-failed metadata analysis constructs and statistics guard cleanly.
_ds_all = MedDataset(
img_list=[str(_tmp / 'no1.nii.gz'), str(_tmp / 'no2.nii.gz')],
apply_reorder=False, use_cache=False,
)
test_eq(len(_ds_all.df), 0)
test_eq(len(_ds_all.failed_files), 2)
test_fail(lambda: _ds_all.summary(), contains='empty')
test_fail(lambda: _ds_all.get_suggestion(), contains='empty')
shutil.rmtree(_tmp)