Series

series

Series-level data loader.

Series

Data and predictions for a single image series.

Attributes:

Name	Type	Description
series_name	str	Unique identifier for the series.
h5_path	Optional[str]	Optional path to the HDF5-formatted image series.
primary_path	Optional[str]	Optional path to the primary root predictions file. At least one of the primary, lateral, or crown paths must be provided.
lateral_path	Optional[str]	Optional path to the lateral root predictions file. At least one of the primary, lateral, or crown paths must be provided.
crown_path	Optional[str]	Optional path to the crown predictions file. At least one of the primary, lateral, or crown paths must be provided.
primary_labels	Optional[Labels]	Optional sio.Labels corresponding to the primary root predictions.
lateral_labels	Optional[Labels]	Optional sio.Labels corresponding to the lateral root predictions.
crown_labels	Optional[Labels]	Optional sio.Labels corresponding to the crown predictions.
video	Optional[Video]	Optional sio.Video corresponding to the image series.
csv_path	Optional[str]	Optional path to a metadata CSV. The CSV is keyed on a plant_qr_code column (matched against sample_uid). Any column in the CSV is readable via Series.get_metadata(column). The expected_count, group, qc_fail, and timepoint properties are thin wrappers that read specific well-known columns; users can add arbitrary additional columns and read them via get_metadata. See sleap_roots.metadata.build_metadata_csv for a helper that emits a CSV with the canonical column ordering.
sample_uid	Optional[str]	Optional cross-scan stable identity for this series. Defaults to series_name when unset or empty. Coerced to str so CSV plant_qr_code lookups have predictable equality semantics.

Methods:

Name	Description
load	Load a set of predictions for this series.
get_metadata	Generic CSV-column accessor keyed on sample_uid (and optional plant_id).
__len__	Length of the series (number of images).
__getitem__	Return labeled frames for predictions.
__iter__	Iterator for looping through predictions.
get_frame	Return labeled frames for predictions.
plot	Plot predictions on top of the image.
get_primary_points	Get primary root points.
get_lateral_points	Get lateral root points.
get_crown_points	Get crown root points.

Properties

expected_count: Fetch the expected plant count for this series from the CSV. group: Group name for the series from the CSV. qc_fail: Flag to indicate if the series failed QC from the CSV. timepoint: Numeric time-axis value for this series, coerced to float.

expected_count property

expected_count: Union[float, int]

Fetch the expected plant count for this series from the CSV.

group property

group: str

Group name for the series from the CSV.

qc_fail property

qc_fail: Union[int, float]

Flag to indicate if the series failed QC from the CSV.

timepoint property

timepoint: float

Numeric time-axis value for this series, coerced to float.

Returns np.nan when the CSV is absent, the timepoint column is missing, or no row matches sample_uid. Raises ValueError when a matching row contains a non-numeric string value (e.g. a date) OR a non-finite value (inf, -inf) — failing loudly at the metadata layer beats silently producing nonsensical results in downstream timepoint arithmetic.

__attrs_post_init__

__attrs_post_init__() -> None

Default sample_uid to series_name and coerce to str.

Source code in sleap_roots/series.py

def __attrs_post_init__(self) -> None:
    """Default `sample_uid` to `series_name` and coerce to `str`."""
    if self.sample_uid is None or self.sample_uid == "":
        self.sample_uid = self.series_name
    self.sample_uid = str(self.sample_uid)

load classmethod

load(
    series_name: str,
    h5_path: Optional[str] = None,
    primary_path: Optional[str] = None,
    lateral_path: Optional[str] = None,
    crown_path: Optional[str] = None,
    csv_path: Optional[str] = None,
    sample_uid: Optional[str] = None,
) -> Series

Load a set of predictions for this series.

Parameters:

Name	Type	Description	Default
series_name	str	Unique identifier for the series.	required
h5_path	Optional[str]	Optional path to the HDF5-formatted image series, which will be used to load the video.	None
primary_path	Optional[str]	Optional path to the primary root '.slp' predictions file.	None
lateral_path	Optional[str]	Optional path to the lateral root '.slp' predictions file.	None
crown_path	Optional[str]	Optional path to the crown '.slp' predictions file.	None
csv_path	Optional[str]	Optional path to a metadata CSV (keyed on plant_qr_code, matched against sample_uid). Read via get_metadata(column) or the wrapper properties (expected_count, group, qc_fail, timepoint).	None
sample_uid	Optional[str]	Optional cross-scan stable identity. Defaults to series_name when unset or empty. Used as the CSV plant_qr_code lookup key by get_metadata and the expected_count/group/qc_fail properties.	None

Returns:

Type	Description
Series	An instance of Series loaded with the specified predictions.

Source code in sleap_roots/series.py

