resolve cyclic dependency

Author: FynnBe

bioimageio/core/digest_spec.py

@@ -17,10 +17,15 @@
     Union,
 )
 
+import numpy as np
 import xarray as xr
+from loguru import logger
 from numpy.typing import NDArray
 from typing_extensions import Unpack, assert_never
 
+from bioimageio.core.common import MemberId, PerMember, SampleId
+from bioimageio.core.io import load_tensor
+from bioimageio.core.sample import Sample
 from bioimageio.spec._internal.io_utils import HashKwargs, download
 from bioimageio.spec.common import FileSource
 from bioimageio.spec.model import AnyModelDescr, v0_4, v0_5
@@ -32,10 +37,9 @@
 )
 from bioimageio.spec.utils import load_array
 
-from .axis import AxisId, AxisInfo, PerAxis
+from .axis import AxisId, AxisInfo, AxisLike, PerAxis
 from .block_meta import split_multiple_shapes_into_blocks
 from .common import Halo, MemberId, PerMember, SampleId, TotalNumberOfBlocks
-from .io import get_tensor
 from .sample import (
     LinearSampleAxisTransform,
     Sample,
@@ -332,6 +336,27 @@ def get_io_sample_block_metas(
     )
 
 
+def get_tensor(
+    src: Union[Tensor, xr.DataArray, NDArray[Any], Path],
+    ipt: Union[v0_4.InputTensorDescr, v0_5.InputTensorDescr],
+):
+    """helper to cast/load various tensor sources"""
+
+    if isinstance(src, Tensor):
+        return src
+
+    if isinstance(src, xr.DataArray):
+        return Tensor.from_xarray(src)
+
+    if isinstance(src, np.ndarray):
+        return Tensor.from_numpy(src, dims=get_axes_infos(ipt))
+
+    if isinstance(src, Path):
+        return load_tensor(src, axes=get_axes_infos(ipt))
+
+    assert_never(src)
+
+
 def create_sample_for_model(
     model: AnyModelDescr,
     *,
@@ -371,3 +396,47 @@ def create_sample_for_model(
         stat={} if stat is None else stat,
         id=sample_id,
     )
+
+
+def load_sample_for_model(
+    *,
+    model: AnyModelDescr,
+    paths: PerMember[Path],
+    axes: Optional[PerMember[Sequence[AxisLike]]] = None,
+    stat: Optional[Stat] = None,
+    sample_id: Optional[SampleId] = None,
+):
+    """load a single sample from `paths` that can be processed by `model`"""
+
+    if axes is None:
+        axes = {}
+
+    # make sure members are keyed by MemberId, not string
+    paths = {MemberId(k): v for k, v in paths.items()}
+    axes = {MemberId(k): v for k, v in axes.items()}
+
+    model_inputs = {get_member_id(d): d for d in model.inputs}
+
+    if unknown := {k for k in paths if k not in model_inputs}:
+        raise ValueError(f"Got unexpected paths for {unknown}")
+
+    if unknown := {k for k in axes if k not in model_inputs}:
+        raise ValueError(f"Got unexpected axes hints for: {unknown}")
+
+    members: Dict[MemberId, Tensor] = {}
+    for m, p in paths.items():
+        if m not in axes:
+            axes[m] = get_axes_infos(model_inputs[m])
+            logger.warning(
+                "loading paths with {}'s default input axes {} for input '{}'",
+                axes[m],
+                model.id or model.name,
+                m,
+            )
+        members[m] = load_tensor(p, axes[m])
+
+    return Sample(
+        members=members,
+        stat={} if stat is None else stat,
+        id=sample_id or tuple(sorted(paths.values())),
+    )

bioimageio/core/io.py

@@ -1,21 +1,13 @@
 from pathlib import Path
-from typing import Any, Dict, Optional, Sequence, Union
+from typing import Any, Optional, Sequence, Union
 
 import imageio
-import numpy as np
-import xarray as xr
-from loguru import logger
 from numpy.typing import NDArray
-from typing_extensions import assert_never
 
-from bioimageio.spec.model import AnyModelDescr, v0_4, v0_5
 from bioimageio.spec.utils import load_array, save_array
 
 from .axis import Axis, AxisLike
-from .common import MemberId, PerMember, SampleId
-from .digest_spec import get_axes_infos, get_member_id
 from .sample import Sample
-from .stat_measures import Stat
 from .tensor import Tensor
 
 
@@ -40,27 +32,6 @@ def load_tensor(path: Path, axes: Optional[Sequence[AxisLike]] = None) -> Tensor
     return Tensor.from_numpy(array, dims=axes)
 
 
-def get_tensor(
-    src: Union[Tensor, xr.DataArray, NDArray[Any], Path],
-    ipt: Union[v0_4.InputTensorDescr, v0_5.InputTensorDescr],
-):
-    """helper to cast/load various tensor sources"""
-
-    if isinstance(src, Tensor):
-        return src
-
-    if isinstance(src, xr.DataArray):
-        return Tensor.from_xarray(src)
-
-    if isinstance(src, np.ndarray):
-        return Tensor.from_numpy(src, dims=get_axes_infos(ipt))
-
-    if isinstance(src, Path):
-        return load_tensor(src, axes=get_axes_infos(ipt))
-
-    assert_never(src)
-
-
 def save_tensor(path: Path, tensor: Tensor) -> None:
     # TODO: save axis meta data
     data: NDArray[Any] = tensor.data.to_numpy()
@@ -82,47 +53,3 @@ def save_sample(path: Union[Path, str], sample: Sample) -> None:
 
     for m, t in sample.members.items():
         save_tensor(Path(path.format(member_id=m)), t)
-
-
-def load_sample_for_model(
-    *,
-    model: AnyModelDescr,
-    paths: PerMember[Path],
-    axes: Optional[PerMember[Sequence[AxisLike]]] = None,
-    stat: Optional[Stat] = None,
-    sample_id: Optional[SampleId] = None,
-):
-    """load a single sample from `paths` that can be processed by `model`"""
-
-    if axes is None:
-        axes = {}
-
-    # make sure members are keyed by MemberId, not string
-    paths = {MemberId(k): v for k, v in paths.items()}
-    axes = {MemberId(k): v for k, v in axes.items()}
-
-    model_inputs = {get_member_id(d): d for d in model.inputs}
-
-    if unknown := {k for k in paths if k not in model_inputs}:
-        raise ValueError(f"Got unexpected paths for {unknown}")
-
-    if unknown := {k for k in axes if k not in model_inputs}:
-        raise ValueError(f"Got unexpected axes hints for: {unknown}")
-
-    members: Dict[MemberId, Tensor] = {}
-    for m, p in paths.items():
-        if m not in axes:
-            axes[m] = get_axes_infos(model_inputs[m])
-            logger.warning(
-                "loading paths with {}'s default input axes {} for input '{}'",
-                axes[m],
-                model.id or model.name,
-                m,
-            )
-        members[m] = load_tensor(p, axes[m])
-
-    return Sample(
-        members=members,
-        stat={} if stat is None else stat,
-        id=sample_id or tuple(sorted(paths.values())),
-    )