Trait pipelines

trait_pipelines

Extract traits in a pipeline based on a trait graph.

NumpyArrayEncoder

Bases: JSONEncoder

Custom encoder for NumPy array types.

default

default(obj: Any) -> Any

Serialize NumPy arrays to lists.

Parameters:

Name	Type	Description	Default
obj	Any	The object to serialize.	required

Returns:

Type	Description
Any	A list representation of the NumPy array or the object itself.

Source code in sleap_roots/trait_pipelines.py

def default(self, obj: Any) -> Any:
    """Serialize NumPy arrays to lists.

    Args:
        obj: The object to serialize.

    Returns:
        A list representation of the NumPy array or the object itself.
    """
    if isinstance(obj, np.ndarray):
        return obj.tolist()
    elif isinstance(obj, np.int64):
        return int(obj)
    # Let the base class default method raise the TypeError
    return json.JSONEncoder.default(self, obj)

TraitDef

Definition of how to compute a trait.

Attributes:

Name	Type	Description
name	str	Unique identifier for the trait.
fn	Callable	Function used to compute the trait's value.
input_traits	List[str]	List of trait names that should be computed before the current trait and are expected as input positional arguments to fn.
scalar	bool	Indicates if the trait is scalar (has a dimension of 0 per frame). If True, the trait is also listed in SCALAR_TRAITS.
include_in_csv	bool	Trueindicates the trait should be included in downstream CSV files.
kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the fn function. These arguments are not reused from previously computed traits.
description	Optional[str]	String describing the trait for documentation purposes.

Notes

The fn specified will be called with a pattern like:

trait_def = TraitDef(
    name="my_trait",
    fn=compute_my_trait,
    input_traits=["input_trait_1", "input_trait_2"],
    scalar=True,
    include_in_csv=True,
    kwargs={"kwarg1": True}
)
traits[trait_def.name] = trait_def.fn(
    *[traits[input_trait] for input_trait in trait_def.input_traits],
    **trait_def.kwargs
)

For this example, the last line is equivalent to:

traits["my_trait"] = trait_def.fn(
    traits["input_trait_1"], traits["input_trait_2"],
    kwarg1=True
)

Pipeline

Pipeline for computing traits.

Attributes:

Name	Type	Description
traits	List[TraitDef]	List of TraitDef objects.
trait_map	Dict[str, TraitDef]	Dictionary mapping trait names to their definitions.
trait_computation_order	List[str]	List of trait names in the order they should be computed.

summary_traits property

summary_traits: List[str]

List of traits to include in the summary CSV.

csv_traits property

csv_traits: List[str]

List of frame-level traits to include in the CSV.

csv_traits_multiple_plants property

csv_traits_multiple_plants: List[str]

List of frame-level traits to include in the CSV for multiple plants.

__attrs_post_init__

__attrs_post_init__()

Build pipeline objects from traits list.

Source code in sleap_roots/trait_pipelines.py

def __attrs_post_init__(self):
    """Build pipeline objects from traits list."""
    # Build list of trait definitions.
    self.traits = self.define_traits()

    # Check that trait names are unique.
    trait_names = [trait.name for trait in self.traits]
    if len(trait_names) != len(set(trait_names)):
        raise ValueError("Trait names must be unique.")

    # Map trait names to their definitions.
    self.trait_map = {trait_def.name: trait_def for trait_def in self.traits}

    # Determine computation order by topologically sorting the nodes.
    self.trait_computation_order = self.get_computation_order()

define_traits

define_traits() -> List[TraitDef]

Return list of TraitDef objects.

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Return list of `TraitDef` objects."""
    raise NotImplementedError

get_computation_order

get_computation_order() -> List[str]

Determine computation order by topologically sorting the nodes.

Returns:

Type	Description
List[str]	A list of trait names in the order they should be computed.

Source code in sleap_roots/trait_pipelines.py

def get_computation_order(self) -> List[str]:
    """Determine computation order by topologically sorting the nodes.

    Returns:
        A list of trait names in the order they should be computed.
    """
    # Infer edges from trait map.
    edges = []
    for trait_def in self.traits:
        for input_trait in trait_def.input_traits:
            edges.append((input_trait, trait_def.name))

    # Build networkx graph from inferred edges.
    G = nx.DiGraph()
    G.add_edges_from(edges)

    # Determine computation order by topologically sorting the nodes.
    trait_computation_order = list(nx.topological_sort(G))

    return trait_computation_order

compute_frame_traits

compute_frame_traits(
    traits: Dict[str, Any],
) -> Dict[str, Any]

Compute traits based on the pipeline.

Parameters:

Name	Type	Description	Default
traits	Dict[str, Any]	Dictionary of traits where keys are trait names and values are the trait values.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of computed traits.

Source code in sleap_roots/trait_pipelines.py

def compute_frame_traits(self, traits: Dict[str, Any]) -> Dict[str, Any]:
    """Compute traits based on the pipeline.

    Args:
        traits: Dictionary of traits where keys are trait names and values are
            the trait values.

    Returns:
        A dictionary of computed traits.
    """
    # Initialize traits container with initial data.
    traits = traits.copy()

    # Compute traits!
    for trait_name in self.trait_computation_order:
        if trait_name in traits:
            # Skip traits already computed.
            continue

        # Get trait definition.
        trait_def = self.trait_map[trait_name]

        # Compute trait based on trait definition.
        traits[trait_name] = trait_def.fn(
            *[traits[input_trait] for input_trait in trait_def.input_traits],
            **trait_def.kwargs,
        )

    return traits

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits.
Dict[str, Any]	This is defined on a per-pipeline basis as different plant species will have
Dict[str, Any]	different initial points to be used as starting traits.
Dict[str, Any]	Most commonly, this will be the primary and lateral root points for the
Dict[str, Any]	current frame.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits.

        This is defined on a per-pipeline basis as different plant species will have
        different initial points to be used as starting traits.

        Most commonly, this will be the primary and lateral root points for the
        current frame.
    """
    raise NotImplementedError

compute_plant_traits

compute_plant_traits(
    plant: Series,
    write_csv: bool = False,
    output_dir: str = ".",
    csv_suffix: str = ".traits.csv",
    return_non_scalar: bool = False,
) -> DataFrame

Compute traits for a plant.

Parameters:

Name	Type	Description	Default
plant	Series	The plant image series as a Series object.	required
write_csv	bool	A boolean value. If True, it writes per plant detailed CSVs with traits for every instance on every frame.	False
output_dir	str	The directory to write the CSV files to.	'.'
csv_suffix	str	If write_csv is True, a CSV file will be saved with the same name as the plant's {plant.series_name}{csv_suffix}.	'.traits.csv'
return_non_scalar	bool	If True, return all non-scalar traits as well as the summarized traits.	False

Returns:

Type	Description
DataFrame	The computed traits as a pandas DataFrame.

Source code in sleap_roots/trait_pipelines.py

def compute_plant_traits(
    self,
    plant: Series,
    write_csv: bool = False,
    output_dir: str = ".",
    csv_suffix: str = ".traits.csv",
    return_non_scalar: bool = False,
) -> pd.DataFrame:
    """Compute traits for a plant.

    Args:
        plant: The plant image series as a `Series` object.
        write_csv: A boolean value. If True, it writes per plant detailed
            CSVs with traits for every instance on every frame.
        output_dir: The directory to write the CSV files to.
        csv_suffix: If `write_csv` is `True`, a CSV file will be saved with the same
            name as the plant's `{plant.series_name}{csv_suffix}`.
        return_non_scalar: If `True`, return all non-scalar traits as well as the
            summarized traits.

    Returns:
        The computed traits as a pandas DataFrame.
    """
    traits = []
    for frame in range(len(plant)):
        # Get initial traits for the frame.
        initial_traits = self.get_initial_frame_traits(plant, frame)

        # Compute traits via the frame-level pipeline.
        frame_traits = self.compute_frame_traits(initial_traits)

        # Compute trait summaries.
        for trait_name in self.summary_traits:
            trait_summary = get_summary(
                frame_traits[trait_name], prefix=f"{trait_name}_"
            )
            frame_traits.update(trait_summary)

        # Add metadata.
        frame_traits["plant_name"] = plant.series_name
        frame_traits["frame_idx"] = frame
        traits.append(frame_traits)
    traits = pd.DataFrame(traits)

    # Move metadata columns to the front.
    plant_name = traits.pop("plant_name")
    frame_idx = traits.pop("frame_idx")
    traits = pd.concat([plant_name, frame_idx, traits], axis=1)

    if write_csv:
        csv_name = Path(output_dir) / f"{plant.series_name}{csv_suffix}"
        traits[["plant_name", "frame_idx"] + self.csv_traits].to_csv(
            csv_name, index=False
        )

    if return_non_scalar:
        return traits
    else:
        return traits[["plant_name", "frame_idx"] + self.csv_traits]

compute_multiple_dicots_traits

compute_multiple_dicots_traits(
    series: Series,
    write_json: bool = False,
    json_suffix: str = ".all_frames_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".all_frames_summary.csv",
) -> Dict[str, Any]

Computes plant traits for pipelines with multiple plants over all frames in a series.

Parameters:

Name	Type	Description	Default
series	Series	The Series object containing the primary and lateral root points.	required
write_json	bool	Whether to write the aggregated traits to a JSON file. Default is False.	False
json_suffix	str	The suffix to append to the JSON file name. Default is ".all_frames_traits.json".	'.all_frames_traits.json'
write_csv	bool	Whether to write the summary statistics to a CSV file. Default is False.	False
csv_suffix	str	The suffix to append to the CSV file name. Default is ".all_frames_summary.csv".	'.all_frames_summary.csv'

Returns:

Type	Description
Dict[str, Any]	Dictionary containing the series name, group, qc_fail, aggregated traits, and summary statistics.

Source code in sleap_roots/trait_pipelines.py

def compute_multiple_dicots_traits(
    self,
    series: Series,
    write_json: bool = False,
    json_suffix: str = ".all_frames_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".all_frames_summary.csv",
) -> Dict[str, Any]:
    """Computes plant traits for pipelines with multiple plants over all frames in a series.

    Args:
        series: The Series object containing the primary and lateral root points.
        write_json: Whether to write the aggregated traits to a JSON file. Default is False.
        json_suffix: The suffix to append to the JSON file name. Default is ".all_frames_traits.json".
        write_csv: Whether to write the summary statistics to a CSV file. Default is False.
        csv_suffix: The suffix to append to the CSV file name. Default is ".all_frames_summary.csv".

    Returns:
        Dictionary containing the series name, group, qc_fail, aggregated traits, and summary statistics.
    """
    # Initialize the return structure with the series name and group
    result = {
        "series": str(series.series_name),
        "group": str(series.group),
        "qc_fail": series.qc_fail,
        "traits": {},
        "summary_stats": {},
    }

    # Check if the series has frames to process
    if len(series) == 0:
        print(f"Series '{series.series_name}' contains no frames to process.")
        # Return early with the initialized structure
        return result

    # Initialize a separate dictionary to hold the aggregated traits across all frames
    aggregated_traits = {}

    # Iterate over frames in series
    for frame in range(len(series)):
        # Get initial points and number of plants per frame
        initial_frame_traits = self.get_initial_frame_traits(series, frame)
        # Compute initial associations and perform filter operations
        frame_traits = self.compute_frame_traits(initial_frame_traits)

        # Instantiate DicotPipeline
        dicot_pipeline = DicotPipeline()

        # Extract the plant associations for this frame
        associations = frame_traits["plant_associations_dict"]

        for primary_idx, assoc in associations.items():
            primary_pts = assoc["primary_points"]
            lateral_pts = assoc["lateral_points"]
            # Get the initial frame traits for this plant using the primary and lateral points
            initial_frame_traits = {
                "primary_pts": primary_pts,
                "lateral_pts": lateral_pts,
            }
            # Use the dicot pipeline to compute the plant traits on this frame
            plant_traits = dicot_pipeline.compute_frame_traits(initial_frame_traits)

            # For each plant's traits in the frame
            for trait_name, trait_value in plant_traits.items():
                # Not all traits are added to the aggregated traits dictionary
                if trait_name in dicot_pipeline.csv_traits_multiple_plants:
                    if trait_name not in aggregated_traits:
                        # Initialize the trait array if it's the first frame
                        aggregated_traits[trait_name] = [np.atleast_1d(trait_value)]
                    else:
                        # Append new trait values for subsequent frames
                        aggregated_traits[trait_name].append(
                            np.atleast_1d(trait_value)
                        )

    # After processing, update the result dictionary with computed traits
    for trait, arrays in aggregated_traits.items():
        aggregated_traits[trait] = np.concatenate(arrays, axis=0)
    result["traits"] = aggregated_traits

    # Write to JSON if requested
    if write_json:
        json_name = f"{series.series_name}{json_suffix}"
        try:
            with open(json_name, "w") as f:
                json.dump(
                    result, f, cls=NumpyArrayEncoder, ensure_ascii=False, indent=4
                )
            print(f"Aggregated traits saved to {json_name}")
        except IOError as e:
            print(f"Error writing JSON file '{json_name}': {e}")

    # Compute summary statistics and update result
    summary_stats = {}
    for trait_name, trait_values in aggregated_traits.items():
        trait_stats = get_summary(trait_values, prefix=f"{trait_name}_")
        summary_stats.update(trait_stats)
    result["summary_stats"] = summary_stats

    # Optionally write summary stats to CSV
    if write_csv:
        csv_name = f"{series.series_name}{csv_suffix}"
        try:
            summary_df = pd.DataFrame([summary_stats])
            summary_df.insert(0, "series", series.series_name)
            summary_df.to_csv(csv_name, index=False)
            print(f"Summary statistics saved to {csv_name}")
        except IOError as e:
            print(f"Failed to write CSV file '{csv_name}': {e}")

    # Return the final result structure
    return result

