Skip to content

position_trodes_position.py

TrodesPosParams

Bases: SpyglassMixin, Manual

Parameters for calculating the position (centroid, velocity, orientation)

Source code in src/spyglass/position/v1/position_trodes_position.py 
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
@schema
class TrodesPosParams(SpyglassMixin, dj.Manual):
    """
    Parameters for calculating the position (centroid, velocity, orientation)
    """

    definition = """
    trodes_pos_params_name: varchar(80) # name for this set of parameters
    ---
    params: longblob
    """

    @property
    def default_pk(self) -> dict:
        """Return the default primary key for this table."""
        return {"trodes_pos_params_name": "default"}

    @property
    def default_params(self) -> dict:
        """Return the default parameters for this table."""
        return {
            "max_LED_separation": 9.0,
            "max_plausible_speed": 300.0,
            "position_smoothing_duration": 0.125,
            "speed_smoothing_std_dev": 0.100,
            "orient_smoothing_std_dev": 0.001,
            "led1_is_front": 1,
            "is_upsampled": 0,
            "upsampling_sampling_rate": None,
            "upsampling_interpolation_method": "linear",
        }

    @classmethod
    def insert_default(cls, **kwargs) -> None:
        """
        Insert default parameter set for position determination
        """
        cls.insert1(
            {**cls().default_pk, "params": cls().default_params},
            skip_duplicates=True,
        )

    @classmethod
    def get_default(cls) -> dict:
        """Return the default set of parameters for position calculation"""
        query = cls & cls().default_pk
        if not len(query) > 0:
            cls().insert_default(skip_duplicates=True)
            return (cls & cls().default_pk).fetch1()

        return query.fetch1()

    @classmethod
    def get_accepted_params(cls) -> list:
        """Return a list of accepted parameters for position calculation"""
        return [k for k in cls().default_params.keys()]

default_pk: dict property

Return the default primary key for this table.

default_params: dict property

Return the default parameters for this table.

insert_default(**kwargs) classmethod

Insert default parameter set for position determination

Source code in src/spyglass/position/v1/position_trodes_position.py 
52
53
54
55
56
57
58
59
60
@classmethod
def insert_default(cls, **kwargs) -> None:
    """
    Insert default parameter set for position determination
    """
    cls.insert1(
        {**cls().default_pk, "params": cls().default_params},
        skip_duplicates=True,
    )

get_default() classmethod

Return the default set of parameters for position calculation

Source code in src/spyglass/position/v1/position_trodes_position.py 
62
63
64
65
66
67
68
69
70
@classmethod
def get_default(cls) -> dict:
    """Return the default set of parameters for position calculation"""
    query = cls & cls().default_pk
    if not len(query) > 0:
        cls().insert_default(skip_duplicates=True)
        return (cls & cls().default_pk).fetch1()

    return query.fetch1()

get_accepted_params() classmethod

Return a list of accepted parameters for position calculation

Source code in src/spyglass/position/v1/position_trodes_position.py 
72
73
74
75
@classmethod
def get_accepted_params(cls) -> list:
    """Return a list of accepted parameters for position calculation"""
    return [k for k in cls().default_params.keys()]

TrodesPosSelection

Bases: SpyglassMixin, Manual

Table to pair an interval with position data and position determination parameters

Source code in src/spyglass/position/v1/position_trodes_position.py 
 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
