Skip to content

common_sensors.py

Schema for headstage or other environmental sensors.

SensorData

Bases: SpyglassMixin, Imported

Source code in src/spyglass/common/common_sensors.py
@schema
class SensorData(SpyglassMixin, 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

    def make(self, key):
        """Populate SensorData using the analog BehavioralEvents from the NWB."""

        nwb_file_name = key["nwb_file_name"]
        nwb_file_abspath = Nwbfile().get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)

        sensor = get_data_interface(
            nwbf, "analog", pynwb.behavior.BehavioralEvents
        )
        if sensor is None:
            logger.info(f"No conforming sensor data found in {nwb_file_name}\n")
            return

        key["sensor_data_object_id"] = sensor.time_series["analog"].object_id

        # the valid times for these data are the same as the valid times for
        # the raw ephys data

        key["interval_list_name"] = (
            Raw & {"nwb_file_name": nwb_file_name}
        ).fetch1("interval_list_name")
        self.insert1(key)

make(key)

Populate SensorData using the analog BehavioralEvents from the NWB.

Source code in src/spyglass/common/common_sensors.py
def make(self, key):
    """Populate SensorData using the analog BehavioralEvents from the NWB."""

    nwb_file_name = key["nwb_file_name"]
    nwb_file_abspath = Nwbfile().get_abs_path(nwb_file_name)
    nwbf = get_nwb_file(nwb_file_abspath)

    sensor = get_data_interface(
        nwbf, "analog", pynwb.behavior.BehavioralEvents
    )
    if sensor is None:
        logger.info(f"No conforming sensor data found in {nwb_file_name}\n")
        return

    key["sensor_data_object_id"] = sensor.time_series["analog"].object_id

    # the valid times for these data are the same as the valid times for
    # the raw ephys data

    key["interval_list_name"] = (
        Raw & {"nwb_file_name": nwb_file_name}
    ).fetch1("interval_list_name")
    self.insert1(key)