@classmethod
def load(
    cls,
    series_name: str,
    h5_path: Optional[str] = None,
    primary_path: Optional[str] = None,
    lateral_path: Optional[str] = None,
    crown_path: Optional[str] = None,
    csv_path: Optional[str] = None,
    sample_uid: Optional[str] = None,
) -> "Series":
    """Load a set of predictions for this series.

    Args:
        series_name: Unique identifier for the series.
        h5_path: Optional path to the HDF5-formatted image series, which will be
            used to load the video.
        primary_path: Optional path to the primary root '.slp' predictions file.
        lateral_path: Optional path to the lateral root '.slp' predictions file.
        crown_path: Optional path to the crown '.slp' predictions file.
        csv_path: Optional path to a metadata CSV (keyed on `plant_qr_code`,
            matched against `sample_uid`). Read via `get_metadata(column)`
            or the wrapper properties (`expected_count`, `group`, `qc_fail`,
            `timepoint`).
        sample_uid: Optional cross-scan stable identity. Defaults to `series_name`
            when unset or empty. Used as the CSV `plant_qr_code` lookup key by
            `get_metadata` and the `expected_count`/`group`/`qc_fail` properties.

    Returns:
        An instance of Series loaded with the specified predictions.
    """
    # Initialize the labels as None
    primary_labels, lateral_labels, crown_labels = None, None, None

    # Attempt to load the predictions, with error handling
    try:
        if primary_path:
            # Make path object
            primary_path = Path(primary_path)
            # Check if the file exists
            if primary_path.exists():
                # Make the primary_path POSIX-compliant
                primary_path = primary_path.as_posix()
                # Load the primary predictions
                primary_labels = sio.load_slp(primary_path)
            else:
                print(f"Primary prediction file not found: {primary_path}")
        if lateral_path:
            # Make path object
            lateral_path = Path(lateral_path)
            # Check if the file exists
            if lateral_path.exists():
                # Make the lateral_path POSIX-compliant
                lateral_path = lateral_path.as_posix()
                # Load the lateral predictions
                lateral_labels = sio.load_slp(lateral_path)
            else:
                print(f"Lateral prediction file not found: {lateral_path}")
        if crown_path:
            # Make path object
            crown_path = Path(crown_path)
            # Check if the file exists
            if crown_path.exists():
                # Make the crown_path POSIX-compliant
                crown_path = crown_path.as_posix()
                # Load the crown predictions
                crown_labels = sio.load_slp(crown_path)
            else:
                print(f"Crown prediction file not found: {crown_path}")
    except Exception as e:
        print(f"Error loading prediction files: {e}")

    # Attempt to load the video, with error handling
    video = None
    try:
        if h5_path:
            # Make path object
            h5_path = Path(h5_path)
            # Check if the file exists
            if h5_path.exists():
                # Make the h5_path POSIX-compliant
                h5_path = h5_path.as_posix()
                # Load the video
                video = sio.Video.from_filename(h5_path)
                # Replace the filename in the labels with the h5_path
                for labels in [primary_labels, lateral_labels, crown_labels]:
                    if labels is not None:
                        labels.video.replace_filename(h5_path)
            else:
                print(f"Video file not found: {h5_path}")
    except Exception as e:
        print(f"Error loading video file {h5_path}: {e}")

    # Make the csv path POSIX-compliant
    if csv_path:
        csv_path = Path(csv_path).as_posix()

    return cls(
        series_name=series_name,
        h5_path=h5_path,
        primary_path=primary_path,
        lateral_path=lateral_path,
        crown_path=crown_path,
        primary_labels=primary_labels,
        lateral_labels=lateral_labels,
        crown_labels=crown_labels,
        video=video,
        csv_path=csv_path,
        sample_uid=sample_uid,
    )

get_metadata

get_metadata(
    column: str, plant_id: Optional[int] = None
) -> Any

Fetch a CSV column value keyed on sample_uid (and optional plant_id).

Looks up df[df["plant_qr_code"] == self.sample_uid]. If plant_id is given AND the CSV has a plant_id column, the lookup is the composite (sample_uid, plant_id) match. If plant_id is given but the CSV has no plant_id column, the argument is silently ignored, the sample-uid-only lookup is used, and a one-shot WARNING is emitted (per Series instance).

Parameters:

Name	Type	Description	Default
column	str	The CSV column name to fetch.	required
plant_id	Optional[int]	Optional per-plant disambiguator. Used only when the CSV has a plant_id column.	None

Returns:

Type	Description
Any	The value in the first matching row's column field. Returns np.nan
Any	when the CSV is missing, the requested column is missing, the
Any	plant_qr_code lookup-key column is missing (with WARNING logged),
Any	or no row matches.

Source code in sleap_roots/series.py

def get_metadata(self, column: str, plant_id: Optional[int] = None) -> Any:
    """Fetch a CSV column value keyed on `sample_uid` (and optional `plant_id`).

    Looks up `df[df["plant_qr_code"] == self.sample_uid]`. If `plant_id` is given
    AND the CSV has a `plant_id` column, the lookup is the composite
    (sample_uid, plant_id) match. If `plant_id` is given but the CSV has no
    `plant_id` column, the argument is silently ignored, the sample-uid-only
    lookup is used, and a one-shot WARNING is emitted (per Series instance).

    Args:
        column: The CSV column name to fetch.
        plant_id: Optional per-plant disambiguator. Used only when the CSV has a
            `plant_id` column.

    Returns:
        The value in the first matching row's `column` field. Returns `np.nan`
        when the CSV is missing, the requested `column` is missing, the
        `plant_qr_code` lookup-key column is missing (with WARNING logged),
        or no row matches.
    """
    # Reject bool-typed plant_id — Python booleans are ints (False == 0,
    # True == 1) and pandas equality matches them against integer plant_id
    # columns. A caller passing plant_id=False (e.g. from a CLI arg parser
    # or a `flag or False` ternary) would silently fetch the row for
    # plant_id=0. Fail loudly instead of producing wrong rows.
    if isinstance(plant_id, bool):
        raise TypeError(
            f"Series.get_metadata: plant_id must be int or None, not "
            f"bool ({plant_id!r}). Booleans match int 0/1 in pandas "
            f"equality and would silently fetch the wrong row."
        )
    if not self.csv_path or not Path(self.csv_path).exists():
        return np.nan
    df = pd.read_csv(self.csv_path)
    if column not in df.columns:
        return np.nan
    if "plant_qr_code" not in df.columns:
        # Lookup key absent — preserve fail-soft contract instead of
        # raising KeyError. Surface the misconfiguration via WARNING so
        # users notice they wrote a CSV without the required column.
        logger.warning(
            "Series '%s': metadata CSV %s has no 'plant_qr_code' column; "
            "cannot perform sample_uid lookup. Returning NaN. The CSV "
            "MUST contain a 'plant_qr_code' column to be useful (see "
            "build_metadata_csv).",
            self.series_name,
            self.csv_path,
        )
        return np.nan
    sample_match = df[df["plant_qr_code"] == self.sample_uid]
    if plant_id is not None and "plant_id" in df.columns:
        sample_match = sample_match[sample_match["plant_id"] == plant_id]
    elif plant_id is not None and not self._warned_missing_plant_id_column:
        logger.warning(
            "Series '%s': plant_id=%r ignored because CSV %s has no "
            "'plant_id' column; falling back to sample_uid-only lookup. "
            "Add a 'plant_id' column to disambiguate per-plant rows.",
            self.series_name,
            plant_id,
            self.csv_path,
        )
        self._warned_missing_plant_id_column = True
    if len(sample_match) == 0:
        return np.nan
    return sample_match[column].iloc[0]

