diff --git a/.gitignore b/.gitignore index 30b51f1..960f98e 100644 --- a/.gitignore +++ b/.gitignore @@ -135,4 +135,6 @@ pythonlib* # Blender *.blend1 blendcache_* -*.cats.txt~ \ No newline at end of file +*.cats.txt~ + +tests/data \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md index d409372..661944b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,6 +24,11 @@ uv sync uv run build.py # platforms: windows-x64, linux-x64, macos-arm64, macos-x64 ``` +### Unit Test +```bash +uv run pytest tests/ +``` + ### Code Quality ```bash # Format code (autopep8) diff --git a/src/bioxelnodes/bioxel/__init__.py b/src/bioxelnodes/bioxel/__init__.py index e69de29..039ca35 100644 --- a/src/bioxelnodes/bioxel/__init__.py +++ b/src/bioxelnodes/bioxel/__init__.py @@ -0,0 +1,34 @@ +from .layer import Layer +from .container import Container +from .io import load_container, save_container +from .data import ( + Data, + read, + read_meta, + calc_layer_shape, + calc_layer_size, + get_ext, + SUPPORT_EXTS, + DICOM_EXTS, + OME_EXTS, + MRC_EXTS, + SEQUENCE_EXTS +) + +__all__ = [ + 'Layer', + 'Container', + 'load_container', + 'save_container', + 'Data', + 'read', + 'read_meta', + 'calc_layer_shape', + 'calc_layer_size', + 'get_ext', + 'SUPPORT_EXTS', + 'DICOM_EXTS', + 'OME_EXTS', + 'MRC_EXTS', + 'SEQUENCE_EXTS' +] diff --git a/src/bioxelnodes/bioxel/data.py b/src/bioxelnodes/bioxel/data.py new file mode 100644 index 0000000..a9c199d --- /dev/null +++ b/src/bioxelnodes/bioxel/data.py @@ -0,0 +1,831 @@ +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional, Callable, List +import numpy as np +import SimpleITK as sitk +from pyometiff import OMETIFFReader +import mrcfile +import transforms3d + +from .layer import Layer + +# 类型定义 +ProgressCallback = Optional[Callable[[float, str], None]] + + +SUPPORT_EXTS = [ + "", + ".dcm", + ".DCM", + ".DICOM", + ".ima", + ".IMA", + ".ome.tiff", + ".ome.tif", + ".tif", + ".TIF", + ".tiff", + ".TIFF", + ".mrc", + ".mrc.gz", + ".map", + ".map.gz", + ".bmp", + ".BMP", + ".png", + ".PNG", + ".jpg", + ".JPG", + ".jpeg", + ".JPEG", + ".PIC", + ".pic", + ".gipl", + ".gipl.gz", + ".lsm", + ".LSM", + ".mnc", + ".MNC", + ".mrc", + ".rec", + ".mha", + ".mhd", + ".hdf", + ".h4", + ".hdf4", + ".he2", + ".h5", + ".hdf5", + ".he5", + ".nia", + ".nii", + ".nii.gz", + ".hdr", + ".img", + ".img.gz", + ".nrrd", + ".nhdr", + ".vtk", + ".gz", +] + +OME_EXTS = [".ome.tiff", ".ome.tif", ".tif", ".TIF", ".tiff", ".TIFF"] + +MRC_EXTS = [".mrc", ".mrc.gz", ".map", ".map.gz"] + +DICOM_EXTS = ["", ".dcm", ".DCM", ".DICOM", ".ima", ".IMA"] + +SEQUENCE_EXTS = [ + ".bmp", + ".BMP", + ".jpg", + ".JPG", + ".jpeg", + ".JPEG", + ".tif", + ".TIF", + ".tiff", + ".TIFF", + ".png", + ".PNG", + ".mrc", +] + + +@dataclass +class Data: + filepath: str + series_id: str + _data: Optional[np.ndarray] = field(default=None, repr=False) + _meta: Optional[dict] = field(default=None, repr=False) + + @property + def data(self) -> np.ndarray: + if self._data is None: + self.load_data() + return self._data + + @property + def meta(self) -> dict: + if self._meta is None: + self.load_meta() + return self._meta + + @property + def name(self) -> str: + return self.meta["name"] + + @property + def description(self) -> str: + return self.meta["description"] + + @property + def shape(self) -> tuple: + return ( + self.meta["frame_count"], + *self.meta["xyz_shape"], + self.meta["channel_count"], + ) + + @property + def xyz_shape(self) -> tuple: + return self.meta["xyz_shape"] + + @property + def spacing(self) -> tuple: + return self.meta["spacing"] + + @property + def affine(self) -> np.ndarray: + return self.meta["affine"] + + @property + def frame_count(self) -> int: + return self.meta["frame_count"] + + @property + def channel_count(self) -> int: + return self.meta["channel_count"] + + @property + def dtype(self) -> np.dtype: + return self.meta.get("dtype", np.float32) + + def load_meta(self, progress_callback: ProgressCallback = None): + data, meta = self._parse_file(progress_callback) + self._meta = meta + del data + + def load_data(self, progress_callback: ProgressCallback = None): + if not self.is_loaded(): + data, meta = self._parse_file(progress_callback) + self._data = data + self._meta = meta + + def is_loaded(self) -> bool: + return self._data is not None + + def to_layers( + self, + kind: str = "scalar", + layer_name: str = "", + bioxel_size: float = 1.0, + smooth: int = 0, + remap: bool = False, + split_channel: bool = False, + frame_source: str = "-1", + progress_callback: ProgressCallback = None, + ) -> List[Layer]: + from .layer import Layer + + data = self.data.copy() + + data, layer_shape = self._transform_shape(data, frame_source) + + data = self._preprocess(data, kind, remap) + + mat_scale = transforms3d.zooms.zfdir2aff(bioxel_size) + affine = np.dot(self.meta["affine"], mat_scale) + + base_name = layer_name or kind.capitalize() + + if kind == "label": + return self._create_label_layers( + data, base_name, layer_shape, affine, smooth, progress_callback + ) + elif kind == "color": + return self._create_color_layers( + data, base_name, layer_shape, affine, progress_callback + ) + elif kind == "scalar": + return self._create_scalar_layers( + data, + base_name, + layer_shape, + affine, + smooth, + split_channel, + progress_callback, + ) + + return [] + + def _parse_file(self, progress_callback: ProgressCallback = None): + data_path = Path(self.filepath).resolve() + ext = self._get_ext(data_path) + + if progress_callback: + progress_callback(0.0, "Reading Data...") + + is_sequence = False + sequence = None + if ext in SEQUENCE_EXTS: + sequence = self._collect_sequence(data_path) + if sequence and len(sequence) > 1: + is_sequence = True + + data = None + name = "" + description = "" + affine = np.identity(4) + spacing = (1, 1, 1) + origin = (0, 0, 0) + direction = (1, 0, 0, 0, 1, 0, 0, 0, 1) + + if data is None and ext in MRC_EXTS and not is_sequence: + data, name, spacing = self._parse_mrc(data_path) + + if data is None and ext in OME_EXTS and not is_sequence: + data, name, spacing = self._parse_ome_tiff(data_path, progress_callback) + + if data is None: + data, name, description, spacing, origin, direction = self._parse_sitk( + data_path, ext, is_sequence, sequence, progress_callback + ) + + t = origin + r = np.array(direction).reshape((3, 3)) + affine = np.dot(affine, transforms3d.affines.compose(t, r, [1, 1, 1])) + + meta = { + "name": name, + "description": description, + "spacing": spacing, + "affine": affine, + "xyz_shape": data.shape[1:4], + "frame_count": data.shape[0], + "channel_count": data.shape[-1], + "dtype": data.dtype, + } + + return data, meta + + def _parse_mrc(self, data_path: Path): + print("Parsing with mrcfile...") + with mrcfile.open(data_path, "r") as mrc: + mrc_data = mrc.data + + if mrc.is_single_image(): + data = np.expand_dims(np.asarray(mrc_data), axis=0) + data = np.expand_dims(data, axis=-1) + data = np.expand_dims(data, axis=-1) + elif mrc.is_image_stack(): + data = np.expand_dims(np.asarray(mrc_data), axis=-1) + data = np.expand_dims(data, axis=-1) + elif mrc.is_volume(): + data = np.expand_dims(np.asarray(mrc_data), axis=0) + data = np.expand_dims(data, axis=-1) + elif mrc.is_volume_stack(): + data = np.expand_dims(np.asarray(mrc_data), axis=-1) + + name = self._get_file_no_digits_name(data_path) + spacing = (mrc.voxel_size.x, mrc.voxel_size.y, mrc.voxel_size.z) + + return data, name, spacing + + def _parse_ome_tiff(self, data_path: Path, progress_callback=None): + print("Parsing with OMETIFFReader...") + reader = OMETIFFReader(fpath=data_path) + ome_image, metadata, xml_metadata = reader.read() + + if progress_callback: + progress_callback(0.5, "Transpose to 'TXYZC'...") + + try: + ome_order = metadata["DimOrder BF Array"] + data = self._transpose_ome_image(ome_image, ome_order) + + try: + spacing = ( + metadata["PhysicalSizeX"], + metadata["PhysicalSizeY"], + metadata["PhysicalSizeZ"], + ) + except: + spacing = (1, 1, 1) + + name = self._get_file_no_digits_name(data_path) + except Exception as e: + print(f"Error parsing OME-TIFF: {e}") + data = np.zeros((1, 1, 1, 1, 1), dtype=np.float32) + name = "" + spacing = (1, 1, 1) + + return data, name, spacing + + def _parse_sitk( + self, + data_path: Path, + ext: str, + is_sequence: bool, + sequence: Optional[list], + progress_callback=None, + ): + print("Parsing with SimpleITK...") + + if ext in DICOM_EXTS: + data_dirpath = data_path.parent + reader = sitk.ImageSeriesReader() + reader.MetaDataDictionaryArrayUpdateOn() + reader.LoadPrivateTagsOn() + series_files = reader.GetGDCMSeriesFileNames( + str(data_dirpath), self.series_id + ) + reader.SetFileNames(series_files) + + itk_image = reader.Execute() + name, description = self._extract_dicom_meta(reader, data_dirpath) + + elif ext in SEQUENCE_EXTS and is_sequence and sequence: + sequence_list = sequence if sequence else [] + itk_image = sitk.ReadImage(sequence_list) + name = self._get_file_no_digits_name(data_path) + description = "" + + else: + itk_image = sitk.ReadImage(data_path) + name = self._get_filename(data_path) + description = "" + + if progress_callback: + progress_callback(0.5, "Transpose to 'TXYZC'...") + + data = self._convert_sitk_to_numpy(itk_image, ext, is_sequence) + + if itk_image.GetDimension() >= 3 and ext not in SEQUENCE_EXTS: + spacing = tuple(itk_image.GetSpacing()) + origin = tuple(itk_image.GetOrigin()) + direction = tuple(itk_image.GetDirection()) + else: + spacing = (1, 1, 1) + origin = (0, 0, 0) + direction = (1, 0, 0, 0, 1, 0, 0, 0, 1) + + return data, name, description, spacing, origin, direction + + def _transpose_ome_image(self, ome_image, ome_order: str): + if ome_image.ndim == 2: + ome_order = ome_order.replace("T", "").replace("C", "").replace("Z", "") + bioxel_order = (ome_order.index("X"), ome_order.index("Y")) + data = np.transpose(ome_image, bioxel_order) + data = np.expand_dims(data, axis=0) + data = np.expand_dims(data, axis=-1) + data = np.expand_dims(data, axis=-1) + + elif ome_image.ndim == 3: + ome_order = ome_order.replace("T", "").replace("C", "") + bioxel_order = ( + ome_order.index("X"), + ome_order.index("Y"), + ome_order.index("Z"), + ) + data = np.transpose(ome_image, bioxel_order) + data = np.expand_dims(data, axis=0) + data = np.expand_dims(data, axis=-1) + + elif ome_image.ndim == 4: + ome_order = ome_order.replace("T", "") + bioxel_order = ( + ome_order.index("X"), + ome_order.index("Y"), + ome_order.index("Z"), + ome_order.index("C"), + ) + data = np.transpose(ome_image, bioxel_order) + data = np.expand_dims(data, axis=0) + + elif ome_image.ndim == 5: + bioxel_order = ( + ome_order.index("T"), + ome_order.index("X"), + ome_order.index("Y"), + ome_order.index("Z"), + ome_order.index("C"), + ) + data = np.transpose(ome_image, bioxel_order) + + return data + + def _convert_sitk_to_numpy(self, itk_image, ext: str, is_sequence: bool): + if itk_image.GetDimension() == 2: + data = sitk.GetArrayFromImage(itk_image) + + if data.ndim == 3: + data = np.transpose(data, (1, 0, 2)) + data = np.expand_dims(data, axis=-2) + else: + data = np.transpose(data) + data = np.expand_dims(data, axis=-1) + data = np.expand_dims(data, axis=-1) + + data = np.expand_dims(data, axis=0) + + elif itk_image.GetDimension() == 3: + if ext not in SEQUENCE_EXTS: + itk_image = sitk.DICOMOrient(itk_image, "RAS") + + data = sitk.GetArrayFromImage(itk_image) + + if data.ndim == 4: + data = np.transpose(data, (2, 1, 0, 3)) + else: + data = np.transpose(data) + data = np.expand_dims(data, axis=-1) + + data = np.expand_dims(data, axis=0) + + elif itk_image.GetDimension() == 4: + data = sitk.GetArrayFromImage(itk_image) + + if data.ndim == 5: + data = np.transpose(data, (0, 3, 2, 1, 4)) + else: + data = np.transpose(data, (0, 3, 2, 1)) + data = np.expand_dims(data, axis=-1) + + return data + + def _extract_dicom_meta(self, reader, data_dirpath: Path): + def get_meta(key): + try: + string = reader.GetMetaData(0, key).removesuffix(" ") + string.encode("utf-8") + if string in ["No study description", "No series description", ""]: + return None + else: + return string + except: + return None + + study_description = get_meta("0008|1030") or data_dirpath.name + series_description = get_meta("0008|103e") + series_modality = get_meta("0008|0060") + + if series_description and series_modality: + description = f"{series_description}-{series_modality}" + elif series_description: + description = series_description + elif series_modality: + description = series_modality + else: + description = "" + + name = study_description.replace(" ", "-") + description = description.replace(" ", "-") + + return name, description + + @staticmethod + def _get_ext(filepath: Path) -> str: + if filepath.name.endswith(".nii.gz"): + return ".nii.gz" + elif filepath.name.endswith(".img.gz"): + return ".img.gz" + elif filepath.name.endswith(".gipl.gz"): + return ".gipl.gz" + elif filepath.name.endswith(".ome.tiff"): + return ".ome.tiff" + elif filepath.name.endswith(".ome.tif"): + return ".ome.tif" + elif filepath.name.endswith(".mrc.gz"): + return ".mrc.gz" + elif filepath.name.endswith(".map.gz"): + return ".map.gz" + else: + suffix = filepath.suffix + return "" if len(suffix) > 5 else suffix + + @staticmethod + def _get_filename(filepath: Path) -> str: + ext = Data._get_ext(filepath) + return filepath.name.removesuffix(ext) + + @staticmethod + def _get_file_no_digits_name(filepath: Path) -> str: + prefix, digits, suffix = Data._get_filename_parts(filepath) + prefix = Data._remove_end_str(prefix, "_") + prefix = Data._remove_end_str(prefix, ".") + prefix = Data._remove_end_str(prefix, "-") + prefix = Data._remove_end_str(prefix, " ") + return prefix + suffix + + @staticmethod + def _get_filename_parts(filepath: Path) -> tuple: + def has_digits(s): + return any(char.isdigit() for char in s) + + name = Data._get_filename(filepath) + parts = name.replace(".", " ").replace("_", " ").split(" ") + skip_prefixs = ["CH", "ch", "channel"] + number_part = None + number_part_i = None + + for i, part in enumerate(parts[::-1]): + if has_digits(part): + if not any([part.startswith(prefix) for prefix in skip_prefixs]): + number_part = part + number_part_i = len(parts) - i + break + + if number_part is None: + return name, "", "" + + prefix_parts = parts[: number_part_i - 1] + prefix_parts_count = sum([len(part) + 1 for part in prefix_parts]) + + digits = "" + suffix = "" + + started = False + for char in number_part[::-1]: + if char.isdigit(): + started = True + digits += char + else: + if started: + break + else: + suffix += char + + digits = digits[::-1] + + prefix_parts_count += len(number_part) - len(digits) - len(suffix) + + prefix = name[:prefix_parts_count] + suffix = name[prefix_parts_count + len(digits) :] + + return prefix, digits, suffix + + @staticmethod + def _get_file_index(filepath: Path) -> int: + prefix, digits, suffix = Data._get_filename_parts(filepath) + return int(digits) if digits != "" else 0 + + @staticmethod + def _collect_sequence(filepath: Path): + file_dict = {} + for f in filepath.parent.iterdir(): + if ( + f.is_file() + and Data._get_ext(filepath) == Data._get_ext(f) + and Data._get_file_no_digits_name(filepath) + == Data._get_file_no_digits_name(f) + ): + index = Data._get_file_index(f) + file_dict[index] = f + + for key in list(file_dict.keys()): + if not file_dict.get(key + 1) and not file_dict.get(key - 1): + del file_dict[key] + + file_dict = dict(sorted(file_dict.items())) + sequence = [str(f) for f in file_dict.values()] + + if len(sequence) == 0: + sequence = [str(filepath)] + + return sequence + + @staticmethod + def _remove_end_str(string: str, end: str) -> str: + while string.endswith(end) and len(string) > 0: + string = string.removesuffix(end) + return string + + def _transform_shape(self, data, frame_source: str) -> tuple: + orig_shape = self.xyz_shape + + if frame_source == "-1": + data = data[0:1, :, :, :, :] + new_shape = orig_shape + elif frame_source == "0": + new_shape = orig_shape + elif frame_source == "1": + data = data.transpose(1, 0, 2, 3, 4) + new_shape = (1, orig_shape[1], orig_shape[2]) + elif frame_source == "2": + data = data.transpose(2, 1, 0, 3, 4) + new_shape = (orig_shape[0], 1, orig_shape[2]) + elif frame_source == "3": + data = data.transpose(3, 1, 2, 0, 4) + new_shape = (orig_shape[0], orig_shape[1], 1) + else: + data = data.transpose(4, 1, 2, 3, 0) + new_shape = orig_shape + + return data, new_shape + + def _preprocess(self, data, kind: str, remap: bool) -> np.ndarray: + if kind == "color": + if np.issubdtype(data.dtype, np.uint8): + data = np.multiply(data, 1.0 / 256, dtype=np.float32) + elif data.dtype.kind in ["u", "i"]: + data = data.astype(np.float32) + min_val, max_val = data.min(), data.max() + if max_val != min_val: + data = (data - min_val) / (max_val - min_val) + else: + data = np.zeros_like(data, dtype=np.float32) + else: + data = data.astype(np.float32) + + if data.shape[4] == 1: + data = np.repeat(data, repeats=3, axis=4) + elif data.shape[4] == 2: + zeros_shape = list(data.shape[:4]) + [1] + zeros = np.zeros(zeros_shape, dtype=np.float32) + data = np.concatenate((data, zeros), axis=-1) + elif data.shape[4] > 3: + data = data[:, :, :, :, :3] + + elif kind == "scalar": + if remap: + data = data.astype(np.float32) + min_val, max_val = data.min(), data.max() + if max_val != min_val: + data = (data - min_val) / (max_val - min_val) + else: + data = np.zeros_like(data, dtype=np.float32) + + elif kind == "label": + data = data.astype(int) + + return data + + def _create_progress_cb_factory(self, progress_callback): + def factory(name, progress, progress_step): + def callback(frame, total): + if progress_callback: + sub_progress = progress + frame * (progress_step / total) + progress_callback( + sub_progress, f"Processing {name} Frame {frame+1}..." + ) + + return callback + + return factory + + def _create_label_layers( + self, + data, + base_name, + layer_shape, + affine, + smooth, + progress_callback: ProgressCallback, + ): + from .layer import Layer + + layers = [] + label_count = int(np.max(data)) + if label_count == 0: + return layers + + progress_step = 0.7 / label_count + cb_factory = self._create_progress_cb_factory(progress_callback) + + for i in range(label_count): + name_i = f"{base_name}_{i+1}" + progress = 0.2 + i * progress_step + if progress_callback: + progress_callback(progress, f"Processing {name_i}...") + + label_data = data == np.full_like(data, i + 1) + progress_cb = cb_factory(name_i, progress, progress_step) + + layer = Layer(data=label_data, name=name_i, kind="label") + layer.resize( + shape=layer_shape, smooth=smooth, progress_callback=progress_cb + ) + layer.affine = affine + layers.append(layer) + + return layers + + def _create_color_layers( + self, data, base_name, layer_shape, affine, progress_callback: ProgressCallback + ): + from .layer import Layer + + if progress_callback: + progress_callback(0.2, f"Processing {base_name}...") + + cb_factory = self._create_progress_cb_factory(progress_callback) + progress_cb = cb_factory(base_name, 0.2, 0.7) + + layer = Layer(data=data, name=base_name, kind="color") + layer.resize(shape=layer_shape, progress_callback=progress_cb) + layer.affine = affine + + return [layer] + + def _create_scalar_layers( + self, + data, + base_name, + layer_shape, + affine, + smooth, + split_channel, + progress_callback: ProgressCallback, + ): + from .layer import Layer + + layers = [] + cb_factory = self._create_progress_cb_factory(progress_callback) + + if split_channel: + channel_count = data.shape[-1] + progress_step = 0.7 / channel_count + + for i in range(channel_count): + name_i = f"{base_name}_{i+1}" + progress = 0.2 + i * progress_step + if progress_callback: + progress_callback(progress, f"Processing {name_i}...") + + progress_cb = cb_factory(name_i, progress, progress_step) + layer = Layer( + data=data[:, :, :, :, i : i + 1], name=name_i, kind="scalar" + ) + layer.resize( + shape=layer_shape, smooth=smooth, progress_callback=progress_cb + ) + layer.affine = affine + layers.append(layer) + else: + if progress_callback: + progress_callback(0.2, f"Processing {base_name}...") + + progress_cb = cb_factory(base_name, 0.2, 0.7) + layer = Layer(data=data, name=base_name, kind="scalar") + layer.resize( + shape=layer_shape, smooth=smooth, progress_callback=progress_cb + ) + layer.affine = affine + layers.append(layer) + + return layers + + +# 模块级函数签名 +def read( + filepath: str, series_id: str = "", progress_callback: ProgressCallback = None +) -> Data: + filepath = Path(filepath).resolve() + if not filepath.exists(): + raise FileNotFoundError(f"File not found: {filepath}") + + data_obj = Data(filepath=filepath, series_id=series_id) + data_obj.load_data(progress_callback) + return data_obj + + +def read_meta(filepath: str, series_id: str = "") -> Data: + data_obj = Data(filepath=filepath, series_id=series_id) + data_obj.load_meta() + return data_obj + + +def calc_layer_shape( + bioxel_size: float, orig_shape: tuple, orig_spacing: tuple +) -> tuple: + shape = ( + int(orig_shape[0] / bioxel_size * orig_spacing[0]), + int(orig_shape[1] / bioxel_size * orig_spacing[1]), + int(orig_shape[2] / bioxel_size * orig_spacing[2]), + ) + return ( + shape[0] if shape[0] > 0 else 1, + shape[1] if shape[1] > 0 else 1, + shape[2] if shape[2] > 0 else 1, + ) + + +def calc_layer_size(shape: tuple, bioxel_size: float, scale: float = 1.0) -> tuple: + size = ( + float(shape[0] * bioxel_size * scale), + float(shape[1] * bioxel_size * scale), + float(shape[2] * bioxel_size * scale), + ) + return size + + +def get_ext(filepath: Path) -> str: + if filepath.name.endswith(".nii.gz"): + return ".nii.gz" + elif filepath.name.endswith(".img.gz"): + return ".img.gz" + elif filepath.name.endswith(".gipl.gz"): + return ".gipl.gz" + elif filepath.name.endswith(".ome.tiff"): + return ".ome.tiff" + elif filepath.name.endswith(".ome.tif"): + return ".ome.tif" + elif filepath.name.endswith(".mrc.gz"): + return ".mrc.gz" + elif filepath.name.endswith(".map.gz"): + return ".map.gz" + else: + suffix = filepath.suffix + return "" if len(suffix) > 5 else suffix diff --git a/src/bioxelnodes/bioxel/layer.py b/src/bioxelnodes/bioxel/layer.py index d9449df..5dd9a32 100644 --- a/src/bioxelnodes/bioxel/layer.py +++ b/src/bioxelnodes/bioxel/layer.py @@ -1,6 +1,12 @@ import copy import numpy as np +# 条件导入 openvdb（只在 Blender 环境中存在） +try: + import openvdb as vdb +except ImportError: + vdb = None + from . import scipy from . import scipy as ndi diff --git a/src/bioxelnodes/layer.py b/src/bioxelnodes/layer.py index c6fb53d..a3926a0 100644 --- a/src/bioxelnodes/layer.py +++ b/src/bioxelnodes/layer.py @@ -6,7 +6,11 @@ import matplotlib.pyplot as plt import bpy import numpy as np -import openvdb as vdb + +try: + import openvdb as vdb +except ImportError: + vdb = None from .bioxel.layer import Layer from .utils import ndarray_to_png @@ -57,12 +61,10 @@ def cache_layer_data(layer: Layer, cache_path: str): grid.copyFromArray(data[f, :, :, :].copy().astype(np.float32)) else: # 颜色类型 grid = vdb.Vec3SGrid() - grid.copyFromArray( - data[f, :, :, :, :].copy().astype(np.float32)) + grid.copyFromArray(data[f, :, :, :, :].copy().astype(np.float32)) # 仅设置transform，不存储metadata - grid.transform = vdb.createLinearTransform( - layer.affine.transpose()) + grid.transform = vdb.createLinearTransform(layer.affine.transpose()) grid.name = layer.kind # 保存序列帧VDB @@ -272,6 +274,5 @@ def save_layers_to_json(layers: List[Layer], cache_dir: str) -> List[int]: added_ids.append(cache_id) set_layer_caches(existing_data) - print( - f"Successfully added {len(added_ids)} layers to the internal text datablock") + print(f"Successfully added {len(added_ids)} layers to the internal text datablock") return added_ids diff --git a/src/bioxelnodes/node.py b/src/bioxelnodes/node.py index 566bee6..b09b66b 100644 --- a/src/bioxelnodes/node.py +++ b/src/bioxelnodes/node.py @@ -105,7 +105,7 @@ def add_bioxel_node(name: str): bpy.ops.object.modifier_add_node_group( asset_library_type="CUSTOM", asset_library_identifier="O Bioxel", - relative_asset_identifier=f"Nodes.blend\\NodeTree\\{name}", + relative_asset_identifier=f"BioxelNodes.blend\\NodeTree\\{name}", ) bpy.ops.object.modifier_remove(modifier=name) try: diff --git a/src/bioxelnodes/operators/io.py b/src/bioxelnodes/operators/io.py index 96e1b5f..a94fdd2 100644 --- a/src/bioxelnodes/operators/io.py +++ b/src/bioxelnodes/operators/io.py @@ -12,37 +12,20 @@ import transforms3d from ..props import BIOXEL_Series from ..utils import get_layer_obj, wrapped_label -from ..bioxel.layer import Layer -from ..bioxel.parse import DICOM_EXTS, SUPPORT_EXTS, get_ext, parse_volumetric_data +from ..bioxel import ( + DICOM_EXTS, + SUPPORT_EXTS, + read, + read_meta, + calc_layer_shape, + calc_layer_size, + get_ext, +) from ..utils import get_cache_dir, progress_update, progress_bar from ..layer import get_layer_caches, save_layers_to_json -def get_layer_shape(bioxel_size: float, orig_shape: tuple, orig_spacing: tuple): - shape = ( - int(orig_shape[0] / bioxel_size * orig_spacing[0]), - int(orig_shape[1] / bioxel_size * orig_spacing[1]), - int(orig_shape[2] / bioxel_size * orig_spacing[2]), - ) - - return ( - shape[0] if shape[0] > 0 else 1, - shape[1] if shape[1] > 0 else 1, - shape[2] if shape[2] > 0 else 1, - ) - - -def get_layer_size(shape: tuple, bioxel_size: float, scale: float = 1.0): - size = ( - float(shape[0] * bioxel_size * scale), - float(shape[1] * bioxel_size * scale), - float(shape[2] * bioxel_size * scale), - ) - - return size - - """ ImportData -> ParseVolumetricData -> ImportDataDialog start import parse data execute import @@ -211,11 +194,7 @@ class ParseVolumetricData(bpy.types.Operator): try: series_id = self.series_id if self.series_id != "empty" else "" - data, meta = parse_volumetric_data( - data_file=self.filepath, - series_id=series_id, - progress_callback=progress_callback, - ) + data = read_meta(self.filepath, series_id=series_id) except KeyboardInterrupt: return except Exception as e: @@ -225,8 +204,8 @@ class ParseVolumetricData(bpy.types.Operator): if cancel(): return - self.meta = meta - self.label_count = int(np.max(data)) + self.meta = data.meta + self.label_count = int(np.max(data.data)) self.dtype = data.dtype # Init cancel flag @@ -334,8 +313,8 @@ class ParseVolumetricData(bpy.types.Operator): bioxel_size = max(min(*orig_spacing), 1.0) - layer_shape = get_layer_shape(bioxel_size, orig_shape, orig_spacing) - layer_size = get_layer_size(layer_shape, bioxel_size, 0.01) + layer_shape = calc_layer_shape(bioxel_size, orig_shape, orig_spacing) + layer_size = calc_layer_size(layer_shape, bioxel_size, 0.01) min_log10 = math.floor(math.log10(min(*layer_size))) max_log10 = math.floor(math.log10(max(*layer_size))) @@ -547,33 +526,23 @@ class ImportDataDialog(bpy.types.Operator): def execute(self, context): def import_volumetric_data_func(self, context, cancel): - progress_update(context, 0.0, "Parsing Volumetirc Data...") - def progress_callback(factor, text): if cancel(): raise KeyboardInterrupt("Cancelled by user") progress_update(context, factor * 0.2, text) - def progress_callback_factory(layer_name, progress, progress_step): - def progress_callback(frame, total): - if cancel(): - raise KeyboardInterrupt("Cancelled by user") - sub_progress_step = progress_step / total - sub_progress = progress + frame * sub_progress_step - progress_update( - context, - sub_progress, - f"Processing {layer_name} Frame {frame+1}...", - ) - print(f"Processing {layer_name} Frame {frame+1}...") - - return progress_callback - try: - data, meta = parse_volumetric_data( - data_file=self.filepath, - series_id=self.series_id, - progress_callback=progress_callback, + data = read(self.filepath, self.series_id, progress_callback) + + self.layers = data.to_layers( + kind=self.read_as.lower(), + layer_name=self.layer_name, + bioxel_size=self.bioxel_size, + smooth=self.smooth, + remap=self.remap, + split_channel=self.split_channel, + frame_source=self.frame_source, + progress_callback=lambda f, t: progress_callback(0.2 + f * 0.7, t), ) except KeyboardInterrupt: @@ -585,203 +554,6 @@ class ImportDataDialog(bpy.types.Operator): if cancel(): return - shape = get_layer_shape( - self.bioxel_size, self.orig_shape, self.orig_spacing - ) - - mat_scale = transforms3d.zooms.zfdir2aff(self.bioxel_size) - affine = np.dot(meta["affine"], mat_scale) - kind = self.read_as.lower() - - if cancel(): - return - - # change shape as sequence or not - if self.frame_source == "-1": - data = data[0:1, :, :, :, :] - elif self.frame_source == "0": - # frame as frame - pass - elif self.frame_source == "1": - # X as frame - data = data.transpose(1, 0, 2, 3, 4) - shape = (1, shape[1], shape[2]) - elif self.frame_source == "2": - # Y as frame - data = data.transpose(2, 1, 0, 3, 4) - shape = (shape[0], 1, shape[2]) - elif self.frame_source == "3": - # Z as frame - data = data.transpose(3, 1, 2, 0, 4) - shape = (shape[0], shape[1], 1) - else: - # channel as frame - data = data.transpose(4, 1, 2, 3, 0) - - layers = [] - if kind == "label": - name = self.layer_name or "Label" - data = data.astype(int) - label_count = int(np.max(data)) - progress_step = 0.7 / label_count - - for i in range(label_count): - if cancel(): - return - - name_i = f"{name}_{i+1}" - progress = 0.2 + i * progress_step - progress_update(context, progress, f"Processing {name_i}...") - - progress_callback = progress_callback_factory( - name_i, progress, progress_step - ) - label_data = data == np.full_like(data, i + 1) - # label_data = label_data.astype(np.float32) - try: - layer = Layer(data=label_data, name=name_i, kind=kind) - - layer.resize( - shape=shape, - smooth=self.smooth, - progress_callback=progress_callback, - ) - - layer.affine = affine - - layers.append(layer) - except KeyboardInterrupt: - return - except Exception as e: - self.has_error = e - return - - if kind == "color": - if np.issubdtype(np.uint8, data.dtype): - data = np.multiply(data, 1.0 / 256, dtype=np.float32) - elif data.dtype.kind in ["u", "i"]: - # Convert the normalized array to float dtype - data = data.astype(np.float32) - - min_val = data.min() - max_val = data.max() - # Avoid division by zero if all values are the same - if max_val != min_val: - # Normalize the array to the range (0,1) - data = (data - min_val) / (max_val - min_val) - else: - # If all values are the same, the normalized array will be all zeros - data = np.zeros_like(data, dtype=np.float32) - - else: - data = data.astype(np.float32) - - # Gamma Correct - # data = data ** 2.2 - - name = self.layer_name or "Color" - if data.shape[4] == 1: - data = np.repeat(data, repeats=3, axis=4) - elif data.shape[4] == 2: - d_shape = list(data.shape) - d_shape = d_shape[:4] + [1] - zore = np.zeros(tuple(d_shape), dtype=np.float32) - data = np.concatenate((data, zore), axis=-1) - elif data.shape[4] > 3: - data = data[:, :, :, :, :3] - - if cancel(): - return - - progress_update(context, 0.2, f"Processing {name}...") - progress_callback = progress_callback_factory(name, 0.2, 0.7) - - try: - layer = Layer(data=data, name=name, kind=kind) - - layer.resize(shape=shape, progress_callback=progress_callback) - - layer.affine = affine - - layers.append(layer) - except KeyboardInterrupt: - return - except Exception as e: - self.has_error = e - return - - elif kind == "scalar": - name = self.layer_name or "Scalar" - - if self.remap: - # Convert the normalized array to float dtype - data = data.astype(np.float32) - - min_val = data.min() - max_val = data.max() - # Avoid division by zero if all values are the same - if max_val != min_val: - # Normalize the array to the range (0,1) - data = (data - min_val) / (max_val - min_val) - else: - # If all values are the same, the normalized array will be all zeros - data = np.zeros_like(data, dtype=np.float32) - - if self.split_channel: - progress_step = 0.7 / self.channel_count - - for i in range(self.channel_count): - if cancel(): - return - - name_i = f"{name}_{i+1}" - progress = 0.2 + i * progress_step - progress_update(context, progress, f"Processing {name_i}...") - progress_callback = progress_callback_factory( - name_i, progress, progress_step - ) - try: - layer = Layer( - data=data[:, :, :, :, i : i + 1], name=name_i, kind=kind - ) - - layer.resize( - shape=shape, progress_callback=progress_callback - ) - - layer.affine = affine - - layers.append(layer) - except KeyboardInterrupt: - return - except Exception as e: - self.has_error = e - return - else: - if cancel(): - return - - progress_update(context, 0.2, f"Processing {name}...") - progress_callback = progress_callback_factory(name, 0.2, 0.7) - - try: - layer = Layer(data=data, name=name, kind=kind) - - layer.resize(shape=shape, progress_callback=progress_callback) - - layer.affine = affine - - layers.append(layer) - except KeyboardInterrupt: - return - except Exception as e: - self.has_error = e - return - - if cancel(): - return - - self.layers = layers progress_update(context, 0.9, "Creating Layers...") self.is_cancelled = False @@ -851,7 +623,7 @@ class ImportDataDialog(bpy.types.Operator): return {"RUNNING_MODAL"} def draw(self, context): - layer_shape = get_layer_shape( + layer_shape = calc_layer_shape( self.bioxel_size, self.orig_shape, self.orig_spacing ) diff --git a/src/bioxelnodes/operators/layer.py b/src/bioxelnodes/operators/layer.py index 6825012..cc95bdc 100644 --- a/src/bioxelnodes/operators/layer.py +++ b/src/bioxelnodes/operators/layer.py @@ -16,13 +16,11 @@ class RenameLayer(bpy.types.Operator): bl_options = {"REGISTER", "UNDO"} cache_id: bpy.props.StringProperty(options={"HIDDEN"}) # type: ignore - new_name: bpy.props.StringProperty( - name="New name", default="") # type: ignore + new_name: bpy.props.StringProperty(name="New name", default="") # type: ignore def invoke(self, context, event): caches = get_layer_caches() - entry = next((c for c in caches if str( - c.get("id", "")) == self.cache_id), None) + entry = next((c for c in caches if str(c.get("id", "")) == self.cache_id), None) if entry: self.new_name = str(entry.get("name", "")) return context.window_manager.invoke_props_dialog(self) @@ -127,8 +125,7 @@ class AddLayerNode(bpy.types.Operator): layer_node.inputs["Frame Count"].default_value = entry["frame_count"] layer_node.inputs["Animation"].default_value = True else: - hidden_sockets = hidden_sockets + \ - ["Frame Count", "Frame Offset", "Cycle"] + hidden_sockets = hidden_sockets + ["Frame Count", "Frame Offset", "Cycle"] # 将特定属性隐藏到hidden面板 for socket_name in hidden_sockets: @@ -177,8 +174,7 @@ class RelocatePath(bpy.types.Operator): return {"CANCELLED"} caches = get_layer_caches() - entry = next((c for c in caches if str( - c.get("id", "")) == self.cache_id), None) + entry = next((c for c in caches if str(c.get("id", "")) == self.cache_id), None) if not entry: self.report({"ERROR"}, "Layer not found") return {"CANCELLED"} @@ -228,8 +224,7 @@ class SaveLayer(bpy.types.Operator): def invoke(self, context, event): # Default to Blender file directory if possible - blend_dir = Path( - bpy.data.filepath).parent if bpy.data.filepath else Path.cwd() + blend_dir = Path(bpy.data.filepath).parent if bpy.data.filepath else Path.cwd() context.window_manager.fileselect_add(self) if not self.directory: self.directory = str(blend_dir) @@ -244,8 +239,7 @@ class SaveLayer(bpy.types.Operator): return {"CANCELLED"} caches = get_layer_caches() - entry = next((c for c in caches if str( - c.get("id")) == self.cache_id), None) + entry = next((c for c in caches if str(c.get("id")) == self.cache_id), None) cache_id = entry.get("id") if not entry: diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..db4d862 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,43 @@ +# BioxelNodes Unit Tests + +This directory contains unit tests for the bioxel module. + +## Test Data + +Test files should be placed in `tests/data/`: + +## Running Tests + +Run all tests: +```bash +uv run pytest +``` + +Run specific test file: +```bash +uv run pytest tests/test_bioxel.py +``` + +Run specific test class: +```bash +uv run pytest tests/test_bioxel.py::TestDataClass +``` + +Run with verbose output: +```bash +uv run pytest -v +``` + +Run with coverage: +```bash +uv run pytest --cov=bioxelnodes.bioxel --cov-report=html +``` + +## Test Categories + +- **TestDataClass**: Data class initialization and properties +- **TestDataToLayers**: Data.to_layers() method +- **TestUtilityFunctions**: Utility functions +- **TestLayerClass**: Layer class functionality +- **TestConstants**: Module constants +- **TestEdgeCases**: Edge cases and error handling diff --git a/tests/test_bioxel.py b/tests/test_bioxel.py new file mode 100644 index 0000000..43cbe6c --- /dev/null +++ b/tests/test_bioxel.py @@ -0,0 +1,539 @@ +""" +Unit tests for bioxel module. + +Tests cover: +- Data class functionality +- Parsing NRRD and DICOM formats +- Layer creation and transformation +- Utility functions +""" + +from bioxelnodes.bioxel import ( + Data, + read, + read_meta, + calc_layer_shape, + calc_layer_size, + Layer, +) +import pytest +import numpy as np +from pathlib import Path + +import sys + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + + +# Test data files +NRRD_FILE = Path(__file__).parent / "data/VHP_M.nrrd" +DICOM_FILE = Path(__file__).parent / "data/VHP_M_CT_Head/IM000001.dcm" +DICOM_DIR = Path(__file__).parent / "data/VHP_M_CT_Head/" + + +class TestDataClass: + """Test Data class initialization and properties""" + + def test_data_initialization(self): + """Test basic Data initialization with NRRD""" + data = Data(filepath=str(NRRD_FILE), series_id="") + + assert data.filepath is not None + assert data.series_id == "" + assert data._data is None + assert data._meta is None + + def test_read_meta_nrrd(self): + """Test reading only meta from NRRD file""" + data_obj = read_meta(str(NRRD_FILE)) + + assert data_obj._data is None + assert data_obj._meta is not None + assert "name" in data_obj.meta + assert "spacing" in data_obj.meta + assert "affine" in data_obj.meta + assert "xyz_shape" in data_obj.meta + assert "frame_count" in data_obj.meta + assert "channel_count" in data_obj.meta + assert "dtype" in data_obj.meta + + def test_read_meta_dicom(self): + """Test reading only meta from DICOM file""" + data_obj = read_meta(str(DICOM_FILE), series_id="") + + assert data_obj._data is None + assert data_obj._meta is not None + assert "name" in data_obj.meta + + def test_read_full_data_nrrd(self): + """Test reading full NRRD data""" + data_obj = read(str(NRRD_FILE)) + + assert data_obj._data is not None + assert data_obj._meta is not None + assert isinstance(data_obj.data, np.ndarray) + assert data_obj.data.ndim == 5 + + def test_read_full_data_dicom(self): + """Test reading full DICOM data""" + data_obj = read(str(DICOM_FILE), series_id="") + + assert data_obj._data is not None + assert data_obj._meta is not None + assert isinstance(data_obj.data, np.ndarray) + + def test_data_properties_nrrd(self): + """Test Data property accessors""" + data_obj = read(str(NRRD_FILE)) + + assert hasattr(data_obj, "name") + assert hasattr(data_obj, "description") + assert hasattr(data_obj, "shape") + assert hasattr(data_obj, "xyz_shape") + assert hasattr(data_obj, "spacing") + assert hasattr(data_obj, "affine") + assert hasattr(data_obj, "frame_count") + assert hasattr(data_obj, "channel_count") + assert hasattr(data_obj, "dtype") + + assert isinstance(data_obj.shape, tuple) + assert isinstance(data_obj.xyz_shape, tuple) + assert isinstance(data_obj.spacing, tuple) + assert isinstance(data_obj.affine, np.ndarray) + assert isinstance(data_obj.dtype, np.dtype) + + def test_data_lazy_loading(self): + """Test lazy loading behavior""" + data_obj = Data(filepath=str(NRRD_FILE), series_id="") + + assert data_obj.is_loaded() is False + + data_obj.load_meta() + assert data_obj._data is None + assert data_obj._meta is not None + + data_obj.load_data() + assert data_obj._data is not None + assert data_obj.is_loaded() is True + + def test_shape_property(self): + """Test shape property returns TXYZC format""" + data_obj = read(str(NRRD_FILE)) + + shape = data_obj.shape + assert len(shape) == 5 + assert shape[0] == data_obj.meta["frame_count"] + assert shape[1:4] == data_obj.meta["xyz_shape"] + assert shape[4] == data_obj.meta["channel_count"] + + def test_load_meta_method(self): + """Test load_meta() method exists and works""" + data_obj = Data(filepath=str(NRRD_FILE), series_id="") + + assert hasattr(data_obj, "load_meta") + + data_obj.load_meta() + assert data_obj._meta is not None + assert data_obj._data is None + + +class TestDataToLayers: + """Test Data.to_layers() method""" + + def test_to_layers_dicom_scalar(self): + """Test converting DICOM data to scalar layers""" + data_obj = read(str(DICOM_FILE), series_id="") + + layers = data_obj.to_layers(kind="scalar", bioxel_size=2.0) + + assert len(layers) == 1 + assert layers[0].kind == "scalar" + assert isinstance(layers[0], Layer) + + def test_to_layers_nrrd_scalar(self): + """Test converting NRRD data to scalar layers""" + data_obj = read(str(NRRD_FILE)) + + layers = data_obj.to_layers(kind="scalar", bioxel_size=2.0) + + assert len(layers) == 1 + assert layers[0].kind == "scalar" + assert isinstance(layers[0], Layer) + + def test_to_layers_with_remap(self): + """Test data remapping""" + data_obj = read(str(NRRD_FILE)) + + layers = data_obj.to_layers(kind="scalar", bioxel_size=2.0, remap=True) + + assert len(layers) == 1 + assert layers[0].data.dtype == np.float32 + assert np.all(layers[0].data >= 0) + assert np.all(layers[0].data <= 1) + + def test_to_layers_with_bioxel_size(self): + """Test with different bioxel sizes""" + data_obj = read(str(NRRD_FILE)) + + layers_1x = data_obj.to_layers(kind="scalar", bioxel_size=1.0) + layers_2x = data_obj.to_layers(kind="scalar", bioxel_size=2.0) + + assert len(layers_1x) == 1 + assert len(layers_2x) == 1 + + # Larger bioxel size means smaller layer + shape_1x = layers_1x[0].shape + shape_2x = layers_2x[0].shape + assert shape_2x[0] <= shape_1x[0] + assert shape_2x[1] <= shape_1x[1] + assert shape_2x[2] <= shape_1x[2] + + def test_to_layers_with_smooth(self): + """Test with smoothing parameter""" + data_obj = read(str(NRRD_FILE)) + + layers = data_obj.to_layers(kind="scalar", bioxel_size=2.0, smooth=1) + + assert len(layers) == 1 + assert isinstance(layers[0], Layer) + + def test_to_layers_frame_source_first_frame(self): + """Test frame_source = '-1' (first frame only)""" + data_obj = read(str(NRRD_FILE)) + + layers = data_obj.to_layers(kind="scalar", bioxel_size=2.0, frame_source="-1") + + assert len(layers) == 1 + assert layers[0].frame_count == 1 + + +class TestUtilityFunctions: + """Test utility functions""" + + def test_calc_layer_shape(self): + """Test layer shape calculation""" + shape = calc_layer_shape( + bioxel_size=1.0, orig_shape=(100, 100, 100), orig_spacing=(1.0, 1.0, 1.0) + ) + + assert len(shape) == 3 + assert shape[0] == 100 + assert shape[1] == 100 + assert shape[2] == 100 + + def test_calc_layer_shape_small_bioxel(self): + """Test layer shape with small bioxel size""" + shape = calc_layer_shape( + bioxel_size=2.0, orig_shape=(100, 100, 100), orig_spacing=(1.0, 1.0, 1.0) + ) + + assert len(shape) == 3 + assert shape[0] == 50 + assert shape[1] == 50 + assert shape[2] == 50 + + def test_calc_layer_shape_minimum(self): + """Test layer shape doesn't go below 1""" + shape = calc_layer_shape( + bioxel_size=100.0, orig_shape=(100, 100, 100), orig_spacing=(1.0, 1.0, 1.0) + ) + + assert shape == (1, 1, 1) + + def test_calc_layer_size(self): + """Test layer size calculation""" + size = calc_layer_size(shape=(100, 100, 100), bioxel_size=1.0, scale=1.0) + + assert len(size) == 3 + assert size[0] == 100.0 + assert size[1] == 100.0 + assert size[2] == 100.0 + + def test_calc_layer_size_with_scale(self): + """Test layer size with scale factor""" + size = calc_layer_size(shape=(100, 100, 100), bioxel_size=1.0, scale=0.01) + + assert len(size) == 3 + assert size[0] == 1.0 + assert size[1] == 1.0 + assert size[2] == 1.0 + + +class TestLayerClass: + """Test Layer class functionality""" + + def test_layer_initialization(self): + """Test Layer initialization""" + data = np.random.rand(1, 8, 8, 8, 1).astype(np.float32) + layer = Layer(data=data, name="test_layer", kind="scalar") + + assert layer.name == "test_layer" + assert layer.kind == "scalar" + assert np.array_equal(layer.data, data) + + def test_layer_properties(self): + """Test Layer property accessors""" + data = np.random.rand(1, 8, 8, 8, 1).astype(np.float32) + layer = Layer( + data=data, name="test_layer", kind="scalar", affine=np.identity(4) + ) + + assert hasattr(layer, "bioxel_size") + assert hasattr(layer, "shape") + assert hasattr(layer, "dtype") + assert hasattr(layer, "origin") + assert hasattr(layer, "euler") + assert hasattr(layer, "frame_count") + assert hasattr(layer, "channel_count") + assert hasattr(layer, "min") + assert hasattr(layer, "max") + + def test_layer_shape_property(self): + """Test layer shape returns XYZ (not TXYZC)""" + data = np.random.rand(1, 8, 8, 8, 1).astype(np.float32) + layer = Layer(data=data, name="test", kind="scalar") + + shape = layer.shape + assert len(shape) == 3 + assert shape[0] == 8 + assert shape[1] == 8 + assert shape[2] == 8 + + def test_layer_affine_property(self): + """Test affine transformation matrix""" + data = np.random.rand(1, 8, 8, 8, 1).astype(np.float32) + affine = np.array( + [ + [1.0, 0.0, 0.0, 0.0], + [0.0, 1.0, 0.0, 0.0], + [0.0, 0.0, 1.0, 0.0], + [0.0, 0.0, 0.0, 1.0], + ] + ) + layer = Layer(data=data, name="test", kind="scalar", affine=affine) + + result_affine = layer.affine + assert result_affine.shape == (4, 4) + assert np.allclose(result_affine, affine) + + def test_layer_min_max(self): + """Test layer min/max properties""" + data = np.random.rand(1, 8, 8, 8, 1).astype(np.float32) + layer = Layer(data=data, name="test", kind="scalar") + + assert isinstance(layer.min, float) + assert isinstance(layer.max, float) + assert layer.min <= layer.max + + def test_layer_copy(self): + """Test layer deep copy""" + data = np.random.rand(1, 8, 8, 8, 1).astype(np.float32) + layer = Layer(data=data, name="test", kind="scalar") + layer_copy = layer.copy() + + assert layer_copy.name == layer.name + assert layer_copy.kind == layer.kind + assert np.array_equal(layer_copy.data, layer.data) + + assert layer_copy is not layer + layer_copy.data[0, 0, 0, 0, 0] = 999.0 + assert layer.data[0, 0, 0, 0, 0] != 999.0 + + def test_layer_invalid_data_shape(self): + """Test Layer raises error for invalid data shape""" + data = np.random.rand(8, 8, 8).astype(np.float32) + + with pytest.raises(Exception): + Layer(data=data, name="test", kind="scalar") + + def test_layer_invalid_affine_shape(self): + """Test Layer raises error for invalid affine shape""" + data = np.random.rand(1, 8, 8, 8, 1).astype(np.float32) + affine = np.identity(3) + + with pytest.raises(Exception): + Layer(data=data, name="test", kind="scalar", affine=affine) + + +class TestConstants: + """Test module constants""" + + def test_support_exts_exists(self): + """Test SUPPORT_EXTS constant""" + from bioxelnodes.bioxel import SUPPORT_EXTS + + assert isinstance(SUPPORT_EXTS, list) + assert len(SUPPORT_EXTS) > 0 + assert ".h5" in SUPPORT_EXTS + assert ".nrrd" in SUPPORT_EXTS + assert ".nii.gz" in SUPPORT_EXTS + assert ".dcm" in SUPPORT_EXTS + + def test_dicom_exts_exists(self): + """Test DICOM_EXTS constant""" + from bioxelnodes.bioxel import DICOM_EXTS + + assert isinstance(DICOM_EXTS, list) + assert "" in DICOM_EXTS + assert ".dcm" in DICOM_EXTS + assert ".DCM" in DICOM_EXTS + + def test_ome_exts_exists(self): + """Test OME_EXTS constant""" + from bioxelnodes.bioxel import OME_EXTS + + assert isinstance(OME_EXTS, list) + assert ".tiff" in OME_EXTS + assert ".ome.tiff" in OME_EXTS + + def test_mrc_exts_exists(self): + """Test MRC_EXTS constant""" + from bioxelnodes.bioxel import MRC_EXTS + + assert isinstance(MRC_EXTS, list) + assert ".mrc" in MRC_EXTS + + +class TestEdgeCases: + """Test edge cases and error handling""" + + def test_nonexistent_file(self): + """Test handling of nonexistent file""" + with pytest.raises(FileNotFoundError): + data_obj = read("/nonexistent/file.nii.gz", "") + + def test_frame_count_property_multiframe(self): + """Test frame_count property for multi-frame data""" + data_obj = read(str(NRRD_FILE)) + + assert data_obj.frame_count >= 1 + assert data_obj.frame_count == data_obj.meta["frame_count"] + + def test_channel_count_property(self): + """Test channel_count property""" + data_obj = read(str(NRRD_FILE)) + + assert data_obj.channel_count >= 1 + assert data_obj.channel_count == data_obj.meta["channel_count"] + + def test_nrrd_format_support(self): + """Test NRRD format is in supported extensions""" + from bioxelnodes.bioxel import SUPPORT_EXTS + + assert ".nrrd" in SUPPORT_EXTS + + def test_dicom_format_support(self): + """Test DICOM format is in supported extensions""" + from bioxelnodes.bioxel import SUPPORT_EXTS + + assert ".dcm" in SUPPORT_EXTS + assert "" in SUPPORT_EXTS + + +class TestRealDataFiles: + """Test with real data files""" + + def test_nrrd_file_exists(self): + """Test NRRD test file exists""" + assert NRRD_FILE.exists() + assert NRRD_FILE.is_file() + + def test_dicom_file_exists(self): + """Test DICOM test file exists""" + assert DICOM_FILE.exists() + assert DICOM_FILE.is_file() + + def test_dicom_directory_exists(self): + """Test DICOM directory exists""" + assert DICOM_DIR.exists() + assert DICOM_DIR.is_dir() + + def test_read_nrrd_data(self): + """Test reading real NRRD data""" + data_obj = read(str(NRRD_FILE)) + + assert data_obj._data is not None + assert data_obj._meta is not None + assert isinstance(data_obj.data, np.ndarray) + assert data_obj.data.size > 0 + + def test_read_dicom_data(self): + """Test reading real DICOM data""" + data_obj = read(str(DICOM_FILE), series_id="") + + assert data_obj._data is not None + assert data_obj._meta is not None + assert isinstance(data_obj.data, np.ndarray) + assert data_obj.data.size > 0 + + def test_nrrd_meta_contains_required_fields(self): + """Test NRRD meta contains all required fields""" + data_obj = read(str(NRRD_FILE)) + meta = data_obj.meta + + required_fields = [ + "name", + "description", + "spacing", + "affine", + "xyz_shape", + "frame_count", + "channel_count", + "dtype", + ] + + for field in required_fields: + assert field in meta, f"Meta missing field: {field}" + + def test_dicom_meta_contains_required_fields(self): + """Test DICOM meta contains all required fields""" + data_obj = read(str(DICOM_FILE), series_id="") + meta = data_obj.meta + + required_fields = [ + "name", + "description", + "spacing", + "affine", + "xyz_shape", + "frame_count", + "channel_count", + "dtype", + ] + + for field in required_fields: + assert field in meta, f"Meta missing field: {field}" + + def test_data_to_layers_with_real_nrrd(self): + """Test to_layers with real NRRD data""" + data_obj = read(str(NRRD_FILE)) + + layers = data_obj.to_layers(kind="scalar", bioxel_size=3.0) + + assert len(layers) > 0 + assert all(isinstance(layer, Layer) for layer in layers) + + def test_data_to_layers_with_real_dicom(self): + """Test to_layers with real DICOM data""" + data_obj = read(str(DICOM_FILE), series_id="") + + layers = data_obj.to_layers(kind="scalar", bioxel_size=3.0) + + assert len(layers) > 0 + assert all(isinstance(layer, Layer) for layer in layers) + + def test_nrrd_spacing_values(self): + """Test NRRD spacing values are reasonable""" + data_obj = read(str(NRRD_FILE)) + spacing = data_obj.spacing + + assert len(spacing) == 3 + assert all(s > 0 for s in spacing) + + def test_dicom_spacing_values(self): + """Test DICOM spacing values are reasonable""" + data_obj = read(str(DICOM_FILE), series_id="") + spacing = data_obj.spacing + + assert len(spacing) == 3 + assert all(s > 0 for s in spacing)