@schema
class TrodesPosSelection(SpyglassMixin, dj.Manual):
    """
    Table to pair an interval with position data
    and position determination parameters
    """

    definition = """
    -> RawPosition
    -> TrodesPosParams
    """

    @classmethod
    def insert_with_default(
        cls,
        key: dict,
        skip_duplicates: bool = False,
        edit_defaults: dict = {},
        edit_name: str = None,
    ) -> None:
        """Insert key with default parameters.

        To change defaults, supply a dict as edit_defaults with a name for
        the new paramset as edit_name.

        Parameters
        ----------
        key: Union[dict, str]
            Restriction uniquely identifying entr(y/ies) in RawPosition.
        skip_duplicates: bool, optional
            Skip duplicate entries.
        edit_defaults: dict, optional
            Dictionary of overrides to default parameters.
        edit_name: str, optional
            If edit_defauts is passed, the name of the new entry

        Raises
        ------
        ValueError
            Key does not identify any entries in RawPosition.
        """
        query = RawPosition & key
        if not query:
            raise ValueError(f"Found no entries found for {key}")

        param_pk, param_name = list(TrodesPosParams().default_pk.items())[0]

        if bool(edit_defaults) ^ bool(edit_name):  # XOR: only one of them
            raise ValueError("Must specify both edit_defauts and edit_name")

        elif edit_defaults and edit_name:
            TrodesPosParams.insert1(
                {
                    param_pk: edit_name,
                    "params": {
                        **TrodesPosParams().default_params,
                        **edit_defaults,
                    },
                },
                skip_duplicates=skip_duplicates,
            )

        cls.insert(
            [
                {**k, param_pk: edit_name or param_name}
                for k in query.fetch("KEY", as_dict=True)
            ],
            skip_duplicates=skip_duplicates,
        )

insert_with_default(key, skip_duplicates=False, edit_defaults={}, edit_name=None) classmethod

Insert key with default parameters.

To change defaults, supply a dict as edit_defaults with a name for the new paramset as edit_name.

Parameters:

Name Type Description Default
key dict

Restriction uniquely identifying entr(y/ies) in RawPosition.

 required
skip_duplicates bool

Skip duplicate entries.

 False
edit_defaults dict

Dictionary of overrides to default parameters.

 {}
edit_name str

If edit_defauts is passed, the name of the new entry

 None

Raises:

Type Description
ValueError

Key does not identify any entries in RawPosition.
Source code in src/spyglass/position/v1/position_trodes_position.py 
 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
@classmethod
def insert_with_default(
    cls,
    key: dict,
    skip_duplicates: bool = False,
    edit_defaults: dict = {},
    edit_name: str = None,
) -> None:
    """Insert key with default parameters.

    To change defaults, supply a dict as edit_defaults with a name for
    the new paramset as edit_name.

    Parameters
    ----------
    key: Union[dict, str]
        Restriction uniquely identifying entr(y/ies) in RawPosition.
    skip_duplicates: bool, optional
        Skip duplicate entries.
    edit_defaults: dict, optional
        Dictionary of overrides to default parameters.
    edit_name: str, optional
        If edit_defauts is passed, the name of the new entry

    Raises
    ------
    ValueError
        Key does not identify any entries in RawPosition.
    """
    query = RawPosition & key
    if not query:
        raise ValueError(f"Found no entries found for {key}")

    param_pk, param_name = list(TrodesPosParams().default_pk.items())[0]

    if bool(edit_defaults) ^ bool(edit_name):  # XOR: only one of them
        raise ValueError("Must specify both edit_defauts and edit_name")

    elif edit_defaults and edit_name:
        TrodesPosParams.insert1(
            {
                param_pk: edit_name,
                "params": {
                    **TrodesPosParams().default_params,
                    **edit_defaults,
                },
            },
            skip_duplicates=skip_duplicates,
        )

    cls.insert(
        [
            {**k, param_pk: edit_name or param_name}
            for k in query.fetch("KEY", as_dict=True)
        ],
        skip_duplicates=skip_duplicates,
    )

TrodesPosV1

Bases: SpyglassMixin, Computed

Table to calculate the position based on Trodes tracking

