|
1 | 1 | import collections.abc
|
| 2 | +from os import PathLike |
2 | 3 | from pathlib import Path
|
3 | 4 | from typing import Any, Mapping, Optional, Sequence, Union
|
4 | 5 |
|
5 | 6 | import imageio
|
| 7 | +from imageio.v3 import imread, imwrite |
6 | 8 | from loguru import logger
|
7 | 9 | from numpy.typing import NDArray
|
8 | 10 | from pydantic import BaseModel, ConfigDict, TypeAdapter
|
|
16 | 18 | from .tensor import Tensor
|
17 | 19 |
|
18 | 20 |
|
19 |
| -def load_image(path: Path, is_volume: bool) -> NDArray[Any]: |
20 |
| - """load a single image as numpy array""" |
| 21 | +def load_image(path: Path, is_volume: Optional[bool] = None) -> NDArray[Any]: |
| 22 | + """load a single image as numpy array |
| 23 | +
|
| 24 | + Args: |
| 25 | + path: image path |
| 26 | + is_volume: deprecated |
| 27 | + """ |
21 | 28 | ext = path.suffix
|
22 | 29 | if ext == ".npy":
|
23 | 30 | return load_array(path)
|
24 | 31 | else:
|
25 |
| - return imageio.volread(path) if is_volume else imageio.imread(path) |
| 32 | + return imread(path) # pyright: ignore[reportUnknownVariableType] |
26 | 33 |
|
27 | 34 |
|
28 | 35 | def load_tensor(path: Path, axes: Optional[Sequence[AxisLike]] = None) -> Tensor:
|
29 | 36 | # TODO: load axis meta data
|
30 |
| - array = load_image( |
31 |
| - path, |
32 |
| - is_volume=( |
33 |
| - axes is None or sum(Axis.create(a).type != "channel" for a in axes) > 2 |
34 |
| - ), |
35 |
| - ) |
| 37 | + array = load_image(path) |
36 | 38 |
|
37 | 39 | return Tensor.from_numpy(array, dims=axes)
|
38 | 40 |
|
39 | 41 |
|
40 |
| -def save_tensor(path: Path, tensor: Tensor) -> None: |
| 42 | +def save_tensor(path: PathLike[str], tensor: Tensor) -> None: |
41 | 43 | # TODO: save axis meta data
|
42 | 44 |
|
43 | 45 | data: NDArray[Any] = tensor.data.to_numpy()
|
| 46 | + path = Path(path) |
44 | 47 | path.parent.mkdir(exist_ok=True, parents=True)
|
45 | 48 | if path.suffix == ".npy":
|
46 | 49 | save_array(path, data)
|
47 | 50 | else:
|
48 |
| - if singleton_axes := [a for a, s in tensor.tagged_shape.items() if s == 1]: |
49 |
| - tensor = tensor[{a: 0 for a in singleton_axes}] |
50 |
| - singleton_axes_msg = f"(without singleton axes {singleton_axes}) " |
51 |
| - else: |
52 |
| - singleton_axes_msg = "" |
53 |
| - |
54 |
| - # attempt to write a volume or an image with imageio |
55 |
| - error = None |
56 |
| - for d in (data, data.T): |
57 |
| - for write in ( # pyright: ignore[reportUnknownVariableType] |
58 |
| - imageio.volwrite, |
59 |
| - imageio.imwrite, |
60 |
| - ): |
61 |
| - try: |
62 |
| - write(path, d) |
63 |
| - except ValueError as e: |
64 |
| - error = e |
65 |
| - else: |
66 |
| - logger.info( |
67 |
| - "wrote tensor {} {}to {} using imageio.{}", |
68 |
| - dict(tensor.tagged_shape), |
69 |
| - singleton_axes_msg, |
70 |
| - path, |
71 |
| - write.__name__, |
72 |
| - ) |
73 |
| - |
74 |
| - if error is not None: |
75 |
| - raise error |
| 51 | + # if singleton_axes := [a for a, s in tensor.tagged_shape.items() if s == 1]: |
| 52 | + # tensor = tensor[{a: 0 for a in singleton_axes}] |
| 53 | + # singleton_axes_msg = f"(without singleton axes {singleton_axes}) " |
| 54 | + # else: |
| 55 | + singleton_axes_msg = "" |
| 56 | + |
| 57 | + logger.debug( |
| 58 | + "writing tensor {} {}to {}", |
| 59 | + dict(tensor.tagged_shape), |
| 60 | + singleton_axes_msg, |
| 61 | + path, |
| 62 | + ) |
| 63 | + imwrite(path, data) |
76 | 64 |
|
77 | 65 |
|
78 | 66 | def save_sample(path: Union[Path, str, PerMember[Path]], sample: Sample) -> None:
|
|
0 commit comments