compute_multiple_dicots_traits_for_groups

compute_multiple_dicots_traits_for_groups(
    series_list: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    json_suffix: str = ".grouped_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".grouped_summary.csv",
) -> List[
    Dict[
        str,
        Union[
            str,
            List[str],
            Dict[str, Union[List[float], ndarray]],
        ],
    ]
]

Aggregates plant traits over groups of samples.

Parameters:

Name	Type	Description	Default
series_list	List[Series]	A list of Series objects containing the primary and lateral root points for each sample.	required
output_dir	str	The directory to write the JSON and CSV files to. Default is "grouped_traits".	'grouped_traits'
write_json	bool	Whether to write the aggregated traits to a JSON file. Default is False.	False
json_suffix	str	The suffix to append to the JSON file name. Default is ".grouped_traits.json".	'.grouped_traits.json'
write_csv	bool	Whether to write the summary statistics to a CSV file. Default is False.	False
csv_suffix	str	The suffix to append to the CSV file name. Default is ".grouped_summary.csv".	'.grouped_summary.csv'

Returns:

Type	Description
List[Dict[str, Union[str, List[str], Dict[str, Union[List[float], ndarray]]]]]	A list of dictionaries containing the aggregated traits and summary statistics for each group.

Source code in sleap_roots/trait_pipelines.py

def compute_multiple_dicots_traits_for_groups(
    self,
    series_list: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    json_suffix: str = ".grouped_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".grouped_summary.csv",
) -> List[
    Dict[str, Union[str, List[str], Dict[str, Union[List[float], np.ndarray]]]]
]:
    """Aggregates plant traits over groups of samples.

    Args:
        series_list: A list of Series objects containing the primary and lateral root points for each sample.
        output_dir: The directory to write the JSON and CSV files to. Default is "grouped_traits".
        write_json: Whether to write the aggregated traits to a JSON file. Default is False.
        json_suffix: The suffix to append to the JSON file name. Default is ".grouped_traits.json".
        write_csv: Whether to write the summary statistics to a CSV file. Default is False.
        csv_suffix: The suffix to append to the CSV file name. Default is ".grouped_summary.csv".

    Returns:
        A list of dictionaries containing the aggregated traits and summary statistics for each group.
    """
    # Input Validation
    if not isinstance(series_list, list) or not all(
        isinstance(series, Series) for series in series_list
    ):
        raise ValueError("series_list must be a list of Series objects.")

    # Group series by their group property
    series_groups = {}
    for series in series_list:
        # Exclude series with qc_fail flag set to 1
        if int(series.qc_fail) == 1:
            print(f"Skipping series '{series.series_name}' due to qc_fail flag.")
            continue
        # Get the group name from the series object
        group_name = str(series.group)
        if group_name not in series_groups:
            series_groups[group_name] = {"names": [], "series": []}
        # Store series names and objects in the dictionary
        series_groups[group_name]["names"].append(str(series.series_name))
        series_groups[group_name]["series"].append(series)  # Store Series objects

    # Initialize the list to hold the results for each group
    grouped_results = []
    # Iterate over each group of series
    for group_name, group_data in series_groups.items():
        # Initialize the return structure with the group name
        group_result = {
            "group": group_name,
            "series": group_data["names"],  # Use series names
            "traits": {},
        }

        # Aggregate traits over all samples in the group
        aggregated_traits = {}
        # Iterate over each series in the group
        for series in group_data["series"]:
            print(f"Processing series '{series.series_name}'")
            # Get the trait results for each series in the group
            result = self.compute_multiple_dicots_traits(
                series=series, write_json=False, write_csv=False
            )
            # Aggregate the series traits into the group traits
            for trait, values in result["traits"].items():
                # Ensure values are at least 1D
                values = np.atleast_1d(values)
                if trait not in aggregated_traits:
                    aggregated_traits[trait] = values
                else:
                    # Concatenate the current values with the existing array
                    aggregated_traits[trait] = np.concatenate(
                        (aggregated_traits[trait], values)
                    )

        group_result["traits"] = aggregated_traits
        print(f"Finished processing group '{group_name}'")

        # Write to JSON if requested
        if write_json:
            # Make the output directory if it doesn't exist
            Path(output_dir).mkdir(parents=True, exist_ok=True)
            # Construct the JSON file name
            json_name = f"{group_name}{json_suffix}"
            # Join the output directory with the JSON file name
            json_path = Path(output_dir) / json_name
            try:
                with open(json_path, "w") as f:
                    json.dump(
                        group_result,
                        f,
                        cls=NumpyArrayEncoder,
                        ensure_ascii=False,
                        indent=4,
                    )
                print(
                    f"Aggregated traits for group {group_name} saved to {str(json_path)}"
                )
            except IOError as e:
                print(f"Error writing JSON file '{str(json_path)}': {e}")

        # Compute summary statistics
        summary_stats = {}
        for trait, trait_values in aggregated_traits.items():
            trait_stats = get_summary(trait_values, prefix=f"{trait}_")
            summary_stats.update(trait_stats)

        group_result["summary_stats"] = summary_stats

        # Write summary stats to CSV if requested
        if write_csv:
            # Make the output directory if it doesn't exist
            Path(output_dir).mkdir(parents=True, exist_ok=True)
            # Construct the CSV file name
            csv_name = f"{group_name}{csv_suffix}"
            # Join the output directory with the CSV file name
            csv_path = Path(output_dir) / csv_name
            try:
                summary_df = pd.DataFrame([summary_stats])
                summary_df.insert(0, "genotype", group_name)
                summary_df.to_csv(csv_path, index=False)
                print(
                    f"Summary statistics for group {group_name} saved to {str(csv_path)}"
                )
            except IOError as e:
                print(f"Failed to write CSV file '{str(csv_path)}': {e}")

        # Append the group result to the list of results
        grouped_results.append(group_result)

    return grouped_results

compute_multiple_primary_roots_traits

compute_multiple_primary_roots_traits(
    series: Series,
    write_json: bool = False,
    json_suffix: str = ".all_frames_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".all_frames_summary.csv",
    per_instance: bool = False,
    flattened_csv_suffix: str = ".flattened_traits.csv",
) -> Union[Dict[str, Any], DataFrame]

Computes plant traits for pipelines with multiple primary roots over all frames in a series.

Parameters:

Name	Type	Description	Default
series	Series	The Series object containing the primary and lateral root points.	required
write_json	bool	Whether to write the aggregated traits to a JSON file. Default is False.	False
json_suffix	str	The suffix to append to the JSON file name. Default is ".all_frames_traits.json".	'.all_frames_traits.json'
write_csv	bool	Whether to write the summary statistics to a CSV file. Default is False.	False
csv_suffix	str	The suffix to append to the CSV file name. Default is ".all_frames_summary.csv".	'.all_frames_summary.csv'

Returns:

Type	Description
Union[Dict[str, Any], DataFrame]	Dictionary with aggregated traits, summary stats, and optionally per-instance traits.

Source code in sleap_roots/trait_pipelines.py

def compute_multiple_primary_roots_traits(
    self,
    series: Series,
    write_json: bool = False,
    json_suffix: str = ".all_frames_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".all_frames_summary.csv",
    per_instance: bool = False,
    flattened_csv_suffix: str = ".flattened_traits.csv",
) -> Union[Dict[str, Any], pd.DataFrame]:
    """Computes plant traits for pipelines with multiple primary roots over all frames in a series.

    Args:
        series: The Series object containing the primary and lateral root points.
        write_json: Whether to write the aggregated traits to a JSON file. Default is False.
        json_suffix: The suffix to append to the JSON file name. Default is ".all_frames_traits.json".
        write_csv: Whether to write the summary statistics to a CSV file. Default is False.
        csv_suffix: The suffix to append to the CSV file name. Default is ".all_frames_summary.csv".

    Returns:
        Dictionary with aggregated traits, summary stats, and optionally per-instance traits.
    """
    result = {
        "series": str(series.series_name),
        "group": str(series.group),
        "qc_fail": series.qc_fail,
        "traits": {},
        "summary_stats": {},
    }

    if per_instance:
        result["per_instance_traits"] = []

    if len(series) == 0:
        print(f"Series '{series.series_name}' contains no frames to process.")
        return result

    aggregated_traits = {}
    primary_root_pipeline = PrimaryRootPipeline()

    for frame_idx in range(len(series)):
        initial_frame_traits = self.get_initial_frame_traits(series, frame_idx)
        frame_traits = self.compute_frame_traits(initial_frame_traits)

        primary_root_instances = frame_traits[
            "filtered_primary_pts_with_expected_ct"
        ]

        for instance_idx, primary_root_inst in enumerate(primary_root_instances):
            inst_input = {"primary_pts": primary_root_inst}
            plant_traits = primary_root_pipeline.compute_frame_traits(inst_input)

            if per_instance:
                result["per_instance_traits"].append(
                    {
                        "frame": frame_idx,
                        "instance": instance_idx,
                        "traits": plant_traits,
                    }
                )

            for trait_name, trait_value in plant_traits.items():
                if trait_name in primary_root_pipeline.csv_traits_multiple_plants:
                    if trait_name not in aggregated_traits:
                        aggregated_traits[trait_name] = [np.atleast_1d(trait_value)]
                    else:
                        aggregated_traits[trait_name].append(
                            np.atleast_1d(trait_value)
                        )

    for trait, arrays in aggregated_traits.items():
        aggregated_traits[trait] = np.concatenate(arrays, axis=0)
    result["traits"] = aggregated_traits

    summary_stats = {}
    for trait_name, trait_values in aggregated_traits.items():
        trait_stats = get_summary(trait_values, prefix=f"{trait_name}_")
        summary_stats.update(trait_stats)
    result["summary_stats"] = summary_stats

    if write_json:
        json_name = f"{series.series_name}{json_suffix}"
        try:
            with open(json_name, "w") as f:
                json.dump(
                    result, f, cls=NumpyArrayEncoder, ensure_ascii=False, indent=4
                )
            print(f"Aggregated traits saved to {json_name}")
        except IOError as e:
            print(f"Error writing JSON file '{json_name}': {e}")

    if write_csv:
        csv_name = f"{series.series_name}{csv_suffix}"
        try:
            summary_df = pd.DataFrame([summary_stats])
            summary_df.insert(0, "series", series.series_name)
            summary_df.to_csv(csv_name, index=False)
            print(f"Summary statistics saved to {csv_name}")
        except IOError as e:
            print(f"Failed to write summary CSV '{csv_name}': {e}")

    flat_df = None
    if per_instance:
        try:
            rows = []
            for inst in result["per_instance_traits"]:
                row = {
                    "series": series.series_name,
                    "frame": inst["frame"],
                    "instance": inst["instance"],
                }
                for trait_name, trait_value in inst["traits"].items():
                    if isinstance(
                        trait_value, (int, float, np.integer, np.floating)
                    ):
                        row[trait_name] = trait_value
                    elif isinstance(trait_value, np.ndarray):
                        if trait_value.ndim == 0:
                            row[trait_name] = trait_value.item()
                        elif trait_value.ndim == 1 and trait_value.shape[0] == 1:
                            row[trait_name] = trait_value[0]
                        else:
                            continue
                    else:
                        continue
                rows.append(row)

            flat_df = pd.DataFrame(rows)

            if write_csv:
                flat_csv_name = f"{series.series_name}{flattened_csv_suffix}"
                flat_df.to_csv(flat_csv_name, index=False)
                print(f"Flattened per-instance traits saved to {flat_csv_name}")
        except Exception as e:
            print(f"Failed to process flattened traits: {e}")
        return flat_df
    else:
        return result

compute_multiple_primary_roots_traits_for_groups

compute_multiple_primary_roots_traits_for_groups(
    series_list: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    json_suffix: str = ".grouped_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".grouped_summary.csv",
) -> List[
    Dict[
        str,
        Union[
            str,
            List[str],
            Dict[str, Union[List[float], ndarray]],
        ],
    ]
]

Aggregates plant traits over groups of samples.

