io

Functions:

Name	Description
ensure_unzipped	unzip a (downloaded) source to a file in folder if source is a zip archive
get_suffix	Deprecated use source.suffix instead.
load_dataset_stat	DEPRECATED alias for load_stat(): use load_stat() instead.
load_image	load a single image as numpy array
load_stat	Load sample and dataset statistics from JSON
load_tensor
save_dataset_stat	DEPRECATED alias for save_stat(): use save_stats() instead.
save_sample	Save a sample to a path pattern
save_stat	Save sample and dataset statistics as a JSON file
save_tensor
serialize_stat	Serialize a stat mapping to a JSON string

Attributes:

Name	Type	Description
JsonValue	TypeAlias
Suffix

JsonValue module-attribute

JsonValue: TypeAlias = Union[bool, int, float, str, None, List['JsonValue'], Dict[str, 'JsonValue']]

Returned by:

• API Reference  io  serialize_stat

Used by:

•  client  SerializedSampleBlock
•  serializer  SerializedSampleBlock
• API Reference  io  load_stat

Suffix module-attribute

Suffix = str

Returned by:

• API Reference  io  get_suffix

ensure_unzipped

ensure_unzipped(source: Union[PermissiveFileSource, ZipPath, BytesReader], folder: Path)

unzip a (downloaded) source to a file in folder if source is a zip archive otherwise copy source to a file in folder.

Source code in src/bioimageio/core/io.py

def ensure_unzipped(
    source: Union[PermissiveFileSource, ZipPath, BytesReader], folder: Path
):
    """unzip a (downloaded) **source** to a file in **folder** if source is a zip archive
    otherwise copy **source** to a file in **folder**."""
    if isinstance(source, BytesReader):
        weights_reader = source
    else:
        weights_reader = get_reader(source)

    out_path = folder / (
        weights_reader.original_file_name or f"file{weights_reader.suffix}"
    )

    if zipfile.is_zipfile(weights_reader):
        out_path = out_path.with_name(out_path.name + ".unzipped")
        out_path.parent.mkdir(exist_ok=True, parents=True)
        # source itself is a zipfile
        with zipfile.ZipFile(weights_reader, "r") as f:
            f.extractall(out_path)

    else:
        out_path.parent.mkdir(exist_ok=True, parents=True)
        with out_path.open("wb") as f:
            copyfileobj(weights_reader, f)

    return out_path

get_suffix

get_suffix(source: Union[ZipPath, FileSource]) -> Suffix

Deprecated

use source.suffix instead.

Source code in src/bioimageio/core/io.py

def get_suffix(source: Union[ZipPath, FileSource]) -> Suffix:
    """DEPRECATED: use source.suffix instead."""
    return source.suffix

load_dataset_stat

load_dataset_stat(path: Path) -> Stat

DEPRECATED alias for load_stat(): use load_stat() instead.

Source code in src/bioimageio/core/io.py

def load_dataset_stat(path: Path) -> Stat:
    """DEPRECATED alias for `load_stat()`: use `load_stat()` instead."""
    warnings.warn("`load_dataset_stat()` is deprecated, use `load_stats()` instead.")
    return load_stat(path)

load_image

load_image(source: Union[PermissiveFileSource, ZipPath], is_volume: Optional[bool] = None) -> NDArray[Any]

load a single image as numpy array

Parameters:

Name	Type	Description	Default
source	Union[PermissiveFileSource, ZipPath]	image source	required
is_volume	Optional[bool]	deprecated	None

Source code in src/bioimageio/core/io.py

def load_image(
    source: Union[PermissiveFileSource, ZipPath], is_volume: Optional[bool] = None
) -> NDArray[Any]:
    """load a single image as numpy array

    Args:
        source: image source
        is_volume: deprecated
    """
    if is_volume is not None:
        warnings.warn("**is_volume** is deprecated and will be removed soon.")

    if isinstance(source, (FileDescr, ZipPath)):
        parsed_source = source
    else:
        parsed_source = interprete_file_source(source)

    if isinstance(parsed_source, RelativeFilePath):
        parsed_source = parsed_source.absolute()

    if parsed_source.suffix == ".npy":
        image = load_array(parsed_source)
    else:
        reader = download(parsed_source)
        image = imread(  # pyright: ignore[reportUnknownVariableType]
            reader.read(), extension=parsed_source.suffix
        )

    assert is_ndarray(image)
    return image

load_stat

load_stat(source: Union[Path, str, Sequence[JsonValue]]) -> Stat

Load sample and dataset statistics from JSON

Source code in src/bioimageio/core/io.py

def load_stat(source: Union[Path, str, Sequence[JsonValue]]) -> Stat:
    """Load sample and dataset statistics from JSON"""
    if isinstance(source, Path):
        source = source.read_text(encoding="utf-8")

    if isinstance(source, str):
        seq = _StatList.model_validate_json(source)
    else:
        seq = _StatList.model_validate(source)

    return {e.measure: e.value for e in seq.root}

load_tensor

load_tensor(source: Union[PermissiveFileSource, ZipPath], /, axes: Optional[Sequence[AxisLike]] = None) -> Tensor

Source code in src/bioimageio/core/io.py

def load_tensor(
    source: Union[PermissiveFileSource, ZipPath],
    /,
    axes: Optional[Sequence[AxisLike]] = None,
) -> Tensor:
    # TODO: load axis meta data
    array = load_image(source)

    return Tensor.from_numpy(array, dims=axes)

