Skip to content

decoding_merge.py

create_1D_decode_view(*args, **kwargs)

Lazily dispatch to non_local_detector's 1D figurl view.

Source code in src/spyglass/decoding/decoding_merge.py 
20
21
22
23
24
25
26
def create_1D_decode_view(*args, **kwargs):
    """Lazily dispatch to non_local_detector's 1D figurl view."""
    from non_local_detector.visualization.figurl_1D import (
        create_1D_decode_view as _view,
    )

    return _view(*args, **kwargs)

create_2D_decode_view(*args, **kwargs)

Lazily dispatch to non_local_detector's 2D figurl view.

Source code in src/spyglass/decoding/decoding_merge.py 
29
30
31
32
33
34
35
def create_2D_decode_view(*args, **kwargs):
    """Lazily dispatch to non_local_detector's 2D figurl view."""
    from non_local_detector.visualization.figurl_2D import (
        create_2D_decode_view as _view,
    )

    return _view(*args, **kwargs)

DecodingOutput

Bases: _Merge, SpyglassMixin

Source code in src/spyglass/decoding/decoding_merge.py 
 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
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
@schema
class DecodingOutput(_Merge, SpyglassMixin):
    definition = """
    merge_id: uuid
    ---
    source: varchar(32)
    """

    class ClusterlessDecodingV1(SpyglassMixin, dj.Part):  # noqa: F811
        definition = """
        -> master
        ---
        -> ClusterlessDecodingV1
        """

    class SortedSpikesDecodingV1(SpyglassMixin, dj.Part):  # noqa: F811
        definition = """
        -> master
        ---
        -> SortedSpikesDecodingV1
        """

    def _fetch_registered_paths(self, attr):
        """Fetch a filepath attribute from all part parents, skipping missing."""
        paths = []
        for tbl in self.merge_get_parent(multi_source=True):
            paths.extend(tbl.fetch(attr).tolist())
        return paths

    def cleanup(self, dry_run=False):
        """Remove any decoding outputs that are not in the merge table"""
        if dry_run:
            self._info_msg("Dry run, not removing any files")
        else:
            self._info_msg("Cleaning up decoding outputs")
        table_results_paths = self._fetch_registered_paths("results_path")
        for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.nc"):
            if str(path) not in table_results_paths:
                self._info_msg(f"Removing {path}")
                if not dry_run:
                    try:
                        path.unlink(missing_ok=True)  # Ignore FileNotFoundError
                    except PermissionError:
                        logger.warning(f"Unable to remove {path}, skipping")

        table_model_paths = self._fetch_registered_paths("classifier_path")
        for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.pkl"):
            if str(path) not in table_model_paths:
                self._info_msg(f"Removing {path}")
                if not dry_run:
                    try:
                        path.unlink()
                    except (PermissionError, FileNotFoundError):
                        logger.warning(f"Unable to remove {path}, skipping")

    @classmethod
    def fetch_results(cls, key):
        """Fetch the decoding results for a given key."""
        return cls().merge_restrict_class(key).fetch_results()

    @classmethod
    def fetch_model(cls, key):
        """Fetch the decoding model for a given key."""
        return cls().merge_restrict_class(key).fetch_model()

    @classmethod
    def fetch_environments(cls, key):
        """Fetch the decoding environments for a given key."""
        restr_parent = cls().merge_restrict_class(key)
        decoding_selection_key = restr_parent.fetch1("KEY")
        return restr_parent.fetch_environments(decoding_selection_key)

    @classmethod
    def fetch_position_info(cls, key):
        """Fetch the decoding position info for a given key."""
        restr_parent = cls().merge_restrict_class(key)
        decoding_selection_key = restr_parent.fetch1("KEY")
        return restr_parent.fetch_position_info(decoding_selection_key)

    @classmethod
    def fetch_linear_position_info(cls, key):
        """Fetch the decoding linear position info for a given key."""
        restr_parent = cls().merge_restrict_class(key)
        decoding_selection_key = restr_parent.fetch1("KEY")
        return restr_parent.fetch_linear_position_info(decoding_selection_key)

    @classmethod
    def fetch_spike_data(cls, key, filter_by_interval=True):
        """Fetch the decoding spike data for a given key."""
        restr_parent = cls().merge_restrict_class(key)
        decoding_selection_key = restr_parent.fetch1("KEY")
        return restr_parent.fetch_spike_data(
            decoding_selection_key, filter_by_interval=filter_by_interval
        )

    @classmethod
    def create_decoding_view(
        cls, key, head_direction_name="head_orientation", interval_idx=None
    ):
        """Create a decoding view for a given key.

        Parameters
        ----------
        key : dict
            Key identifying the decoding output
        head_direction_name : str, optional
            Name of the head direction column for 2D views, by default
            "head_orientation". If the column is absent (e.g. v1 Trodes/DLC
            position uses "orientation"), it is auto-detected; if no
            orientation column is present, head direction is omitted from
            the view.
        interval_idx : int, optional
            If specified, only visualize this interval (0-indexed).
            If None (default), visualize all intervals together.

        Returns
        -------
        view
            Figurl visualization view (1D or 2D depending on decoder)
        """
        results = cls.fetch_results(key)

        # Filter to specific interval if requested
        if interval_idx is not None:
            if "interval_labels" in results.coords:
                results = results.where(
                    results.interval_labels == interval_idx, drop=True
                )
            else:
                logger.warning(
                    f"interval_idx={interval_idx} specified but results do not "
                    "have 'interval_labels' coordinate. Ignoring interval_idx."
                )

        posterior = (
            results.acausal_posterior.unstack("state_bins")
            .drop_sel(state=["Local", "No-Spike"], errors="ignore")
            .sum("state")
        )
        # Normalize over the spatial dimension(s): 1D posteriors have a
        # "position" dim while 2D posteriors have "x_position"/"y_position".
        is_2d = "x_position" in results.coords
        spatial_dims = ["x_position", "y_position"] if is_2d else ["position"]
        posterior /= posterior.sum(spatial_dims)
        env = cls.fetch_environments(key)[0]

        if is_2d:
            position_info, position_variable_names = cls.fetch_position_info(
                key
            )
            # The orientation column name varies by position source: v1
            # Trodes/DLC use "orientation" while legacy common position uses
            # "head_orientation". Honor the requested column if present;
            # otherwise auto-detect, falling back to None (head_dir is
            # optional). Warn so a silent substitution/drop (e.g. from a typo)
            # is visible.
            requested = head_direction_name
            head_direction_name = resolve_orientation_col(
                position_info, orientation_name=head_direction_name
            )
            if requested not in position_info.columns:
                if head_direction_name is not None:
                    logger.warning(
                        f"head_direction_name='{requested}' not found in "
                        f"position columns; using '{head_direction_name}'."
                    )
                else:
                    logger.warning(
                        f"head_direction_name='{requested}' not found in "
                        f"position columns {list(position_info.columns)} and no "
                        "orientation column detected; omitting head direction "
                        "from the 2D view."
                    )
            head_dir = (
                position_info[head_direction_name]
                if head_direction_name is not None
                else None
            )
            bin_size = (
                np.nanmedian(np.diff(np.unique(results.x_position.values))),
                np.nanmedian(np.diff(np.unique(results.y_position.values))),
            )
            return create_2D_decode_view(
                position_time=position_info.index,
                position=position_info[position_variable_names],
                interior_place_bin_centers=env.place_bin_centers_[
                    env.is_track_interior_.ravel(order="C")
                ],
                place_bin_size=bin_size,
                posterior=posterior,
                head_dir=head_dir,
            )
        else:
            # create_1D_decode_view expects a 1D linear position, but
            # fetch_linear_position_info returns a multi-column DataFrame.
            return create_1D_decode_view(
                posterior=posterior,
                linear_position=cls.fetch_linear_position_info(key)[
                    "linear_position"
                ],
            )

