Skip to content

imported_pose.py

ImportedPose

Bases: SpyglassIngestion, Manual

Table to ingest pose data generated prior to spyglass. Each entry corresponds to on ndx_pose.PoseEstimation object in an NWB file. PoseEstimation objects should be stored in nwb.processing.behavior Assumptions: - Single skeleton object per PoseEstimation object

Source code in src/spyglass/position/v1/imported_pose.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@schema
class ImportedPose(SpyglassIngestion, dj.Manual):
    """
    Table to ingest pose data generated prior to spyglass.
    Each entry corresponds to on ndx_pose.PoseEstimation object in an NWB file.
    PoseEstimation objects should be stored in nwb.processing.behavior
    Assumptions:
    - Single skeleton object per PoseEstimation object
    """

    _nwb_table = Nwbfile

    definition = """
    -> IntervalList
    ---
    pose_object_id: varchar(80) # unique identifier for the pose object
    skeleton_object_id: varchar(80) # unique identifier for the skeleton object
    """

    class BodyPart(SpyglassIngestion, dj.Part):
        definition = """
        -> master
        part_name: varchar(80)
        ---
        part_object_id: varchar(80)
        """

        table_key_to_obj_attr = {"self": {"part_object_id": "object_id"}}

    _source_nwb_object_type = ndx_pose.PoseEstimation

    table_key_to_obj_attr = {
        "self": {"pose_object_id": "object_id"},
        "skeleton": {"skeleton_object_id": "object_id"},
    }

    def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
        """Generate the interval, the pose entry, and one entry per body part.

        The interval comes first: it is this table's parent, derived from the
        timestamps of the object's first body part.
        """
        nwb_file_name = base_key["nwb_file_name"]

        # use the timestamps from the first body part to define valid times
        timestamps = list(nwb_obj.pose_estimation_series.values())[
            0
        ].get_timestamps()
        sampling_rate = estimate_sampling_rate(
            timestamps, filename=nwb_file_name
        )
        valid_intervals = get_valid_intervals(
            timestamps,
            sampling_rate=sampling_rate,
            min_valid_len=sampling_rate,
            warn=not self._test_mode,
        )

        interval_pk = {
            "nwb_file_name": nwb_file_name,
            "interval_list_name": f"pose_{nwb_obj.name}_valid_intervals",
        }
        part_attr = self.BodyPart.table_key_to_obj_attr["self"]

        return {
            IntervalList: [
                {
                    **interval_pk,
                    "valid_times": valid_intervals,
                    "pipeline": "ImportedPose",
                }
            ],
            **super().generate_entries_from_nwb_object(nwb_obj, interval_pk),
            self.BodyPart: [
                dict(
                    interval_pk,
                    part_name=part,
                    **{k: getattr(part_obj, v) for k, v in part_attr.items()},
                )
                for part, part_obj in nwb_obj.pose_estimation_series.items()
            ],
        }

    def make(self, key):
        """Deprecated in favor of insert_from_nwbfile."""
        raise NotImplementedError(
            "ImportedPose.make is deprecated. Use insert_from_nwbfile."
        )

    def fetch_pose_dataframe(self, key=None):
        """Fetch pose data as a pandas DataFrame

        Parameters
        ----------
        key : dict
            Key to fetch pose data for

        Returns
        -------
        pd.DataFrame
            DataFrame containing pose data
        """
        _ = self.ensure_single_entry()
        key = key or self.fetch1("KEY")
        query = self & key
        if len(query) != 1:
            raise ValueError(f"Key selected {len(query)} entries: {query}")
        key = query.fetch1("KEY")
        pose_estimations = (
            (self & key).fetch_nwb()[0]["pose"].pose_estimation_series
        )

        index = None
        pose_df = {}
        body_parts = list(pose_estimations.keys())
        index = pose_estimations[body_parts[0]].get_timestamps()
        for body_part in body_parts:
            bp_data = pose_estimations[body_part].data
            part_df = {
                "video_frame_ind": np.nan,
                "x": bp_data[:, 0],
                "y": bp_data[:, 1],
                "likelihood": pose_estimations[body_part].confidence[:],
            }

            pose_df[body_part] = pd.DataFrame(part_df, index=index)

        pose_df
        return pd.concat(pose_df, axis=1)

    def fetch_skeleton(self, key=None):
        _ = self.ensure_single_entry()
        key = key or self.fetch1("KEY")
        skeleton = (self & key).fetch_nwb()[0]["skeleton"]
        nodes = skeleton.nodes[:]
        int_edges = skeleton.edges[:]
        named_edges = [[nodes[i], nodes[j]] for i, j in int_edges]
        return {"nodes": nodes, "edges": named_edges}

