Skip to content

core.py

DecodingParameters

Bases: SpyglassMixin, Lookup

Params for decoding mental position and some category of interest

Source code in src/spyglass/decoding/v1/core.py 
 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
@schema
class DecodingParameters(SpyglassMixin, dj.Lookup):
    """Params for decoding mental position and some category of interest"""

    definition = """
    decoding_param_name : varchar(80)  # a name for this set of parameters
    ---
    decoding_params : BLOB             # initialization parameters for model
    decoding_kwargs = NULL : BLOB      # additional keyword arguments
    """

    pk = "decoding_param_name"
    sk = "decoding_params"

    contents = [
        {
            pk: f"contfrag_clusterless_{non_local_detector_version}",
            sk: ContFragClusterlessClassifier(),
        },
        {
            pk: f"nonlocal_clusterless_{non_local_detector_version}",
            sk: NonLocalClusterlessDetector(),
        },
        {
            pk: f"contfrag_sorted_{non_local_detector_version}",
            sk: ContFragSortedSpikesClassifier(),
        },
        {
            pk: f"nonlocal_sorted_{non_local_detector_version}",
            sk: NonLocalSortedSpikesDetector(),
        },
    ]

    @classmethod
    def insert_default(cls):
        """Insert default decoding parameters"""
        cls.super().insert(cls.contents, skip_duplicates=True)

    def insert(self, rows, *args, **kwargs):
        """Override insert to convert classes to dict before inserting"""
        for row in rows:
            params = row["decoding_params"]
            if hasattr(params, "__dict__"):
                params = vars(params)
            row["decoding_params"] = convert_classes_to_dict(params)
        super().insert(rows, *args, **kwargs)

    def fetch(self, *args, **kwargs):
        """Return decoding parameters as a list of classes."""
        rows = super().fetch(*args, **kwargs)
        if kwargs.get("format", None) == "array":
            # case when recalled by dj.fetch(), class conversion performed later in stack
            return rows

        if not len(args):
            # infer args from table heading
            args = tuple(self.heading)

        if "decoding_params" not in args:
            return rows

        params_index = args.index("decoding_params")
        if len(args) == 1:
            # only fetching decoding_params
            content = [restore_classes(r) for r in rows]
        elif len(rows):
            content = []
            for row in zip(*rows):
                row = list(row)
                row[params_index] = restore_classes(row[params_index])
                content.append(tuple(row))
        else:
            content = rows
        return content

    def fetch1(self, *args, **kwargs):
        """Return one decoding paramset as a class."""
        row = super().fetch1(*args, **kwargs)

        if len(args) == 0:
            row["decoding_params"] = restore_classes(row["decoding_params"])
            return row

        if "decoding_params" in args:
            if len(args) == 1:
                return restore_classes(row)
            row = list(row)
            row[args.index("decoding_params")] = restore_classes(
                row[args.index("decoding_params")]
            )
            return tuple(row)

        return row

insert_default() classmethod

Insert default decoding parameters

Source code in src/spyglass/decoding/v1/core.py 
56
57
58
59
@classmethod
def insert_default(cls):
    """Insert default decoding parameters"""
    cls.super().insert(cls.contents, skip_duplicates=True)

insert(rows, *args, **kwargs)

Override insert to convert classes to dict before inserting

Source code in src/spyglass/decoding/v1/core.py 
61
62
63
64
65
66
67
68
def insert(self, rows, *args, **kwargs):
    """Override insert to convert classes to dict before inserting"""
    for row in rows:
        params = row["decoding_params"]
        if hasattr(params, "__dict__"):
            params = vars(params)
        row["decoding_params"] = convert_classes_to_dict(params)
    super().insert(rows, *args, **kwargs)

fetch(*args, **kwargs)

Return decoding parameters as a list of classes.

Source code in src/spyglass/decoding/v1/core.py 
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
def fetch(self, *args, **kwargs):
    """Return decoding parameters as a list of classes."""
    rows = super().fetch(*args, **kwargs)
    if kwargs.get("format", None) == "array":
        # case when recalled by dj.fetch(), class conversion performed later in stack
        return rows

    if not len(args):
        # infer args from table heading
        args = tuple(self.heading)

    if "decoding_params" not in args:
        return rows

    params_index = args.index("decoding_params")
    if len(args) == 1:
        # only fetching decoding_params
        content = [restore_classes(r) for r in rows]
    elif len(rows):
        content = []
        for row in zip(*rows):
            row = list(row)
            row[params_index] = restore_classes(row[params_index])
            content.append(tuple(row))
    else:
        content = rows
    return content