cleanup(dry_run=False)

Remove any decoding outputs that are not in the merge table

Source code in src/spyglass/decoding/decoding_merge.py 
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 cleanup(self, dry_run=False):
    """Remove any decoding outputs that are not in the merge table"""
    if dry_run:
        self._info_msg("Dry run, not removing any files")
    else:
        self._info_msg("Cleaning up decoding outputs")
    table_results_paths = self._fetch_registered_paths("results_path")
    for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.nc"):
        if str(path) not in table_results_paths:
            self._info_msg(f"Removing {path}")
            if not dry_run:
                try:
                    path.unlink(missing_ok=True)  # Ignore FileNotFoundError
                except PermissionError:
                    logger.warning(f"Unable to remove {path}, skipping")

    table_model_paths = self._fetch_registered_paths("classifier_path")
    for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.pkl"):
        if str(path) not in table_model_paths:
            self._info_msg(f"Removing {path}")
            if not dry_run:
                try:
                    path.unlink()
                except (PermissionError, FileNotFoundError):
                    logger.warning(f"Unable to remove {path}, skipping")

fetch_results(key) classmethod

Fetch the decoding results for a given key.

Source code in src/spyglass/decoding/decoding_merge.py 
93
94
95
96
@classmethod
def fetch_results(cls, key):
    """Fetch the decoding results for a given key."""
    return cls().merge_restrict_class(key).fetch_results()