Source code in src/spyglass/position/v1/position_trodes_position.py 
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
@schema
class TrodesPosV1(SpyglassMixin, dj.Computed):
    """
    Table to calculate the position based on Trodes tracking
    """

    definition = """
    -> TrodesPosSelection
    ---
    -> AnalysisNwbfile
    position_object_id : varchar(80)
    orientation_object_id : varchar(80)
    velocity_object_id : varchar(80)
    """

    def make(self, key):
        """Populate the table with position data.

        1. Fetch the raw position data and parameters from the RawPosition
            table and TrodesPosParams table, respectively.
        2. Inherit methods from IntervalPositionInfo to calculate the position
            and generate position components (position, orientation, velocity).
        3. Generate AnalysisNwbfile and insert the key into the table.
        4. Insert the key into the PositionOutput Merge table.
        """
        logger.info(f"Computing position for: {key}")
        orig_key = copy.deepcopy(key)

        analysis_file_name = AnalysisNwbfile().create(  # logged
            key["nwb_file_name"]
        )

        raw_position = RawPosition.PosObject & key
        spatial_series = raw_position.fetch_nwb()[0]["raw_position"]
        spatial_df = raw_position.fetch1_dataframe()

        position_info_parameters = (TrodesPosParams() & key).fetch1("params")
        position_info = self.calculate_position_info(
            spatial_df=spatial_df,
            meters_to_pixels=spatial_series.conversion,
            **position_info_parameters,
        )

        key.update(
            dict(
                analysis_file_name=analysis_file_name,
                **self.generate_pos_components(
                    spatial_series=spatial_series,
                    position_info=position_info,
                    analysis_fname=analysis_file_name,
                    prefix="",
                    add_frame_ind=True,
                    video_frame_ind=getattr(
                        spatial_df, "video_frame_ind", None
                    ),
                ),
            )
        )

        AnalysisNwbfile().add(key["nwb_file_name"], analysis_file_name)

        self.insert1(key)

        from ..position_merge import PositionOutput

        # TODO: change to mixin camelize function
        part_name = to_camel_case(self.table_name.split("__")[-1])

        # TODO: The next line belongs in a merge table function
        PositionOutput._merge_insert(
            [orig_key], part_name=part_name, skip_duplicates=True
        )
        AnalysisNwbfile().log(key, table=self.full_table_name)

    @staticmethod
    def generate_pos_components(*args, **kwargs):
        """Generate position components from 2D spatial series."""
        return IntervalPositionInfo().generate_pos_components(*args, **kwargs)

    @staticmethod
    def calculate_position_info(*args, **kwargs):
        """Calculate position info from 2D spatial series."""
        return IntervalPositionInfo().calculate_position_info(*args, **kwargs)

    def fetch1_dataframe(self, add_frame_ind=True) -> DataFrame:
        """Fetch the position data as a pandas DataFrame."""
        pos_params = self.fetch1("trodes_pos_params_name")
        if (
            add_frame_ind
            and (
                TrodesPosParams & {"trodes_pos_params_name": pos_params}
            ).fetch1("params")["is_upsampled"]
        ):
            logger.warning(
                "Upsampled position data, frame indices are invalid. "
                + "Setting add_frame_ind=False"
            )
            add_frame_ind = False
        return IntervalPositionInfo._data_to_df(
            self.fetch_nwb()[0], prefix="", add_frame_ind=add_frame_ind
        )

make(key)

Populate the table with position data.

  1. Fetch the raw position data and parameters from the RawPosition table and TrodesPosParams table, respectively.
  2. Inherit methods from IntervalPositionInfo to calculate the position and generate position components (position, orientation, velocity).
  3. Generate AnalysisNwbfile and insert the key into the table.
  4. Insert the key into the PositionOutput Merge table.
Source code in src/spyglass/position/v1/position_trodes_position.py 
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
def make(self, key):
    """Populate the table with position data.

    1. Fetch the raw position data and parameters from the RawPosition
        table and TrodesPosParams table, respectively.
    2. Inherit methods from IntervalPositionInfo to calculate the position
        and generate position components (position, orientation, velocity).
    3. Generate AnalysisNwbfile and insert the key into the table.
    4. Insert the key into the PositionOutput Merge table.
    """
    logger.info(f"Computing position for: {key}")
    orig_key = copy.deepcopy(key)

    analysis_file_name = AnalysisNwbfile().create(  # logged
        key["nwb_file_name"]
    )

    raw_position = RawPosition.PosObject & key
    spatial_series = raw_position.fetch_nwb()[0]["raw_position"]
    spatial_df = raw_position.fetch1_dataframe()

    position_info_parameters = (TrodesPosParams() & key).fetch1("params")
    position_info = self.calculate_position_info(
        spatial_df=spatial_df,
        meters_to_pixels=spatial_series.conversion,
        **position_info_parameters,
    )

    key.update(
        dict(
            analysis_file_name=analysis_file_name,
            **self.generate_pos_components(
                spatial_series=spatial_series,
                position_info=position_info,
                analysis_fname=analysis_file_name,
                prefix="",
                add_frame_ind=True,
                video_frame_ind=getattr(
                    spatial_df, "video_frame_ind", None
                ),
            ),
        )
    )

    AnalysisNwbfile().add(key["nwb_file_name"], analysis_file_name)

    self.insert1(key)

    from ..position_merge import PositionOutput

    # TODO: change to mixin camelize function
    part_name = to_camel_case(self.table_name.split("__")[-1])

    # TODO: The next line belongs in a merge table function
    PositionOutput._merge_insert(
        [orig_key], part_name=part_name, skip_duplicates=True
    )
    AnalysisNwbfile().log(key, table=self.full_table_name)