__len__

__len__() -> int

Length of the series (number of images).

Source code in sleap_roots/series.py

def __len__(self) -> int:
    """Length of the series (number of images)."""
    if self.video is not None:
        return len(self.video)
    else:
        # Check all labels if video is None
        for labels in [self.primary_labels, self.lateral_labels, self.crown_labels]:
            if labels is not None:
                return len(labels)
        # If all labels are None, return 0
        return 0

__getitem__

__getitem__(idx: int) -> Dict[str, Optional[LabeledFrame]]

Return labeled frames for primary and/or lateral and/or crown predictions.

Source code in sleap_roots/series.py

def __getitem__(self, idx: int) -> Dict[str, Optional[sio.LabeledFrame]]:
    """Return labeled frames for primary and/or lateral and/or crown predictions."""
    return self.get_frame(idx)

__iter__

__iter__()

Iterator for looping through predictions.

Source code in sleap_roots/series.py

def __iter__(self):
    """Iterator for looping through predictions."""
    for i in range(len(self)):
        yield self[i]

get_frame

get_frame(frame_idx: int) -> dict

Return labeled frames for primary, lateral, and crown predictions.

Parameters:

Name	Type	Description	Default
frame_idx	int	Integer frame number.	required

Returns:

Type	Description
dict	Dictionary with keys 'primary', 'lateral', and 'crown', each corresponding
dict	to the sio.LabeledFrame from each set of predictions on the same frame. If
dict	any set of predictions is not available, its value will be None.

Source code in sleap_roots/series.py

def get_frame(self, frame_idx: int) -> dict:
    """Return labeled frames for primary, lateral, and crown predictions.

    Args:
        frame_idx: Integer frame number.

    Returns:
        Dictionary with keys 'primary', 'lateral', and 'crown', each corresponding
        to the `sio.LabeledFrame` from each set of predictions on the same frame. If
        any set of predictions is not available, its value will be None.
    """
    frames = {}

    # For primary predictions
    if self.primary_labels is not None:
        frames["primary"] = self.primary_labels.find(
            self.primary_labels.video, frame_idx, return_new=True
        )[0]
    else:
        frames["primary"] = None

    # For lateral predictions
    if self.lateral_labels is not None:
        frames["lateral"] = self.lateral_labels.find(
            self.lateral_labels.video, frame_idx, return_new=True
        )[0]
    else:
        frames["lateral"] = None

    # For crown predictions
    if self.crown_labels is not None:
        frames["crown"] = self.crown_labels.find(
            self.crown_labels.video, frame_idx, return_new=True
        )[0]
    else:
        frames["crown"] = None

    return frames

plot

plot(
    frame_idx: int, scale: float = 1.0, **kwargs
) -> Figure

Plot predictions on top of the image.

Parameters:

Name	Type	Description	Default
frame_idx	int	Frame index to visualize.	required
scale	float	Relative size of the visualized image. Useful for plotting smaller images within notebooks.	1.0

Returns:

Type	Description
Figure	matplotlib.figure.Figure object that shows predictions on top of images.

Source code in sleap_roots/series.py

def plot(
    self, frame_idx: int, scale: float = 1.0, **kwargs
) -> matplotlib.figure.Figure:
    """Plot predictions on top of the image.

    Args:
        frame_idx: Frame index to visualize.
        scale: Relative size of the visualized image. Useful for plotting smaller
            images within notebooks.

    Returns:
        matplotlib.figure.Figure object that shows predictions on top of images.
    """
    # Check if the video is available
    if self.video is None:
        raise ValueError("Video is not available. Specify the h5_path to load it.")

    # Retrieve all available frames
    frames = self.get_frame(frame_idx)

    # Generate the color palette from seaborn
    cmap = sns.color_palette("tab10")

    # Define the order of preference for the predictions for plotting the image
    prediction_order = ["primary", "lateral", "crown"]

    # Variable to keep track if the image has been plotted
    image_plotted = False

    # First, find the first available prediction to plot the image
    for prediction in prediction_order:
        labeled_frame = frames.get(prediction)
        if labeled_frame is not None and not image_plotted:
            # Plot the image
            plot_img(labeled_frame.image, scale=scale)
            # Set the flag to True to avoid plotting the image again
            image_plotted = True

    # Then, iterate through all predictions to plot instances
    for i, prediction in enumerate(prediction_order):
        labeled_frame = frames.get(prediction)
        if labeled_frame is not None:
            # Use the color map index for each prediction type
            # Modulo the length of the color map to avoid index out of range
            color = cmap[i % len(cmap)]

            # Plot the instances
            plot_instances(labeled_frame.instances, cmap=[color], **kwargs)

    # Capture the current plot after calling plot_img and plot_instances.
    fig = plt.gcf()

    # Close the captured fig to avoid duplicate rendering from Jupyter.
    plt.close(fig)

    # Return the figure. In a cell, Jupyter will automatically render the plot.
    return fig

get_primary_points

get_primary_points(frame_idx: int) -> ndarray

Get primary root points.

Parameters:

Name	Type	Description	Default
frame_idx	int	Frame index.	required

Returns:

Type	Description
ndarray	Primary root points as array of shape (n_instances, n_nodes, 2).

Source code in sleap_roots/series.py