fetch_model(key) classmethod

Fetch the decoding model for a given key.

Source code in src/spyglass/decoding/decoding_merge.py 
 98
 99
100
101
@classmethod
def fetch_model(cls, key):
    """Fetch the decoding model for a given key."""
    return cls().merge_restrict_class(key).fetch_model()

fetch_environments(key) classmethod

Fetch the decoding environments for a given key.

Source code in src/spyglass/decoding/decoding_merge.py 
103
104
105
106
107
108
@classmethod
def fetch_environments(cls, key):
    """Fetch the decoding environments for a given key."""
    restr_parent = cls().merge_restrict_class(key)
    decoding_selection_key = restr_parent.fetch1("KEY")
    return restr_parent.fetch_environments(decoding_selection_key)

fetch_position_info(key) classmethod

Fetch the decoding position info for a given key.

Source code in src/spyglass/decoding/decoding_merge.py 
110
111
112
113
114
115
@classmethod
def fetch_position_info(cls, key):
    """Fetch the decoding position info for a given key."""
    restr_parent = cls().merge_restrict_class(key)
    decoding_selection_key = restr_parent.fetch1("KEY")
    return restr_parent.fetch_position_info(decoding_selection_key)

fetch_linear_position_info(key) classmethod

Fetch the decoding linear position info for a given key.

Source code in src/spyglass/decoding/decoding_merge.py 
117
118
119
120
121
122
@classmethod
def fetch_linear_position_info(cls, key):
    """Fetch the decoding linear position info for a given key."""
    restr_parent = cls().merge_restrict_class(key)
    decoding_selection_key = restr_parent.fetch1("KEY")
    return restr_parent.fetch_linear_position_info(decoding_selection_key)

fetch_spike_data(key, filter_by_interval=True) classmethod

Fetch the decoding spike data for a given key.

Source code in src/spyglass/decoding/decoding_merge.py 
124
125
126
127
128
129
130
131
@classmethod
def fetch_spike_data(cls, key, filter_by_interval=True):
    """Fetch the decoding spike data for a given key."""
    restr_parent = cls().merge_restrict_class(key)
    decoding_selection_key = restr_parent.fetch1("KEY")
    return restr_parent.fetch_spike_data(
        decoding_selection_key, filter_by_interval=filter_by_interval
    )

create_decoding_view(key, head_direction_name='head_orientation', interval_idx=None) classmethod

Create a decoding view for a given key.

Parameters:

Name Type Description Default
key dict

Key identifying the decoding output

 required
head_direction_name str

