From ea9dd555d826eafe5d3e180509b755a92ac5b5a2 Mon Sep 17 00:00:00 2001 From: wendtalexander Date: Tue, 21 Oct 2025 11:56:07 +0200 Subject: [PATCH] [docstring] --- oephys2nix/stimulus_recreation.py | 158 +++++++++++++++++++++++++++--- 1 file changed, 144 insertions(+), 14 deletions(-) diff --git a/oephys2nix/stimulus_recreation.py b/oephys2nix/stimulus_recreation.py index e08b935..fce3fcf 100644 --- a/oephys2nix/stimulus_recreation.py +++ b/oephys2nix/stimulus_recreation.py @@ -1,7 +1,6 @@ import logging import pathlib import sys -import tomllib import matplotlib.pyplot as plt import nixio @@ -22,6 +21,47 @@ console = Console() class StimulusToNix: + """Processing the stimulus recreation from the relax dataset and the open-ephys data. + + Parameters + ---------- + open_ephys_path: pathlib.Path + Path to open-ephys recording + relacs_nix_path : str + Path to relacs nix file + nix_file : str + Path to new nix file + + Attributes + ---------- + relacs_nix_path : str + Path to relacs nix file + nix_file_path : str + Path to new nix file + dataset : rlx.Dataset + Dataset from the relacs file + relacs_nix_file : nixio.File + Relacs nix file + relacs_block : nixio.Block + Relacs nix block + relacs_sections :nixio.Section + Relacs nix Section + neo_data : neo.OpenEphysBinaryIO + Open-ephys data + fs : float + Sample rate of the open-ephys recording + nix_file : nixio.File + New nix file + block : nixio.Block + New nix block + threshold : float + Threshold for TTL line + new_start_jiggle : float + For finding the new start, ensuring finding next TTL pulse + + + """ + def __init__(self, open_ephys_path: pathlib.Path, relacs_nix_path: str, nix_file: str): self.relacs_nix_path = relacs_nix_path self.nix_file_path = nix_file @@ -43,7 +83,8 @@ class StimulusToNix: self.threshold = 2 self.new_start_jiggle = 0.1 - def _append_relacs_tag_mtags(self): + def _append_relacs_tag_mtags(self) -> None: + """Append relacs tags and multi tags to new nix file.""" for t in self.relacs_block.tags: log.debug(f"Appending relacs tags {t.name}") tag = self.block.create_tag(f"relacs_{t.name}", t.type, position=t.position) @@ -80,7 +121,31 @@ class StimulusToNix: pass mtag.metadata = self.nix_file.sections[sec.name] - def _find_peak_ttl(self, time_ttl, peaks_ttl, lower, upper): + def _find_next_ttl( + self, time_ttl: np.ndarray, peaks_ttl: np.ndarray, lower: float, upper: float + ) -> np.ndarray: + """Find the next TTL pulse within a specific duration constrained by lower and upper. + + Parameters + ---------- + time_ttl : np.ndarray + Time array of the TTL line + + peaks_ttl : np.ndarray + Detected peaks indeces on the TTL line + + lower : float + lower bound for searching the TTL pulse + + upper : float + upper bound for searching the TTL pulse + + Returns + ------- + np.ndarray + time point of the new TTL pulse + + """ peak = time_ttl[peaks_ttl[(time_ttl[peaks_ttl] > lower) & (time_ttl[peaks_ttl] < upper)]] if not peak.size > 0: @@ -92,7 +157,28 @@ class StimulusToNix: peak = np.mean(peak) return peak - def _find_peak_ttl_index(self, time_ttl, peaks_ttl, current_position): + def _find_peak_ttl_index( + self, time_ttl: np.ndarray, peaks_ttl: np.ndarray, current_position: np.ndarray + ) -> np.ndarray: + """Find the next TTL pulse from the indeces of the detected TTL pulses. + + Parameters + ---------- + time_ttl : np.ndarray + Time array of the TTL line + + peaks_ttl : np.ndarray + Detected peaks indeces on the TTL line + + current_position : np.ndarray + Current time of the TTL pulse + + Returns + ------- + np.ndarray + Next time of TTL pulse + + """ new_repro_start_index = peaks_ttl[ (time_ttl[peaks_ttl] > current_position - self.new_start_jiggle) & (time_ttl[peaks_ttl] < current_position + self.new_start_jiggle) @@ -112,13 +198,34 @@ class StimulusToNix: @property def _reference_groups(self) -> list[nixio.Group]: + """Holds the reference groups. + + Returns + ------- + list[nixio.Group] + + """ return [ self.block.groups["neuronal-data"], self.block.groups["efish"], self.block.groups["relacs"], ] - def _append_mtag(self, repro, positions, extents): + def _append_mtag(self, repro: rlx.Dataset, positions: np.ndarray, extents: np.ndarray) -> None: + """Apped multi tags of the current repro to the nix file. + + Parameters + ---------- + repro : rlx.Dataset + Current Repro + + positions : np.ndarray + postions of the multi tags + + extents : np.ndarray + extents of the multi tags + + """ try: nix_mtag = self.block.create_multi_tag( f"{repro.name}", @@ -159,7 +266,20 @@ class StimulusToNix: nix_group = self.nix_file.blocks[0].groups[repro.name] nix_group.multi_tags.append(nix_mtag) - def _append_tag(self, repro, position, extent): + def _append_tag(self, repro: rlx.Dataset, position: np.ndarray, extent: np.ndarray) -> None: + """Append tag of the current repro. + + Parameters + ---------- + repro : rlx.Dataset + Current Repro + + position : np.ndarray + positions of the multi tags + + extent : np.ndarray + extents of the multi tags + """ try: nix_tag = self.block.create_tag( f"{repro.name}", @@ -201,7 +321,8 @@ class StimulusToNix: nix_group = self.nix_file.blocks[0].groups[repro.name] nix_group.tags.append(nix_tag) - def create_repros_automatically(self): + def create_repros_automatically(self) -> None: + """Create the repros form relacs with the TTL pulses.""" ttl_oeph = self.block.data_arrays["ttl-line"][:] time_ttl = np.arange(len(ttl_oeph)) / self.fs time_index = np.arange(len(ttl_oeph)) @@ -214,7 +335,7 @@ class StimulusToNix: if close_peaks.size > 0: peaks_ttl = np.delete(peaks_ttl, close_peaks) - first_peak = self._find_peak_ttl( + first_peak = self._find_next_ttl( time_ttl, peaks_ttl, time_ttl[peaks_ttl[0]] - self.new_start_jiggle, @@ -229,6 +350,8 @@ class StimulusToNix: log.warning(f"Skipping repro {repro.name} because it is two short") continue + embed() + exit() if repro.stimuli: log.debug("Processing MultiTag") repetition = len(repro.stimuli) @@ -244,7 +367,9 @@ class StimulusToNix: time_ttl, peaks_ttl, current_position ) - # current_position = position_mtags[-1] + if "FICurve" in repro.name: + position_mtags += 0.2 + self._append_mtag(repro, position_mtags, extents_mtag) extent = position_mtags[-1] + extents_mtag[-1] - position_mtags[0] self._append_tag(repro, position_mtags[0], extent) @@ -282,9 +407,11 @@ class StimulusToNix: last_repro_position.reshape(-1, 1), (current_position - last_repro_position).reshape(-1, 1), ) - # self.close() - def create_repros_from_config_file(self): + def create_repros_from_config_file(self) -> None: + """Creates repros form a config file. + NOT MAINTAINED. + """ ttl_oeph = self.block.data_arrays["ttl-line"][:] peaks_ttl = signal.find_peaks( ttl_oeph.flatten(), @@ -429,7 +556,8 @@ class StimulusToNix: ] nix_group.multi_tags.append(nix_mtag) - def print_table(self): + def print_table(self) -> None: + """Print the converted times in a rich table.""" nix_data_set = rlx.Dataset(self.nix_file_path) table = Table("Repro Name", "start", "stop", "duration") for repro_r, repro_n in zip(self.dataset.repro_runs(), nix_data_set.repro_runs()): @@ -443,7 +571,8 @@ class StimulusToNix: console.print(table) nix_data_set.close() - def checks(self): + def checks(self) -> None: + """Just for debugging currently.""" important_repros = ["FileStimulus", "SAM", "FICurve"] nix_data_set = rlx.Dataset(self.nix_file_path) @@ -508,7 +637,8 @@ class StimulusToNix: v_eod, t_eod = repro_n.trace_data("global-eod") v_eodr, t_eodr = repro_n.trace_data("EOD") - def plot_stimulus(self): + def plot_stimulus(self) -> None: + """Plot the relacs stimulus, open-epyhs and TTL line.""" ttl_oeph = self.block.data_arrays["ttl-line"][:] time_index = np.arange(len(ttl_oeph))