Skip to content

position_dlc_cohort.py

DLCSmoothInterpCohortSelection

Bases: SpyglassMixin, Manual

Table to specify which combination of bodyparts from DLCSmoothInterp get combined into a cohort

Source code in src/spyglass/position/v1/position_dlc_cohort.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
@schema
class DLCSmoothInterpCohortSelection(SpyglassMixin, dj.Manual):
    """
    Table to specify which combination of bodyparts from DLCSmoothInterp
    get combined into a cohort
    """

    definition = """
    dlc_si_cohort_selection_name : varchar(120)
    -> DLCPoseEstimation
    ---
    bodyparts_params_dict   : blob      # Dict with bodypart as key and desired dlc_si_params_name as value
    """

DLCSmoothInterpCohort

Bases: SpyglassMixin, Computed

Table to combine multiple bodyparts from DLCSmoothInterp to enable centroid/orientation calculations

Source code in src/spyglass/position/v1/position_dlc_cohort.py 
 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
@schema
class DLCSmoothInterpCohort(SpyglassMixin, dj.Computed):
    """
    Table to combine multiple bodyparts from DLCSmoothInterp
    to enable centroid/orientation calculations
    """

    # Need to ensure that nwb_file_name/epoch/interval list name endure as primary keys
    definition = """
    -> DLCSmoothInterpCohortSelection
    """
    log_path = None

    class BodyPart(SpyglassMixin, dj.Part):
        definition = """
        -> DLCSmoothInterpCohort
        -> DLCSmoothInterp
        ---
        -> AnalysisNwbfile
        dlc_smooth_interp_position_object_id : varchar(80)
        dlc_smooth_interp_info_object_id : varchar(80)
        """

        def fetch1_dataframe(self) -> pd.DataFrame:
            """Fetch a single dataframe."""
            nwb_data = self.fetch_nwb()[0]
            index = pd.Index(
                np.asarray(
                    nwb_data["dlc_smooth_interp_position"]
                    .get_spatial_series()
                    .timestamps
                ),
                name="time",
            )
            COLUMNS = [
                "video_frame_ind",
                "x",
                "y",
            ]
            return pd.DataFrame(
                np.concatenate(
                    (
                        np.asarray(
                            nwb_data["dlc_smooth_interp_info"]
                            .time_series["video_frame_ind"]
                            .data,
                            dtype=int,
                        )[:, np.newaxis],
                        np.asarray(
                            nwb_data["dlc_smooth_interp_position"]
                            .get_spatial_series()
                            .data
                        ),
                    ),
                    axis=1,
                ),
                columns=COLUMNS,
                index=index,
            )

    def make(self, key):
        """Populate DLCSmoothInterpCohort table with the combined bodyparts.

        Calls _logged_make to log the process to a log.log file while...
        1. Fetching the cohort selection and smooted interpolated data for each
              bodypart.
        2. Ensuring the number of bodyparts match across data and parameters.
        3. Inserting the combined bodyparts into DLCSmoothInterpCohort.
        """
        output_dir = infer_output_dir(key=key, makedir=False)
        self.log_path = Path(output_dir) / "log.log"
        self._logged_make(key)
        logger.info("Inserted entry into DLCSmoothInterpCohort")

    @file_log(logger, console=False)
    def _logged_make(self, key):
        logger.info("-----------------------")
        logger.info("Bodypart Cohort")

        cohort_selection = (DLCSmoothInterpCohortSelection & key).fetch1()
        table_entries = []
        bp_params_dict = cohort_selection.pop("bodyparts_params_dict")
        temp_key = cohort_selection.copy()
        for bodypart, params in bp_params_dict.items():
            temp_key["bodypart"] = bodypart
            temp_key["dlc_si_params_name"] = params
            table_entries.append((DLCSmoothInterp & temp_key).fetch())

        if not len(table_entries) == len(bp_params_dict):
            raise ValueError(
                f"Mismatch: DLCSmoothInterp {len(table_entries)} vs "
                + f"bodyparts_params_dict {len(bp_params_dict)}"
            )

        # TODO: change to DLCSmoothInterp.heading.names
        table_column_names = list(table_entries[0].dtype.fields.keys())

        part_keys = [
            {
                **{k: v for k, v in zip(table_column_names, table_entry[0])},
                **key,
            }
            for table_entry in table_entries
        ]

        self.insert1(key)
        self.BodyPart.insert(part_keys, skip_duplicates=True)