Name of the head direction column for 2D views, by default "head_orientation". If the column is absent (e.g. v1 Trodes/DLC position uses "orientation"), it is auto-detected; if no orientation column is present, head direction is omitted from the view.

 'head_orientation'
interval_idx int

If specified, only visualize this interval (0-indexed). If None (default), visualize all intervals together.

 None

Returns:

Type Description
view

Figurl visualization view (1D or 2D depending on decoder)
Source code in src/spyglass/decoding/decoding_merge.py 
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
@classmethod
def create_decoding_view(
    cls, key, head_direction_name="head_orientation", interval_idx=None
):
    """Create a decoding view for a given key.

    Parameters
    ----------
    key : dict
        Key identifying the decoding output
    head_direction_name : str, optional
        Name of the head direction column for 2D views, by default
        "head_orientation". If the column is absent (e.g. v1 Trodes/DLC
        position uses "orientation"), it is auto-detected; if no
        orientation column is present, head direction is omitted from
        the view.
    interval_idx : int, optional
        If specified, only visualize this interval (0-indexed).
        If None (default), visualize all intervals together.

    Returns
    -------
    view
        Figurl visualization view (1D or 2D depending on decoder)
    """
    results = cls.fetch_results(key)

    # Filter to specific interval if requested
    if interval_idx is not None:
        if "interval_labels" in results.coords:
            results = results.where(
                results.interval_labels == interval_idx, drop=True
            )
        else:
            logger.warning(
                f"interval_idx={interval_idx} specified but results do not "
                "have 'interval_labels' coordinate. Ignoring interval_idx."
            )

    posterior = (
        results.acausal_posterior.unstack("state_bins")
        .drop_sel(state=["Local", "No-Spike"], errors="ignore")
        .sum("state")
    )
    # Normalize over the spatial dimension(s): 1D posteriors have a
    # "position" dim while 2D posteriors have "x_position"/"y_position".
    is_2d = "x_position" in results.coords
    spatial_dims = ["x_position", "y_position"] if is_2d else ["position"]
    posterior /= posterior.sum(spatial_dims)
    env = cls.fetch_environments(key)[0]

    if is_2d:
        position_info, position_variable_names = cls.fetch_position_info(
            key
        )
        # The orientation column name varies by position source: v1
        # Trodes/DLC use "orientation" while legacy common position uses
        # "head_orientation". Honor the requested column if present;
        # otherwise auto-detect, falling back to None (head_dir is
        # optional). Warn so a silent substitution/drop (e.g. from a typo)
        # is visible.
        requested = head_direction_name
        head_direction_name = resolve_orientation_col(
            position_info, orientation_name=head_direction_name
        )
        if requested not in position_info.columns:
            if head_direction_name is not None:
                logger.warning(
                    f"head_direction_name='{requested}' not found in "
                    f"position columns; using '{head_direction_name}'."
                )
            else:
                logger.warning(
                    f"head_direction_name='{requested}' not found in "
                    f"position columns {list(position_info.columns)} and no "
                    "orientation column detected; omitting head direction "
                    "from the 2D view."
                )
        head_dir = (
            position_info[head_direction_name]
            if head_direction_name is not None
            else None
        )
        bin_size = (
            np.nanmedian(np.diff(np.unique(results.x_position.values))),
            np.nanmedian(np.diff(np.unique(results.y_position.values))),
        )
        return create_2D_decode_view(
            position_time=position_info.index,
            position=position_info[position_variable_names],
            interior_place_bin_centers=env.place_bin_centers_[
                env.is_track_interior_.ravel(order="C")
            ],
            place_bin_size=bin_size,
            posterior=posterior,
            head_dir=head_dir,
        )
    else:
        # create_1D_decode_view expects a 1D linear position, but
        # fetch_linear_position_info returns a multi-column DataFrame.
        return create_1D_decode_view(
            posterior=posterior,
            linear_position=cls.fetch_linear_position_info(key)[
                "linear_position"
            ],
        )