fetch1(*args, **kwargs)

Return one decoding paramset as a class.

Source code in src/spyglass/decoding/v1/core.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def fetch1(self, *args, **kwargs):
    """Return one decoding paramset as a class."""
    row = super().fetch1(*args, **kwargs)

    if len(args) == 0:
        row["decoding_params"] = restore_classes(row["decoding_params"])
        return row

    if "decoding_params" in args:
        if len(args) == 1:
            return restore_classes(row)
        row = list(row)
        row[args.index("decoding_params")] = restore_classes(
            row[args.index("decoding_params")]
        )
        return tuple(row)

    return row

PositionGroup

Bases: SpyglassMixin, Manual

Source code in src/spyglass/decoding/v1/core.py 
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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
@schema
class PositionGroup(SpyglassMixin, dj.Manual):
    definition = """
    -> Session
    position_group_name: varchar(80)
    ----
    position_variables = NULL: longblob # list of position variables to decode
    upsample_rate = NULL: float # upsampling rate for position data (Hz)
    """

    class Position(SpyglassMixinPart):
        definition = """
        -> PositionGroup
        -> PositionOutput.proj(pos_merge_id='merge_id')
        """

    def create_group(
        self,
        nwb_file_name: str,
        group_name: str,
        keys: list[dict],
        position_variables: list[str] = ["position_x", "position_y"],
        upsample_rate: float = np.nan,
    ):
        """Create a new position group."""
        group_key = {
            "nwb_file_name": nwb_file_name,
            "position_group_name": group_name,
        }
        if self & group_key:
            logger.error(  # Easier for pytests to not raise error on duplicate
                f"Group {nwb_file_name}: {group_name} already exists. "
                + "Please delete the group before creating a new one"
            )
            return
        self.insert1(
            {
                **group_key,
                "position_variables": position_variables,
                "upsample_rate": upsample_rate,
            },
            skip_duplicates=True,
        )
        for key in keys:
            self.Position.insert1(
                {
                    **key,
                    **group_key,
                },
                skip_duplicates=True,
            )

    def fetch_position_info(
        self, key: dict = None, min_time: float = None, max_time: float = None
    ) -> tuple[pd.DataFrame, list[str]]:
        """fetch position information for decoding

        Parameters
        ----------
        key : dict, optional
            restriction to a single entry in PositionGroup, by default None
        min_time : float, optional
            restrict position information to times greater than min_time,
            by default None
        max_time : float, optional
            restrict position information to times less than max_time,
            by default None

        Returns
        -------
        tuple[pd.DataFrame, list[str]]
            position information and names of position variables
        """
        if key is None:
            key = {}
        key = (self & key).fetch1("KEY")
        position_variable_names = (self & key).fetch1("position_variables")

        position_info = []
        upsample_rate = (self & key).fetch1("upsample_rate")
        for pos_merge_id in (self.Position & key).fetch("pos_merge_id"):
            if not np.isnan(upsample_rate):
                position_info.append(
                    self._upsample(
                        (
                            PositionOutput & {"merge_id": pos_merge_id}
                        ).fetch1_dataframe(),
                        upsampling_sampling_rate=upsample_rate,
                        position_variable_names=position_variable_names,
                    )
                )
            else:
                position_info.append(
                    (
                        PositionOutput & {"merge_id": pos_merge_id}
                    ).fetch1_dataframe()
                )

        if min_time is None:
            min_time = min([df.index.min() for df in position_info])
        if max_time is None:
            max_time = max([df.index.max() for df in position_info])
        position_info = pd.concat(position_info, axis=0).loc[min_time:max_time]

        return position_info, position_variable_names

    @staticmethod
    def _upsample(
        position_df: pd.DataFrame,
        upsampling_sampling_rate: float,
        upsampling_interpolation_method: str = "linear",
        position_variable_names: list[str] = None,
    ) -> pd.DataFrame:
        """upsample position data to a fixed sampling rate

        Parameters
        ----------
        position_df : pd.DataFrame
            dataframe containing position data
        upsampling_sampling_rate : float
            sampling rate to upsample to
        upsampling_interpolation_method : str, optional
            pandas method for interpolation, by default "linear"
        position_variable_names : list[str], optional
            names of position variables of focus, for which nan values will not be
            interpolated, by default None includes all columns

        Returns
        -------
        pd.DataFrame
            upsampled position data
        """

        upsampling_start_time = position_df.index[0]
        upsampling_end_time = position_df.index[-1]

        n_samples = (
            int(
                np.ceil(
                    (upsampling_end_time - upsampling_start_time)
                    * upsampling_sampling_rate
                )
            )
            + 1
        )
        new_time = np.linspace(
            upsampling_start_time, upsampling_end_time, n_samples
        )
        new_index = pd.Index(
            np.unique(np.concatenate((position_df.index, new_time))),
            name="time",
        )

        # Find NaN intervals
        nan_intervals = {}
        if position_variable_names is None:
            position_variable_names = position_df.columns
        for column in position_variable_names:
            is_nan = position_df[column].isna().to_numpy().astype(int)
            st = np.where(np.diff(is_nan) == 1)[0] + 1
            en = np.where(np.diff(is_nan) == -1)[0]
            if is_nan[0]:
                st = np.insert(st, 0, 0)
            if is_nan[-1]:
                en = np.append(en, len(is_nan) - 1)
            st = position_df.index[st].to_numpy()
            en = position_df.index[en].to_numpy()
            nan_intervals[column] = list(zip(st, en))

        # upsample and interpolate
        position_df = (
            position_df.reindex(index=new_index)
            .interpolate(method=upsampling_interpolation_method)
            .reindex(index=new_time)
        )

        # Fill NaN intervals
        for column, intervals in nan_intervals.items():
            for st, en in intervals:
                position_df[column][st:en] = np.nan

        return position_df

