Spaces:
Paused
Paused
| """Contract smoke test: convert the example to several formats, check the hub tools.""" | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import sys | |
| from core.io import APP_TMP_DIR | |
| from core.processing import catalog, convert, dicom_info, recommend | |
| def main() -> int: | |
| tif = APP_TMP_DIR / "example.tif" | |
| if not tif.exists(): | |
| import scripts.make_example as me | |
| me.main() | |
| src = str(tif) | |
| # convert RGB TIFF -> png, jpg, npy; grayscale + 16-bit variants | |
| for fmt in ("png", "jpg", "bmp", "webp", "npy", "tiff"): | |
| out, rep = convert(src, target_format=fmt) | |
| assert os.path.exists(out) and os.path.getsize(out) > 0, f"{fmt} produced nothing" | |
| assert rep["target_format"] in (fmt, "tif") or fmt == "tiff" | |
| g_out, g_rep = convert(src, target_format="png", to_gray=True, bit_depth="8") | |
| assert g_rep["to_gray"] and g_rep["output_dtype"] == "uint8" | |
| # DICOM: synthesize a CT series zip, convert -> TIFF stack, check (Z,H,W) + HU | |
| from core.dicom import write_synth_series_zip | |
| zip_path = str(APP_TMP_DIR / "ct_series.zip") | |
| write_synth_series_zip(zip_path, n=6, size=64) | |
| d_out, d_rep = convert(zip_path, target_format="tiff") | |
| assert d_rep["is_stack"] and d_rep["output_shape"][0] == 6, d_rep | |
| import tifffile as _tf | |
| hu = _tf.imread(d_out) | |
| assert hu.min() < -500 and hu.max() > 500, ("HU not recovered", float(hu.min()), float(hu.max())) | |
| drec = recommend(zip_path) | |
| assert "ct-baggage" in [a["app"] for a in drec["suggested_apps"]], drec | |
| # DICOM study (2 series) + DICOMDIR: largest series chosen; overview lists both | |
| study_zip = str(APP_TMP_DIR / "ct_study.zip") | |
| write_synth_series_zip(study_zip, n=6, size=48, two_series=True) | |
| s_out, s_rep = convert(study_zip, target_format="tiff") | |
| assert s_rep["output_shape"][0] == 6, ("largest series not chosen", s_rep) | |
| ov = dicom_info(study_zip) | |
| assert ov["n_series"] == 2, ov | |
| dd_zip = str(APP_TMP_DIR / "ct_dicomdir.zip") | |
| write_synth_series_zip(dd_zip, n=5, size=48, with_dicomdir=True) | |
| dd_out, dd_rep = convert(dd_zip, target_format="npy") | |
| assert dd_rep["is_stack"] and dd_rep["output_shape"][0] == 5, ("DICOMDIR read", dd_rep) | |
| # DICOM SEG: read as a labeled mask; overview flags the segmentation | |
| from core.dicom import write_synth_seg | |
| seg_path = str(APP_TMP_DIR / "seg.dcm") | |
| write_synth_seg(seg_path, n_slices=2, size=32) | |
| seg_out, seg_rep = convert(seg_path, target_format="npy") | |
| seg_arr = __import__("numpy").load(seg_out) | |
| assert seg_arr.max() >= 2, ("two segment labels expected", float(seg_arr.max())) | |
| sov = dicom_info(seg_path) | |
| assert sov["has_segmentation"] and "lesion" in sov["studies"][0]["series"][0]["segments"], sov | |
| # RTSTRUCT: rasterize contours over the referenced series -> labeled volume | |
| from core.dicom import write_synth_rtstruct_zip | |
| rt_zip = str(APP_TMP_DIR / "ct_rtstruct.zip") | |
| write_synth_rtstruct_zip(rt_zip, n=4, size=32) | |
| rt_out, rt_rep = convert(rt_zip, target_format="npy") | |
| rt_arr = __import__("numpy").load(rt_out) | |
| assert rt_arr.ndim == 3 and int(rt_arr.max()) >= 2, ("RTSTRUCT rasterize", rt_arr.shape) | |
| rov = dicom_info(rt_zip) | |
| rt_series = [se for st in rov["studies"] for se in st["series"] if se["modality"] == "RTSTRUCT"] | |
| assert rov["has_rtstruct"] and "lesion" in rt_series[0]["rois"], rov | |
| # NIfTI: synthesize a .nii.gz volume, convert -> npy, check (Z,H,W) + HU | |
| from core.nifti import write_synth_nifti | |
| nii_path = str(APP_TMP_DIR / "vol.nii.gz") | |
| write_synth_nifti(nii_path, n=6, size=64) | |
| n_out, n_rep = convert(nii_path, target_format="npy") | |
| assert n_rep["is_stack"] and n_rep["output_shape"][0] == 6, n_rep | |
| nvol = __import__("numpy").load(n_out) | |
| assert nvol.min() < -500 and nvol.max() > 500, ("NIfTI HU not recovered", float(nvol.min())) | |
| # hub surfaces | |
| cat = catalog() | |
| assert len(cat["apps"]) >= 14, "catalog missing apps" | |
| rec = recommend(src) | |
| assert rec["detected"] == "a single 2D image", rec | |
| assert any(a["app"] == "skimage-classic" for a in rec["suggested_apps"]) | |
| print("SMOKE OK:", json.dumps({"recommend": rec["detected"], | |
| "dicom_series": d_rep["output_shape"], | |
| "dicom_study_series": ov["n_series"], | |
| "dicomdir_volume": dd_rep["output_shape"], | |
| "seg_labels": int(seg_arr.max()), | |
| "rtstruct_labels": int(rt_arr.max()), | |
| "nifti_volume": n_rep["output_shape"], | |
| "n_apps": len(cat["apps"]), | |
| "gray_png_bytes": g_rep["output_bytes"]})) | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |