Data objects

Return these from filters to control how outputs are displayed.

Image

Bases: Data

Image payload with save, load and display helpers.

Wraps a float numpy array with values expected in [0, 1], of shape (H, W) (grayscale) or (H, W, 3) (RGB). Saving and loading go through Pillow or OpenCV, whichever is installed; pass backend="pillow" or backend="opencv" to force one.

Parameters:

Name	Type	Description	Default
data	ndarray	Image array, float values in [0, 1].	required
title	str	Title used when displaying or appended to the file stem when saving.	''

Source code in src/interactive_pipe/data_objects/image.py

class Image(Data):
    """Image payload with save, load and display helpers.

    Wraps a float numpy array with values expected in ``[0, 1]``, of shape
    ``(H, W)`` (grayscale) or ``(H, W, 3)`` (RGB). Saving and loading go
    through Pillow or OpenCV, whichever is installed; pass
    ``backend="pillow"`` or ``backend="opencv"`` to force one.

    Args:
        data: Image array, float values in ``[0, 1]``.
        title: Title used when displaying or appended to the file stem
            when saving.
    """

    def __init__(self, data: np.ndarray, title: str = "") -> None:
        super().__init__(data)
        self.title = title
        self.path = None

    def _set_file_extensions(self):
        self.file_extensions = [".png", ".jpg", ".tif"]

    def _save(self, path: Path, backend=None):
        if path is None:
            raise ValueError("Save requires a path")
        if self.title is not None:
            self.path = self.append_with_stem(path, self.title)
        else:
            self.path = path
        self.save_image(self.data, self.path, backend=backend)

    def _load(self, path: Path, backend=None, title=None) -> np.ndarray:
        if title is not None:
            self.title = title
        self.path = path
        return self.load_image(path, backend=backend)

    @staticmethod
    def save_image(data: np.ndarray, path: Path, precision: int = 8, backend: Optional[str] = None) -> None:
        """Save a ``[0, 1]`` float image array to disk.

        Args:
            data: Image array, float values in ``[0, 1]``.
            path: Destination file path (extension picks the format).
            precision: Bit depth of the written file (8 or 16;
                Pillow supports 8 only).
            backend: ``"pillow"`` or ``"opencv"``; auto-selected when None.
        """
        backend = _resolve_image_backend(backend)
        if backend == IMAGE_BACKEND_OPENCV:
            Image.save_image_cv2(data, path, precision)
        if backend == IMAGE_BACKEND_PILLOW:
            Image.save_image_PIL(data, path, precision)

    @staticmethod
    def rescale_dynamic(data: np.ndarray, precision: int = 8) -> np.ndarray:
        """Scale a ``[0, 1]`` float image to integer dynamic (``[0, 2^precision - 1]``)."""
        if len(data.shape) == 2:
            # Black & white image
            data = np.expand_dims(data, axis=-1)  # add channel dimension
            data = np.repeat(data, 3, axis=-1)  # repeat for RGB
        amplitude = 2**precision - 1
        return np.round(data * amplitude).clip(0, amplitude)

    @staticmethod
    def normalize_dynamic(img: np.ndarray, precision: int = 8) -> np.ndarray:
        """Scale an integer image (``[0, 2^precision - 1]``) to ``[0, 1]`` floats."""
        return img / (2.0**precision - 1)  # scale image data to [0, 1]

    @staticmethod
    def save_image_cv2(data, path: Path, precision=8):
        if not isinstance(path, Path):
            raise TypeError(f"path must be a Path object, got {type(path)}")
        out = Image.rescale_dynamic(data, precision=precision)
        out = out.astype(np.uint8 if precision == 8 else np.uint16)
        out = cv2.cvtColor(out, cv2.COLOR_BGR2RGB)
        cv2.imwrite(str(path), out)

    @staticmethod
    def save_image_PIL(data, path: Path, precision=8):
        if precision != 8:
            raise ValueError(f"PIL backend requires precision=8, got {precision}")
        if not isinstance(path, Path):
            raise TypeError(f"path must be a Path object, got {type(path)}")
        out = Image.rescale_dynamic(data, precision=precision)
        out = out.astype(np.uint8)  # PIL requires image data in uint8 format
        out = PilImage.fromarray(out, "RGB")
        out.save(str(path))

    @staticmethod
    def load_image_cv2(path: Path, precision=8) -> np.ndarray:
        img = cv2.imread(str(path))
        if img is None:
            raise ValueError(f"Could not load image from {path}")
        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)  # type: ignore
        return Image.normalize_dynamic(img, precision=precision)

    @staticmethod
    def load_image_PIL(path: Path, precision=8) -> np.ndarray:
        img = PilImage.open(path)
        return Image.normalize_dynamic(np.array(img), precision=precision)

    @staticmethod
    def load_image(path: Path, precision: int = 8, backend: Optional[str] = None) -> np.ndarray:
        """Load an image file as a ``[0, 1]`` float RGB array.

        Args:
            path: Image file path.
            precision: Bit depth of the stored file used for normalization.
            backend: ``"pillow"`` or ``"opencv"``; auto-selected when None.
        """
        backend = _resolve_image_backend(backend)
        if backend == IMAGE_BACKEND_OPENCV:
            return Image.load_image_cv2(path)
        return Image.load_image_PIL(path, precision)

    def show(self) -> None:
        """Display the image in a matplotlib figure (title includes the shape)."""
        plt.figure()
        plt.imshow(self.data)
        ttl = (f"{self.title} -" if self.title else "") + f"{self.data.shape}"
        plt.title(ttl)
        plt.show()

save_image staticmethod

save_image(data: ndarray, path: Path, precision: int = 8, backend: Optional[str] = None) -> None

Save a [0, 1] float image array to disk.

Parameters:

Name	Type	Description	Default
data	ndarray	Image array, float values in [0, 1].	required
path	Path	Destination file path (extension picks the format).	required
precision	int	Bit depth of the written file (8 or 16; Pillow supports 8 only).	8
backend	Optional[str]	"pillow" or "opencv"; auto-selected when None.	None

Source code in src/interactive_pipe/data_objects/image.py

@staticmethod
def save_image(data: np.ndarray, path: Path, precision: int = 8, backend: Optional[str] = None) -> None:
    """Save a ``[0, 1]`` float image array to disk.

    Args:
        data: Image array, float values in ``[0, 1]``.
        path: Destination file path (extension picks the format).
        precision: Bit depth of the written file (8 or 16;
            Pillow supports 8 only).
        backend: ``"pillow"`` or ``"opencv"``; auto-selected when None.
    """
    backend = _resolve_image_backend(backend)
    if backend == IMAGE_BACKEND_OPENCV:
        Image.save_image_cv2(data, path, precision)
    if backend == IMAGE_BACKEND_PILLOW:
        Image.save_image_PIL(data, path, precision)

rescale_dynamic staticmethod

rescale_dynamic(data: ndarray, precision: int = 8) -> np.ndarray

Scale a [0, 1] float image to integer dynamic ([0, 2^precision - 1]).

Source code in src/interactive_pipe/data_objects/image.py

@staticmethod
def rescale_dynamic(data: np.ndarray, precision: int = 8) -> np.ndarray:
    """Scale a ``[0, 1]`` float image to integer dynamic (``[0, 2^precision - 1]``)."""
    if len(data.shape) == 2:
        # Black & white image
        data = np.expand_dims(data, axis=-1)  # add channel dimension
        data = np.repeat(data, 3, axis=-1)  # repeat for RGB
    amplitude = 2**precision - 1
    return np.round(data * amplitude).clip(0, amplitude)

normalize_dynamic staticmethod

normalize_dynamic(img: ndarray, precision: int = 8) -> np.ndarray

Scale an integer image ([0, 2^precision - 1]) to [0, 1] floats.

Source code in src/interactive_pipe/data_objects/image.py

@staticmethod
def normalize_dynamic(img: np.ndarray, precision: int = 8) -> np.ndarray:
    """Scale an integer image (``[0, 2^precision - 1]``) to ``[0, 1]`` floats."""
    return img / (2.0**precision - 1)  # scale image data to [0, 1]

load_image staticmethod

load_image(path: Path, precision: int = 8, backend: Optional[str] = None) -> np.ndarray

Load an image file as a [0, 1] float RGB array.

Parameters:

Name	Type	Description	Default
path	Path	Image file path.	required
precision	int	Bit depth of the stored file used for normalization.	8
backend	Optional[str]	"pillow" or "opencv"; auto-selected when None.	None

Source code in src/interactive_pipe/data_objects/image.py

@staticmethod
def load_image(path: Path, precision: int = 8, backend: Optional[str] = None) -> np.ndarray:
    """Load an image file as a ``[0, 1]`` float RGB array.

    Args:
        path: Image file path.
        precision: Bit depth of the stored file used for normalization.
        backend: ``"pillow"`` or ``"opencv"``; auto-selected when None.
    """
    backend = _resolve_image_backend(backend)
    if backend == IMAGE_BACKEND_OPENCV:
        return Image.load_image_cv2(path)
    return Image.load_image_PIL(path, precision)

show

show() -> None

Display the image in a matplotlib figure (title includes the shape).

Source code in src/interactive_pipe/data_objects/image.py

def show(self) -> None:
    """Display the image in a matplotlib figure (title includes the shape)."""
    plt.figure()
    plt.imshow(self.data)
    ttl = (f"{self.title} -" if self.title else "") + f"{self.data.shape}"
    plt.title(ttl)
    plt.show()

Curve

Bases: Data

Attributes - .grid - .xlabel - .ylabel - .xlim - .ylim - .title - .curves

It is possible to access a curve by simply using the bracket operator
`.curves[index]` is equivalent to `[index]`

Source code in src/interactive_pipe/data_objects/curves.py

class Curve(Data):
    """
    Attributes
        - `.grid`
        - `.xlabel`
        - `.ylabel`
        - `.xlim`
        - `.ylim`
        - `.title`
        - `.curves`

        It is possible to access a curve by simply using the bracket operator
        `.curves[index]` is equivalent to `[index]`

    """

    def __init__(
        self,
        curves: Union[
            None,
            SingleCurve,
            np.ndarray,
            dict,
            List[Union[SingleCurve, list, tuple, dict, np.ndarray]],
            Tuple[Union[SingleCurve, list, tuple, dict, np.ndarray], ...],
        ],
        xlabel: Optional[str] = None,
        ylabel: Optional[str] = None,
        title: Optional[str] = None,
        grid: Optional[bool] = None,
        xlim: Optional[Tuple[Union[int, float], Union[int, float]]] = None,
        ylim: Optional[Tuple[Union[int, float], Union[int, float]]] = None,
    ) -> None:
        if curves is None:
            curve_list = []
        elif isinstance(curves, SingleCurve):
            curve_list = [curves]
            logging.debug("Empty curve")
        elif isinstance(curves, np.ndarray):
            curve_list = [SingleCurve(y=curves)]
            logging.debug("Empty curve")
        elif isinstance(curves, dict):
            curve_list = [SingleCurve(**curves)]  # type: ignore
            logging.debug("Empty curve")
        elif isinstance(curves, (list, tuple)):
            if all(isinstance(item, SingleCurve) for item in curves):
                curve_list = curves
            else:
                curve_list = []
                for curve in curves:
                    current_curve = self.__create_curve_from_abbreviation(curve)
                    curve_list.append(current_curve)
        data = {
            "curves": curve_list,
            "xlabel": xlabel,
            "ylabel": ylabel,
            "title": title,
            "grid": grid,
            "xlim": xlim,
            "ylim": ylim,
        }
        super().__init__(data)

    def __create_curve_from_abbreviation(self, curve: Union[list, tuple, dict, SingleCurve, np.ndarray]):
        current_curve = None
        if isinstance(curve, (list, tuple)):
            if len(curve) < 2:
                raise ValueError(f"curve must have at least 2 elements (x, y), got {len(curve)} elements: {curve}")
            style = None
            if len(curve) >= 3:
                style = curve[2]
                if style is None or isinstance(style, str):
                    label = None
                    if len(curve) >= 4:
                        label = curve[3]
                        if label is not None:
                            if not isinstance(label, str):
                                raise TypeError(f"label must be a string, got {type(label)}")
                    current_curve = SingleCurve(x=curve[0], y=curve[1], style=style, label=label)
                elif isinstance(style, dict):
                    current_curve = SingleCurve(x=curve[0], y=curve[1], **style)
            else:
                current_curve = SingleCurve(x=curve[0], y=curve[1])
        elif isinstance(curve, dict):
            current_curve = SingleCurve(**curve)
        elif isinstance(curve, SingleCurve):
            current_curve = curve
        elif isinstance(curve, np.ndarray):
            current_curve = SingleCurve(x=None, y=curve)
        if current_curve is None:
            raise ValueError(f"could not create a single curve from abbreviation: {curve}")
        return current_curve

    # .grid
    # ---------------------------------------

    @property
    def grid(self) -> bool:
        return self.data["grid"]

    @grid.setter
    def grid(self, grid: bool):
        self.data["grid"] = grid

    # .title
    # ---------------------------------------
    @property
    def title(self) -> str:
        return self.data["title"]

    @title.setter
    def title(self, title: str):
        self.data["title"] = title

    # .xlim / .ylim
    # ---------------------------------------
    @property
    def xlim(self) -> Optional[Tuple[Union[int, float], Union[int, float]]]:
        return self.data["xlim"]

    @xlim.setter
    def xlim(self, xlim: Optional[Tuple[Union[int, float], Union[int, float]]]):
        self.data["xlim"] = xlim

    @property
    def ylim(self) -> Optional[Tuple[Union[int, float], Union[int, float]]]:
        return self.data["ylim"]

    @ylim.setter
    def ylim(self, ylim: Optional[Tuple[Union[int, float], Union[int, float]]]):
        self.data["ylim"] = ylim

    # .xlabel / .ylabel
    # ---------------------------------------
    @property
    def xlabel(self) -> str:
        return self.data["xlabel"]

    @xlabel.setter
    def xlabel(self, xlabel: str):
        self.data["xlabel"] = xlabel

    @property
    def ylabel(self) -> str:
        return self.data["ylabel"]

    @ylabel.setter
    def ylabel(self, ylabel: str):
        self.data["ylabel"] = ylabel

    # .curves
    # ---------------------------------------
    @property
    def curves(self) -> List[SingleCurve]:
        return self.data["curves"]

    @curves.setter
    def curves(self, curves: List[SingleCurve]):
        self.data["curves"] = curves

    # brackets []
    # ---------------------------------------
    def __getitem__(self, key: int) -> SingleCurve:
        if isinstance(key, slice):
            return [self.data["curves"][idx] for idx in range(*key.indices(len(self.data["curves"])))]
        if not isinstance(key, int):
            raise TypeError(f"key must be an integer, got {type(key)}")
        if key >= len(self.data["curves"]):
            raise IndexError(f"curve index {key} out of range (max: {len(self.data['curves']) - 1})")
        return self.data["curves"][key]

    def __setitem__(self, key: int, value):
        if isinstance(key, slice):
            for lin_index, idx in enumerate(range(*key.indices(len(value)))):
                if idx >= len(self.data["curves"]):
                    raise IndexError(f"curve index {idx} out of range (max: {len(self.data['curves']) - 1})")
                self.data["curves"][idx] = value[lin_index]
            return
        if not isinstance(key, int):
            raise TypeError(f"key must be an integer, got {type(key)}")
        if key >= len(self.data["curves"]):
            raise IndexError(f"curve index {key} out of range (max: {len(self.data['curves']) - 1})")
        if not isinstance(value, SingleCurve):
            raise TypeError(f"value must be a SingleCurve instance, got {type(value)}")
        self.data["curves"][key] = value

    def append(self, new_curve: SingleCurve):
        self.data["curves"].append(new_curve)

    def prepend(self, new_curve: SingleCurve):
        self.data["curves"].insert(0, new_curve)

    def _set_file_extensions(self):
        self.file_extensions = [".png", ".jpg", ".pkl"]

    def _load(self, path: Path) -> dict:
        if path.suffix != ".pkl":
            raise ValueError(f"Unsupported file extension: {path.suffix}, expected .pkl")
        self.path = path
        if path.suffix == ".pkl":
            # App-controlled round-trip of files written by save_binary
            data = Data.load_binary(path, allow_pickle=True)
        return data

    def _save(self, path: Path, backend=None, figsize=None):
        if path is None:
            raise ValueError("Save requires a path")
        self.path = path
        if path.suffix == ".pkl":
            Data.save_binary(self.data, path)
        elif path.suffix in [".png", ".jpg"]:
            self.save_figure(self.data, self.path, backend=backend, figsize=figsize)

    @staticmethod
    def save_figure(data, path: Path, backend=None, figsize=None):
        if backend is None:
            backend = SIGNAL_BACKEND_MPL
        if backend not in [SIGNAL_BACKEND_MPL]:
            raise ValueError(f"backend must be {SIGNAL_BACKEND_MPL}, got {backend}")
        if backend == SIGNAL_BACKEND_MPL:
            Curve.save_figure_mpl(data, path, figsize=figsize)

    @staticmethod
    def save_figure_mpl(data, path: Path, figsize=None):
        fig, ax = plt.subplots(figsize=figsize)
        Curve._plot_curve(data, ax=ax)
        plt.savefig(path)
        plt.close(fig)

    def create_plot(self, ax=None):
        plt_obj = Curve._plot_curve(self.data, ax=ax)
        return plt_obj

    def update_plot(self, plt_obj, ax=None):
        Curve._update_plot(self.data, plt_obj, ax=ax)

    @staticmethod
    def _update_plot(data, plt_obj, ax=None):
        legend = ax.get_legend() if ax is not None else None
        texts = []
        if legend is not None:
            texts = legend.get_texts()
        linear_index = 0
        for curve_idx, curve in enumerate(data["curves"]):
            if curve.data.get("x", None) is not None:
                plt_obj[curve_idx][0].set_xdata(curve.data["x"])
            if curve.data.get("y", None) is not None:
                plt_obj[curve_idx][0].set_ydata(curve.data["y"])
            if curve.data["label"] is not None:
                texts[linear_index].set_text(curve.data["label"])
                linear_index += 1

        if ax is not None:
            if data.get("title", None) is not None:
                ax.set_title(data["title"])
            if data.get("xlabel", None) is not None:
                ax.set_xlabel(data["xlabel"])
            if data.get("ylabel", None) is not None:
                ax.set_ylabel(data["ylabel"])
            if data.get("grid", None) is not None:
                ax.grid(data["grid"])
            if data.get("xlim", None) is not None:
                ax.set_xlim(data["xlim"])
            if data.get("ylim", None) is not None:
                ax.set_ylim(data["ylim"])

        return plt_obj

    @staticmethod
    def _plot_curve(data, ax=None):
        legend_flag = False
        plt_obj = []
        for curve in data["curves"]:
            if curve.data.get("x", None) is not None:
                inps = [
                    curve.data["x"],
                    curve.data["y"],
                ]
            else:
                inps = [
                    curve.data["y"],
                ]
            has_style_fmt = curve.data.get("style", None) is not None
            if has_style_fmt:
                inps.append(curve.data["style"])
            if curve.data.get("label", None) is not None:
                legend_flag = True

            # Build kwargs - avoid redundant linestyle when style fmt string is provided
            # The fmt string (e.g., "r-o") already contains linestyle info, so passing
            # linestyle keyword would be redundant and trigger a warning
            plot_kwargs = {
                "label": curve.data.get("label", None),
                "alpha": curve.data.get("alpha", None),
            }
            # Only add linestyle if style fmt string is not provided (to avoid redundancy warning)
            if not has_style_fmt:
                linestyle = curve.data.get("linestyle", None)
                if linestyle is not None:
                    plot_kwargs["linestyle"] = linestyle

            linewidth = curve.data.get("linewidth", None)
            if linewidth is not None:
                plot_kwargs["linewidth"] = linewidth

            markersize = curve.data.get("markersize", None)
            if markersize is not None:
                plot_kwargs["markersize"] = markersize

            if ax is not None:
                plt_obj.append(ax.plot(*inps, **plot_kwargs))
        if ax is not None:
            if legend_flag:
                ax.legend(loc="upper right")
            if data.get("title", None) is not None:
                ax.set_title(data["title"])
            if data.get("xlabel", None) is not None:
                ax.set_xlabel(data["xlabel"])
            if data.get("ylabel", None) is not None:
                ax.set_ylabel(data["ylabel"])
            if data.get("grid", None) is not None:
                ax.grid(data["grid"])
            if data.get("xlim", None) is not None:
                ax.set_xlim(data["xlim"])
            if data.get("ylim", None) is not None:
                ax.set_ylim(data["ylim"])
        return plt_obj

    def show(self, figsize=None):
        fig, ax = plt.subplots(figsize=figsize)
        Curve._plot_curve(self.data, ax=ax)
        plt.show()