def get_primary_points(self, frame_idx: int) -> np.ndarray:
    """Get primary root points.

    Args:
        frame_idx: Frame index.

    Returns:
        Primary root points as array of shape `(n_instances, n_nodes, 2)`.
    """
    # Check that self.primary_labels is not None
    if self.primary_labels is None:
        raise ValueError("Primary labels are not available.")
    # Retrieve all available frames
    frames = self.get_frame(frame_idx)
    # Get the primary labeled frame
    primary_lf = frames.get("primary")
    # Get the ground truth instances and unused predictions
    gt_instances_pr = primary_lf.user_instances + primary_lf.unused_predictions
    # If there are no instances, return an empty array
    if len(gt_instances_pr) == 0:
        primary_pts = np.array([[(np.nan, np.nan), (np.nan, np.nan)]])
    # Otherwise, stack the instances into an array
    else:
        primary_pts = np.stack([inst.numpy() for inst in gt_instances_pr], axis=0)
    return primary_pts

get_lateral_points

get_lateral_points(frame_idx: int) -> ndarray

Get lateral root points.

Parameters:

Name	Type	Description	Default
frame_idx	int	Frame index.	required

Returns:

Type	Description
ndarray	Lateral root points as array of shape (n_instances, n_nodes, 2).

Source code in sleap_roots/series.py

def get_lateral_points(self, frame_idx: int) -> np.ndarray:
    """Get lateral root points.

    Args:
        frame_idx: Frame index.

    Returns:
        Lateral root points as array of shape `(n_instances, n_nodes, 2)`.
    """
    # Check that self.lateral_labels is not None
    if self.lateral_labels is None:
        raise ValueError("Lateral labels are not available.")
    # Retrieve all available frames
    frames = self.get_frame(frame_idx)
    # Get the lateral labeled frame
    lateral_lf = frames.get("lateral")
    # Get the ground truth instances and unused predictions
    gt_instances_lr = lateral_lf.user_instances + lateral_lf.unused_predictions
    # If there are no instances, return an empty array
    if len(gt_instances_lr) == 0:
        lateral_pts = np.array([[(np.nan, np.nan), (np.nan, np.nan)]])
    # Otherwise, stack the instances into an array
    else:
        lateral_pts = np.stack([inst.numpy() for inst in gt_instances_lr], axis=0)
    return lateral_pts

get_crown_points

get_crown_points(frame_idx: int) -> ndarray

Get crown root points.

Parameters:

Name	Type	Description	Default
frame_idx	int	Frame index.	required

Returns:

Type	Description
ndarray	Crown root points as array of shape (n_instances, n_nodes, 2).

Source code in sleap_roots/series.py

def get_crown_points(self, frame_idx: int) -> np.ndarray:
    """Get crown root points.

    Args:
        frame_idx: Frame index.

    Returns:
        Crown root points as array of shape `(n_instances, n_nodes, 2)`.
    """
    # Check that self.crown_labels is not None
    if self.crown_labels is None:
        raise ValueError("Crown labels are not available.")
    # Retrieve all available frames
    frames = self.get_frame(frame_idx)
    # Get the crown labeled frame
    crown_lf = frames.get("crown")
    # Get the ground truth instances and unused predictions
    gt_instances_cr = crown_lf.user_instances + crown_lf.unused_predictions
    # If there are no instances, return an empty array
    if len(gt_instances_cr) == 0:
        crown_pts = np.array([[(np.nan, np.nan), (np.nan, np.nan)]])
    # Otherwise, stack the instances into an array
    else:
        crown_pts = np.stack([inst.numpy() for inst in gt_instances_cr], axis=0)
    return crown_pts

get_tracked_tips

get_tracked_tips(
    root_type: Optional[str] = None,
) -> DataFrame

Return per-frame tracked-tip rows in long format.

Parameters:

Name	Type	Description	Default
root_type	Optional[str]	One of "primary", "lateral", "crown". When None (default), auto-detects from whichever single <root_type>_path is populated. Raises ValueError when zero or more than one path is populated and root_type is None.	None

Returns:

Type	Description
DataFrame	A pandas.DataFrame with columns `["track_id", "frame",
DataFrame	"tip_x", "tip_y"]in that order, sorted by(track_id, frame)`
DataFrame	with a clean range index. One row per (track_id, frame) where
DataFrame	the track has an instance.
DataFrame	Coordinates tip_x, tip_y are in image pixels (origin
DataFrame	top-left, y increases downward), matching the SLEAP
DataFrame	convention. The tip coordinate is the LAST node of the
DataFrame	skeleton (inst.numpy()[-1]), matching the convention used
DataFrame	by sleap_roots.tips.get_tips. This works for both
DataFrame	single-node skeletons (e.g. ["r0"]) and multi-node
DataFrame	skeletons (e.g. ["base", "mid", "tip"]).

Raises:

Type	Description
ValueError	When root_type is None and zero or more than one of primary_path / lateral_path / crown_path is populated. Also raised when any instance has inst.track is None or empty inst.track.name — the pipeline requires SLEAP-tracked predictions.

Source code in sleap_roots/series.py