generate_entries_from_nwb_object(nwb_obj, base_key=None)

Generate the interval, the pose entry, and one entry per body part.

The interval comes first: it is this table's parent, derived from the timestamps of the object's first body part.

Source code in src/spyglass/position/v1/imported_pose.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
    """Generate the interval, the pose entry, and one entry per body part.

    The interval comes first: it is this table's parent, derived from the
    timestamps of the object's first body part.
    """
    nwb_file_name = base_key["nwb_file_name"]

    # use the timestamps from the first body part to define valid times
    timestamps = list(nwb_obj.pose_estimation_series.values())[
        0
    ].get_timestamps()
    sampling_rate = estimate_sampling_rate(
        timestamps, filename=nwb_file_name
    )
    valid_intervals = get_valid_intervals(
        timestamps,
        sampling_rate=sampling_rate,
        min_valid_len=sampling_rate,
        warn=not self._test_mode,
    )

    interval_pk = {
        "nwb_file_name": nwb_file_name,
        "interval_list_name": f"pose_{nwb_obj.name}_valid_intervals",
    }
    part_attr = self.BodyPart.table_key_to_obj_attr["self"]

    return {
        IntervalList: [
            {
                **interval_pk,
                "valid_times": valid_intervals,
                "pipeline": "ImportedPose",
            }
        ],
        **super().generate_entries_from_nwb_object(nwb_obj, interval_pk),
        self.BodyPart: [
            dict(
                interval_pk,
                part_name=part,
                **{k: getattr(part_obj, v) for k, v in part_attr.items()},
            )
            for part, part_obj in nwb_obj.pose_estimation_series.items()
        ],
    }

make(key)

Deprecated in favor of insert_from_nwbfile.

Source code in src/spyglass/position/v1/imported_pose.py 
100
101
102
103
104
def make(self, key):
    """Deprecated in favor of insert_from_nwbfile."""
    raise NotImplementedError(
        "ImportedPose.make is deprecated. Use insert_from_nwbfile."
    )

fetch_pose_dataframe(key=None)

Fetch pose data as a pandas DataFrame

Parameters:

Name Type Description Default
key dict

Key to fetch pose data for

 None

Returns:

Type Description
DataFrame

DataFrame containing pose data
Source code in src/spyglass/position/v1/imported_pose.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def fetch_pose_dataframe(self, key=None):
    """Fetch pose data as a pandas DataFrame

    Parameters
    ----------
    key : dict
        Key to fetch pose data for

    Returns
    -------
    pd.DataFrame
        DataFrame containing pose data
    """
    _ = self.ensure_single_entry()
    key = key or self.fetch1("KEY")
    query = self & key
    if len(query) != 1:
        raise ValueError(f"Key selected {len(query)} entries: {query}")
    key = query.fetch1("KEY")
    pose_estimations = (
        (self & key).fetch_nwb()[0]["pose"].pose_estimation_series
    )

    index = None
    pose_df = {}
    body_parts = list(pose_estimations.keys())
    index = pose_estimations[body_parts[0]].get_timestamps()
    for body_part in body_parts:
        bp_data = pose_estimations[body_part].data
        part_df = {
            "video_frame_ind": np.nan,
            "x": bp_data[:, 0],
            "y": bp_data[:, 1],
            "likelihood": pose_estimations[body_part].confidence[:],
        }

        pose_df[body_part] = pd.DataFrame(part_df, index=index)

    pose_df
    return pd.concat(pose_df, axis=1)