SingleCurve

Bases: Data

SingleCurve are 2D signals you can plot, defined by x and y numpy arrays Attributes: .x .y .label .alpha .style You'd normally use matplotlib.plot(x, y, label) - .from_file("data.csv") containing x,y columns - save as: - .csv (not style, label or markers exported) - .pkl (dict)

Source code in src/interactive_pipe/data_objects/curves.py

class SingleCurve(Data):
    """SingleCurve are 2D signals you can plot, defined by x and y numpy arrays
    Attributes:
        .x
        .y
        .label
        .alpha
        .style
    You'd normally use matplotlib.plot(x, y, label)
    - .from_file("data.csv") containing x,y columns
    - save as:
        - .csv (not style, label or markers exported)
        - .pkl (dict)
    """

    def __init__(
        self,
        x: Optional[Union[List[np.ndarray], np.ndarray]] = None,
        y: Optional[Union[List[np.ndarray], np.ndarray]] = None,
        style: Optional[str] = None,
        label: Optional[str] = None,
        linestyle: Optional[str] = None,
        linewidth: Optional[int] = None,
        markersize: Optional[int] = None,
        alpha: Optional[float] = None,
    ):
        if x is not None and y is not None:
            if len(x) != len(y):
                raise ValueError(f"x and y lengths must match: x has {len(x)} elements, y has {len(y)} elements")
        data = {
            "x": x,
            "y": y,
            "label": label,
            "style": style,
            "linestyle": linestyle,
            "linewidth": linewidth,
            "markersize": markersize,
            "alpha": alpha,
        }
        super().__init__(data)

    @property
    def x(self) -> np.ndarray:
        return self.data["x"]

    @x.setter
    def x(self, x: np.ndarray):
        self.data["x"] = x

    @property
    def y(self) -> np.ndarray:
        return self.data["y"]

    @y.setter
    def y(self, y: np.ndarray):
        self.data["y"] = y

    @property
    def label(self) -> str:
        return self.data["label"]

    @label.setter
    def label(self, label):
        if label is not None and not isinstance(label, str):
            raise TypeError(f"label must be a string or None, got {type(label)}")
        self.data["label"] = label

    @property
    def style(self) -> str:
        return self.data["style"]

    @style.setter
    def style(self, style):
        if style is not None and not isinstance(style, str):
            raise TypeError(f"style must be a string or None, got {type(style)}")
        self.data["style"] = style

    @property
    def alpha(self) -> float:
        return self.data["alpha"]

    @alpha.setter
    def alpha(self, alpha):
        if alpha is None:
            self.data["alpha"] = None
        else:
            if not isinstance(alpha, (float, int)):
                raise TypeError(f"alpha must be a number, got {type(alpha)}")
            if not (0 <= alpha <= 1):
                raise ValueError(f"alpha must be between [0, 1], got {alpha}")
            self.data["alpha"] = alpha

    def _set_file_extensions(self):
        self.file_extensions = [".png", ".jpg", ".csv", ".pkl"]

    def _save(
        self,
        path: Path,
        backend=None,
        figsize=None,
        xlabel: Optional[str] = None,
        ylabel: Optional[str] = None,
        title: Optional[str] = None,
        grid: Optional[bool] = None,
    ):
        if path is None:
            raise ValueError("Save requires a path")
        self.path = path
        Curve([self], xlabel=xlabel, ylabel=ylabel, title=title, grid=grid)
        if path.suffix == ".csv":
            SingleCurve.save_tabular(self.data, path)
        elif path.suffix == ".pkl":
            Data.save_binary(self.data, path)
        elif path.suffix in [".png", "jpg"]:
            Curve.save_figure(data=self.data, path=path, backend=backend, figsize=figsize)

    def _load(
        self,
        path: Path,
        style: Optional[str] = None,
        label: Optional[str] = None,
        linestyle: Optional[str] = None,
        linewidth: Optional[int] = None,
        markersize: Optional[int] = None,
        alpha: Optional[float] = None,
    ) -> dict:
        if path.suffix not in [".csv", ".pkl"]:
            raise ValueError(f"Unsupported file extension: {path.suffix}, expected .csv or .pkl")
        self.path = path
        if path.suffix == ".csv":
            df = pd.read_csv(self.path)
            data = {
                "x": df["x"],
                "y": df["y"],
                "label": label,
                "style": style,
                "linestyle": linestyle,
                "linewidth": linewidth,
                "markersize": markersize,
                "alpha": alpha,
            }
        elif path.suffix == ".pkl":
            # App-controlled round-trip of files written by save_binary
            data = Data.load_binary(path, allow_pickle=True)
            if style is not None:
                data["style"] = style
            if label is not None:
                data["label"] = label
            if linestyle is not None:
                data["linestyle"] = linestyle
            if linewidth is not None:
                data["linewidth"] = linewidth
            if markersize is not None:
                data["markersize"] = markersize
            if alpha is not None:
                data["alpha"] = alpha
        return data

    def show(
        self,
        figsize=None,
        xlabel: Optional[str] = None,
        ylabel: Optional[str] = None,
        title: Optional[str] = None,
        grid: Optional[bool] = None,
    ):
        crv = Curve([self], xlabel=xlabel, ylabel=ylabel, title=title, grid=grid)
        crv.show(figsize=figsize)

    def as_dataframe(self):  # -> pd.DataFrame
        df = SingleCurve.dataframe_from_data(self.data)
        self.df = df
        return df

    @staticmethod
    def dataframe_from_data(data):  # -> pd.DataFrame
        if SIGNAL_BACKEND_PD not in signal_backends:
            raise RuntimeError("pandas backend is not available. Install pandas to use this feature.")
        df = pd.DataFrame.from_dict(dict(x=data["x"], y=data["y"]))
        return df

    @staticmethod
    def save_tabular(data, path: Path):
        if path.suffix != ".csv":
            raise ValueError(f"save_tabular requires .csv extension, got {path.suffix}")
        df = SingleCurve.dataframe_from_data(data)
        df.to_csv(path, index=False)

