Skip to content

common_sensors.py

Schema for headstage or other environmental sensors.

SensorData

Bases: SpyglassIngestion, Imported

Source code in src/spyglass/common/common_sensors.py 
 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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@schema
class SensorData(SpyglassIngestion, dj.Imported):
    definition = """
    -> Session
    ---
    sensor_data_object_id: varchar(40)  # object id of the data in the NWB file
    -> IntervalList                     # the list of intervals for this object
    """

    _nwb_table = Nwbfile
    # The enclosing ProcessingModule and the inner TimeSeries share this name,
    # so the type below is what distinguishes the three.
    _source_nwb_object_name = "analog"
    _only_ingest_first = True  # first match wins, as get_data_interface did

    _source_nwb_object_type = pynwb.behavior.BehavioralEvents

    @property
    def table_key_to_obj_attr(self):
        return {"self": {"sensor_data_object_id": self._analog_object_id}}

    @staticmethod
    def _analog_series(nwb_obj):
        """Return the analog TimeSeries held by a BehavioralEvents object.

        Parameters
        ----------
        nwb_obj : pynwb.behavior.BehavioralEvents
            The sensor data container.

        Returns
        -------
        pynwb.base.TimeSeries
            The series named 'analog'.
        """
        return nwb_obj.time_series["analog"]

    def _analog_object_id(self, nwb_obj) -> str:
        """Validate the analog series and return its object id.

        The id stored is the series', not the enclosing container's.

        Parameters
        ----------
        nwb_obj : pynwb.behavior.BehavioralEvents
            The sensor data container.

        Returns
        -------
        str
            Object id of the analog series.

        Raises
        ------
        ValueError
            If the description does not name one column per column of data.
        """
        series = self._analog_series(nwb_obj)

        columns = [
            col
            for col in series.description.split()
            if col not in ["time", "timestamps"]
        ]
        n_cols = series.data.shape[1]

        if len(columns) != n_cols:
            raise ValueError(
                f"Number of columns in description ({len(columns)}) "
                f"does not match number of columns in data ({n_cols}). "
                f"Columns: {series.description}. "
                "Please check the NWB file."
            )

        return series.object_id

    def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
        """Attach the raw ephys interval, which these data share."""
        super_ins = super().generate_entries_from_nwb_object(nwb_obj, base_key)
        self_key = super_ins[self][0]

        # the valid times for these data are the same as the valid times for
        # the raw ephys data
        self_key["interval_list_name"] = (
            Raw & {"nwb_file_name": self_key["nwb_file_name"]}
        ).fetch1("interval_list_name")

        return super_ins

    def make(self, key: dict) -> None:
        """Deprecated in favor of insert_from_nwbfile."""
        raise NotImplementedError(
            "SensorData.make is deprecated. Use insert_from_nwbfile."
        )

    def fetch1_dataframe(
        self, interval_list_name: Optional[str] = None
    ) -> Optional[pd.DataFrame]:
        """Fetch the sensor data as a DataFrame.

        Parameters
        ----------
        interval_list_name: str, optional
            The name of the interval list to filter the data by.
            If None, no filtering is applied.

        Returns
        -------
        pd.DataFrame
            A DataFrame containing the sensor data, indexed by time.
            If no data is found, None is returned.

        Raises
        -------
        ValueError
            If more than one sensor data object is found or
            if the specified interval list is not found.

        """
        if len(self) == 0:
            return None
        _ = self.ensure_single_entry()

        nwb = self.fetch_nwb()[0]
        columns = nwb["sensor_data"].description.split()
        columns = [col for col in columns if col not in ["time", "timestamps"]]

        if interval_list_name is None:
            # corresponds to `raw data valid times` interval
            return pd.DataFrame(
                nwb["sensor_data"].data,
                index=pd.Index(nwb["sensor_data"].timestamps, name="time"),
                columns=columns,
            )

        nwb_file_name = self.fetch1("nwb_file_name")
        valid_times = (
            IntervalList
            & {
                "nwb_file_name": nwb_file_name,
                "interval_list_name": interval_list_name,
            }
        ).fetch1("valid_times")

        if len(valid_times) == 0:
            raise ValueError(
                f"No valid times found for {nwb_file_name} "
                f"and {interval_list_name}"
            )

        sensor_data_df = []
        for start_time, end_time in valid_times:
            start_ind, end_ind = np.searchsorted(
                nwb["sensor_data"].timestamps, [start_time, end_time]
            )
            sensor_data_df.append(
                pd.DataFrame(
                    nwb["sensor_data"].data[start_ind:end_ind, :],
                    index=pd.Index(
                        nwb["sensor_data"].timestamps[start_ind:end_ind],
                        name="time",
                    ),
                    columns=columns,
                )
            )

        return pd.concat(sensor_data_df, axis=0).sort_index()