create_group(nwb_file_name, group_name, keys, position_variables=['position_x', 'position_y'], upsample_rate=np.nan)

Create a new position group.

Source code in src/spyglass/decoding/v1/core.py 
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
def create_group(
    self,
    nwb_file_name: str,
    group_name: str,
    keys: list[dict],
    position_variables: list[str] = ["position_x", "position_y"],
    upsample_rate: float = np.nan,
):
    """Create a new position group."""
    group_key = {
        "nwb_file_name": nwb_file_name,
        "position_group_name": group_name,
    }
    if self & group_key:
        logger.error(  # Easier for pytests to not raise error on duplicate
            f"Group {nwb_file_name}: {group_name} already exists. "
            + "Please delete the group before creating a new one"
        )
        return
    self.insert1(
        {
            **group_key,
            "position_variables": position_variables,
            "upsample_rate": upsample_rate,
        },
        skip_duplicates=True,
    )
    for key in keys:
        self.Position.insert1(
            {
                **key,
                **group_key,
            },
            skip_duplicates=True,
        )

fetch_position_info(key=None, min_time=None, max_time=None)

fetch position information for decoding

Parameters:

Name Type Description Default
key dict

restriction to a single entry in PositionGroup, by default None

 None
min_time float

restrict position information to times greater than min_time, by default None

 None
max_time float

restrict position information to times less than max_time, by default None

 None

Returns:

Type Description
tuple[DataFrame, list[str]]

position information and names of position variables
Source code in src/spyglass/decoding/v1/core.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def fetch_position_info(
    self, key: dict = None, min_time: float = None, max_time: float = None
) -> tuple[pd.DataFrame, list[str]]:
    """fetch position information for decoding

    Parameters
    ----------
    key : dict, optional
        restriction to a single entry in PositionGroup, by default None
    min_time : float, optional
        restrict position information to times greater than min_time,
        by default None
    max_time : float, optional
        restrict position information to times less than max_time,
        by default None

    Returns
    -------
    tuple[pd.DataFrame, list[str]]
        position information and names of position variables
    """
    if key is None:
        key = {}
    key = (self & key).fetch1("KEY")
    position_variable_names = (self & key).fetch1("position_variables")

    position_info = []
    upsample_rate = (self & key).fetch1("upsample_rate")
    for pos_merge_id in (self.Position & key).fetch("pos_merge_id"):
        if not np.isnan(upsample_rate):
            position_info.append(
                self._upsample(
                    (
                        PositionOutput & {"merge_id": pos_merge_id}
                    ).fetch1_dataframe(),
                    upsampling_sampling_rate=upsample_rate,
                    position_variable_names=position_variable_names,
                )
            )
        else:
            position_info.append(
                (
                    PositionOutput & {"merge_id": pos_merge_id}
                ).fetch1_dataframe()
            )

    if min_time is None:
        min_time = min([df.index.min() for df in position_info])
    if max_time is None:
        max_time = max([df.index.max() for df in position_info])
    position_info = pd.concat(position_info, axis=0).loc[min_time:max_time]

    return position_info, position_variable_names