ndv.models #

def is_multichannel(self) -> bool:
    """Return whether this mode displays multiple channels.

    If `is_multichannel` is True, then the `channel_axis` slider should be hidden.
    """
    return self in (ChannelMode.COMPOSITE, ChannelMode.RGBA)

@abstractmethod
def get_limits(self, image: npt.NDArray) -> tuple[float, float]:
    """Return the contrast limits for the given image."""

def axis_scales(self) -> dict[Hashable, float]:
    """Return per-axis scale factors from coordinate spacing."""
    scales: dict[Hashable, float] = {}
    for dim in self.dims:
        coord = self.coords.get(dim)
        if coord is None or len(coord) < 2:
            continue
        try:
            values = [float(v) for v in coord]
        except (TypeError, ValueError):
            continue
        if any(not np.isfinite(v) for v in values):
            continue
        diffs = [values[i + 1] - values[i] for i in range(len(values) - 1)]
        if all(abs(d - diffs[0]) < 1e-10 for d in diffs):
            scales[dim] = diffs[0]
    return scales

def channel_names(self, channel_axis: int | None) -> dict[int, str]:
    """Return channel display names from data metadata."""
    if channel_axis is None:
        return {}
    dim = self.dims[channel_axis]
    coords = self.coords.get(dim, None)
    # range coords are not informative, so return empty dict in that case
    # and let resolution logic fallback to generic channel naming
    if coords is None or isinstance(coords, range):
        return {}
    return {
        i: str(v.item() if hasattr(v, "item") else v) for i, v in enumerate(coords)
    }

def clear_cache(self) -> None:
    """Clear any cached properties."""
    if hasattr(self, "axis_map"):
        del self.axis_map

@classmethod
def create(cls, data: ArrayT) -> DataWrapper[ArrayT]:
    """Create a DataWrapper instance for the given data.

    This method will detect all subclasses of DataWrapper and check them in order of
    their `PRIORITY` class variable. The first subclass that
    [`supports`][ndv.DataWrapper.supports] the given data will be used to wrap it.

    !!! tip

        This means that you can subclass DataWrapper to handle new data types.
        Just make sure that your subclass is imported before calling `create`.

    If no subclasses support the data, a `NotImplementedError` is raised.

    If an instance of `DataWrapper` is passed in, it will be returned as-is.
    """
    if isinstance(data, DataWrapper):
        return data

    # check subclasses for support
    # This allows users to define their own DataWrapper subclasses which will
    # be automatically detected (assuming they have been imported by this point)
    for subclass in sorted(_recurse_subclasses(cls), key=lambda x: x.PRIORITY):
        try:
            if subclass.supports(data):
                logger.debug("Using %s to wrap %s", subclass.__name__, type(data))
                return subclass(data)
        except Exception as e:
            warnings.warn(
                f"Error checking DataWrapper subclass {subclass.__name__}: {e}",
                RuntimeWarning,
                stacklevel=2,
            )
    raise NotImplementedError(f"Don't know how to wrap type {type(data)}")

def guess_channel_axis(self) -> Hashable | None:
    """Return the (best guess) axis name for the channel dimension."""
    # for arrays with labeled dimensions,
    # see if any of the dimensions are named "channel"
    sizes = self.sizes()
    if len(sizes) < 3 or min(sizes.values()) > self.MAX_CHANNELS:
        return None

    for dimkey, val in sizes.items():
        if str(dimkey).lower() in self.COMMON_CHANNEL_NAMES:
            if val <= self.MAX_CHANNELS:
                return self.normalize_axis_key(dimkey)

    # otherwise use the smallest dimension as the channel axis
    return min(sizes, key=sizes.get)  # type: ignore [arg-type]

def guess_z_axis(self) -> Hashable | None:
    """Return the (best guess) axis name for the z (3rd spatial) dimension."""
    sizes = self.sizes()
    ch = self.guess_channel_axis()
    for dimkey in sizes:
        if str(dimkey).lower() in self.COMMON_Z_AXIS_NAMES:
            if (normed := self.normalize_axis_key(dimkey)) != ch:
                return normed

    # otherwise return the LAST axis that is neither in the last two dimensions
    # or the channel axis guess
    return next(
        (self.normalize_axis_key(x) for x in reversed(self.dims[:-2]) if x != ch),
        None,
    )