generate_entries_from_nwb_object(nwb_obj, base_key=None)

Attach the raw ephys interval, which these data share.

Source code in src/spyglass/common/common_sensors.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
    """Attach the raw ephys interval, which these data share."""
    super_ins = super().generate_entries_from_nwb_object(nwb_obj, base_key)
    self_key = super_ins[self][0]

    # the valid times for these data are the same as the valid times for
    # the raw ephys data
    self_key["interval_list_name"] = (
        Raw & {"nwb_file_name": self_key["nwb_file_name"]}
    ).fetch1("interval_list_name")

    return super_ins

make(key)

Deprecated in favor of insert_from_nwbfile.

Source code in src/spyglass/common/common_sensors.py 
108
109
110
111
112
def make(self, key: dict) -> None:
    """Deprecated in favor of insert_from_nwbfile."""
    raise NotImplementedError(
        "SensorData.make is deprecated. Use insert_from_nwbfile."
    )

fetch1_dataframe(interval_list_name=None)

Fetch the sensor data as a DataFrame.

Parameters:

Name Type Description Default
interval_list_name Optional[str]

The name of the interval list to filter the data by. If None, no filtering is applied.

 None

Returns:

Type Description
DataFrame

A DataFrame containing the sensor data, indexed by time. If no data is found, None is returned.

Raises:

Type Description
ValueError

If more than one sensor data object is found or if the specified interval list is not found.
Source code in src/spyglass/common/common_sensors.py 
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def fetch1_dataframe(
    self, interval_list_name: Optional[str] = None
) -> Optional[pd.DataFrame]:
    """Fetch the sensor data as a DataFrame.

    Parameters
    ----------
    interval_list_name: str, optional
        The name of the interval list to filter the data by.
        If None, no filtering is applied.

    Returns
    -------
    pd.DataFrame
        A DataFrame containing the sensor data, indexed by time.
        If no data is found, None is returned.

    Raises
    -------
    ValueError
        If more than one sensor data object is found or
        if the specified interval list is not found.

    """
    if len(self) == 0:
        return None
    _ = self.ensure_single_entry()

    nwb = self.fetch_nwb()[0]
    columns = nwb["sensor_data"].description.split()
    columns = [col for col in columns if col not in ["time", "timestamps"]]

    if interval_list_name is None:
        # corresponds to `raw data valid times` interval
        return pd.DataFrame(
            nwb["sensor_data"].data,
            index=pd.Index(nwb["sensor_data"].timestamps, name="time"),
            columns=columns,
        )

    nwb_file_name = self.fetch1("nwb_file_name")
    valid_times = (
        IntervalList
        & {
            "nwb_file_name": nwb_file_name,
            "interval_list_name": interval_list_name,
        }
    ).fetch1("valid_times")

    if len(valid_times) == 0:
        raise ValueError(
            f"No valid times found for {nwb_file_name} "
            f"and {interval_list_name}"
        )

    sensor_data_df = []
    for start_time, end_time in valid_times:
        start_ind, end_ind = np.searchsorted(
            nwb["sensor_data"].timestamps, [start_time, end_time]
        )
        sensor_data_df.append(
            pd.DataFrame(
                nwb["sensor_data"].data[start_ind:end_ind, :],
                index=pd.Index(
                    nwb["sensor_data"].timestamps[start_ind:end_ind],
                    name="time",
                ),
                columns=columns,
            )
        )

    return pd.concat(sensor_data_df, axis=0).sort_index()