BodyPart

Bases: SpyglassMixin, Part

Source code in src/spyglass/position/v1/position_dlc_cohort.py 
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
class BodyPart(SpyglassMixin, dj.Part):
    definition = """
    -> DLCSmoothInterpCohort
    -> DLCSmoothInterp
    ---
    -> AnalysisNwbfile
    dlc_smooth_interp_position_object_id : varchar(80)
    dlc_smooth_interp_info_object_id : varchar(80)
    """

    def fetch1_dataframe(self) -> pd.DataFrame:
        """Fetch a single dataframe."""
        nwb_data = self.fetch_nwb()[0]
        index = pd.Index(
            np.asarray(
                nwb_data["dlc_smooth_interp_position"]
                .get_spatial_series()
                .timestamps
            ),
            name="time",
        )
        COLUMNS = [
            "video_frame_ind",
            "x",
            "y",
        ]
        return pd.DataFrame(
            np.concatenate(
                (
                    np.asarray(
                        nwb_data["dlc_smooth_interp_info"]
                        .time_series["video_frame_ind"]
                        .data,
                        dtype=int,
                    )[:, np.newaxis],
                    np.asarray(
                        nwb_data["dlc_smooth_interp_position"]
                        .get_spatial_series()
                        .data
                    ),
                ),
                axis=1,
            ),
            columns=COLUMNS,
            index=index,
        )

fetch1_dataframe()

Fetch a single dataframe.

Source code in src/spyglass/position/v1/position_dlc_cohort.py 
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
def fetch1_dataframe(self) -> pd.DataFrame:
    """Fetch a single dataframe."""
    nwb_data = self.fetch_nwb()[0]
    index = pd.Index(
        np.asarray(
            nwb_data["dlc_smooth_interp_position"]
            .get_spatial_series()
            .timestamps
        ),
        name="time",
    )
    COLUMNS = [
        "video_frame_ind",
        "x",
        "y",
    ]
    return pd.DataFrame(
        np.concatenate(
            (
                np.asarray(
                    nwb_data["dlc_smooth_interp_info"]
                    .time_series["video_frame_ind"]
                    .data,
                    dtype=int,
                )[:, np.newaxis],
                np.asarray(
                    nwb_data["dlc_smooth_interp_position"]
                    .get_spatial_series()
                    .data
                ),
            ),
            axis=1,
        ),
        columns=COLUMNS,
        index=index,
    )

make(key)

Populate DLCSmoothInterpCohort table with the combined bodyparts.

Calls _logged_make to log the process to a log.log file while... 1. Fetching the cohort selection and smooted interpolated data for each bodypart. 2. Ensuring the number of bodyparts match across data and parameters. 3. Inserting the combined bodyparts into DLCSmoothInterpCohort.

Source code in src/spyglass/position/v1/position_dlc_cohort.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def make(self, key):
    """Populate DLCSmoothInterpCohort table with the combined bodyparts.

    Calls _logged_make to log the process to a log.log file while...
    1. Fetching the cohort selection and smooted interpolated data for each
          bodypart.
    2. Ensuring the number of bodyparts match across data and parameters.
    3. Inserting the combined bodyparts into DLCSmoothInterpCohort.
    """
    output_dir = infer_output_dir(key=key, makedir=False)
    self.log_path = Path(output_dir) / "log.log"
    self._logged_make(key)
    logger.info("Inserted entry into DLCSmoothInterpCohort")