generate_pos_components(*args, **kwargs) staticmethod

Generate position components from 2D spatial series.

Source code in src/spyglass/position/v1/position_trodes_position.py 
223
224
225
226
@staticmethod
def generate_pos_components(*args, **kwargs):
    """Generate position components from 2D spatial series."""
    return IntervalPositionInfo().generate_pos_components(*args, **kwargs)

calculate_position_info(*args, **kwargs) staticmethod

Calculate position info from 2D spatial series.

Source code in src/spyglass/position/v1/position_trodes_position.py 
228
229
230
231
@staticmethod
def calculate_position_info(*args, **kwargs):
    """Calculate position info from 2D spatial series."""
    return IntervalPositionInfo().calculate_position_info(*args, **kwargs)

fetch1_dataframe(add_frame_ind=True)

Fetch the position data as a pandas DataFrame.

Source code in src/spyglass/position/v1/position_trodes_position.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def fetch1_dataframe(self, add_frame_ind=True) -> DataFrame:
    """Fetch the position data as a pandas DataFrame."""
    pos_params = self.fetch1("trodes_pos_params_name")
    if (
        add_frame_ind
        and (
            TrodesPosParams & {"trodes_pos_params_name": pos_params}
        ).fetch1("params")["is_upsampled"]
    ):
        logger.warning(
            "Upsampled position data, frame indices are invalid. "
            + "Setting add_frame_ind=False"
        )
        add_frame_ind = False
    return IntervalPositionInfo._data_to_df(
        self.fetch_nwb()[0], prefix="", add_frame_ind=add_frame_ind
    )

TrodesPosVideo

Bases: SpyglassMixin, Computed

Creates a video of the computed head position and orientation as well as the original LED positions overlaid on the video of the animal.

Use for debugging the effect of position extraction parameters.

Source code in src/spyglass/position/v1/position_trodes_position.py 
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
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
@schema
class TrodesPosVideo(SpyglassMixin, dj.Computed):
    """Creates a video of the computed head position and orientation as well as
    the original LED positions overlaid on the video of the animal.

    Use for debugging the effect of position extraction parameters.
    """

    definition = """
    -> TrodesPosV1
    ---
    has_video : bool
    """

    def make(self, key):
        """Generate a video with overlaid position data.

        Fetches...
            - Raw position data from the RawPosition table
            - Position data from the TrodesPosV1 table
            - Video data from the VideoFile table
        Generates a video using opencv and the VideoMaker class.
        """
        M_TO_CM = 100

        logger.info("Loading position data...")
        raw_df = (
            RawPosition.PosObject
            & {
                "nwb_file_name": key["nwb_file_name"],
                "interval_list_name": key["interval_list_name"],
            }
        ).fetch1_dataframe()
        pos_df = (TrodesPosV1() & key).fetch1_dataframe()

        logger.info("Loading video data...")
        epoch = (
            int(
                key["interval_list_name"]
                .replace("pos ", "")
                .replace(" valid times", "")
            )
            + 1
        )

        (
            video_path,
            video_filename,
            meters_per_pixel,
            video_time,
        ) = get_video_info(
            {"nwb_file_name": key["nwb_file_name"], "epoch": epoch}
        )

        if not video_path:
            self.insert1(dict(**key, has_video=False))
            return

        video_path = find_mp4(
            video_path=os.path.dirname(video_path) + "/",
            video_filename=video_filename,
        )

        output_video_filename = (
            key["nwb_file_name"].replace(".nwb", "")
            + f"_{epoch:02d}_"
            + f'{key["trodes_pos_params_name"]}.mp4'
        )

        adj_df = _fix_col_names(raw_df)  # adjust 'xloc1' to 'xloc'

        if test_mode:
            # pytest video data has mismatched shapes in some cases
            min_len = min(len(adj_df), len(pos_df), len(video_time))
            adj_df = adj_df[:min_len]
            pos_df = pos_df[:min_len]
            video_time = video_time[:min_len]

        make_video(
            processor="opencv-trodes",
            video_filename=video_path,
            centroids={
                "red": np.asarray(adj_df[["xloc", "yloc"]]),
                "green": np.asarray(adj_df[["xloc2", "yloc2"]]),
            },
            position_mean=np.asarray(pos_df[["position_x", "position_y"]]),
            orientation_mean=np.asarray(pos_df[["orientation"]]),
            video_time=video_time,
            position_time=np.asarray(pos_df.index),
            output_video_filename=output_video_filename,
            cm_to_pixels=meters_per_pixel * M_TO_CM,
            disable_progressbar=False,
        )
        self.insert1(dict(**key, has_video=True))

