digest_spec

Classes:

Name	Description
IO_SampleBlockMeta

Functions:

Name	Description
create_sample_for_model	Create a sample from a single set of input(s) for a specific bioimage.io model
get_axes_infos	get a unified, simplified axis representation from spec axes
get_block_transform	returns how a model's output tensor shapes relates to its input shapes
get_input_halo	returns which halo input tensors need to be divided into blocks with, such that
get_io_sample_block_metas	returns an iterable yielding meta data for corresponding input and output samples
get_member_id	get the normalized tensor ID, usable as a sample member ID
get_member_ids	get normalized tensor IDs to be used as sample member IDs
get_tensor	helper to cast/load various tensor sources
get_test_input_sample
get_test_output_sample	returns a model's test output sample
import_callable	import a callable (e.g. a torch.nn.Module) from a spec node describing it
load_sample_for_model	load a single sample from paths that can be processed by model

Attributes:

Name	Type	Description
TensorSource
get_test_inputs		Deprecated use get_test_input_sample instead
get_test_outputs		Deprecated use get_test_input_sample instead
tmp_dirs_in_use	List[TemporaryDirectory[str]]	keep global reference to temporary directories created during import to delay cleanup

TensorSource module-attribute

TensorSource = Union[Tensor, xr.DataArray, NDArray[Any], Path]

Used by:

• API Reference
  •  digest_spec
    •  create_sample_for_model
    •  get_tensor
  •  prediction
    •  predict
    •  predict_many

get_test_inputs module-attribute

get_test_inputs = get_test_input_sample

Deprecated

use get_test_input_sample instead

get_test_outputs module-attribute

get_test_outputs = get_test_output_sample

Deprecated

use get_test_input_sample instead

tmp_dirs_in_use module-attribute

tmp_dirs_in_use: List[TemporaryDirectory[str]] = []

keep global reference to temporary directories created during import to delay cleanup

IO_SampleBlockMeta

Bases: NamedTuple

              flowchart TD
              bioimageio.core.digest_spec.IO_SampleBlockMeta[IO_SampleBlockMeta]

              

              click bioimageio.core.digest_spec.IO_SampleBlockMeta href "" "bioimageio.core.digest_spec.IO_SampleBlockMeta"
            

Returned by:

• API Reference  digest_spec  get_io_sample_block_metas

Attributes:

Name	Type	Description
input	SampleBlockMeta
output	SampleBlockMeta

input instance-attribute

input: SampleBlockMeta

output instance-attribute

output: SampleBlockMeta

create_sample_for_model

create_sample_for_model(model: AnyModelDescr, *, stat: Optional[Stat] = None, sample_id: SampleId = None, inputs: Union[PerMember[TensorSource], TensorSource]) -> Sample

Create a sample from a single set of input(s) for a specific bioimage.io model

Parameters:

Name	Type	Description	Default
model	AnyModelDescr	a bioimage.io model description	required
stat	Optional[Stat]	dictionary with sample and dataset statistics (may be updated in-place!)	None
inputs	Union[PerMember[TensorSource], TensorSource]	the input(s) constituting a single sample.	required

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

def create_sample_for_model(
    model: AnyModelDescr,
    *,
    stat: Optional[Stat] = None,
    sample_id: SampleId = None,
    inputs: Union[PerMember[TensorSource], TensorSource],
) -> Sample:
    """Create a sample from a single set of input(s) for a specific bioimage.io model

    Args:
        model: a bioimage.io model description
        stat: dictionary with sample and dataset statistics (may be updated in-place!)
        inputs: the input(s) constituting a single sample.
    """

    model_inputs = {get_member_id(d): d for d in model.inputs}
    if isinstance(inputs, collections.abc.Mapping):
        inputs = {MemberId(k): v for k, v in inputs.items()}
    elif len(model_inputs) == 1:
        inputs = {list(model_inputs)[0]: inputs}
    else:
        raise TypeError(
            f"Expected `inputs` to be a mapping with keys {tuple(model_inputs)}"
        )

    if unknown := {k for k in inputs if k not in model_inputs}:
        raise ValueError(f"Got unexpected inputs: {unknown}")

    if missing := {
        k
        for k, v in model_inputs.items()
        if k not in inputs and not (isinstance(v, v0_5.InputTensorDescr) and v.optional)
    }:
        raise ValueError(f"Missing non-optional model inputs: {missing}")

    return Sample(
        members={
            m: get_tensor(inputs[m], ipt)
            for m, ipt in model_inputs.items()
            if m in inputs
        },
        stat={} if stat is None else stat,
        id=sample_id,
    )