def get_tracked_tips(
    self,
    root_type: Optional[str] = None,
) -> pd.DataFrame:
    """Return per-frame tracked-tip rows in long format.

    Args:
        root_type: One of `"primary"`, `"lateral"`, `"crown"`. When
            `None` (default), auto-detects from whichever single
            `<root_type>_path` is populated. Raises `ValueError` when
            zero or more than one path is populated and `root_type` is
            `None`.

    Returns:
        A `pandas.DataFrame` with columns `["track_id", "frame",
        "tip_x", "tip_y"]` in that order, sorted by `(track_id, frame)`
        with a clean range index. One row per `(track_id, frame)` where
        the track has an instance.

        Coordinates `tip_x`, `tip_y` are in image pixels (origin
        top-left, y increases downward), matching the SLEAP
        convention. The tip coordinate is the LAST node of the
        skeleton (`inst.numpy()[-1]`), matching the convention used
        by `sleap_roots.tips.get_tips`. This works for both
        single-node skeletons (e.g. `["r0"]`) and multi-node
        skeletons (e.g. `["base", "mid", "tip"]`).

    Raises:
        ValueError: When `root_type` is `None` and zero or more than
            one of `primary_path` / `lateral_path` / `crown_path`
            is populated. Also raised when any instance has
            `inst.track is None` or empty `inst.track.name` — the
            pipeline requires SLEAP-tracked predictions.
    """
    # Resolve root_type via the shared helper (validates explicit
    # values + auto-detects when None).
    path_attrs = {
        "primary": self.primary_path,
        "lateral": self.lateral_path,
        "crown": self.crown_path,
    }
    root_type = _resolve_root_type(path_attrs, root_type)
    # Get the labels object for the resolved root type.
    labels_attr = f"{root_type}_labels"
    labels = getattr(self, labels_attr, None)
    if labels is None:
        raise ValueError(
            f"{labels_attr} is not available — was {root_type}_path set "
            f"on Series.load?"
        )

    # Iterate per-instance — tracker output does NOT preserve positional
    # ordering across frames (verified during the #129 brainstorm: frame 0
    # may be [track_0, 1, 2, 3, 4, 5] but frame 1 may be
    # [track_0, 3, 4, 2, 1, 5]). We must read inst.track.name per instance.
    #
    # On proofread .slp files, SLEAP retains BOTH the original
    # PredictedInstance (tracker's prediction; score is a float) AND the
    # user-corrected Instance (score is None) for every frame the user
    # manually corrected. Per
    # docs/circumnutation/preliminary_results_2026-05-07.md §3.1:
    # "Where an instance_type=0 (user-corrected) and instance_type=1
    # (predicted) instance exist for the same frame and track, the
    # user-corrected value takes precedence." We honor that convention so
    # callers get exactly one row per (track_id, frame_idx). Without this
    # dedup, downstream consumers (e.g.
    # sleap_roots.circumnutation.kinematics.compute) compute Δframe=0
    # divides and propagate ±inf / NaN through every velocity-bearing
    # trait on every proofread frame.
    # (Instance / PredictedInstance imported at top of module — line 8.)

    # First pass: collect every tracked instance plus the set of
    # (frame_idx, track_name) keys for which a user-corrected instance
    # exists.
    candidates: List[Tuple[Tuple[int, str], "Instance", bool]] = []
    user_corrected_keys: set = set()
    offending_frames = []
    for lf in labels.labeled_frames:
        for inst in lf.instances:
            track = inst.track
            if track is None or not getattr(track, "name", None):
                offending_frames.append(lf.frame_idx)
                continue
            key = (lf.frame_idx, track.name)
            # PredictedInstance subclasses Instance in sleap-io, so the
            # isinstance check MUST exclude PredictedInstance to detect
            # user corrections.
            is_user = isinstance(inst, Instance) and not isinstance(
                inst, PredictedInstance
            )
            candidates.append((key, inst, is_user))
            if is_user:
                user_corrected_keys.add(key)

    if offending_frames:
        unique_frames = sorted(set(offending_frames))
        raise ValueError(
            f"TrackedTipPipeline requires tracked .slp predictions; "
            f"untracked instances found at frame indices {unique_frames}. "
            f"See https://sleap.ai/tutorials/tracking.html for tracking "
            f"setup."
        )

    # Second pass: drop PredictedInstance rows where a user-corrected
    # Instance exists for the same (track, frame). All other duplicates
    # are preserved (e.g., two user-corrected instances from a buggy
    # proofreader, or two PredictedInstances from a buggy tracker — these
    # remain visible in the long-format DataFrame for downstream
    # debugging, matching the existing `TrackedTipPipeline` contract
    # tested in `tests/test_tracked_tip_pipeline.py` §6.14).
    rows = []
    for key, inst, is_user in candidates:
        if not is_user and key in user_corrected_keys:
            # Predicted instance shadowed by a user-corrected version
            # at the same (track, frame). Drop per prelim §3.1.
            continue
        tip_xy = inst.numpy()[-1]  # last node = tip by skeleton convention
        rows.append(
            {
                "track_id": key[1],
                "frame": key[0],
                "tip_x": float(tip_xy[0]),
                "tip_y": float(tip_xy[1]),
            }
        )

    df = pd.DataFrame(rows, columns=["track_id", "frame", "tip_x", "tip_y"])
    df = df.sort_values(["track_id", "frame"]).reset_index(drop=True)
    return df

find_all_h5_paths

find_all_h5_paths(
    data_folders: Union[str, List[str]],
) -> List[str]

Find all .h5 paths from a list of folders.

Parameters:

Name	Type	Description	Default
data_folders	Union[str, List[str]]	Path or list of paths to folders containing .h5 paths.	required

Returns:

Type	Description
List[str]	A list of filenames to .h5 paths.

Source code in sleap_roots/series.py

def find_all_h5_paths(data_folders: Union[str, List[str]]) -> List[str]:
    """Find all .h5 paths from a list of folders.

    Args:
        data_folders: Path or list of paths to folders containing .h5 paths.

    Returns:
        A list of filenames to .h5 paths.
    """
    if type(data_folders) != list:
        data_folders = [data_folders]

    h5_paths = []
    for data_folder in data_folders:
        h5_paths.extend([Path(p).as_posix() for p in Path(data_folder).glob("*.h5")])
    return h5_paths

find_all_slp_paths

find_all_slp_paths(
    data_folders: Union[str, List[str]],
) -> List[str]

Find all .slp paths from a list of folders.

Parameters:

Name	Type	Description	Default
data_folders	Union[str, List[str]]	Path or list of paths to folders containing .slp paths.	required

Returns:

Type	Description
List[str]	A list of filenames to .slp paths.

Source code in sleap_roots/series.py

def find_all_slp_paths(data_folders: Union[str, List[str]]) -> List[str]:
    """Find all .slp paths from a list of folders.

    Args:
        data_folders: Path or list of paths to folders containing .slp paths.

    Returns:
        A list of filenames to .slp paths.
    """
    if type(data_folders) != list:
        data_folders = [data_folders]

    slp_paths = []
    for data_folder in data_folders:
        slp_paths.extend([Path(p).as_posix() for p in Path(data_folder).glob("*.slp")])
    return slp_paths

load_series_from_h5s

load_series_from_h5s(
    h5_paths: List[str],
    model_id: str,
    csv_path: Optional[str] = None,
) -> List[Series]

Load a list of Series from a list of .h5 paths.