Parameters:

Name	Type	Description	Default
series_list	List[Series]	A list of Series objects containing the primary root points for each sample.	required
output_dir	str	The directory to write the JSON and CSV files to. Default is "grouped_traits".	'grouped_traits'
write_json	bool	Whether to write the aggregated traits to a JSON file. Default is False.	False
json_suffix	str	The suffix to append to the JSON file name. Default is ".grouped_traits.json".	'.grouped_traits.json'
write_csv	bool	Whether to write the summary statistics to a CSV file. Default is False.	False
csv_suffix	str	The suffix to append to the CSV file name. Default is ".grouped_summary.csv".	'.grouped_summary.csv'

Returns:

Type	Description
List[Dict[str, Union[str, List[str], Dict[str, Union[List[float], ndarray]]]]]	A list of dictionaries containing the aggregated traits and summary statistics for each group.

Source code in sleap_roots/trait_pipelines.py

def compute_multiple_primary_roots_traits_for_groups(
    self,
    series_list: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    json_suffix: str = ".grouped_traits.json",
    write_csv: bool = False,
    csv_suffix: str = ".grouped_summary.csv",
) -> List[
    Dict[str, Union[str, List[str], Dict[str, Union[List[float], np.ndarray]]]]
]:
    """Aggregates plant traits over groups of samples.

    Args:
        series_list: A list of Series objects containing the primary root points for each sample.
        output_dir: The directory to write the JSON and CSV files to. Default is "grouped_traits".
        write_json: Whether to write the aggregated traits to a JSON file. Default is False.
        json_suffix: The suffix to append to the JSON file name. Default is ".grouped_traits.json".
        write_csv: Whether to write the summary statistics to a CSV file. Default is False.
        csv_suffix: The suffix to append to the CSV file name. Default is ".grouped_summary.csv".

    Returns:
        A list of dictionaries containing the aggregated traits and summary statistics for each group.
    """
    # Input Validation
    if not isinstance(series_list, list) or not all(
        isinstance(series, Series) for series in series_list
    ):
        raise ValueError("series_list must be a list of Series objects.")

    # Group series by their group property
    series_groups = {}
    for series in series_list:
        # Exclude series with qc_fail flag set to 1
        if int(series.qc_fail) == 1:
            print(f"Skipping series '{series.series_name}' due to qc_fail flag.")
            continue
        # Get the group name from the series object
        group_name = str(series.group)
        if group_name not in series_groups:
            series_groups[group_name] = {"names": [], "series": []}
        # Store series names and objects in the dictionary
        series_groups[group_name]["names"].append(str(series.series_name))
        series_groups[group_name]["series"].append(series)  # Store Series objects

    # Initialize the list to hold the results for each group
    grouped_results = []
    # Iterate over each group of series
    for group_name, group_data in series_groups.items():
        # Initialize the return structure with the group name
        group_result = {
            "group": group_name,
            "series": group_data["names"],  # Use series names
            "traits": {},
        }

        # Aggregate traits over all samples in the group
        aggregated_traits = {}
        # Iterate over each series in the group
        for series in group_data["series"]:
            print(f"Processing series '{series.series_name}'")
            # Get the trait results for each series in the group
            result = self.compute_multiple_primary_roots_traits(
                series=series, write_json=False, write_csv=False
            )
            # Aggregate the series traits into the group traits
            for trait, values in result["traits"].items():
                # Ensure values are at least 1D
                values = np.atleast_1d(values)
                if trait not in aggregated_traits:
                    aggregated_traits[trait] = values
                else:
                    # Concatenate the current values with the existing array
                    aggregated_traits[trait] = np.concatenate(
                        (aggregated_traits[trait], values)
                    )

        group_result["traits"] = aggregated_traits
        print(f"Finished processing group '{group_name}'")

        # Write to JSON if requested
        if write_json:
            # Make the output directory if it doesn't exist
            Path(output_dir).mkdir(parents=True, exist_ok=True)
            # Construct the JSON file name
            json_name = f"{group_name}{json_suffix}"
            # Join the output directory with the JSON file name
            json_path = Path(output_dir) / json_name
            try:
                with open(json_path, "w") as f:
                    json.dump(
                        group_result,
                        f,
                        cls=NumpyArrayEncoder,
                        ensure_ascii=False,
                        indent=4,
                    )
                print(
                    f"Aggregated traits for group {group_name} saved to {str(json_path)}"
                )
            except IOError as e:
                print(f"Error writing JSON file '{str(json_path)}': {e}")

        # Compute summary statistics
        summary_stats = {}
        for trait, trait_values in aggregated_traits.items():
            trait_stats = get_summary(trait_values, prefix=f"{trait}_")
            summary_stats.update(trait_stats)

        group_result["summary_stats"] = summary_stats

        # Write summary stats to CSV if requested
        if write_csv:
            # Make the output directory if it doesn't exist
            Path(output_dir).mkdir(parents=True, exist_ok=True)
            # Construct the CSV file name
            csv_name = f"{group_name}{csv_suffix}"
            # Join the output directory with the CSV file name
            csv_path = Path(output_dir) / csv_name
            try:
                summary_df = pd.DataFrame([summary_stats])
                summary_df.insert(0, "genotype", group_name)
                summary_df.to_csv(csv_path, index=False)
                print(
                    f"Summary statistics for group {group_name} saved to {str(csv_path)}"
                )
            except IOError as e:
                print(f"Failed to write CSV file '{str(csv_path)}': {e}")

        # Append the group result to the list of results
        grouped_results.append(group_result)

    return grouped_results

compute_batch_traits

compute_batch_traits(
    plants: List[Series],
    write_csv: bool = False,
    csv_path: str = "traits.csv",
) -> DataFrame

Compute traits for a batch of plants.

Parameters:

Name	Type	Description	Default
plants	List[Series]	List of Series objects.	required
write_csv	bool	If True, write the computed traits to a CSV file.	False
csv_path	str	Path to write the CSV file to.	'traits.csv'

Returns:

Type	Description
DataFrame	A pandas DataFrame of computed traits summarized over all frames of each
DataFrame	plant. The resulting dataframe will have a row for each plant and a column
DataFrame	for each plant-level summarized trait.
DataFrame	Summarized traits are prefixed with the trait name and an underscore,
DataFrame	followed by the summary statistic.

Source code in sleap_roots/trait_pipelines.py

def compute_batch_traits(
    self,
    plants: List[Series],
    write_csv: bool = False,
    csv_path: str = "traits.csv",
) -> pd.DataFrame:
    """Compute traits for a batch of plants.

    Args:
        plants: List of `Series` objects.
        write_csv: If `True`, write the computed traits to a CSV file.
        csv_path: Path to write the CSV file to.

    Returns:
        A pandas DataFrame of computed traits summarized over all frames of each
        plant. The resulting dataframe will have a row for each plant and a column
        for each plant-level summarized trait.

        Summarized traits are prefixed with the trait name and an underscore,
        followed by the summary statistic.
    """
    all_traits = []
    for plant in plants:
        print(f"Processing series: {plant.series_name}")
        # Compute frame level traits for the plant.
        plant_traits = self.compute_plant_traits(plant)

        # Summarize frame level traits.
        plant_summary = {"plant_name": plant.series_name}
        for trait_name in self.csv_traits:
            trait_summary = get_summary(
                plant_traits[trait_name], prefix=f"{trait_name}_"
            )
            plant_summary.update(trait_summary)
        all_traits.append(plant_summary)

    # Build dataframe from list of frame-level summaries.
    all_traits = pd.DataFrame(all_traits)

    if write_csv:
        all_traits.to_csv(csv_path, index=False)
        print(f"Batch traits saved to {csv_path}")
    return all_traits

compute_batch_multiple_dicots_traits

compute_batch_multiple_dicots_traits(
    all_series: List[Series],
    write_csv: bool = False,
    csv_path: str = "traits.csv",
) -> DataFrame

Compute traits for a batch of series with multiple dicots.

Parameters:

Name	Type	Description	Default
all_series	List[Series]	List of Series objects.	required
write_csv	bool	If True, write the computed traits to a CSV file.	False
csv_path	str	Path to write the CSV file to.	'traits.csv'

Returns:

Type	Description
DataFrame	A pandas DataFrame of computed traits summarized over all frames of each
DataFrame	series. The resulting dataframe will have a row for each series and a column
DataFrame	for each series-level summarized trait.
DataFrame	Summarized traits are prefixed with the trait name and an underscore,
DataFrame	followed by the summary statistic.

Source code in sleap_roots/trait_pipelines.py

def compute_batch_multiple_dicots_traits(
    self,
    all_series: List[Series],
    write_csv: bool = False,
    csv_path: str = "traits.csv",
) -> pd.DataFrame:
    """Compute traits for a batch of series with multiple dicots.

    Args:
        all_series: List of `Series` objects.
        write_csv: If `True`, write the computed traits to a CSV file.
        csv_path: Path to write the CSV file to.

    Returns:
        A pandas DataFrame of computed traits summarized over all frames of each
        series. The resulting dataframe will have a row for each series and a column
        for each series-level summarized trait.

        Summarized traits are prefixed with the trait name and an underscore,
        followed by the summary statistic.
    """
    all_series_summaries = []

    for series in all_series:
        print(f"Processing series '{series.series_name}'")
        # Use the updated function and access its return value
        series_result = self.compute_multiple_dicots_traits(
            series, write_json=False, write_csv=False
        )
        # Prepare the series-level summary.
        series_summary = {
            "series_name": series_result["series"],
            **series_result["summary_stats"],  # Unpack summary_stats
        }
        all_series_summaries.append(series_summary)

    # Convert list of dictionaries to a DataFrame
    all_series_summaries_df = pd.DataFrame(all_series_summaries)

    # Write to CSV if requested
    if write_csv:
        all_series_summaries_df.to_csv(csv_path, index=False)
        print(f"Computed traits for all series saved to {csv_path}")

    return all_series_summaries_df

compute_batch_multiple_dicots_traits_for_groups

compute_batch_multiple_dicots_traits_for_groups(
    all_series: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    write_csv: bool = False,
    csv_path: str = "group_summarized_traits.csv",
) -> DataFrame

Compute traits for a batch of grouped series with multiple dicots.

Parameters:

Name	Type	Description	Default
all_series	List[Series]	List of Series objects.	required
output_dir	str	The directory to write the JSON and CSV files to. Default is "grouped_traits".	'grouped_traits'
write_json	bool	If True, write each set of group traits to a JSON file.	False
write_csv	bool	If True, write the computed traits to a CSV file.	False
csv_path	str	Path to write the CSV file to.	'group_summarized_traits.csv'

Returns:

Type	Description
DataFrame	A pandas DataFrame of computed traits summarized over all frames of each
DataFrame	group. The resulting dataframe will have a row for each series and a column
DataFrame	for each series-level summarized trait.
DataFrame	Summarized traits are prefixed with the trait name and an underscore,
DataFrame	followed by the summary statistic.

Source code in sleap_roots/trait_pipelines.py

def compute_batch_multiple_dicots_traits_for_groups(
    self,
    all_series: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    write_csv: bool = False,
    csv_path: str = "group_summarized_traits.csv",
) -> pd.DataFrame:
    """Compute traits for a batch of grouped series with multiple dicots.

    Args:
        all_series: List of `Series` objects.
        output_dir: The directory to write the JSON and CSV files to. Default is "grouped_traits".
        write_json: If `True`, write each set of group traits to a JSON file.
        write_csv: If `True`, write the computed traits to a CSV file.
        csv_path: Path to write the CSV file to.

    Returns:
        A pandas DataFrame of computed traits summarized over all frames of each
        group. The resulting dataframe will have a row for each series and a column
        for each series-level summarized trait.

        Summarized traits are prefixed with the trait name and an underscore,
        followed by the summary statistic.
    """
    # Check if the input list is empty
    if not all_series:
        raise ValueError("The input list 'all_series' is empty.")

    try:
        # Compute traits for each group of series
        grouped_results = self.compute_multiple_dicots_traits_for_groups(
            all_series,
            output_dir=output_dir,
            write_json=write_json,
            write_csv=False,
        )
    except Exception as e:
        raise RuntimeError(f"Error computing traits for groups: {e}")

    # Prepare the list of dictionaries for the DataFrame
    all_group_summaries = []
    for group_result in grouped_results:
        # Validate the expected key exists in the result
        if "summary_stats" not in group_result:
            raise KeyError(
                "Expected key 'summary_stats' not found in group result."
            )

        # Assuming 'group' key exists in group_result and it indicates the genotype
        genotype = group_result.get(
            "group", "Unknown Genotype"
        )  # Default to "Unknown Genotype" if not found

        # Start with a dictionary containing the genotype
        group_summary = {"genotype": genotype}

        # Add each trait statistic from the summary_stats dictionary to the group_summary
        # This assumes summary_stats is a dictionary where keys are trait names and values are the statistics
        for trait, statistic in group_result["summary_stats"].items():
            group_summary[trait] = statistic

        all_group_summaries.append(group_summary)

    # Create a DataFrame from the list of dictionaries
    all_group_summaries_df = pd.DataFrame(all_group_summaries)

    # Write to CSV if requested
    if write_csv:
        try:
            all_group_summaries_df.to_csv(csv_path, index=False)
            print(f"Computed traits for all groups saved to {csv_path}")
        except Exception as e:
            raise IOError(f"Failed to write computed traits to CSV: {e}")

    return all_group_summaries_df