get_axes_infos

get_axes_infos(io_descr: Union[v0_4.InputTensorDescr, v0_4.OutputTensorDescr, v0_5.InputTensorDescr, v0_5.OutputTensorDescr]) -> List[AxisInfo]

get a unified, simplified axis representation from spec axes

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

def get_axes_infos(
    io_descr: Union[
        v0_4.InputTensorDescr,
        v0_4.OutputTensorDescr,
        v0_5.InputTensorDescr,
        v0_5.OutputTensorDescr,
    ],
) -> List[AxisInfo]:
    """get a unified, simplified axis representation from spec axes"""
    ret: List[AxisInfo] = []
    for a in io_descr.axes:
        if isinstance(a, v0_5.AxisBase):
            ret.append(AxisInfo.create(Axis(id=a.id, type=a.type)))
        else:
            assert a in ("b", "i", "t", "c", "z", "y", "x")
            ret.append(AxisInfo.create(a))

    return ret

get_block_transform

get_block_transform(model: v0_5.ModelDescr) -> PerMember[PerAxis[Union[LinearSampleAxisTransform, int]]]

returns how a model's output tensor shapes relates to its input shapes

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

def get_block_transform(
    model: v0_5.ModelDescr,
) -> PerMember[PerAxis[Union[LinearSampleAxisTransform, int]]]:
    """returns how a model's output tensor shapes relates to its input shapes"""
    ret: Dict[MemberId, Dict[AxisId, Union[LinearSampleAxisTransform, int]]] = {}
    batch_axis_trf = None
    for ipt in model.inputs:
        for a in ipt.axes:
            if a.type == "batch":
                batch_axis_trf = LinearSampleAxisTransform(
                    axis=a.id, scale=1, offset=0, member=ipt.id
                )
                break
        if batch_axis_trf is not None:
            break
    axis_scales = {
        t.id: {a.id: a.scale for a in t.axes}
        for t in chain(model.inputs, model.outputs)
    }
    for out in model.outputs:
        new_axes: Dict[AxisId, Union[LinearSampleAxisTransform, int]] = {}
        for a in out.axes:
            if a.size is None:
                assert a.type == "batch"
                if batch_axis_trf is None:
                    raise ValueError(
                        "no batch axis found in any input tensor, but output tensor"
                        + f" '{out.id}' has one."
                    )
                s = batch_axis_trf
            elif isinstance(a.size, int):
                s = a.size
            elif isinstance(a.size, v0_5.DataDependentSize):
                s = -1
            elif isinstance(a.size, v0_5.SizeReference):
                s = LinearSampleAxisTransform(
                    axis=a.size.axis_id,
                    scale=axis_scales[a.size.tensor_id][a.size.axis_id] / a.scale,
                    offset=a.size.offset,
                    member=a.size.tensor_id,
                )
            else:
                assert_never(a.size)

            new_axes[a.id] = s

        ret[out.id] = new_axes

    return ret

get_input_halo

get_input_halo(model: v0_5.ModelDescr, output_halo: PerMember[PerAxis[Halo]])

returns which halo input tensors need to be divided into blocks with, such that output_halo can be cropped from their outputs without introducing gaps.

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