To load the Series, the files must be named with the following convention: h5_path: '/path/to/scan/series_name.h5' primary_path: '/path/to/scan/series_name.model{model_id}.rootprimary.slp' lateral_path: '/path/to/scan/series_name.model{model_id}.rootlateral.slp' crown_path: '/path/to/scan/series_name.model{model_id}.rootcrown.slp'

Our pipeline outputs prediction files with this format: //scan{scan_id}.model{model_id}.root{model_type}.slp

Parameters:

Name	Type	Description	Default
h5_paths	List[str]	List of paths to .h5 files.	required
csv_path	Optional[str]	Optional path to the CSV file containing the expected plant count.	None

Returns:

Type	Description
List[Series]	A list of Series loaded with the specified .h5 files.

Source code in sleap_roots/series.py

def load_series_from_h5s(
    h5_paths: List[str], model_id: str, csv_path: Optional[str] = None
) -> List[Series]:
    """Load a list of Series from a list of .h5 paths.

    To load the `Series`, the files must be named with the following convention:
    h5_path: '/path/to/scan/series_name.h5'
    primary_path: '/path/to/scan/series_name.model{model_id}.rootprimary.slp'
    lateral_path: '/path/to/scan/series_name.model{model_id}.rootlateral.slp'
    crown_path: '/path/to/scan/series_name.model{model_id}.rootcrown.slp'

    Our pipeline outputs prediction files with this format:
    /<output_folder>/scan{scan_id}.model{model_id}.root{model_type}.slp

    Args:
        h5_paths: List of paths to .h5 files.
        csv_path: Optional path to the CSV file containing the expected plant count.

    Returns:
        A list of Series loaded with the specified .h5 files.
    """
    series_list = []
    for h5_path in h5_paths:
        # Extract the series name from the h5 path
        series_name = Path(h5_path).name.split(".")[0]
        # Generate the paths for the primary, lateral, and crown predictions
        primary_path = h5_path.replace(".h5", f".model{model_id}.rootprimary.slp")
        lateral_path = h5_path.replace(".h5", f".model{model_id}.rootlateral.slp")
        crown_path = h5_path.replace(".h5", f".model{model_id}.rootcrown.slp")
        # Load the Series
        series = Series.load(
            series_name,
            h5_path=h5_path,
            primary_path=primary_path,
            lateral_path=lateral_path,
            crown_path=crown_path,
            csv_path=csv_path,
        )
        series_list.append(series)
    return series_list

load_series_from_slps

load_series_from_slps(
    slp_paths: List[str],
    h5s: bool = False,
    csv_path: Optional[str] = None,
) -> List[Series]

Load a list of Series from a list of .slp paths.

To load the Series, the files must be named with the following convention. The slp_paths are expeted to have the series_name in the filename and "primary", "lateral", or "crown" in the filename to differentiate the predictions. h5_path: '/path/to/scan/series_name.h5' Note that everything is expected to be in the same folder.

Our pipeline outputs prediction files with this format: //scan{scan_id}.model{model_id}.root{model_type}.slp

Parameters:

Name	Type	Description	Default
slp_paths	List[str]	List of paths to .slp files.	required
h5s	bool	Boolean flag to indicate if the .h5 files are available. Default is False.	False
csv_path	Optional[str]	Optional path to the CSV file containing the expected plant count.	None

Source code in sleap_roots/series.py

def load_series_from_slps(
    slp_paths: List[str], h5s: bool = False, csv_path: Optional[str] = None
) -> List[Series]:
    """Load a list of Series from a list of .slp paths.

    To load the `Series`, the files must be named with the following convention.
    The `slp_paths` are expeted to have the `series_name` in the filename and "primary",
    "lateral", or "crown" in the filename to differentiate the predictions.
    h5_path: '/path/to/scan/series_name.h5'
    Note that everything is expected to be in the same folder.

    Our pipeline outputs prediction files with this format:
    /<output_folder>/scan{scan_id}.model{model_id}.root{model_type}.slp


    Args:
        slp_paths: List of paths to .slp files.
        h5s: Boolean flag to indicate if the .h5 files are available. Default is False.
        csv_path: Optional path to the CSV file containing the expected plant count.
    """
    series_list = []
    series_names = list(set([Path(p).name.split(".")[0] for p in slp_paths]))
    for series_name in series_names:
        # Generate the paths for the primary, lateral, and crown predictions
        primary_path = [p for p in slp_paths if series_name in p and "primary" in p]
        lateral_path = [p for p in slp_paths if series_name in p and "lateral" in p]
        crown_path = [p for p in slp_paths if series_name in p and "crown" in p]
        # Check if the .h5 files are available
        if h5s:
            # Get directory of the h5s
            h5_dir = Path(slp_paths[0]).parent
            # Create path to the .h5 file
            h5_path = h5_dir / f"{series_name}.h5"
        else:
            h5_path = None
        # Load the Series
        series = Series.load(
            series_name,
            primary_path=primary_path[0] if primary_path else None,
            lateral_path=lateral_path[0] if lateral_path else None,
            crown_path=crown_path[0] if crown_path else None,
            h5_path=h5_path,
            csv_path=csv_path,
        )
        series_list.append(series)
    return series_list

imgfig

imgfig(
    size: Union[float, Tuple] = 6,
    dpi: int = 72,
    scale: float = 1.0,
) -> Figure

Create a tight figure for image plotting.

Parameters:

Name	Type	Description	Default
size	Union[float, Tuple]	Scalar or 2-tuple specifying the (width, height) of the figure in inches. If scalar, will assume equal width and height.	6
dpi	int	Dots per inch, controlling the resolution of the image.	72
scale	float	Factor to scale the size of the figure by. This is a convenience for increasing the size of the plot at the same DPI.	1.0

Returns:

Type	Description
Figure	A matplotlib.figure.Figure to use for plotting.

Source code in sleap_roots/series.py