compute_batch_multiple_primary_roots_traits

compute_batch_multiple_primary_roots_traits(
    all_series: List[Series],
    write_csv: bool = False,
    csv_path: str = "traits.csv",
) -> DataFrame

Compute traits for a batch of series with multiple primary roots.

Parameters:

Name	Type	Description	Default
all_series	List[Series]	List of Series objects.	required
write_csv	bool	If True, write the computed traits to a CSV file.	False
csv_path	str	Path to write the CSV file to.	'traits.csv'

Returns:

Type	Description
DataFrame	A pandas DataFrame of computed traits summarized over all frames of each
DataFrame	series. The resulting dataframe will have a row for each series and a column
DataFrame	for each series-level summarized trait.
DataFrame	Summarized traits are prefixed with the trait name and an underscore,
DataFrame	followed by the summary statistic.

Source code in sleap_roots/trait_pipelines.py

def compute_batch_multiple_primary_roots_traits(
    self,
    all_series: List[Series],
    write_csv: bool = False,
    csv_path: str = "traits.csv",
) -> pd.DataFrame:
    """Compute traits for a batch of series with multiple primary roots.

    Args:
        all_series: List of `Series` objects.
        write_csv: If `True`, write the computed traits to a CSV file.
        csv_path: Path to write the CSV file to.

    Returns:
        A pandas DataFrame of computed traits summarized over all frames of each
        series. The resulting dataframe will have a row for each series and a column
        for each series-level summarized trait.

        Summarized traits are prefixed with the trait name and an underscore,
        followed by the summary statistic.
    """
    all_series_summaries = []

    for series in all_series:
        print(f"Processing series '{series.series_name}'")
        # Use the updated function and access its return value
        series_result = self.compute_multiple_primary_roots_traits(
            series, write_json=False, write_csv=False
        )
        # Prepare the series-level summary.
        series_summary = {
            "series_name": series_result["series"],
            **series_result["summary_stats"],  # Unpack summary_stats
        }
        all_series_summaries.append(series_summary)

    # Convert list of dictionaries to a DataFrame
    all_series_summaries_df = pd.DataFrame(all_series_summaries)

    # Write to CSV if requested
    if write_csv:
        all_series_summaries_df.to_csv(csv_path, index=False)
        print(f"Computed traits for all series saved to {csv_path}")

    return all_series_summaries_df

compute_batch_multiple_primary_roots_traits_for_groups

compute_batch_multiple_primary_roots_traits_for_groups(
    all_series: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    write_csv: bool = False,
    csv_path: str = "group_summarized_traits.csv",
) -> DataFrame

Compute traits for a batch of grouped series with multiple primary roots.

Parameters:

Name	Type	Description	Default
all_series	List[Series]	List of Series objects.	required
output_dir	str	The directory to write the JSON and CSV files to. Default is "grouped_traits".	'grouped_traits'
write_json	bool	If True, write each set of group traits to a JSON file.	False
write_csv	bool	If True, write the computed traits to a CSV file.	False
csv_path	str	Path to write the CSV file to.	'group_summarized_traits.csv'

Returns:

Type	Description
DataFrame	A pandas DataFrame of computed traits summarized over all frames of each
DataFrame	group. The resulting dataframe will have a row for each series and a column
DataFrame	for each series-level summarized trait.
DataFrame	Summarized traits are prefixed with the trait name and an underscore,
DataFrame	followed by the summary statistic.

Source code in sleap_roots/trait_pipelines.py

def compute_batch_multiple_primary_roots_traits_for_groups(
    self,
    all_series: List[Series],
    output_dir: str = "grouped_traits",
    write_json: bool = False,
    write_csv: bool = False,
    csv_path: str = "group_summarized_traits.csv",
) -> pd.DataFrame:
    """Compute traits for a batch of grouped series with multiple primary roots.

    Args:
        all_series: List of `Series` objects.
        output_dir: The directory to write the JSON and CSV files to. Default is "grouped_traits".
        write_json: If `True`, write each set of group traits to a JSON file.
        write_csv: If `True`, write the computed traits to a CSV file.
        csv_path: Path to write the CSV file to.

    Returns:
        A pandas DataFrame of computed traits summarized over all frames of each
        group. The resulting dataframe will have a row for each series and a column
        for each series-level summarized trait.

        Summarized traits are prefixed with the trait name and an underscore,
        followed by the summary statistic.
    """
    # Check if the input list is empty
    if not all_series:
        raise ValueError("The input list 'all_series' is empty.")

    try:
        # Compute traits for each group of series
        grouped_results = self.compute_multiple_primary_roots_traits_for_groups(
            all_series,
            output_dir=output_dir,
            write_json=write_json,
            write_csv=False,
        )
    except Exception as e:
        raise RuntimeError(f"Error computing traits for groups: {e}")

    # Prepare the list of dictionaries for the DataFrame
    all_group_summaries = []
    for group_result in grouped_results:
        # Validate the expected key exists in the result
        if "summary_stats" not in group_result:
            raise KeyError(
                "Expected key 'summary_stats' not found in group result."
            )

        # Assuming 'group' key exists in group_result and it indicates the genotype
        genotype = group_result.get(
            "group", "Unknown Genotype"
        )  # Default to "Unknown Genotype" if not found

        # Start with a dictionary containing the genotype
        group_summary = {"genotype": genotype}

        # Add each trait statistic from the summary_stats dictionary to the group_summary
        # This assumes summary_stats is a dictionary where keys are trait names and values are the statistics
        for trait, statistic in group_result["summary_stats"].items():
            group_summary[trait] = statistic

        all_group_summaries.append(group_summary)

    # Create a DataFrame from the list of dictionaries
    all_group_summaries_df = pd.DataFrame(all_group_summaries)

    # Write to CSV if requested
    if write_csv:
        try:
            all_group_summaries_df.to_csv(csv_path, index=False)
            print(f"Computed traits for all groups saved to {csv_path}")
        except Exception as e:
            raise IOError(f"Failed to write computed traits to CSV: {e}")

    return all_group_summaries_df

DicotPipeline

Bases: Pipeline

Pipeline for computing traits for dicot plants (primary + lateral roots).

Attributes:

Name	Type	Description
img_height	int	Image height.
root_width_tolerance	float	Difference in projection norm between right and left side.
n_scanlines	int	Number of scan lines, np.nan for no interaction.
network_fraction	float	Length found in the lower fraction value of the network.

define_traits

define_traits() -> List[TraitDef]

Define the trait computation pipeline for dicot plants.

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Define the trait computation pipeline for dicot plants."""
    trait_definitions = [
        TraitDef(
            name="primary_max_length_pts",
            fn=get_max_length_pts,
            input_traits=["primary_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Points of the primary root with maximum length.",
        ),
        TraitDef(
            name="pts_all_array",
            fn=get_all_pts_array,
            input_traits=["primary_max_length_pts", "lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Landmark points within a given frame as a flat array"
            "of coordinates.",
        ),
        TraitDef(
            name="pts_list",
            fn=join_pts,
            input_traits=["primary_max_length_pts", "lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="A list of instance arrays, each having shape `(nodes, 2)`.",
        ),
        TraitDef(
            name="root_widths",
            fn=get_root_widths,
            input_traits=["primary_max_length_pts", "lateral_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={
                "tolerance": self.root_width_tolerance,
                "return_inds": False,
            },
            description="Estimate root width using bases of lateral roots.",
        ),
        TraitDef(
            name="lateral_count",
            fn=get_count,
            input_traits=["lateral_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Get the number of lateral roots.",
        ),
        TraitDef(
            name="lateral_proximal_node_inds",
            fn=get_node_ind,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": True},
            description="Get the indices of the proximal nodes of lateral roots.",
        ),
        TraitDef(
            name="lateral_distal_node_inds",
            fn=get_node_ind,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": False},
            description="Get the indices of the distal nodes of lateral roots.",
        ),
        TraitDef(
            name="lateral_lengths",
            fn=get_root_lengths,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of lateral root lengths of shape `(instances,)`.",
        ),
        TraitDef(
            name="lateral_base_pts",
            fn=get_bases,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of lateral bases `(instances, (x, y))`.",
        ),
        TraitDef(
            name="lateral_tip_pts",
            fn=get_tips,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of lateral tips `(instances, (x, y))`.",
        ),
        TraitDef(
            name="scanline_intersection_counts",
            fn=count_scanline_intersections,
            input_traits=["pts_list"],
            scalar=False,
            include_in_csv=True,
            kwargs={
                "height": self.img_height,
                "n_line": self.n_scanlines,
            },
            description="Array of intersections of each scanline `(n_scanlines,)`.",
        ),
        TraitDef(
            name="lateral_angles_distal",
            fn=get_root_angle,
            input_traits=["lateral_pts", "lateral_distal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": False, "base_ind": 0},
            description="Array of lateral distal angles in degrees `(instances,)`.",
        ),
        TraitDef(
            name="lateral_angles_proximal",
            fn=get_root_angle,
            input_traits=["lateral_pts", "lateral_proximal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": True, "base_ind": 0},
            description="Array of lateral proximal angles in degrees "
            "`(instances,)`.",
        ),
        TraitDef(
            name="network_solidity",
            fn=get_network_solidity,
            input_traits=["network_length", "chull_area"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the total network length divided by the network"
            "convex area.",
        ),
        TraitDef(
            name="ellipse",
            fn=fit_ellipse,
            input_traits=["pts_all_array"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of (a, b, ratio) containing the semi-major axis "
            "length, semi-minor axis length, and the ratio of the major to minor "
            "lengths.",
        ),
        TraitDef(
            name="bounding_box",
            fn=get_bbox,
            input_traits=["pts_all_array"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of four parameters in bounding box.",
        ),
        TraitDef(
            name="convex_hull",
            fn=get_convhull,
            input_traits=["pts_all_array"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Convex hull of the points.",
        ),
        TraitDef(
            name="primary_proximal_node_ind",
            fn=get_node_ind,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=False,
            kwargs={"proximal": True},
            description="Get the indices of the proximal nodes of primary roots.",
        ),
        TraitDef(
            name="primary_angle_proximal",
            fn=get_root_angle,
            input_traits=["primary_max_length_pts", "primary_proximal_node_ind"],
            scalar=True,
            include_in_csv=True,
            kwargs={"proximal": True, "base_ind": 0},
            description="Array of primary proximal angles in degrees "
            "`(instances,)`.",
        ),
        TraitDef(
            name="primary_distal_node_ind",
            fn=get_node_ind,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=False,
            kwargs={"proximal": False},
            description="Get the indices of the distal nodes of primary roots.",
        ),
        TraitDef(
            name="primary_angle_distal",
            fn=get_root_angle,
            input_traits=["primary_max_length_pts", "primary_distal_node_ind"],
            scalar=True,
            include_in_csv=True,
            kwargs={"proximal": False, "base_ind": 0},
            description="Array of primary distal angles in degrees `(instances,)`.",
        ),
        TraitDef(
            name="primary_length",
            fn=get_root_lengths,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of primary root length.",
        ),
        TraitDef(
            name="primary_base_pt",
            fn=get_bases,
            input_traits=["primary_max_length_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Primary root base point.",
        ),
        TraitDef(
            name="primary_tip_pt",
            fn=get_tips,
            input_traits=["primary_max_length_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Primary root tip point.",
        ),
        TraitDef(
            name="network_length_lower",
            fn=get_network_distribution,
            input_traits=["pts_list", "bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={
                "fraction": self.network_fraction,
            },
            description="Scalar of the root network length in the lower fraction "
            "of the plant.",
        ),
        TraitDef(
            name="lateral_base_xs",
            fn=get_base_xs,
            input_traits=["lateral_base_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Get x coordinates of the base of each lateral root.",
        ),
        TraitDef(
            name="lateral_base_ys",
            fn=get_base_ys,
            input_traits=["lateral_base_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the y-coordinates of lateral bases "
            "`(instances,)`.",
        ),
        TraitDef(
            name="base_ct_density",
            fn=get_base_ct_density,
            input_traits=["primary_length", "lateral_base_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of base count density.",
        ),
        TraitDef(
            name="lateral_tip_xs",
            fn=get_tip_xs,
            input_traits=["lateral_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the x-coordinates of lateral tips `(instance,)`.",
        ),
        TraitDef(
            name="lateral_tip_ys",
            fn=get_tip_ys,
            input_traits=["lateral_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the y-coordinates of lateral tips `(instance,)`.",
        ),
        TraitDef(
            name="network_distribution_ratio",
            fn=get_network_distribution_ratio,
            input_traits=["network_length", "network_length_lower"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of ratio of the root network length in the lower "
            "fraction of the plant over all root length.",
        ),
        TraitDef(
            name="network_length",
            fn=get_network_length,
            input_traits=["primary_length", "lateral_lengths"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of all roots network length.",
        ),
        TraitDef(
            name="primary_base_pt_y",
            fn=get_base_ys,
            input_traits=["primary_base_pt"],
            scalar=True,
            include_in_csv=False,
            kwargs={},
            description="Y-coordinate of the primary root base node.",
        ),
        TraitDef(
            name="primary_tip_pt_y",
            fn=get_tip_ys,
            input_traits=["primary_tip_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Y-coordinate of the primary root tip node.",
        ),
        TraitDef(
            name="ellipse_a",
            fn=get_ellipse_a,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of semi-major axis length.",
        ),
        TraitDef(
            name="ellipse_b",
            fn=get_ellipse_b,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of semi-minor axis length.",
        ),
        TraitDef(
            name="network_width_depth_ratio",
            fn=get_network_width_depth_ratio,
            input_traits=["bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of bounding box width to depth ratio of root "
            "network.",
        ),
        TraitDef(
            name="chull_perimeter",
            fn=get_chull_perimeter,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull perimeter.",
        ),
        TraitDef(
            name="chull_area",
            fn=get_chull_area,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull area.",
        ),
        TraitDef(
            name="chull_max_width",
            fn=get_chull_max_width,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull maximum width.",
        ),
        TraitDef(
            name="chull_max_height",
            fn=get_chull_max_height,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull maximum height.",
        ),
        TraitDef(
            name="chull_line_lengths",
            fn=get_chull_line_lengths,
            input_traits=["convex_hull"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of line lengths connecting any two vertices on the"
            "convex hull.",
        ),
        TraitDef(
            name="base_length",
            fn=get_base_length,
            input_traits=["lateral_base_ys"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the distance between the top and deepest base"
            "y-coordinates.",
        ),
        TraitDef(
            name="base_median_ratio",
            fn=get_base_median_ratio,
            input_traits=["lateral_base_ys", "primary_tip_pt_y"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of base median ratio.",
        ),
        TraitDef(
            name="curve_index",
            fn=get_curve_index,
            input_traits=["primary_length", "primary_base_tip_dist"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of primary root curvature index.",
        ),
        TraitDef(
            name="base_length_ratio",
            fn=get_base_length_ratio,
            input_traits=["primary_length", "base_length"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of base length ratio.",
        ),
        TraitDef(
            name="primary_base_tip_dist",
            fn=get_base_tip_dist,
            input_traits=["primary_base_pt", "primary_tip_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of distance from primary root base to tip.",
        ),
        TraitDef(
            name="ellipse_ratio",
            fn=get_ellipse_ratio,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of ratio of the minor to major lengths.",
        ),
        TraitDef(
            name="scanline_last_ind",
            fn=get_scanline_last_ind,
            input_traits=["scanline_intersection_counts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of count_scanline_interaction index for the last"
            "interaction.",
        ),
        TraitDef(
            name="scanline_first_ind",
            fn=get_scanline_first_ind,
            input_traits=["scanline_intersection_counts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of count_scanline_interaction index for the first"
            "interaction.",
        ),
    ]

    return trait_definitions

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits with keys: - "primary_pts": Array of primary root points. - "lateral_pts": Array of lateral root points.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits with keys:
            - "primary_pts": Array of primary root points.
            - "lateral_pts": Array of lateral root points.
    """
    primary_pts = plant.get_primary_points(frame_idx)
    lateral_pts = plant.get_lateral_points(frame_idx)
    return {"primary_pts": primary_pts, "lateral_pts": lateral_pts}