make(key)

Generate a video with overlaid position data.

Fetches... - Raw position data from the RawPosition table - Position data from the TrodesPosV1 table - Video data from the VideoFile table Generates a video using opencv and the VideoMaker class.

Source code in src/spyglass/position/v1/position_trodes_position.py 
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
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def make(self, key):
    """Generate a video with overlaid position data.

    Fetches...
        - Raw position data from the RawPosition table
        - Position data from the TrodesPosV1 table
        - Video data from the VideoFile table
    Generates a video using opencv and the VideoMaker class.
    """
    M_TO_CM = 100

    logger.info("Loading position data...")
    raw_df = (
        RawPosition.PosObject
        & {
            "nwb_file_name": key["nwb_file_name"],
            "interval_list_name": key["interval_list_name"],
        }
    ).fetch1_dataframe()
    pos_df = (TrodesPosV1() & key).fetch1_dataframe()

    logger.info("Loading video data...")
    epoch = (
        int(
            key["interval_list_name"]
            .replace("pos ", "")
            .replace(" valid times", "")
        )
        + 1
    )

    (
        video_path,
        video_filename,
        meters_per_pixel,
        video_time,
    ) = get_video_info(
        {"nwb_file_name": key["nwb_file_name"], "epoch": epoch}
    )

    if not video_path:
        self.insert1(dict(**key, has_video=False))
        return

    video_path = find_mp4(
        video_path=os.path.dirname(video_path) + "/",
        video_filename=video_filename,
    )

    output_video_filename = (
        key["nwb_file_name"].replace(".nwb", "")
        + f"_{epoch:02d}_"
        + f'{key["trodes_pos_params_name"]}.mp4'
    )

    adj_df = _fix_col_names(raw_df)  # adjust 'xloc1' to 'xloc'

    if test_mode:
        # pytest video data has mismatched shapes in some cases
        min_len = min(len(adj_df), len(pos_df), len(video_time))
        adj_df = adj_df[:min_len]
        pos_df = pos_df[:min_len]
        video_time = video_time[:min_len]

    make_video(
        processor="opencv-trodes",
        video_filename=video_path,
        centroids={
            "red": np.asarray(adj_df[["xloc", "yloc"]]),
            "green": np.asarray(adj_df[["xloc2", "yloc2"]]),
        },
        position_mean=np.asarray(pos_df[["position_x", "position_y"]]),
        orientation_mean=np.asarray(pos_df[["orientation"]]),
        video_time=video_time,
        position_time=np.asarray(pos_df.index),
        output_video_filename=output_video_filename,
        cm_to_pixels=meters_per_pixel * M_TO_CM,
        disable_progressbar=False,
    )
    self.insert1(dict(**key, has_video=True))