def imgfig(
    size: Union[float, Tuple] = 6, dpi: int = 72, scale: float = 1.0
) -> matplotlib.figure.Figure:
    """Create a tight figure for image plotting.

    Args:
        size: Scalar or 2-tuple specifying the (width, height) of the figure in inches.
            If scalar, will assume equal width and height.
        dpi: Dots per inch, controlling the resolution of the image.
        scale: Factor to scale the size of the figure by. This is a convenience for
            increasing the size of the plot at the same DPI.

    Returns:
        A matplotlib.figure.Figure to use for plotting.
    """
    if not isinstance(size, (tuple, list)):
        size = (size, size)
    fig = plt.figure(figsize=(scale * size[0], scale * size[1]), dpi=dpi)
    ax = fig.add_axes([0, 0, 1, 1], frameon=False)
    ax.get_xaxis().set_visible(False)
    ax.get_yaxis().set_visible(False)
    plt.autoscale(tight=True)
    ax.set_xticks([])
    ax.set_yticks([])
    ax.grid(False)
    return fig

plot_img

plot_img(
    img: ndarray, dpi: int = 72, scale: float = 1.0
) -> Figure

Plot an image in a tight figure.

Parameters:

Name	Type	Description	Default
img	ndarray	Image to plot. Can be a numpy array or a tf.Tensor.	required
dpi	int	Dots per inch, controlling the resolution of the image.	72
scale	float	Factor to scale the size of the figure by. This is a convenience for increasing the size of the plot at the same DPI.	1.0

Returns:

Type	Description
Figure	A matplotlib.figure.Figure containing the image.

Source code in sleap_roots/series.py

def plot_img(
    img: np.ndarray, dpi: int = 72, scale: float = 1.0
) -> matplotlib.figure.Figure:
    """Plot an image in a tight figure.

    Args:
        img: Image to plot. Can be a numpy array or a `tf.Tensor`.
        dpi: Dots per inch, controlling the resolution of the image.
        scale: Factor to scale the size of the figure by. This is a convenience for
            increasing the size of the plot at the same DPI.

    Returns:
        A matplotlib.figure.Figure containing the image.
    """
    if hasattr(img, "numpy"):
        img = img.numpy()

    if img.shape[0] == 1:
        # Squeeze out batch singleton dimension.
        img = img.squeeze(axis=0)

    # Check if image is grayscale (single channel).
    grayscale = img.shape[-1] == 1
    if grayscale:
        # Squeeze out singleton channel.
        img = img.squeeze(axis=-1)

    # Normalize the range of pixel values.
    img_min = img.min()
    img_max = img.max()
    if img_min < 0.0 or img_max > 1.0:
        img = (img - img_min) / (img_max - img_min)

    fig = imgfig(
        size=(float(img.shape[1]) / dpi, float(img.shape[0]) / dpi),
        dpi=dpi,
        scale=scale,
    )

    ax = fig.gca()
    ax.imshow(
        img,
        cmap="gray" if grayscale else None,
        origin="upper",
        extent=[-0.5, img.shape[1] - 0.5, img.shape[0] - 0.5, -0.5],
    )
    return fig

plot_instance

plot_instance(
    instance,
    skeleton=None,
    cmap=None,
    color_by_node=False,
    lw=2,
    ms=10,
    bbox=None,
    scale=1.0,
    **kwargs
)

Plot a single instance with edge coloring.

Source code in sleap_roots/series.py

def plot_instance(
    instance,
    skeleton=None,
    cmap=None,
    color_by_node=False,
    lw=2,
    ms=10,
    bbox=None,
    scale=1.0,
    **kwargs,
):
    """Plot a single instance with edge coloring."""
    if cmap is None:
        cmap = sns.color_palette("tab20")

    if skeleton is None and hasattr(instance, "skeleton"):
        skeleton = instance.skeleton

    if skeleton is None:
        color_by_node = True
    else:
        if len(skeleton.edges) == 0:
            color_by_node = True

    if hasattr(instance, "numpy"):
        inst_pts = instance.numpy()
    else:
        inst_pts = instance

    h_lines = []
    if color_by_node:
        for k, (x, y) in enumerate(inst_pts):
            if bbox is not None:
                x -= bbox[1]
                y -= bbox[0]

            x *= scale
            y *= scale

            h_lines_k = plt.plot(x, y, ".", ms=ms, c=cmap[k % len(cmap)], **kwargs)
            h_lines.append(h_lines_k)

    else:
        for k, (src_ind, dst_ind) in enumerate(skeleton.edge_inds):
            src_pt = inst_pts[src_ind]
            dst_pt = inst_pts[dst_ind]

            x = np.array([src_pt[0], dst_pt[0]])
            y = np.array([src_pt[1], dst_pt[1]])

            if bbox is not None:
                x -= bbox[1]
                y -= bbox[0]

            x *= scale
            y *= scale

            h_lines_k = plt.plot(
                x, y, ".-", ms=ms, lw=lw, c=cmap[k % len(cmap)], **kwargs
            )

            h_lines.append(h_lines_k)

    return h_lines

plot_instances

plot_instances(
    instances: List,
    skeleton: Optional[Skeleton] = None,
    cmap: Optional[List] = None,
    color_by_track: bool = False,
    tracks: Optional[List] = None,
    **kwargs: Any
) -> List

Plot a list of instances with identity coloring.

Parameters:

Name	Type	Description	Default
instances	List	List of instances to plot.	required
skeleton	Optional[Skeleton]	Skeleton to use for edge coloring. If not provided, will use node coloring.	None
cmap	Optional[List]	Color map to use for coloring. If not provided, will use the default seaborn tab10 color palette.	None
color_by_track	bool	If True, will color instances by their track. If False, will color instances by their identity in the list.	False
tracks	Optional[List]	List of tracks to use for coloring. If not provided, will infer tracks from the instances.	None
**kwargs	Any	Additional keyword arguments to pass to plot_instance.	{}

Returns:

Type	Description
List	A list of handles to the plotted lines.

Source code in sleap_roots/series.py