def get_input_halo(model: v0_5.ModelDescr, output_halo: PerMember[PerAxis[Halo]]):
    """returns which halo input tensors need to be divided into blocks with, such that
    `output_halo` can be cropped from their outputs without introducing gaps."""
    input_halo: Dict[MemberId, Dict[AxisId, Halo]] = {}
    outputs = {t.id: t for t in model.outputs}
    all_tensors = {**{t.id: t for t in model.inputs}, **outputs}

    for t, th in output_halo.items():
        axes = {a.id: a for a in outputs[t].axes}

        for a, ah in th.items():
            s = axes[a].size
            if not isinstance(s, v0_5.SizeReference):
                raise ValueError(
                    f"Unable to map output halo for {t}.{a} to an input axis"
                )

            axis = axes[a]
            ref_axis = {a.id: a for a in all_tensors[s.tensor_id].axes}[s.axis_id]

            input_halo_left = ah.left * axis.scale / ref_axis.scale
            input_halo_right = ah.right * axis.scale / ref_axis.scale
            assert input_halo_left == int(input_halo_left), f"{input_halo_left} not int"
            assert input_halo_right == int(input_halo_right), (
                f"{input_halo_right} not int"
            )

            input_halo.setdefault(s.tensor_id, {})[a] = Halo(
                int(input_halo_left), int(input_halo_right)
            )

    return input_halo

get_io_sample_block_metas

get_io_sample_block_metas(model: v0_5.ModelDescr, input_sample_shape: PerMember[PerAxis[int]], ns: Mapping[Tuple[MemberId, AxisId], ParameterizedSize_N], batch_size: int = 1) -> Tuple[TotalNumberOfBlocks, Iterable[IO_SampleBlockMeta]]

returns an iterable yielding meta data for corresponding input and output samples

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

def get_io_sample_block_metas(
    model: v0_5.ModelDescr,
    input_sample_shape: PerMember[PerAxis[int]],
    ns: Mapping[Tuple[MemberId, AxisId], ParameterizedSize_N],
    batch_size: int = 1,
) -> Tuple[TotalNumberOfBlocks, Iterable[IO_SampleBlockMeta]]:
    """returns an iterable yielding meta data for corresponding input and output samples"""
    if not isinstance(model, v0_5.ModelDescr):
        raise TypeError(f"get_block_meta() not implemented for {type(model)}")

    block_axis_sizes = model.get_axis_sizes(ns=ns, batch_size=batch_size)
    input_block_shape = {
        t: {aa: s for (tt, aa), s in block_axis_sizes.inputs.items() if tt == t}
        for t in {tt for tt, _ in block_axis_sizes.inputs}
    }
    output_halo = {
        t.id: {
            a.id: Halo(a.halo, a.halo) for a in t.axes if isinstance(a, v0_5.WithHalo)
        }
        for t in model.outputs
    }
    input_halo = get_input_halo(model, output_halo)

    n_input_blocks, input_blocks = split_multiple_shapes_into_blocks(
        input_sample_shape, input_block_shape, halo=input_halo
    )
    block_transform = get_block_transform(model)
    return n_input_blocks, (
        IO_SampleBlockMeta(ipt, ipt.get_transformed(block_transform))
        for ipt in sample_block_meta_generator(
            input_blocks, sample_shape=input_sample_shape, sample_id=None
        )
    )

get_member_id

get_member_id(tensor_description: Union[v0_4.InputTensorDescr, v0_4.OutputTensorDescr, v0_5.InputTensorDescr, v0_5.OutputTensorDescr]) -> MemberId

get the normalized tensor ID, usable as a sample member ID

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

def get_member_id(
    tensor_description: Union[
        v0_4.InputTensorDescr,
        v0_4.OutputTensorDescr,
        v0_5.InputTensorDescr,
        v0_5.OutputTensorDescr,
    ],
) -> MemberId:
    """get the normalized tensor ID, usable as a sample member ID"""

    if isinstance(tensor_description, (v0_4.InputTensorDescr, v0_4.OutputTensorDescr)):
        return MemberId(tensor_description.name)
    elif isinstance(
        tensor_description, (v0_5.InputTensorDescr, v0_5.OutputTensorDescr)
    ):
        return tensor_description.id
    else:
        assert_never(tensor_description)