save_dataset_stat

save_dataset_stat(stat: Mapping[DatasetMeasure, MeasureValue], path: Path) -> None

DEPRECATED alias for save_stat(): use save_stats() instead.

Source code in src/bioimageio/core/io.py

def save_dataset_stat(stat: Mapping[DatasetMeasure, MeasureValue], path: Path) -> None:
    """DEPRECATED alias for save_stat(): use `save_stats()` instead."""
    warnings.warn("`save_dataset_stat()` is deprecated, use `save_stats()` instead.")
    save_stat({k: v for k, v in stat.items()}, path)

save_sample

save_sample(path: Union[Path, str, PerMember[Union[Path, str]]], sample: Sample) -> None

Save a sample to a path pattern or all sample members in the path mapping.

If path is a pathlib.Path or a string and the sample has multiple members, path it must contain {member_id} (or {input_id} or {output_id}).

(Each) path may contain {sample_id} to be formatted with the sample object.

Source code in src/bioimageio/core/io.py

def save_sample(
    path: Union[Path, str, PerMember[Union[Path, str]]], sample: Sample
) -> None:
    """Save a **sample** to a **path** pattern
    or all sample members in the **path** mapping.

    If **path** is a pathlib.Path or a string and the **sample** has multiple members,
    **path** it must contain `{member_id}` (or `{input_id}` or `{output_id}`).

    (Each) **path** may contain `{sample_id}` to be formatted with the **sample** object.
    """
    if not isinstance(path, collections.abc.Mapping):
        if len(sample.members) < 2 or any(
            m in str(path) for m in ("{member_id}", "{input_id}", "{output_id}")
        ):
            path = {m: path for m in sample.members}
        else:
            raise ValueError(
                f"path {path} must contain '{{member_id}}' for sample with multiple members {list(sample.members)}."
            )

    for m, p in path.items():
        t = sample.members[m]
        p_formatted = Path(
            str(p).format(sample_id=sample.id, member_id=m, input_id=m, output_id=m)
        )
        save_tensor(p_formatted, t)

save_stat

save_stat(stat: Mapping[Union[DatasetMeasure, SampleMeasure], MeasureValue], output: Union[Path, BytesIO]) -> None

Save sample and dataset statistics as a JSON file

Source code in src/bioimageio/core/io.py

def save_stat(
    stat: Mapping[Union[DatasetMeasure, SampleMeasure], MeasureValue],
    output: Union[Path, BytesIO],
) -> None:
    """Save sample and dataset statistics as a JSON file"""

    if isinstance(output, Path):
        ctxt = output.open("wb")
    else:
        ctxt = nullcontext(output)

    with ctxt as out:
        _ = out.write(json.dumps(serialize_stat(stat), indent=2).encode("utf-8"))

save_tensor

save_tensor(path: Union[Path, str], tensor: Tensor) -> None

Source code in src/bioimageio/core/io.py

def save_tensor(path: Union[Path, str], tensor: Tensor) -> None:
    # TODO: save axis meta data

    path = Path(path)
    if not path.suffix:
        raise ValueError(f"No suffix (needed to decide file format) found in {path}")

    extension = path.suffix.lower()
    path.parent.mkdir(exist_ok=True, parents=True)
    if extension == ".npy":
        save_array(path, tensor.to_numpy())
    elif extension in (".h5", ".hdf", ".hdf5"):
        raise NotImplementedError("Saving to h5 with dataset path is not implemented.")
    else:
        removed_singleton_axes: List[AxisId] = []
        remove_singletons = {
            AxisId("batch"): [
                ".tif",
                ".tiff",
            ],  # remove singleton batch dim for tiff files
            **{
                a: [".png", ".jpg", ".jpeg"] for a in tensor.dims
            },  # remove any singleton axis for png and jpg files
        }
        for rm_a, rm_ext in remove_singletons.items():
            if extension in rm_ext and tensor.tagged_shape.get(rm_a) == 1:
                tensor = tensor[{rm_a: 0}]
                removed_singleton_axes.append(rm_a)

        if removed_singleton_axes:
            singleton_axes_msg = f"(with removed singleton axes {list(map(str, removed_singleton_axes))}) "
        else:
            singleton_axes_msg = ""

        logger.info(
            "writing tensor {} {}to {}",
            dict(tensor.tagged_shape),
            singleton_axes_msg,
            path,
        )
        if extension in (".png", ".jpg", ".jpeg") and tensor.dtype in (
            "float32",
            "float64",
        ):
            logger.warning(
                "converting tensor of dtype {} to uint8 for saving as {}",
                tensor.dtype,
                extension,
            )
            tensor = (
                (tensor - (t_min := tensor.data.min()))
                / xr.ufuncs.maximum(tensor.data.max() - t_min, 1e-8)
                * 255
            ).astype("uint8")

        imwrite(path, tensor, extension=extension)

serialize_stat

serialize_stat(stat: Mapping[Union[DatasetMeasure, SampleMeasure], MeasureValue]) -> List[JsonValue]

Serialize a stat mapping to a JSON string

Source code in src/bioimageio/core/io.py

def serialize_stat(
    stat: Mapping[Union[DatasetMeasure, SampleMeasure], MeasureValue],
) -> List[JsonValue]:
    """Serialize a stat mapping to a JSON string"""
    stat_list = _StatList([_StatEntry(measure=k, value=v) for k, v in stat.items()])
    return stat_list.model_dump(mode="json")