Table

Bases: Data

Tabular data for display in interactive_pipe.

Works without pandas. Pandas features (DataFrame input, csv export) available when pandas is installed.

Supports headerless tables for displaying raw matrices by setting columns=None when using numpy array input.

Attributes:

Name	Type	Description
columns	List[str]	List of column names (empty strings for headerless).
values	List[List]	2D list of values (row-major).
title	Optional[str]	Optional title.
precision	int	Float formatting precision.

Source code in src/interactive_pipe/data_objects/table.py

class Table(Data):
    """Tabular data for display in interactive_pipe.

    Works without pandas. Pandas features (DataFrame input, csv export)
    available when pandas is installed.

    Supports headerless tables for displaying raw matrices by setting
    columns=None when using numpy array input.

    Attributes:
        columns: List of column names (empty strings for headerless).
        values: 2D list of values (row-major).
        title: Optional title.
        precision: Float formatting precision.
    """

    def __init__(
        self,
        data: Union[Dict[str, List], List[Dict], np.ndarray, None] = None,
        columns: Optional[List[str]] = None,
        title: Optional[str] = None,
        precision: int = 2,
    ):
        if data is None:
            # Allow creation from Path (handled by parent class)
            data_dict = {"columns": [], "values": []}
        elif isinstance(data, Path):
            # Handled by parent class
            super().__init__(data, columns=columns, title=title, precision=precision)
            return
        elif PANDAS_AVAILABLE and isinstance(data, pd.DataFrame):
            # Handle pandas DataFrame input
            data_dict = self._normalize_dataframe(data)
        else:
            data_dict = self._normalize_data(data, columns)

        internal_data = {
            "columns": data_dict["columns"],
            "values": data_dict["values"],
            "title": title,
            "precision": precision,
        }
        super().__init__(internal_data)

    def _normalize_data(
        self,
        data: Union[Dict[str, List], List[Dict], np.ndarray],
        columns: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Convert various input formats to internal format."""
        if isinstance(data, dict):
            # Dict of lists: {"col1": [1,2,3], "col2": [4,5,6]}
            if not all(isinstance(v, list) for v in data.values()):
                raise TypeError("Dict input must have list values")
            col_names = list(data.keys())
            if not col_names:
                return {"columns": [], "values": []}
            # Check all lists have same length
            lengths = [len(v) for v in data.values()]
            if len(set(lengths)) > 1:
                raise ValueError(f"All columns must have the same length. Got lengths: {lengths}")
            # Convert to row-major format
            num_rows = lengths[0]
            values = [[data[col][row] for col in col_names] for row in range(num_rows)]
            return {"columns": col_names, "values": values}

        elif isinstance(data, list):
            if not data:
                return {"columns": [], "values": []}
            # List of dicts: [{"col1": 1, "col2": 4}, {"col1": 2, "col2": 5}]
            if isinstance(data[0], dict):
                col_names = list(data[0].keys())
                # Check all dicts have same keys
                for i, row_dict in enumerate(data):
                    if set(row_dict.keys()) != set(col_names):
                        raise ValueError(
                            f"Row {i} has different keys. Expected {col_names}, got {list(row_dict.keys())}"
                        )
                values = [[row_dict[col] for col in col_names] for row_dict in data]
                return {"columns": col_names, "values": values}
            else:
                raise TypeError(
                    "List input must be a list of dictionaries. For 2D arrays, use numpy array with columns parameter."
                )

        elif isinstance(data, np.ndarray):
            # 2D numpy array with optional columns parameter
            if len(data.shape) != 2:
                raise ValueError(f"Numpy array must be 2D, got shape {data.shape}")
            if columns is None:
                # Create empty column names for headerless table
                columns = [""] * data.shape[1]
            if columns is not None and len(columns) != data.shape[1]:
                raise ValueError(f"Number of columns ({len(columns)}) must match array width ({data.shape[1]})")
            values = data.tolist()
            return {"columns": columns, "values": values}

        else:
            raise TypeError(
                f"Unsupported data type: {type(data)}. "
                "Supported: dict of lists, list of dicts, 2D numpy array"
                + (", pandas DataFrame" if PANDAS_AVAILABLE else "")
            )

    def _normalize_dataframe(self, df: "pd.DataFrame") -> Dict[str, Any]:
        """Convert pandas DataFrame to internal format."""
        _require_pandas("DataFrame input")
        col_names = list(df.columns)
        values = df.values.tolist()
        return {"columns": col_names, "values": values}

    @property
    def columns(self) -> List[str]:
        return self.data["columns"]

    @columns.setter
    def columns(self, columns: List[str]):
        if not isinstance(columns, list):
            raise TypeError(f"columns must be a list, got {type(columns)}")
        if not all(isinstance(c, str) for c in columns):
            raise TypeError("All column names must be strings")
        # Validate consistency with existing row width
        if self.values and len(columns) != len(self.values[0]):
            raise ValueError(f"Number of columns ({len(columns)}) must match row width ({len(self.values[0])})")
        self.data["columns"] = columns

    @property
    def values(self) -> List[List]:
        return self.data["values"]

    @values.setter
    def values(self, values: List[List]):
        if not isinstance(values, list):
            raise TypeError(f"values must be a list, got {type(values)}")
        self.data["values"] = values

    @property
    def title(self) -> Optional[str]:
        return self.data["title"]

    @title.setter
    def title(self, title: Optional[str]):
        if title is not None and not isinstance(title, str):
            raise TypeError(f"title must be a string or None, got {type(title)}")
        self.data["title"] = title

    @property
    def precision(self) -> int:
        return self.data["precision"]

    @precision.setter
    def precision(self, precision: int):
        if not isinstance(precision, int):
            raise TypeError(f"precision must be an integer, got {type(precision)}")
        if precision < 0:
            raise ValueError(f"precision must be non-negative, got {precision}")
        self.data["precision"] = precision

    def __len__(self) -> int:
        """Return the number of rows in the table."""
        return len(self.values)

    def __iter__(self):
        """Iterate over rows as dictionaries."""
        for i in range(len(self.values)):
            yield dict(zip(self.columns, self.values[i]))

    def __getitem__(self, key: Union[int, str, slice]):
        """Access data by row index, column name, or slice."""
        if isinstance(key, int):
            # Return row (support negative indexing)
            if key < -len(self.values) or key >= len(self.values):
                raise IndexError(f"Row index {key} out of range (length: {len(self.values)})")
            return dict(zip(self.columns, self.values[key]))
        elif isinstance(key, str):
            # Return column
            if key not in self.columns:
                raise KeyError(f"Column '{key}' not found. Available: {self.columns}")
            col_idx = self.columns.index(key)
            return [row[col_idx] for row in self.values]
        elif isinstance(key, slice):
            # Return slice of rows
            rows = self.values[key]
            return [dict(zip(self.columns, row)) for row in rows]
        else:
            raise TypeError(f"Key must be int, str, or slice, got {type(key)}")

    def _set_file_extensions(self):
        self.file_extensions = [".csv", ".pkl"]

    def as_dataframe(self) -> "pd.DataFrame":
        """Convert Table to pandas DataFrame.

        Returns:
            Table data as a pandas DataFrame

        Raises:
            RuntimeError: If pandas is not installed
        """
        _require_pandas("as_dataframe()")
        return pd.DataFrame(self.values, columns=self.columns)  # type: ignore

    def _save(self, path: Path, **kwargs):
        if path.suffix == ".pkl":
            Data.save_binary(self.data, path)
        elif path.suffix == ".csv":
            _require_pandas("CSV export")
            df = self.as_dataframe()
            df.to_csv(path, index=False)
        else:
            raise ValueError(f"Unsupported file extension: {path.suffix}. Supported: .pkl, .csv")

    def _load(self, path: Path, **kwargs) -> Dict[str, Any]:
        if path.suffix == ".pkl":
            # App-controlled round-trip of files written by save_binary
            data = Data.load_binary(path, allow_pickle=True)
            return data
        elif path.suffix == ".csv":
            _require_pandas("CSV import")
            df = pd.read_csv(path)
            data = {
                "columns": list(df.columns),
                "values": df.values.tolist(),
                "title": kwargs.get("title", None),
                "precision": kwargs.get("precision", 2),
            }
            return data
        else:
            raise ValueError(f"Unsupported file extension: {path.suffix}. Supported: .pkl, .csv")

    def _format_values(self) -> List[List[str]]:
        """Format values with precision for display."""
        formatted = []
        for row in self.values:
            formatted_row = []
            for val in row:
                if isinstance(val, float):
                    formatted_row.append(f"{val:.{self.precision}f}")
                else:
                    formatted_row.append(str(val))
            formatted.append(formatted_row)
        return formatted

    def create_table(self, ax: Optional[Any] = None) -> Any:
        """Create matplotlib table, returns table object.

        Args:
            ax: matplotlib Axes object (optional, will create if None)

        Returns:
            matplotlib Table object
        """
        try:
            import matplotlib.pyplot as plt
        except ImportError:
            raise RuntimeError("Matplotlib is required for table rendering. Install with: pip install matplotlib")

        if ax is None:
            fig, ax = plt.subplots()

        ax.axis("off")

        # Check if this is a headerless table (all columns are empty strings)
        has_headers = any(col != "" for col in self.columns)

        if has_headers:
            table = ax.table(
                cellText=self._format_values(),
                colLabels=self.columns,
                loc="center",
                cellLoc="center",
            )
        else:
            # Headerless table - don't show column labels
            table = ax.table(
                cellText=self._format_values(),
                loc="center",
                cellLoc="center",
            )

        table.auto_set_font_size(False)
        table.set_fontsize(9)
        table.scale(1, 2)  # Make cells taller for better readability

        if self.title:
            ax.set_title(self.title, pad=20)

        return table

    def update_table(self, _table_obj: Optional[Any] = None, ax: Optional[Any] = None) -> Any:
        """Update existing matplotlib table.

        Note: matplotlib tables don't update well, so we clear and recreate.

        Args:
            _table_obj: Previous table object (ignored, kept for API consistency
                with Curve.update_plot signature)
            ax: matplotlib Axes object

        Returns:
            matplotlib Table object
        """
        if ax is None:
            raise ValueError("ax parameter is required for update_table()")
        ax.clear()
        return self.create_table(ax=ax)

__len__

__len__() -> int

Return the number of rows in the table.

Source code in src/interactive_pipe/data_objects/table.py

def __len__(self) -> int:
    """Return the number of rows in the table."""
    return len(self.values)

__iter__

__iter__()

Iterate over rows as dictionaries.

Source code in src/interactive_pipe/data_objects/table.py

def __iter__(self):
    """Iterate over rows as dictionaries."""
    for i in range(len(self.values)):
        yield dict(zip(self.columns, self.values[i]))

__getitem__

__getitem__(key: Union[int, str, slice])

Access data by row index, column name, or slice.

Source code in src/interactive_pipe/data_objects/table.py

def __getitem__(self, key: Union[int, str, slice]):
    """Access data by row index, column name, or slice."""
    if isinstance(key, int):
        # Return row (support negative indexing)
        if key < -len(self.values) or key >= len(self.values):
            raise IndexError(f"Row index {key} out of range (length: {len(self.values)})")
        return dict(zip(self.columns, self.values[key]))
    elif isinstance(key, str):
        # Return column
        if key not in self.columns:
            raise KeyError(f"Column '{key}' not found. Available: {self.columns}")
        col_idx = self.columns.index(key)
        return [row[col_idx] for row in self.values]
    elif isinstance(key, slice):
        # Return slice of rows
        rows = self.values[key]
        return [dict(zip(self.columns, row)) for row in rows]
    else:
        raise TypeError(f"Key must be int, str, or slice, got {type(key)}")

as_dataframe

as_dataframe() -> pd.DataFrame

Convert Table to pandas DataFrame.

Returns:

Type	Description
DataFrame	Table data as a pandas DataFrame

Raises:

Type	Description
RuntimeError	If pandas is not installed

Source code in src/interactive_pipe/data_objects/table.py

def as_dataframe(self) -> "pd.DataFrame":
    """Convert Table to pandas DataFrame.

    Returns:
        Table data as a pandas DataFrame

    Raises:
        RuntimeError: If pandas is not installed
    """
    _require_pandas("as_dataframe()")
    return pd.DataFrame(self.values, columns=self.columns)  # type: ignore

create_table

create_table(ax: Optional[Any] = None) -> Any

Create matplotlib table, returns table object.

Parameters:

Name	Type	Description	Default
ax	Optional[Any]	matplotlib Axes object (optional, will create if None)	None

Returns:

Type	Description
Any	matplotlib Table object

Source code in src/interactive_pipe/data_objects/table.py

def create_table(self, ax: Optional[Any] = None) -> Any:
    """Create matplotlib table, returns table object.

    Args:
        ax: matplotlib Axes object (optional, will create if None)

    Returns:
        matplotlib Table object
    """
    try:
        import matplotlib.pyplot as plt
    except ImportError:
        raise RuntimeError("Matplotlib is required for table rendering. Install with: pip install matplotlib")

    if ax is None:
        fig, ax = plt.subplots()

    ax.axis("off")

    # Check if this is a headerless table (all columns are empty strings)
    has_headers = any(col != "" for col in self.columns)

    if has_headers:
        table = ax.table(
            cellText=self._format_values(),
            colLabels=self.columns,
            loc="center",
            cellLoc="center",
        )
    else:
        # Headerless table - don't show column labels
        table = ax.table(
            cellText=self._format_values(),
            loc="center",
            cellLoc="center",
        )

    table.auto_set_font_size(False)
    table.set_fontsize(9)
    table.scale(1, 2)  # Make cells taller for better readability

    if self.title:
        ax.set_title(self.title, pad=20)

    return table

update_table

update_table(_table_obj: Optional[Any] = None, ax: Optional[Any] = None) -> Any

Update existing matplotlib table.

Note: matplotlib tables don't update well, so we clear and recreate.

Parameters:

Name	Type	Description	Default
_table_obj	Optional[Any]	Previous table object (ignored, kept for API consistency with Curve.update_plot signature)	None
ax	Optional[Any]	matplotlib Axes object	None

Returns:

Type	Description
Any	matplotlib Table object

Source code in src/interactive_pipe/data_objects/table.py

def update_table(self, _table_obj: Optional[Any] = None, ax: Optional[Any] = None) -> Any:
    """Update existing matplotlib table.

    Note: matplotlib tables don't update well, so we clear and recreate.

    Args:
        _table_obj: Previous table object (ignored, kept for API consistency
            with Curve.update_plot signature)
        ax: matplotlib Axes object

    Returns:
        matplotlib Table object
    """
    if ax is None:
        raise ValueError("ax parameter is required for update_table()")
    ax.clear()
    return self.create_table(ax=ax)