Skip to content

decoding_merge.py

DecodingOutput

Bases: _Merge, SpyglassMixin

Source code in src/spyglass/decoding/decoding_merge.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
@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 cleanup(self, dry_run=False):
        """Remove any decoding outputs that are not in the merge table"""
        if dry_run:
            logger.info("Dry run, not removing any files")
        else:
            logger.info("Cleaning up decoding outputs")
        table_results_paths = list(
            chain(
                *[
                    part_parent_table.fetch("results_path").tolist()
                    for part_parent_table in self.merge_get_parent(
                        multi_source=True
                    )
                ]
            )
        )
        for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.nc"):
            if str(path) not in table_results_paths:
                logger.info(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 = list(
            chain(
                *[
                    part_parent_table.fetch("classifier_path").tolist()
                    for part_parent_table in self.merge_get_parent(
                        multi_source=True
                    )
                ]
            )
        )
        for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.pkl"):
            if str(path) not in table_model_paths:
                logger.info(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_get_parent_class(key).fetch_results()

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

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

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

    @classmethod
    def fetch_linear_position_info(cls, key):
        """Fetch the decoding linear position info for a given key."""
        decoding_selection_key = cls.merge_get_parent(key).fetch1("KEY")
        return (
            cls()
            .merge_get_parent_class(key)
            .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."""
        decoding_selection_key = cls.merge_get_parent(key).fetch1("KEY")
        return (
            cls()
            .merge_get_parent_class(key)
            .fetch_linear_position_info(
                decoding_selection_key, filter_by_interval=filter_by_interval
            )
        )

    @classmethod
    def create_decoding_view(cls, key, head_direction_name="head_orientation"):
        """Create a decoding view for a given key."""
        results = cls.fetch_results(key)
        posterior = (
            results.squeeze()
            .acausal_posterior.unstack("state_bins")
            .drop_sel(state=["Local", "No-Spike"], errors="ignore")
            .sum("state")
        )
        posterior /= posterior.sum("position")
        env = cls.fetch_environments(key)[0]

        if "x_position" in results.coords:
            position_info, position_variable_names = cls.fetch_position_info(
                key
            )
            # Not 1D
            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=position_info[head_direction_name],
            )
        else:
            (
                position_info,
                position_variable_names,
            ) = cls.fetch_linear_position_info(key)
            return create_1D_decode_view(
                posterior=posterior,
                linear_position=position_info["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 
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
def cleanup(self, dry_run=False):
    """Remove any decoding outputs that are not in the merge table"""
    if dry_run:
        logger.info("Dry run, not removing any files")
    else:
        logger.info("Cleaning up decoding outputs")
    table_results_paths = list(
        chain(
            *[
                part_parent_table.fetch("results_path").tolist()
                for part_parent_table in self.merge_get_parent(
                    multi_source=True
                )
            ]
        )
    )
    for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.nc"):
        if str(path) not in table_results_paths:
            logger.info(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 = list(
        chain(
            *[
                part_parent_table.fetch("classifier_path").tolist()
                for part_parent_table in self.merge_get_parent(
                    multi_source=True
                )
            ]
        )
    )
    for path in Path(config["SPYGLASS_ANALYSIS_DIR"]).glob("**/*.pkl"):
        if str(path) not in table_model_paths:
            logger.info(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 
85
86
87
88
@classmethod
def fetch_results(cls, key):
    """Fetch the decoding results for a given key."""
    return cls().merge_get_parent_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 
90
91
92
93
@classmethod
def fetch_model(cls, key):
    """Fetch the decoding model for a given key."""
    return cls().merge_get_parent_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 
 95
 96
 97
 98
 99
100
101
102
103
@classmethod
def fetch_environments(cls, key):
    """Fetch the decoding environments for a given key."""
    decoding_selection_key = cls.merge_get_parent(key).fetch1("KEY")
    return (
        cls()
        .merge_get_parent_class(key)
        .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 
105
106
107
108
109
110
111
112
113
@classmethod
def fetch_position_info(cls, key):
    """Fetch the decoding position info for a given key."""
    decoding_selection_key = cls.merge_get_parent(key).fetch1("KEY")
    return (
        cls()
        .merge_get_parent_class(key)
        .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 
115
116
117
118
119
120
121
122
123
@classmethod
def fetch_linear_position_info(cls, key):
    """Fetch the decoding linear position info for a given key."""
    decoding_selection_key = cls.merge_get_parent(key).fetch1("KEY")
    return (
        cls()
        .merge_get_parent_class(key)
        .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 
125
126
127
128
129
130
131
132
133
134
135
@classmethod
def fetch_spike_data(cls, key, filter_by_interval=True):
    """Fetch the decoding spike data for a given key."""
    decoding_selection_key = cls.merge_get_parent(key).fetch1("KEY")
    return (
        cls()
        .merge_get_parent_class(key)
        .fetch_linear_position_info(
            decoding_selection_key, filter_by_interval=filter_by_interval
        )
    )

create_decoding_view(key, head_direction_name='head_orientation') classmethod

Create a decoding view for a given key.

Source code in src/spyglass/decoding/decoding_merge.py 
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
@classmethod
def create_decoding_view(cls, key, head_direction_name="head_orientation"):
    """Create a decoding view for a given key."""
    results = cls.fetch_results(key)
    posterior = (
        results.squeeze()
        .acausal_posterior.unstack("state_bins")
        .drop_sel(state=["Local", "No-Spike"], errors="ignore")
        .sum("state")
    )
    posterior /= posterior.sum("position")
    env = cls.fetch_environments(key)[0]

    if "x_position" in results.coords:
        position_info, position_variable_names = cls.fetch_position_info(
            key
        )
        # Not 1D
        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=position_info[head_direction_name],
        )
    else:
        (
            position_info,
            position_variable_names,
        ) = cls.fetch_linear_position_info(key)
        return create_1D_decode_view(
            posterior=posterior,
            linear_position=position_info["linear_position"],
        )