YoungerMonocotPipeline

Bases: Pipeline

Pipeline for computing traits for young monocot plants (primary + crown roots).

Attributes:

Name	Type	Description
img_height	int	Image height.
n_scanlines	int	Number of scan lines, np.nan for no interaction.
network_fraction	float	Lower fraction value. Defaults to 2/3.

define_traits

define_traits() -> List[TraitDef]

Define the trait computation pipeline for younger monocot plants.

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Define the trait computation pipeline for younger monocot plants."""
    trait_definitions = [
        TraitDef(
            name="primary_max_length_pts",
            fn=get_max_length_pts,
            input_traits=["primary_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Points of the primary root with maximum length.",
        ),
        TraitDef(
            name="pts_all_array",
            fn=get_all_pts_array,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Crown root points within a given frame as a flat array"
            "of coordinates.",
        ),
        TraitDef(
            name="crown_count",
            fn=get_count,
            input_traits=["crown_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Get the number of crown roots.",
        ),
        TraitDef(
            name="crown_proximal_node_inds",
            fn=get_node_ind,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": True},
            description="Get the indices of the proximal nodes of crown roots.",
        ),
        TraitDef(
            name="crown_distal_node_inds",
            fn=get_node_ind,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": False},
            description="Get the indices of the distal nodes of crown roots.",
        ),
        TraitDef(
            name="crown_lengths",
            fn=get_root_lengths,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of crown root lengths of shape `(instances,)`.",
        ),
        TraitDef(
            name="crown_base_pts",
            fn=get_bases,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of crown bases `(instances, (x, y))`.",
        ),
        TraitDef(
            name="crown_tip_pts",
            fn=get_tips,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of crown tips `(instances, (x, y))`.",
        ),
        TraitDef(
            name="scanline_intersection_counts",
            fn=count_scanline_intersections,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={
                "height": self.img_height,
                "n_line": self.n_scanlines,
            },
            description="Array of intersections of each scanline"
            "`(n_scanlines,)`.",
        ),
        TraitDef(
            name="crown_angles_distal",
            fn=get_root_angle,
            input_traits=["crown_pts", "crown_distal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": False, "base_ind": 0},
            description="Array of crown distal angles in degrees `(instances,)`.",
        ),
        TraitDef(
            name="crown_angles_proximal",
            fn=get_root_angle,
            input_traits=["crown_pts", "crown_proximal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": True, "base_ind": 0},
            description="Array of crown proximal angles in degrees "
            "`(instances,)`.",
        ),
        TraitDef(
            name="network_length_lower",
            fn=get_network_distribution,
            input_traits=[
                "crown_pts",
                "bounding_box",
            ],
            scalar=True,
            include_in_csv=True,
            kwargs={
                "fraction": self.network_fraction,
            },
            description="Scalar of the root network length in the lower fraction "
            "of the plant.",
        ),
        TraitDef(
            name="ellipse",
            fn=fit_ellipse,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of (a, b, ratio) containing the semi-major axis "
            "length, semi-minor axis length, and the ratio of the major to minor "
            "lengths.",
        ),
        TraitDef(
            name="bounding_box",
            fn=get_bbox,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of four parameters in bounding box.",
        ),
        TraitDef(
            name="convex_hull",
            fn=get_convhull,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Convex hull of the points.",
        ),
        TraitDef(
            name="primary_proximal_node_ind",
            fn=get_node_ind,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=False,
            kwargs={"proximal": True},
            description="Get the indices of the proximal nodes of primary roots.",
        ),
        TraitDef(
            name="primary_angle_proximal",
            fn=get_root_angle,
            input_traits=[
                "primary_max_length_pts",
                "primary_proximal_node_ind",
            ],
            scalar=True,
            include_in_csv=True,
            kwargs={"proximal": True, "base_ind": 0},
            description="Array of primary proximal angles in degrees "
            "`(instances,)`.",
        ),
        TraitDef(
            name="primary_distal_node_ind",
            fn=get_node_ind,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=False,
            kwargs={"proximal": False},
            description="Get the indices of the distal nodes of primary roots.",
        ),
        TraitDef(
            name="primary_angle_distal",
            fn=get_root_angle,
            input_traits=["primary_max_length_pts", "primary_distal_node_ind"],
            scalar=True,
            include_in_csv=True,
            kwargs={"proximal": False, "base_ind": 0},
            description="Array of primary distal angles in degrees `(instances,)`.",
        ),
        TraitDef(
            name="primary_length",
            fn=get_root_lengths,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of primary root length.",
        ),
        TraitDef(
            name="primary_base_pt",
            fn=get_bases,
            input_traits=["primary_max_length_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Primary root base point.",
        ),
        TraitDef(
            name="primary_tip_pt",
            fn=get_tips,
            input_traits=["primary_max_length_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Primary root tip point.",
        ),
        TraitDef(
            name="crown_tip_xs",
            fn=get_tip_xs,
            input_traits=["crown_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the x-coordinates of crown tips `(instance,)`.",
        ),
        TraitDef(
            name="crown_tip_ys",
            fn=get_tip_ys,
            input_traits=["crown_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the y-coordinates of crown tips `(instance,)`.",
        ),
        TraitDef(
            name="network_distribution_ratio",
            fn=get_network_distribution_ratio,
            input_traits=[
                "network_length",
                "network_length_lower",
            ],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of ratio of the root network length in the lower"
            "fraction of the plant over all root length.",
        ),
        TraitDef(
            name="network_length",
            fn=get_network_length,
            input_traits=["crown_lengths"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of all roots network length.",
        ),
        TraitDef(
            name="crown_base_tip_dists",
            fn=get_base_tip_dist,
            input_traits=["crown_base_pts", "crown_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Straight-line distance(s) from the base(s) to the"
            "tip(s) of the crown root(s).",
        ),
        TraitDef(
            name="crown_curve_indices",
            fn=get_curve_index,
            input_traits=["crown_lengths", "crown_base_tip_dists"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Curvature index for each crown root.",
        ),
        TraitDef(
            name="network_solidity",
            fn=get_network_solidity,
            input_traits=["network_length", "chull_area"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the total network length divided by the"
            "network convex hull area.",
        ),
        TraitDef(
            name="primary_tip_pt_y",
            fn=get_tip_ys,
            input_traits=["primary_tip_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Y-coordinate of the primary root tip node.",
        ),
        TraitDef(
            name="ellipse_a",
            fn=get_ellipse_a,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of semi-major axis length.",
        ),
        TraitDef(
            name="ellipse_b",
            fn=get_ellipse_b,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of semi-minor axis length.",
        ),
        TraitDef(
            name="network_width_depth_ratio",
            fn=get_network_width_depth_ratio,
            input_traits=["bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of bounding box width to depth ratio of root "
            "network.",
        ),
        TraitDef(
            name="chull_perimeter",
            fn=get_chull_perimeter,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull perimeter.",
        ),
        TraitDef(
            name="chull_area",
            fn=get_chull_area,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull area.",
        ),
        TraitDef(
            name="chull_max_width",
            fn=get_chull_max_width,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull maximum width.",
        ),
        TraitDef(
            name="chull_max_height",
            fn=get_chull_max_height,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull maximum height.",
        ),
        TraitDef(
            name="chull_line_lengths",
            fn=get_chull_line_lengths,
            input_traits=["convex_hull"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of line lengths connecting any two vertices on the"
            "convex hull.",
        ),
        TraitDef(
            name="curve_index",
            fn=get_curve_index,
            input_traits=["primary_length", "primary_base_tip_dist"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of primary root curvature index.",
        ),
        TraitDef(
            name="primary_base_tip_dist",
            fn=get_base_tip_dist,
            input_traits=["primary_base_pt", "primary_tip_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of distance from primary root base to tip.",
        ),
        TraitDef(
            name="ellipse_ratio",
            fn=get_ellipse_ratio,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of ratio of the minor to major lengths.",
        ),
        TraitDef(
            name="scanline_last_ind",
            fn=get_scanline_last_ind,
            input_traits=["scanline_intersection_counts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of count_scanline_interaction index for the last"
            "interaction.",
        ),
        TraitDef(
            name="scanline_first_ind",
            fn=get_scanline_first_ind,
            input_traits=["scanline_intersection_counts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of count_scanline_interaction index for the first"
            "interaction.",
        ),
    ]

    return trait_definitions

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits with keys: - "primary_pts": Array of primary root points. - "crown_pts": Array of crown root points.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits with keys:
            - "primary_pts": Array of primary root points.
            - "crown_pts": Array of crown root points.
    """
    primary_pts = plant.get_primary_points(frame_idx)
    crown_pts = plant.get_crown_points(frame_idx)
    return {"primary_pts": primary_pts, "crown_pts": crown_pts}

OlderMonocotPipeline

Bases: Pipeline

Pipeline for computing traits for older monocot plants (crown roots only).

Attributes:

Name	Type	Description
img_height	int	Image height.
n_scanlines	int	Number of scan lines, np.nan for no interaction.
network_fraction	float	Lower fraction value. Defaults to 2/3.

define_traits

define_traits() -> List[TraitDef]

