[docstring]
This commit is contained in:
parent
cd7cd43313
commit
ea9dd555d8
@ -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))
|
||||
|
||||
Loading…
Reference in New Issue
Block a user