megabouts.pipeline#

class megabouts.pipeline.freely_swimming_pipeline.EthogramHeadTracking(segments, bouts, traj)[source]#

Bases: object

Container for head tracking ethogram data.

Parameters:
__init__(segments, bouts, traj)[source]#
compute_df(bouts, traj)[source]#
compute_time_series(x, default_val=0)[source]#
class megabouts.pipeline.freely_swimming_pipeline.EthogramFullTracking(segments, bouts, tail, traj)[source]#

Bases: object

Container for full tracking ethogram data.

Parameters:
__init__(segments, bouts, tail, traj)[source]#
compute_df(bouts, tail, traj)[source]#
compute_time_series(x, default_val=0)[source]#
class megabouts.pipeline.freely_swimming_pipeline.HeadTrackingPipeline(tracking_cfg, exclude_CS=False)[source]#

Bases: Pipeline

Pipeline for processing freely swimming fish with head tracking only.

Parameters:
  • tracking_cfg (TrackingConfig) – Configuration for tracking data

  • exclude_CS (bool, optional) – Whether to exclude capture swim bouts, by default False

Examples

>>> import pandas as pd
>>> from megabouts.tracking_data import TrackingConfig, HeadTrackingData, load_example_data
>>> df, fps, mm_per_unit = load_example_data('fulltracking_posture')
>>> head_x = df["head_x"].values * mm_per_unit
>>> head_y = df["head_y"].values * mm_per_unit
>>> head_yaw = df["head_angle"].values
>>> tracking_data = HeadTrackingData.from_posture(
...     head_x=head_x, head_y=head_y, head_yaw=head_yaw
... )
>>> tracking_cfg = TrackingConfig(fps=fps, tracking='head_tracking')
>>> pipeline = HeadTrackingPipeline(tracking_cfg)
>>> ethogram, bouts, segments, traj = pipeline.run(tracking_data)
>>> isinstance(ethogram.df, pd.DataFrame)
True
__init__(tracking_cfg, exclude_CS=False)[source]#
initialize_parameters_for_pipeline()[source]#

Initialize pipeline parameters.

preprocess_traj(traj_df)[source]#
segment_traj(traj_vigor)[source]#
classify_bouts(traj, segments)[source]#
run(tracking_data)[source]#
class megabouts.pipeline.freely_swimming_pipeline.FullTrackingPipeline(tracking_cfg, exclude_CS=False)[source]#

Bases: Pipeline

Pipeline for processing freely swimming fish with full tracking data.

Parameters:
  • tracking_cfg (TrackingConfig) – Configuration for tracking data

  • exclude_CS (bool, optional) – Whether to exclude capture swim bouts, by default False

Examples

>>> import pandas as pd
>>> from megabouts.tracking_data import TrackingConfig, FullTrackingData, load_example_data
>>> df, fps, mm_per_unit = load_example_data('fulltracking_posture')
>>> head_x = df["head_x"].values * mm_per_unit
>>> head_y = df["head_y"].values * mm_per_unit
>>> head_yaw = df["head_angle"].values
>>> tail_angle = df.filter(like="tail_angle").values
>>> tracking_data = FullTrackingData.from_posture(
...     head_x=head_x, head_y=head_y, head_yaw=head_yaw, tail_angle=tail_angle
... )
>>> tracking_cfg = TrackingConfig(fps=fps, tracking='full_tracking')
>>> pipeline = FullTrackingPipeline(tracking_cfg)
>>> ethogram, bouts, segments, tail, traj = pipeline.run(tracking_data)
>>> isinstance(ethogram.df, pd.DataFrame)
True
__init__(tracking_cfg, exclude_CS=False)[source]#
initialize_parameters_for_pipeline()[source]#

Initialize pipeline parameters.

preprocess_tail(tail_df)[source]#
preprocess_traj(traj_df)[source]#
segment(vigor)[source]#
classify_bouts(tail, traj, segments)[source]#
run(tracking_data)[source]#
class megabouts.pipeline.head_restrained_pipeline.HeadRestrainedPipeline(tracking_cfg)[source]#

Bases: Pipeline

Pipeline for processing head-restrained fish data.

Parameters:

tracking_cfg (TrackingConfig) – Configuration for tracking data

Examples

>>> import pandas as pd
>>> from megabouts.tracking_data import TrackingConfig, TailTrackingData, load_example_data
>>> df, fps, mm_per_unit = load_example_data('HR_DLC')
>>> tail_x = df["DLC_resnet50_Zebrafish"].loc[:, [(f"tail{i}", "x") for i in range(11)]].values * mm_per_unit
>>> tail_y = df["DLC_resnet50_Zebrafish"].loc[:, [(f"tail{i}", "y") for i in range(11)]].values * mm_per_unit
>>> tracking_data = TailTrackingData.from_keypoints(tail_x=tail_x, tail_y=tail_y)
>>> tracking_cfg = TrackingConfig(fps=fps, tracking='tail_tracking')
>>> pipeline = HeadRestrainedPipeline(tracking_cfg)
>>> sparse_coding_result, segments, tail = pipeline.run(tracking_data)
>>> isinstance(sparse_coding_result.df, pd.DataFrame)
True
__init__(tracking_cfg)[source]#
initialize_parameters_for_pipeline()[source]#

Initialize pipeline parameters.

preprocess_tail(tail_df)[source]#

Preprocess tail angle data.

Parameters:

tail_df (pd.DataFrame) – DataFrame containing tail angle data

Returns:

Preprocessed tail data

Return type:

TailPreprocessingResult

segment_tail(tail_vigor)[source]#

Segment tail movement into bouts.

Parameters:

tail_vigor (np.ndarray) – Tail vigor signal

Returns:

Detected segments

Return type:

SegmentationResult

compute_sparse_coding(tail_angle)[source]#

Compute sparse coding of tail angles.

Parameters:

tail_angle (np.ndarray) – Tail angle data

Returns:

Sparse coding results

Return type:

SparseCodingResult

run(tracking_data)[source]#