Define the trait computation pipeline for older monocot plants (crown roots).

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Define the trait computation pipeline for older monocot plants (crown roots)."""
    trait_definitions = [
        TraitDef(
            name="pts_all_array",
            fn=get_all_pts_array,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Landmark points within a given frame as a flat array"
            "of coordinates.",
        ),
        TraitDef(
            name="crown_count",
            fn=get_count,
            input_traits=["crown_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Get the number of crown roots.",
        ),
        TraitDef(
            name="crown_proximal_node_inds",
            fn=get_node_ind,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": True},
            description="Get the indices of the proximal nodes of crown roots.",
        ),
        TraitDef(
            name="crown_distal_node_inds",
            fn=get_node_ind,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": False},
            description="Get the indices of the distal nodes of crown roots.",
        ),
        TraitDef(
            name="crown_lengths",
            fn=get_root_lengths,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of crown root lengths of shape `(instances,)`.",
        ),
        TraitDef(
            name="crown_base_pts",
            fn=get_bases,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of crown bases `(instances, (x, y))`.",
        ),
        TraitDef(
            name="crown_tip_pts",
            fn=get_tips,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of crown tips `(instances, (x, y))`.",
        ),
        TraitDef(
            name="scanline_intersection_counts",
            fn=count_scanline_intersections,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={
                "height": self.img_height,
                "n_line": self.n_scanlines,
            },
            description="Array of intersections of each scanline"
            "`(n_scanlines,)`.",
        ),
        TraitDef(
            name="crown_angles_distal",
            fn=get_root_angle,
            input_traits=["crown_pts", "crown_distal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": False, "base_ind": 0},
            description="Array of crown distal angles in degrees `(instances,)`.",
        ),
        TraitDef(
            name="crown_angles_proximal",
            fn=get_root_angle,
            input_traits=["crown_pts", "crown_proximal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": True, "base_ind": 0},
            description="Array of crown proximal angles in degrees "
            "`(instances,)`.",
        ),
        TraitDef(
            name="network_length_lower",
            fn=get_network_distribution,
            input_traits=[
                "crown_pts",
                "bounding_box",
            ],
            scalar=True,
            include_in_csv=True,
            kwargs={
                "fraction": self.network_fraction,
            },
            description="Scalar of the root network length in the lower fraction "
            "of the plant.",
        ),
        TraitDef(
            name="ellipse",
            fn=fit_ellipse,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of (a, b, ratio) containing the semi-major axis "
            "length, semi-minor axis length, and the ratio of the major to minor "
            "lengths.",
        ),
        TraitDef(
            name="bounding_box",
            fn=get_bbox,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of four parameters representing bounding box.",
        ),
        TraitDef(
            name="convex_hull",
            fn=get_convhull,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Convex hull of the crown points.",
        ),
        TraitDef(
            name="crown_tip_xs",
            fn=get_tip_xs,
            input_traits=["crown_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the x-coordinates of crown tips `(instance,)`.",
        ),
        TraitDef(
            name="crown_tip_ys",
            fn=get_tip_ys,
            input_traits=["crown_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the y-coordinates of crown tips `(instance,)`.",
        ),
        TraitDef(
            name="network_distribution_ratio",
            fn=get_network_distribution_ratio,
            input_traits=[
                "network_length",
                "network_length_lower",
            ],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of ratio of the root network length in the lower"
            "fraction of the plant over all root length.",
        ),
        TraitDef(
            name="network_length",
            fn=get_network_length,
            input_traits=["crown_lengths"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of all roots network length.",
        ),
        TraitDef(
            name="crown_base_tip_dists",
            fn=get_base_tip_dist,
            input_traits=["crown_base_pts", "crown_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Straight-line distance(s) from the base(s) to the"
            "tip(s) of the crown root(s).",
        ),
        TraitDef(
            name="crown_curve_indices",
            fn=get_curve_index,
            input_traits=["crown_lengths", "crown_base_tip_dists"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Curvature index for each crown root.",
        ),
        TraitDef(
            name="network_solidity",
            fn=get_network_solidity,
            input_traits=["network_length", "chull_area"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the total network length divided by the"
            "network convex hull area.",
        ),
        TraitDef(
            name="ellipse_a",
            fn=get_ellipse_a,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of semi-major axis length.",
        ),
        TraitDef(
            name="ellipse_b",
            fn=get_ellipse_b,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of semi-minor axis length.",
        ),
        TraitDef(
            name="network_width_depth_ratio",
            fn=get_network_width_depth_ratio,
            input_traits=["bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of bounding box width to depth ratio of root "
            "network.",
        ),
        TraitDef(
            name="chull_perimeter",
            fn=get_chull_perimeter,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull perimeter.",
        ),
        TraitDef(
            name="chull_area",
            fn=get_chull_area,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull area.",
        ),
        TraitDef(
            name="chull_max_width",
            fn=get_chull_max_width,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull maximum width.",
        ),
        TraitDef(
            name="chull_max_height",
            fn=get_chull_max_height,
            input_traits=["convex_hull"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of convex hull maximum height.",
        ),
        TraitDef(
            name="chull_line_lengths",
            fn=get_chull_line_lengths,
            input_traits=["convex_hull"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of line lengths connecting any two vertices on the"
            "convex hull.",
        ),
        TraitDef(
            name="ellipse_ratio",
            fn=get_ellipse_ratio,
            input_traits=["ellipse"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of ratio of the minor to major lengths.",
        ),
        TraitDef(
            name="scanline_last_ind",
            fn=get_scanline_last_ind,
            input_traits=["scanline_intersection_counts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of count_scanline_interaction index for the last"
            "interaction.",
        ),
        TraitDef(
            name="scanline_first_ind",
            fn=get_scanline_first_ind,
            input_traits=["scanline_intersection_counts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of count_scanline_interaction index for the first"
            "interaction.",
        ),
        TraitDef(
            name="crown_r1_pts",
            fn=get_nodes,
            input_traits=["crown_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"node_index": 1},
            description="Array of crown bases `(instances, (x, y))`.",
        ),
        TraitDef(
            name="chull_r1_intersection_vectors",
            fn=get_chull_intersection_vectors,
            input_traits=[
                "crown_base_pts",
                "crown_r1_pts",
                "crown_pts",
                "convex_hull",
            ],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="A tuple containing vectors from the top left point to the"
            "left intersection point, and from the top right point to the right"
            "intersection point with the convex hull.",
        ),
        TraitDef(
            name="chull_r1_left_intersection_vector",
            fn=get_chull_intersection_vectors_left,
            input_traits=["chull_r1_intersection_vectors"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Vector from the base point to the left"
            "intersection point with the convex hull.",
        ),
        TraitDef(
            name="chull_r1_right_intersection_vector",
            fn=get_chull_intersection_vectors_right,
            input_traits=["chull_r1_intersection_vectors"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Vector from the base point to the right"
            "intersection point with the convex hull.",
        ),
        TraitDef(
            name="angle_chull_r1_left_intersection_vector",
            fn=get_vector_angles_from_gravity,
            input_traits=["chull_r1_left_intersection_vector"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Angle of the left intersection vector from gravity.",
        ),
        TraitDef(
            name="angle_chull_r1_right_intersection_vector",
            fn=get_vector_angles_from_gravity,
            input_traits=["chull_r1_right_intersection_vector"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Angle of the right intersection vector from gravity.",
        ),
        TraitDef(
            name="chull_areas_r1_intersection",
            fn=get_chull_areas_via_intersection,
            input_traits=["crown_r1_pts", "crown_pts", "convex_hull"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of the convex hull areas above and below the r1"
            "intersection.",
        ),
        TraitDef(
            name="chull_area_above_r1_intersection",
            fn=get_chull_area_via_intersection_above,
            input_traits=["chull_areas_r1_intersection"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the convex hull area above the r1 intersection.",
        ),
        TraitDef(
            name="chull_area_below_r1_intersection",
            fn=get_chull_area_via_intersection_below,
            input_traits=["chull_areas_r1_intersection"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the convex hull area below the r1 intersection.",
        ),
    ]

    return trait_definitions

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits with keys: - "crown_pts": Array of crown root points.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits with keys:
            - "crown_pts": Array of crown root points.
    """
    crown_pts = plant.get_crown_points(frame_idx)
    return {"crown_pts": crown_pts}

MultipleDicotPipeline

Bases: Pipeline

Pipeline for computing traits for multiple dicot plants.

define_traits

define_traits() -> List[TraitDef]

Define the trait computation pipeline for primary roots.

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Define the trait computation pipeline for primary roots."""
    trait_definitions = [
        TraitDef(
            name="primary_pts_no_nans",
            fn=filter_roots_with_nans,
            input_traits=["primary_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Primary roots without any NaNs.",
        ),
        TraitDef(
            name="lateral_pts_no_nans",
            fn=filter_roots_with_nans,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Lateral roots without any NaNs.",
        ),
        TraitDef(
            name="filtered_pts_expected_plant_ct",
            fn=filter_plants_with_unexpected_ct,
            input_traits=[
                "primary_pts_no_nans",
                "lateral_pts_no_nans",
                "expected_plant_ct",
            ],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of filtered points with expected plant count.",
        ),
        TraitDef(
            name="primary_pts_expected_plant_ct",
            fn=get_filtered_primary_pts,
            input_traits=["filtered_pts_expected_plant_ct"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Filtered primary root points with expected plant count.",
        ),
        TraitDef(
            name="lateral_pts_expected_plant_ct",
            fn=get_filtered_lateral_pts,
            input_traits=["filtered_pts_expected_plant_ct"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Filtered lateral root points with expected plant count.",
        ),
        TraitDef(
            name="plant_associations_dict",
            fn=associate_lateral_to_primary,
            input_traits=[
                "primary_pts_expected_plant_ct",
                "lateral_pts_expected_plant_ct",
            ],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Dictionary of plant associations.",
        ),
    ]

    return trait_definitions

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits with keys: - "primary_pts": Array of primary root points. - "lateral_pts": Array of lateral root points. - "expected_ct": Expected number of plants as a float.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits with keys:
            - "primary_pts": Array of primary root points.
            - "lateral_pts": Array of lateral root points.
            - "expected_ct": Expected number of plants as a float.
    """
    primary_pts = plant.get_primary_points(frame_idx)
    lateral_pts = plant.get_lateral_points(frame_idx)
    expected_plant_ct = plant.expected_count
    return {
        "primary_pts": primary_pts,
        "lateral_pts": lateral_pts,
        "expected_plant_ct": expected_plant_ct,
    }

MultipleDicotPlatePipeline

Bases: Pipeline

