Skip to content

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
Suffix

JsonValue module-attribute ¤

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

Suffix module-attribute ¤

Suffix = str

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
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
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
275
276
277
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
240
241
242
243
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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
221
222
223
224
225
226
227
228
229
230
231
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
82
83
84
85
86
87
88
89
90
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
234
235
236
237
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
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
206
207
208
209
210
211
212
213
214
215
216
217
218
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
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
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
198
199
200
201
202
203
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")