def isel(self, index: Mapping[int, int | slice]) -> np.ndarray:
    """Return a slice of the data as a numpy array.

    `index` will look like (e.g.) `{0: slice(0, 10), 1: 5}`.
    The default implementation converts the index to a tuple of the same length as
    the self.dims, populating missing keys with `slice(None)`, and then slices the
    data array using __getitem__.
    """
    idx = tuple(index.get(k, slice(None)) for k in range(len(self.dims)))
    # this type ignore is asserted in the __init__ method
    # if the data does not support __getitem__, then the DataWrapper subclass will
    # fail to initialize
    return self._asarray(self._data[idx])  # type: ignore [index]

def normalize_axis_key(self, axis: Hashable) -> int:
    """Return positive index for `axis` (which can be +/- int or str label)."""
    try:
        return self.axis_map[axis]
    except KeyError as e:
        ndims = len(self.dims)
        if isinstance(axis, int):
            raise IndexError(
                f"Axis index {axis} out of bounds for data with {ndims} dimensions"
            ) from e
        raise IndexError(f"Axis label {axis} not found in data dimensions") from e

def sizes(self) -> Mapping[Hashable, int]:
    """Return the sizes of the dimensions."""
    return {dim: len(self.coords[dim]) for dim in self.dims}

def summary_info(self) -> str:
    """Return info label with information about the data."""
    package = getattr(self._data, "__module__", "").split(".")[0]
    info = f"{package}.{getattr(type(self._data), '__qualname__', '')}"

    if sizes := self.sizes():
        # if all of the dimension keys are just integers, omit them from size_str
        if all(isinstance(x, int) for x in sizes):
            size_str = repr(tuple(sizes.values()))
        # otherwise, include the keys in the size_str
        else:
            size_str = ", ".join(f"{k}:{v}" for k, v in sizes.items())
            size_str = f"({size_str})"
        info += f" {size_str}"
    if dtype := getattr(self._data, "dtype", ""):
        info += f", {dtype}"
    if nbytes := getattr(self._data, "nbytes", 0):
        info += f", {_human_readable_size(nbytes)}"
    return info

@classmethod
@abstractmethod
def supports(cls, obj: Any) -> TypeGuard[Any]:
    """Return True if this wrapper can handle the given object.

    Any exceptions raised by this method will be suppressed, so it is safe to
    directly import necessary dependencies without a try/except block.
    """

def append(self, value: npt.ArrayLike) -> None:
    """Append a value to the right end of the buffer."""
    if was_full := self.is_full:
        if not self._allow_overwrite:
            raise IndexError("append to a full RingBuffer with overwrite disabled")
        elif not len(self):
            return
        else:
            self._left_index += 1

    self._arr[self._right_index % self._capacity] = value
    self._right_index += 1
    self._fix_indices()
    if not was_full:
        self.resized.emit(len(self))

def appendleft(self, value: npt.ArrayLike) -> None:
    """Append a value to the left end of the buffer."""
    if was_full := self.is_full:
        if not self._allow_overwrite:
            raise IndexError("append to a full RingBuffer with overwrite disabled")
        elif not len(self):
            return
        else:
            self._right_index -= 1

    self._left_index -= 1
    self._fix_indices()
    self._arr[self._left_index] = value
    if not was_full:
        self.resized.emit(len(self))

def extend(self, values: npt.ArrayLike) -> None:
    """Extend the buffer with the given values."""
    values = np.asarray(values)
    lv = len(values)
    if len(self) + lv > self._capacity:
        if not self._allow_overwrite:
            raise IndexError(
                "Extending a RingBuffer such that it would overflow, "
                "with overwrite disabled."
            )
        elif not len(self):
            return
    if lv >= self._capacity:
        # wipe the entire array! - this may not be threadsafe
        self._arr[...] = values[-self._capacity :]
        self._right_index = self._capacity
        self._left_index = 0
        return

    was_full = self.is_full
    ri = self._right_index % self._capacity
    sl1 = np.s_[ri : min(ri + lv, self._capacity)]
    sl2 = np.s_[: max(ri + lv - self._capacity, 0)]
    self._arr[sl1] = values[: sl1.stop - sl1.start]
    self._arr[sl2] = values[sl1.stop - sl1.start :]
    self._right_index += lv

    self._left_index = max(self._left_index, self._right_index - self._capacity)
    self._fix_indices()
    if not was_full:
        self.resized.emit(len(self))