get_member_ids

get_member_ids(tensor_descriptions: Sequence[Union[v0_4.InputTensorDescr, v0_4.OutputTensorDescr, v0_5.InputTensorDescr, v0_5.OutputTensorDescr]]) -> List[MemberId]

get normalized tensor IDs to be used as sample member IDs

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

def get_member_ids(
    tensor_descriptions: Sequence[
        Union[
            v0_4.InputTensorDescr,
            v0_4.OutputTensorDescr,
            v0_5.InputTensorDescr,
            v0_5.OutputTensorDescr,
        ]
    ],
) -> List[MemberId]:
    """get normalized tensor IDs to be used as sample member IDs"""
    return [get_member_id(descr) for descr in tensor_descriptions]

get_tensor

get_tensor(src: Union[ZipPath, TensorSource], ipt: Union[v0_4.InputTensorDescr, v0_5.InputTensorDescr])

helper to cast/load various tensor sources

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

def get_tensor(
    src: Union[ZipPath, TensorSource],
    ipt: Union[v0_4.InputTensorDescr, v0_5.InputTensorDescr],
):
    """helper to cast/load various tensor sources"""

    if isinstance(src, Tensor):
        return src
    elif isinstance(src, xr.DataArray):
        return Tensor.from_xarray(src)
    elif isinstance(src, np.ndarray):
        return Tensor.from_numpy(src, dims=get_axes_infos(ipt))
    else:
        return load_tensor(src, axes=get_axes_infos(ipt))

get_test_input_sample

get_test_input_sample(model: AnyModelDescr) -> Sample

Used by:

• API Reference  digest_spec  get_test_inputs

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

def get_test_input_sample(model: AnyModelDescr) -> Sample:
    return _get_test_sample(
        model.inputs,
        model.test_inputs if isinstance(model, v0_4.ModelDescr) else model.inputs,
    )

get_test_output_sample

get_test_output_sample(model: AnyModelDescr) -> Sample

returns a model's test output sample

Used by:

• API Reference  digest_spec  get_test_outputs

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

def get_test_output_sample(model: AnyModelDescr) -> Sample:
    """returns a model's test output sample"""
    return _get_test_sample(
        model.outputs,
        model.test_outputs if isinstance(model, v0_4.ModelDescr) else model.outputs,
    )

import_callable

import_callable(node: Union[ArchitectureFromFileDescr, ArchitectureFromLibraryDescr, CallableFromDepencency, CallableFromFile], /, **kwargs: Unpack[HashKwargs]) -> Callable[..., Any]

import a callable (e.g. a torch.nn.Module) from a spec node describing it

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

def import_callable(
    node: Union[
        ArchitectureFromFileDescr,
        ArchitectureFromLibraryDescr,
        CallableFromDepencency,
        CallableFromFile,
    ],
    /,
    **kwargs: Unpack[HashKwargs],
) -> Callable[..., Any]:
    """import a callable (e.g. a torch.nn.Module) from a spec node describing it"""
    if isinstance(node, CallableFromDepencency):
        module = importlib.import_module(node.module_name)
        c = getattr(module, str(node.callable_name))
    elif isinstance(node, ArchitectureFromLibraryDescr):
        module = importlib.import_module(node.import_from)
        c = getattr(module, str(node.callable))
    elif isinstance(node, CallableFromFile):
        c = _import_from_file_impl(node.source_file, str(node.callable_name), **kwargs)
    elif isinstance(node, ArchitectureFromFileDescr):
        c = _import_from_file_impl(node.source, str(node.callable), sha256=node.sha256)
    else:
        assert_never(node)

    if not callable(c):
        raise ValueError(f"{node} (imported: {c}) is not callable")

    return c

load_sample_for_model

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

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

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.debug(
                "loading '{}' from {} with default input axes {} ",
                m,
                p,
                axes[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())),
    )