def plot_instances(
    instances: List,
    skeleton: Optional["sio.Skeleton"] = None,
    cmap: Optional[List] = None,
    color_by_track: bool = False,
    tracks: Optional[List] = None,
    **kwargs: Any,
) -> List:
    """Plot a list of instances with identity coloring.

    Args:
        instances: List of instances to plot.
        skeleton: Skeleton to use for edge coloring. If not provided, will use node
            coloring.
        cmap: Color map to use for coloring. If not provided, will use the default
            seaborn tab10 color palette.
        color_by_track: If True, will color instances by their track. If False, will
            color instances by their identity in the list.
        tracks: List of tracks to use for coloring. If not provided, will infer tracks
            from the instances.
        **kwargs: Additional keyword arguments to pass to `plot_instance`.

    Returns:
        A list of handles to the plotted lines.
    """
    if cmap is None:
        cmap = sns.color_palette("tab10")

    if color_by_track and tracks is None:
        # Infer tracks for ordering if not provided.
        tracks = set()
        for instance in instances:
            tracks.add(instance.track)

        # Sort by spawned frame.
        tracks = sorted(list(tracks), key=lambda track: track.name)

    h_lines = []
    for i, instance in enumerate(instances):
        if color_by_track:
            if instance.track is None:
                raise ValueError(
                    "Instances must have a set track when coloring by track."
                )

            if instance.track not in tracks:
                raise ValueError("Instance has a track not found in specified tracks.")

            color = cmap[tracks.index(instance.track) % len(cmap)]

        else:
            # Color by identity (order in list).
            color = cmap[i % len(cmap)]

        h_lines_i = plot_instance(instance, skeleton=skeleton, cmap=[color], **kwargs)
        h_lines.append(h_lines_i)

    return h_lines

validate_tracked_slp

validate_tracked_slp(slp_path: Union[str, Path]) -> None

Validate that every instance in a .slp file has a non-empty track.

Used as an input precondition for TrackedTipPipeline.

Parameters:

Name	Type	Description	Default
slp_path	Union[str, Path]	Path to the .slp file to validate.	required

Returns:

Type	Description
None	None when every instance has a non-empty inst.track.name.

Raises:

Type	Description
ValueError	When any instance has inst.track is None or empty inst.track.name. The error message lists ALL offending frame indices and points to the SLEAP tracking documentation.

Source code in sleap_roots/series.py

def validate_tracked_slp(slp_path: Union[str, Path]) -> None:
    """Validate that every instance in a .slp file has a non-empty track.

    Used as an input precondition for `TrackedTipPipeline`.

    Args:
        slp_path: Path to the .slp file to validate.

    Returns:
        `None` when every instance has a non-empty `inst.track.name`.

    Raises:
        ValueError: When any instance has `inst.track is None` or empty
            `inst.track.name`. The error message lists ALL offending frame
            indices and points to the SLEAP tracking documentation.
    """
    labels = sio.load_slp(str(slp_path))
    offending = []
    for lf in labels.labeled_frames:
        for inst in lf.instances:
            track = inst.track
            if track is None or not getattr(track, "name", None):
                offending.append(lf.frame_idx)
                break  # one offender per frame is enough to flag the frame
    if offending:
        unique_frames = sorted(set(offending))
        raise ValueError(
            f"TrackedTipPipeline requires tracked .slp predictions; "
            f"untracked instances found at frame indices {unique_frames} in "
            f"{slp_path}. See https://sleap.ai/tutorials/tracking.html for "
            f"tracking setup."
        )

validate_series_for_tracked_tip

validate_series_for_tracked_tip(
    series: Series, root_type: Optional[str] = None
) -> None

Validate a Series is loadable by TrackedTipPipeline.

Composite check: resolves root_type (auto-detect when None), asserts the corresponding <root_type>_path is set and the loaded skeleton has at least one node, then calls validate_tracked_slp on the resolved path.

Parameters:

Name	Type	Description	Default
series	Series	The Series to validate.	required
root_type	Optional[str]	One of "primary", "lateral", "crown". When None, auto-detects from whichever single <root_type>_path is populated.	None

Returns:

Type	Description
None	None when the series is valid for tracked-tip processing.

Raises:

Type	Description
ValueError	When root_type is None and zero or more than one of primary_path / lateral_path / crown_path is populated; when the resolved path's skeleton has zero nodes; or when validate_tracked_slp raises (propagated).

Source code in sleap_roots/series.py

def validate_series_for_tracked_tip(
    series: Series,
    root_type: Optional[str] = None,
) -> None:
    """Validate a `Series` is loadable by `TrackedTipPipeline`.

    Composite check: resolves `root_type` (auto-detect when `None`),
    asserts the corresponding `<root_type>_path` is set and the loaded
    skeleton has at least one node, then calls `validate_tracked_slp` on
    the resolved path.

    Args:
        series: The `Series` to validate.
        root_type: One of `"primary"`, `"lateral"`, `"crown"`. When
            `None`, auto-detects from whichever single `<root_type>_path`
            is populated.

    Returns:
        `None` when the series is valid for tracked-tip processing.

    Raises:
        ValueError: When `root_type` is `None` and zero or more than one
            of `primary_path` / `lateral_path` / `crown_path` is
            populated; when the resolved path's skeleton has zero nodes;
            or when `validate_tracked_slp` raises (propagated).
    """
    path_attrs = {
        "primary": series.primary_path,
        "lateral": series.lateral_path,
        "crown": series.crown_path,
    }
    root_type = _resolve_root_type(path_attrs, root_type)

    resolved_path = path_attrs[root_type]
    if resolved_path is None:
        raise ValueError(
            f"root_type={root_type!r} requested but {root_type}_path is None "
            f"on the series."
        )

    labels_attr = f"{root_type}_labels"
    labels = getattr(series, labels_attr, None)
    if labels is not None:
        # Skeleton sanity check (every Labels has a `skeletons` list).
        sks = getattr(labels, "skeletons", None) or []
        if sks and len(sks[0].nodes) == 0:
            raise ValueError(
                f"{root_type} skeleton has zero nodes — cannot extract tips."
            )

    validate_tracked_slp(resolved_path)