def extendleft(self, values: npt.ArrayLike) -> None:
    """Prepend the buffer with the given values."""
    values = np.asarray(values)
    lv = len(values)
    if len(self) + lv > self._capacity:
        if not self._allow_overwrite:
            raise IndexError(
                "Extending a RingBuffer such that it would overflow, "
                "with overwrite disabled"
            )
        elif not len(self):
            return
    if lv >= self._capacity:
        # wipe the entire array! - this may not be threadsafe
        self._arr[...] = values[: self._capacity]
        self._right_index = self._capacity
        self._left_index = 0
        return

    was_full = self.is_full
    self._left_index -= lv
    self._fix_indices()
    li = self._left_index
    sl1 = np.s_[li : min(li + lv, self._capacity)]
    sl2 = np.s_[: max(li + lv - self._capacity, 0)]
    self._arr[sl1] = values[: sl1.stop - sl1.start]
    self._arr[sl2] = values[sl1.stop - sl1.start :]

    self._right_index = min(self._right_index, self._left_index + self._capacity)
    if not was_full:
        self.resized.emit(len(self))

def pop(self) -> np.ndarray:
    """Pop a value from the right end of the buffer."""
    if len(self) == 0:
        raise IndexError("pop from an empty RingBuffer")
    self._right_index -= 1
    self._fix_indices()
    res = cast("np.ndarray", self._arr[self._right_index % self._capacity])
    self.resized.emit(len(self))
    return res

def popleft(self) -> np.ndarray:
    """Pop a value from the left end of the buffer."""
    if len(self) == 0:
        raise IndexError("pop from an empty RingBuffer")
    res = cast("np.ndarray", self._arr[self._left_index])
    self._left_index += 1
    self._fix_indices()
    self.resized.emit(len(self))
    return res

def append(self, value: npt.ArrayLike) -> None:
    """Append a value to the right end of the buffer."""
    self._ring.append(value)

def appendleft(self, value: npt.ArrayLike) -> None:
    """Append a value to the left end of the buffer."""
    self._ring.appendleft(value)

def axis_scales(self) -> dict[Hashable, float]:
    """Return per-axis scale factors from coordinate spacing."""
    scales: dict[Hashable, float] = {}
    for dim in self.dims:
        coord = self.coords.get(dim)
        if coord is None or len(coord) < 2:
            continue
        try:
            values = [float(v) for v in coord]
        except (TypeError, ValueError):
            continue
        if any(not np.isfinite(v) for v in values):
            continue
        diffs = [values[i + 1] - values[i] for i in range(len(values) - 1)]
        if all(abs(d - diffs[0]) < 1e-10 for d in diffs):
            scales[dim] = diffs[0]
    return scales

def channel_names(self, channel_axis: int | None) -> dict[int, str]:
    """Return channel display names from data metadata."""
    if channel_axis is None:
        return {}
    dim = self.dims[channel_axis]
    coords = self.coords.get(dim, None)
    # range coords are not informative, so return empty dict in that case
    # and let resolution logic fallback to generic channel naming
    if coords is None or isinstance(coords, range):
        return {}
    return {
        i: str(v.item() if hasattr(v, "item") else v) for i, v in enumerate(coords)
    }

def clear_cache(self) -> None:
    """Clear any cached properties."""
    if hasattr(self, "axis_map"):
        del self.axis_map

@classmethod
def create(cls, data: ArrayT) -> DataWrapper[ArrayT]:
    """Create a DataWrapper instance for the given data.

    This method will detect all subclasses of DataWrapper and check them in order of
    their `PRIORITY` class variable. The first subclass that
    [`supports`][ndv.DataWrapper.supports] the given data will be used to wrap it.

    !!! tip

        This means that you can subclass DataWrapper to handle new data types.
        Just make sure that your subclass is imported before calling `create`.

    If no subclasses support the data, a `NotImplementedError` is raised.

    If an instance of `DataWrapper` is passed in, it will be returned as-is.
    """
    if isinstance(data, DataWrapper):
        return data

    # check subclasses for support
    # This allows users to define their own DataWrapper subclasses which will
    # be automatically detected (assuming they have been imported by this point)
    for subclass in sorted(_recurse_subclasses(cls), key=lambda x: x.PRIORITY):
        try:
            if subclass.supports(data):
                logger.debug("Using %s to wrap %s", subclass.__name__, type(data))
                return subclass(data)
        except Exception as e:
            warnings.warn(
                f"Error checking DataWrapper subclass {subclass.__name__}: {e}",
                RuntimeWarning,
                stacklevel=2,
            )
    raise NotImplementedError(f"Don't know how to wrap type {type(data)}")