Pipeline for multi-plant dicot plate images (issue #126 PR 1).

Primary + lateral root handling only; tertiary support is deferred to PR 2. Distinct from MultipleDicotPipeline (cylinder, ~72 rotational frames, cross-frame aggregation) because plates have one frame per timepoint and require per-plant scalar output. The frame loop is designed to support future timelapse (>1 frame per series) without architectural changes — more frames → more per-plant rows.

Design invariants:

(a) Plates skip count-filter (design D2): the TraitDef DAG does NOT include filter_plants_with_unexpected_ct. The pipeline keeps all detected plants regardless of expected_count, and surfaces the discrepancy via expected_count / detected_count output columns.

(b) plant_id is a left-to-right ordering (sorted by primary base x-coordinate) paired with original SLEAP indices (primary_sleap_idx, lateral_sleap_idxs). plant_id is NOT stable across SLEAP model re-prediction.

(c) Per-plant traits reuse DicotPipeline trait names unchanged (design D6). No renaming to primary_root_length etc. — the project-wide convention uses primary_*, lateral_*, network_*, crown_* prefixes; no _root_ infix.

(d) primary_base_tip_dist (Euclidean base-to-tip distance) is the substituted "depth" trait for PR 1. Issue #126 originally defined primary_root_depth as "max y-extent (deepest node y − base node y)" — a dedicated max-y-extent trait is tracked in follow-up F.

(e) qc_fail inherits Series.qc_fail's cylinder-specific semantics (reads CSV column qc_cylinder); plate-aware QC tracked in follow-up E.

(f) expected_count inherits Series.expected_count's cylinder-specific column name (number_of_plants_cylinder); plate-aware naming tracked in follow-up E.

(g) filter_roots_with_nans drops any primary with even one NaN node (whole-root filter); this compounds with plant_id fragility described in (b).

(h) NO back-mapping key is stable across SLEAP model re-prediction. Both plant_id AND primary_sleap_idx can shift if predictions at the confidence threshold flip between runs. For cross-run alignment (e.g., matching plants across re-predicted timelapse frames), use spatial matching (nearest-base-x within tolerance) rather than either identifier.

define_traits

define_traits() -> List[TraitDef]

Return the plate TraitDef DAG (primary + lateral; no count-filter).

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Return the plate TraitDef DAG (primary + lateral; no count-filter)."""
    return [
        TraitDef(
            name="primary_pts_no_nans",
            fn=filter_roots_with_nans,
            input_traits=["primary_pts"],
            scalar=False,
            include_in_csv=False,
            description="Primary roots with NaN instances removed.",
        ),
        TraitDef(
            name="lateral_pts_no_nans",
            fn=filter_roots_with_nans,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            description="Lateral roots with NaN instances removed.",
        ),
        TraitDef(
            name="detected_count",
            fn=get_count,
            input_traits=["primary_pts_no_nans"],
            scalar=True,
            include_in_csv=True,
            description="Number of valid primary roots detected in the frame.",
        ),
        TraitDef(
            name="plant_associations_dict",
            fn=associate_lateral_to_primary,
            input_traits=["primary_pts_no_nans", "lateral_pts_no_nans"],
            scalar=False,
            include_in_csv=False,
            description="Dict mapping primary-instance index to primary + "
            "lateral points for that plant.",
        ),
        TraitDef(
            name="plant_id_order",
            fn=argsort_primaries_by_base_x,
            input_traits=["plant_associations_dict"],
            scalar=False,
            include_in_csv=False,
            description="Primary keys sorted left-to-right by base-node x.",
        ),
    ]

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial per-frame traits plus original SLEAP indices.

Computes primary_sleap_idxs / lateral_sleap_idxs BEFORE filtering so consumers can map post-filter results back to the original .slp instance indices. filter_roots_with_nans collapses indices without preserving a mapping, so the validity mask must be recorded here.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	Frame index.	required

Returns:

Type	Description
Dict[str, Any]	Dict with keys primary_pts, lateral_pts (raw pre-filter arrays
Dict[str, Any]	of shape (n_instances, n_nodes, 2)), primary_sleap_idxs,
Dict[str, Any]	lateral_sleap_idxs (lists of original SLEAP indices of valid
Dict[str, Any]	instances), and expected_count.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial per-frame traits plus original SLEAP indices.

    Computes `primary_sleap_idxs` / `lateral_sleap_idxs` BEFORE filtering so
    consumers can map post-filter results back to the original `.slp` instance
    indices. `filter_roots_with_nans` collapses indices without preserving a
    mapping, so the validity mask must be recorded here.

    Args:
        plant: The plant `Series` object.
        frame_idx: Frame index.

    Returns:
        Dict with keys `primary_pts`, `lateral_pts` (raw pre-filter arrays
        of shape `(n_instances, n_nodes, 2)`), `primary_sleap_idxs`,
        `lateral_sleap_idxs` (lists of original SLEAP indices of valid
        instances), and `expected_count`.
    """
    primary_pts_raw = plant.get_primary_points(frame_idx)
    lateral_pts_raw = plant.get_lateral_points(frame_idx)
    primary_sleap_idxs = [
        i for i, r in enumerate(primary_pts_raw) if not np.isnan(r).any()
    ]
    lateral_sleap_idxs = [
        i for i, r in enumerate(lateral_pts_raw) if not np.isnan(r).any()
    ]
    return {
        "primary_pts": primary_pts_raw,
        "lateral_pts": lateral_pts_raw,
        "primary_sleap_idxs": primary_sleap_idxs,
        "lateral_sleap_idxs": lateral_sleap_idxs,
        "expected_count": plant.expected_count,
    }

compute_plate_traits

compute_plate_traits(
    series: Series,
    write_csv: bool = False,
    write_json: bool = False,
    output_dir: str = ".",
    csv_suffix: str = ".plate_traits.csv",
    json_suffix: str = ".plate_traits.json",
) -> Dict[str, Any]

Run the plate pipeline on a Series (one or more frames).

Returns a per-series dict with a flat per-plant-per-frame plants list. Each plant row carries its plant_id (left-to-right by primary base x), primary_sleap_idx (original SLEAP index for back-mapping to the .slp), raw primary_points / lateral_points, count_validated / count_mismatch booleans (JSON-only; not CSV columns), and the full DicotPipeline trait set.

Parameters:

Name	Type	Description	Default
series	Series	The plant Series.	required
write_csv	bool	If True, write per-plant CSV to output_dir/series_name{csv_suffix}.	False
write_json	bool	If True, write the self-contained JSON to output_dir/series_name{json_suffix}. NaN values are converted to JSON null via pre-serialization sanitization.	False
output_dir	str	Directory for CSV/JSON output. Defaults to CWD.	'.'
csv_suffix	str	Filename suffix for CSV output.	'.plate_traits.csv'
json_suffix	str	Filename suffix for JSON output.	'.plate_traits.json'

Returns:

Type	Description
Dict[str, Any]	Dict with keys schema_version (currently 2), units, series,
Dict[str, Any]	sample_uid, timepoint, group, qc_fail, expected_count,
Dict[str, Any]	and plants (list of per-plant-per-frame dicts). Each plant row
Dict[str, Any]	additionally carries sample_uid and timepoint.
Dict[str, Any]	Note on schema_version: schema_version is JSON-only
Dict[str, Any]	(and present in the in-memory dict). It is NOT written to the
Dict[str, Any]	per-plant CSV — CSV consumers must read by column NAME (not
Dict[str, Any]	position) to remain compatible across schema versions. Positional
Dict[str, Any]	CSV readers cannot programmatically detect the version change
Dict[str, Any]	from schema_version=1 (PR #165) to schema_version=2 (this
Dict[str, Any]	change adds sample_uid/timepoint at columns 1 and 2); they
Dict[str, Any]	will silently misread. Migrate positional readers to name-based
Dict[str, Any]	indexing (df["primary_length"] instead of df.iloc[:, 6]).

Logging

Emits at most one WARNING per call to logger sleap_roots.trait_pipelines when units["time"] == "unspecified" AND series.timepoint is non-NaN — surfaces the missing-unit hazard so downstream consumers don't silently mix days/seconds. The warning fires regardless of len(series) (including empty series with non-NaN CSV-derived timepoint), so a config issue is visible even when no plant rows are produced. Removed once #177 (time_unit population) lands.

Source code in sleap_roots/trait_pipelines.py

def compute_plate_traits(
    self,
    series: Series,
    write_csv: bool = False,
    write_json: bool = False,
    output_dir: str = ".",
    csv_suffix: str = ".plate_traits.csv",
    json_suffix: str = ".plate_traits.json",
) -> Dict[str, Any]:
    """Run the plate pipeline on a Series (one or more frames).

    Returns a per-series dict with a flat per-plant-per-frame `plants` list.
    Each plant row carries its `plant_id` (left-to-right by primary base x),
    `primary_sleap_idx` (original SLEAP index for back-mapping to the .slp),
    raw `primary_points` / `lateral_points`, `count_validated` / `count_mismatch`
    booleans (JSON-only; not CSV columns), and the full DicotPipeline trait set.

    Args:
        series: The plant `Series`.
        write_csv: If True, write per-plant CSV to `output_dir/series_name{csv_suffix}`.
        write_json: If True, write the self-contained JSON to
            `output_dir/series_name{json_suffix}`. NaN values are converted
            to JSON `null` via pre-serialization sanitization.
        output_dir: Directory for CSV/JSON output. Defaults to CWD.
        csv_suffix: Filename suffix for CSV output.
        json_suffix: Filename suffix for JSON output.

    Returns:
        Dict with keys `schema_version` (currently `2`), `units`, `series`,
        `sample_uid`, `timepoint`, `group`, `qc_fail`, `expected_count`,
        and `plants` (list of per-plant-per-frame dicts). Each plant row
        additionally carries `sample_uid` and `timepoint`.

        **Note on `schema_version`:** `schema_version` is JSON-only
        (and present in the in-memory dict). It is NOT written to the
        per-plant CSV — CSV consumers must read by column NAME (not
        position) to remain compatible across schema versions. Positional
        CSV readers cannot programmatically detect the version change
        from `schema_version=1` (PR #165) to `schema_version=2` (this
        change adds `sample_uid`/`timepoint` at columns 1 and 2); they
        will silently misread. Migrate positional readers to name-based
        indexing (`df["primary_length"]` instead of `df.iloc[:, 6]`).

    Logging:
        Emits at most one WARNING per call to logger
        `sleap_roots.trait_pipelines` when `units["time"] == "unspecified"`
        AND `series.timepoint` is non-NaN — surfaces the missing-unit
        hazard so downstream consumers don't silently mix days/seconds.
        The warning fires regardless of `len(series)` (including empty
        series with non-NaN CSV-derived timepoint), so a config issue is
        visible even when no plant rows are produced.
        Removed once #177 (time_unit population) lands.
    """
    # Resolve sample_uid + timepoint ONCE per series call (perf contract).
    # Per-plant rows MUST NOT re-read from `series` inside the frame loop;
    # at plate-timelapse scale (10 series × 100 frames × 6 plants) this
    # would trigger 6000 redundant CSV reads.
    #
    # Scope note: this contract applies to sample_uid + timepoint ONLY.
    # The other CSV-backed properties (group, qc_fail, expected_count)
    # are still read at result-dict construction below — each triggers
    # one pd.read_csv per series. A full lazy-cache fix is tracked by
    # #176; this PR partially mitigates by hoisting expected_count_raw
    # from result["expected_count"] (saves one redundant read per series)
    # but still pays for `series.group`, `series.qc_fail`, and the
    # per-frame `expected_count` read in `get_initial_frame_traits`.
    sample_uid_resolved = str(series.sample_uid)
    timepoint_resolved = series.timepoint

    result: Dict[str, Any] = {
        "schema_version": 2,
        "units": dict(_PLATE_UNITS),
        "series": str(series.series_name),
        "sample_uid": sample_uid_resolved,
        "timepoint": timepoint_resolved,
        "group": series.group,
        "qc_fail": series.qc_fail,
        "expected_count": series.expected_count,
        "plants": [],
    }

    # Per-call warning when units["time"] is "unspecified" AND timepoint is
    # numeric — surfaces the missing-unit hazard at runtime so two
    # collaborators recording timepoint=3 in days vs seconds don't silently
    # produce incompatible data. Each compute_plate_traits call emits at
    # most one such warning; there is no cross-call dedup (intentional —
    # batch processing should still surface each series's hazard).
    # Removed once the time_unit follow-up (#177) lands.
    if result["units"].get("time") == "unspecified" and not pd.isna(
        timepoint_resolved
    ):
        logger.warning(
            "MultipleDicotPlatePipeline: %s timepoint=%r emitted with "
            "units['time']='unspecified'; downstream consumers cannot "
            "interpret this value as a physical duration (e.g. days, "
            "hours, minutes, frames are all valid but mutually "
            "incompatible). Set time_unit once #177 lands.",
            series.series_name,
            timepoint_resolved,
        )

    if len(series) == 0:
        return result

    # Build DicotPipeline once per series call — reused across all plants
    # and frames. `compute_frame_traits` doesn't mutate pipeline state so
    # sharing is safe. Avoids rebuilding the ~25-node trait DAG + topsort
    # per plant row (Copilot review feedback on PR #165).
    dicot_pipeline = DicotPipeline()
    warned_frames: set = set()
    # Reuse the value already resolved into the result dict above to avoid
    # a second pd.read_csv on the same file. The full CSV-read-per-property
    # pattern (group, qc_fail, expected_count) is tracked by #176.
    expected_count_raw = result["expected_count"]

    for frame_idx in range(len(series)):
        initial = self.get_initial_frame_traits(series, frame_idx)
        frame_traits = self.compute_frame_traits(initial)
        associations = frame_traits["plant_associations_dict"]
        plant_id_order = frame_traits["plant_id_order"]
        detected_count = int(frame_traits["detected_count"])

        # Build lateral-sleap-idx map using distance-based association that
        # tracks original SLEAP indices — robust to bit-identical duplicate
        # lateral coordinates.
        lateral_idx_by_primary = self._assign_laterals_to_primaries_by_distance(
            frame_traits["primary_pts_no_nans"],
            frame_traits["lateral_pts_no_nans"],
            initial["lateral_sleap_idxs"],
        )

        # Per-frame mismatch warning (design D5b; per-(series, frame) dedup).
        expected_numeric: Optional[int] = None
        if expected_count_raw is not None:
            try:
                ec_float = float(expected_count_raw)
                if not math.isnan(ec_float):
                    expected_numeric = int(round(ec_float))
            except (TypeError, ValueError):
                expected_numeric = None
        if (
            expected_numeric is not None
            and expected_numeric != detected_count
            and (series.series_name, frame_idx) not in warned_frames
        ):
            logger.warning(
                f"MultipleDicotPlatePipeline: {series.series_name} "
                f"frame {frame_idx} detected {detected_count} primaries "
                f"but expected {expected_count_raw}; no plants dropped"
            )
            warned_frames.add((series.series_name, frame_idx))

        for plant_id, primary_idx in enumerate(plant_id_order):
            assoc = associations[primary_idx]
            primary_sleap_idx = initial["primary_sleap_idxs"][primary_idx]
            lateral_sleap_idxs_for_plant = lateral_idx_by_primary.get(
                primary_idx, []
            )
            plant_row = self._build_plant_row(
                series=series,
                frame_idx=frame_idx,
                plant_id=plant_id,
                primary_sleap_idx=primary_sleap_idx,
                lateral_sleap_idxs_for_plant=lateral_sleap_idxs_for_plant,
                assoc=assoc,
                expected_count=expected_count_raw,
                detected_count=detected_count,
                dicot_pipeline=dicot_pipeline,
                sample_uid=sample_uid_resolved,
                timepoint=timepoint_resolved,
            )
            result["plants"].append(plant_row)

    if write_json:
        json_path = Path(output_dir) / f"{series.series_name}{json_suffix}"
        sanitized = _json_sanitize(result)
        with open(json_path.as_posix(), "w") as f:
            json.dump(
                sanitized,
                f,
                cls=NumpyArrayEncoder,
                allow_nan=False,
                ensure_ascii=False,
                indent=4,
            )

    if write_csv:
        csv_path = Path(output_dir) / f"{series.series_name}{csv_suffix}"
        df = self._build_plate_dataframe(result, dicot_pipeline=dicot_pipeline)
        df.to_csv(csv_path.as_posix(), index=False)

    return result

compute_batch_plate_traits

compute_batch_plate_traits(
    all_series: List[Series],
    write_csv: bool = False,
    write_json: bool = False,
    output_dir: str = ".",
    csv_name: str = "plate_batch_traits.csv",
    json_name: str = "plate_batch_traits.json",
) -> DataFrame

Run compute_plate_traits across multiple Series; concatenate rows.

Parameters:

Name	Type	Description	Default
all_series	List[Series]	List of Series objects to process.	required
write_csv	bool	If True, write concatenated DataFrame to output_dir/csv_name.	False
write_json	bool	If True, write a list of per-series dicts to output_dir/json_name (each per-series dict has the same self-contained shape as compute_plate_traits's return).	False
output_dir	str	Directory for batch output.	'.'
csv_name	str	Filename for batch CSV.	'plate_batch_traits.csv'
json_name	str	Filename for batch JSON.	'plate_batch_traits.json'

Returns:

Type	Description
DataFrame	A concatenated pandas.DataFrame across all input Series.

Source code in sleap_roots/trait_pipelines.py

def compute_batch_plate_traits(
    self,
    all_series: List[Series],
    write_csv: bool = False,
    write_json: bool = False,
    output_dir: str = ".",
    csv_name: str = "plate_batch_traits.csv",
    json_name: str = "plate_batch_traits.json",
) -> pd.DataFrame:
    """Run `compute_plate_traits` across multiple Series; concatenate rows.

    Args:
        all_series: List of `Series` objects to process.
        write_csv: If True, write concatenated DataFrame to
            `output_dir/csv_name`.
        write_json: If True, write a list of per-series dicts to
            `output_dir/json_name` (each per-series dict has the same
            self-contained shape as `compute_plate_traits`'s return).
        output_dir: Directory for batch output.
        csv_name: Filename for batch CSV.
        json_name: Filename for batch JSON.

    Returns:
        A concatenated `pandas.DataFrame` across all input Series.
    """
    # Share one DicotPipeline across all series in the batch — avoids
    # rebuilding its ~25-node trait DAG once per series. compute_plate_traits
    # constructs its own internally (one per call), but the DataFrame builder
    # here explicitly shares across the batch.
    shared_dicot = DicotPipeline()
    per_series_results: List[Dict[str, Any]] = []
    per_series_dfs: List[pd.DataFrame] = []
    for series in all_series:
        result = self.compute_plate_traits(series)
        per_series_results.append(result)
        per_series_dfs.append(
            self._build_plate_dataframe(result, dicot_pipeline=shared_dicot)
        )

    if per_series_dfs:
        batch_df = pd.concat(per_series_dfs, axis=0, ignore_index=True)
    else:
        batch_df = pd.DataFrame()

    if write_csv:
        csv_path = Path(output_dir) / csv_name
        batch_df.to_csv(csv_path.as_posix(), index=False)
    if write_json:
        json_path = Path(output_dir) / json_name
        sanitized = _json_sanitize(per_series_results)
        with open(json_path.as_posix(), "w") as f:
            json.dump(
                sanitized,
                f,
                cls=NumpyArrayEncoder,
                allow_nan=False,
                ensure_ascii=False,
                indent=4,
            )

    return batch_df

PrimaryRootPipeline

Bases: Pipeline

Pipeline for computing traits for a single primary root.

define_traits

define_traits() -> List[TraitDef]

Define the trait computation pipeline for primary roots.

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Define the trait computation pipeline for primary roots."""
    trait_definitions = [
        TraitDef(
            name="primary_max_length_pts",
            fn=get_max_length_pts,
            input_traits=["primary_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Points of the primary root with maximum length.",
        ),
        TraitDef(
            name="primary_proximal_node_ind",
            fn=get_node_ind,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=False,
            kwargs={"proximal": True},
            description="Get the indices of the proximal nodes of primary root.",
        ),
        TraitDef(
            name="primary_distal_node_ind",
            fn=get_node_ind,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=False,
            kwargs={"proximal": False},
            description="Get the indices of the distal nodes of primary roots.",
        ),
        TraitDef(
            name="primary_angle_proximal",
            fn=get_root_angle,
            input_traits=["primary_max_length_pts", "primary_proximal_node_ind"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the primary proximal angle in degrees.",
        ),
        TraitDef(
            name="primary_angle_distal",
            fn=get_root_angle,
            input_traits=["primary_max_length_pts", "primary_distal_node_ind"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the primary distal angle in degrees.",
        ),
        TraitDef(
            name="primary_length",
            fn=get_root_lengths,
            input_traits=["primary_max_length_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the primary root length.",
        ),
        TraitDef(
            name="primary_base_pt",
            fn=get_bases,
            input_traits=["primary_max_length_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Primary root base point.",
        ),
        TraitDef(
            name="primary_tip_pt",
            fn=get_tips,
            input_traits=["primary_max_length_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Primary root tip point.",
        ),
        TraitDef(
            name="primary_base_pt_x",
            fn=get_base_xs,
            input_traits=["primary_base_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="X-coordinate of the primary root base node.",
        ),
        TraitDef(
            name="primary_base_pt_y",
            fn=get_base_ys,
            input_traits=["primary_base_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Y-coordinate of the primary root base node.",
        ),
        TraitDef(
            name="primary_tip_pt_x",
            fn=get_tip_xs,
            input_traits=["primary_tip_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="X-coordinate of the primary root tip node.",
        ),
        TraitDef(
            name="primary_tip_pt_y",
            fn=get_tip_ys,
            input_traits=["primary_tip_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Y-coordinate of the primary root tip node.",
        ),
        TraitDef(
            name="primary_base_tip_dist",
            fn=get_base_tip_dist,
            input_traits=["primary_base_pt", "primary_tip_pt"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of distance from primary root base to tip.",
        ),
        TraitDef(
            name="curve_index",
            fn=get_curve_index,
            input_traits=["primary_length", "primary_base_tip_dist"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the primary root curvature index.",
        ),
        TraitDef(
            name="bounding_box",
            fn=get_bbox,
            input_traits=["primary_max_length_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Tuple of four parameters in bounding box.",
        ),
        TraitDef(
            name="bounding_box_left_x",
            fn=get_bbox_left_x,
            input_traits=["bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the left x-axis value of the bounding box.",
        ),
        TraitDef(
            name="bounding_box_top_y",
            fn=get_bbox_top_y,
            input_traits=["bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the y-axis value of top side of the bounding box.",
        ),
        TraitDef(
            name="bounding_box_width",
            fn=get_bbox_width,
            input_traits=["bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the width of the bounding box.",
        ),
        TraitDef(
            name="bounding_box_height",
            fn=get_bbox_height,
            input_traits=["bounding_box"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of the height of the bounding box.",
        ),
    ]
    return trait_definitions

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits with key: - "primary_pts": Array of primary root points.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits with key:
            - "primary_pts": Array of primary root points.
    """
    primary_pts = plant.get_primary_points(frame_idx)
    return {"primary_pts": primary_pts}

MultiplePrimaryRootPipeline

Bases: Pipeline

Pipeline for computing traits for multiple primary roots.

define_traits

define_traits() -> List[TraitDef]

Define the trait computation pipeline for primary roots.

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Define the trait computation pipeline for primary roots."""
    trait_definitions = [
        TraitDef(
            name="filtered_primary_pts_with_expected_ct",
            fn=filter_primary_roots_with_unexpected_count,
            input_traits=["primary_pts", "expected_plant_ct"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Filtered points of the primary root with expected count.",
        )
    ]
    return trait_definitions

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits with key: - "primary_pts": Array of primary root points.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits with key:
            - "primary_pts": Array of primary root points.
    """
    primary_pts = plant.get_primary_points(frame_idx)
    expected_plant_count = plant.expected_count

    return {"primary_pts": primary_pts, "expected_plant_ct": expected_plant_count}

LateralRootPipeline

Bases: Pipeline

Pipeline just for computing traits for lateral roots.

define_traits

define_traits() -> List[TraitDef]

Define the trait computation pipeline for dicot plants.

Source code in sleap_roots/trait_pipelines.py

def define_traits(self) -> List[TraitDef]:
    """Define the trait computation pipeline for dicot plants."""
    trait_definitions = [
        TraitDef(
            name="lateral_count",
            fn=get_count,
            input_traits=["lateral_pts"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Get the number of lateral roots.",
        ),
        TraitDef(
            name="lateral_proximal_node_inds",
            fn=get_node_ind,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": True},
            description="Get the indices of the proximal nodes of lateral roots.",
        ),
        TraitDef(
            name="lateral_distal_node_inds",
            fn=get_node_ind,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={"proximal": False},
            description="Get the indices of the distal nodes of lateral roots.",
        ),
        TraitDef(
            name="lateral_lengths",
            fn=get_root_lengths,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of lateral root lengths of shape `(instances,)`.",
        ),
        TraitDef(
            name="total_lateral_length",
            fn=get_network_length,
            input_traits=["lateral_lengths"],
            scalar=True,
            include_in_csv=True,
            kwargs={},
            description="Scalar of all lateral root network length.",
        ),
        TraitDef(
            name="lateral_base_pts",
            fn=get_bases,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of lateral bases `(instances, (x, y))`.",
        ),
        TraitDef(
            name="lateral_tip_pts",
            fn=get_tips,
            input_traits=["lateral_pts"],
            scalar=False,
            include_in_csv=False,
            kwargs={},
            description="Array of lateral tips `(instances, (x, y))`.",
        ),
        TraitDef(
            name="lateral_angles_distal",
            fn=get_root_angle,
            input_traits=["lateral_pts", "lateral_distal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": False, "base_ind": 0},
            description="Array of lateral distal angles in degrees `(instances,)`.",
        ),
        TraitDef(
            name="lateral_angles_proximal",
            fn=get_root_angle,
            input_traits=["lateral_pts", "lateral_proximal_node_inds"],
            scalar=False,
            include_in_csv=True,
            kwargs={"proximal": True, "base_ind": 0},
            description="Array of lateral proximal angles in degrees "
            "`(instances,)`.",
        ),
        TraitDef(
            name="lateral_base_xs",
            fn=get_base_xs,
            input_traits=["lateral_base_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Get x coordinates of the base of each lateral root.",
        ),
        TraitDef(
            name="lateral_base_ys",
            fn=get_base_ys,
            input_traits=["lateral_base_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the y-coordinates of lateral bases "
            "`(instances,)`.",
        ),
        TraitDef(
            name="lateral_tip_xs",
            fn=get_tip_xs,
            input_traits=["lateral_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the x-coordinates of lateral tips `(instance,)`.",
        ),
        TraitDef(
            name="lateral_tip_ys",
            fn=get_tip_ys,
            input_traits=["lateral_tip_pts"],
            scalar=False,
            include_in_csv=True,
            kwargs={},
            description="Array of the y-coordinates of lateral tips `(instance,)`.",
        ),
    ]
    return trait_definitions

get_initial_frame_traits

get_initial_frame_traits(
    plant: Series, frame_idx: int
) -> Dict[str, Any]

Return initial traits for a plant frame.

Parameters:

Name	Type	Description	Default
plant	Series	The plant Series object.	required
frame_idx	int	The index of the current frame.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary of initial traits with keys: - "lateral_pts": Array of lateral root points.

Source code in sleap_roots/trait_pipelines.py

def get_initial_frame_traits(self, plant: Series, frame_idx: int) -> Dict[str, Any]:
    """Return initial traits for a plant frame.

    Args:
        plant: The plant `Series` object.
        frame_idx: The index of the current frame.

    Returns:
        A dictionary of initial traits with keys:
            - "lateral_pts": Array of lateral root points.
    """
    lateral_pts = plant.get_lateral_points(frame_idx)
    return {"lateral_pts": lateral_pts}