def guess_channel_axis(self) -> Hashable | None:
    """Return the (best guess) axis name for the channel dimension."""
    # for arrays with labeled dimensions,
    # see if any of the dimensions are named "channel"
    sizes = self.sizes()
    if len(sizes) < 3 or min(sizes.values()) > self.MAX_CHANNELS:
        return None

    for dimkey, val in sizes.items():
        if str(dimkey).lower() in self.COMMON_CHANNEL_NAMES:
            if val <= self.MAX_CHANNELS:
                return self.normalize_axis_key(dimkey)

    # otherwise use the smallest dimension as the channel axis
    return min(sizes, key=sizes.get)  # type: ignore [arg-type]

def guess_z_axis(self) -> Hashable | None:
    """Return the (best guess) axis name for the z (3rd spatial) dimension."""
    sizes = self.sizes()
    ch = self.guess_channel_axis()
    for dimkey in sizes:
        if str(dimkey).lower() in self.COMMON_Z_AXIS_NAMES:
            if (normed := self.normalize_axis_key(dimkey)) != ch:
                return normed

    # otherwise return the LAST axis that is neither in the last two dimensions
    # or the channel axis guess
    return next(
        (self.normalize_axis_key(x) for x in reversed(self.dims[:-2]) if x != ch),
        None,
    )

def isel(self, index: Mapping[int, int | slice]) -> np.ndarray:
    """Return a slice of the data as a numpy array.

    `index` will look like (e.g.) `{0: slice(0, 10), 1: 5}`.
    The default implementation converts the index to a tuple of the same length as
    the self.dims, populating missing keys with `slice(None)`, and then slices the
    data array using __getitem__.
    """
    idx = tuple(index.get(k, slice(None)) for k in range(len(self.dims)))
    # this type ignore is asserted in the __init__ method
    # if the data does not support __getitem__, then the DataWrapper subclass will
    # fail to initialize
    return self._asarray(self._data[idx])  # type: ignore [index]

def normalize_axis_key(self, axis: Hashable) -> int:
    """Return positive index for `axis` (which can be +/- int or str label)."""
    try:
        return self.axis_map[axis]
    except KeyError as e:
        ndims = len(self.dims)
        if isinstance(axis, int):
            raise IndexError(
                f"Axis index {axis} out of bounds for data with {ndims} dimensions"
            ) from e
        raise IndexError(f"Axis label {axis} not found in data dimensions") from e

def pop(self) -> np.ndarray:
    """Pop a value from the right end of the buffer."""
    return self._ring.pop()

def popleft(self) -> np.ndarray:
    """Pop a value from the left end of the buffer."""
    return self._ring.popleft()

def sizes(self) -> Mapping[Hashable, int]:
    """Return the sizes of the dimensions."""
    return {dim: len(self.coords[dim]) for dim in self.dims}

def summary_info(self) -> str:
    """Return info label with information about the data."""
    package = getattr(self._data, "__module__", "").split(".")[0]
    info = f"{package}.{getattr(type(self._data), '__qualname__', '')}"

    if sizes := self.sizes():
        # if all of the dimension keys are just integers, omit them from size_str
        if all(isinstance(x, int) for x in sizes):
            size_str = repr(tuple(sizes.values()))
        # otherwise, include the keys in the size_str
        else:
            size_str = ", ".join(f"{k}:{v}" for k, v in sizes.items())
            size_str = f"({size_str})"
        info += f" {size_str}"
    if dtype := getattr(self._data, "dtype", ""):
        info += f", {dtype}"
    if nbytes := getattr(self._data, "nbytes", 0):
        info += f", {_human_readable_size(nbytes)}"
    return info