184 lines
6.1 KiB
Python
184 lines
6.1 KiB
Python
import logging
|
|
import pathlib
|
|
import sys
|
|
|
|
import nixio
|
|
import numpy as np
|
|
import rlxnix as rlx
|
|
from IPython import embed
|
|
from neo.io import OpenEphysBinaryIO
|
|
|
|
from oephys2nix.logging import setup_logging
|
|
from oephys2nix.metadata import create_dict_from_section, create_metadata_from_dict
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class RawToNix:
|
|
"""Appending all raw data from relacs and open-ephsy to a new nix file.
|
|
|
|
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_file : nixio.File
|
|
Relacs nix file
|
|
dataset : rlx.Dataset
|
|
Dataset of the relacs file
|
|
relacs_block : nixio.Block
|
|
Relacs block
|
|
relacs_sections : nixio.Section
|
|
Relacs section
|
|
neo_data : neo.OpenEphysBinaryIO
|
|
Open Ephys data
|
|
nix_file : nixio.File
|
|
New nix file
|
|
block : nixio.Block
|
|
New nix file block
|
|
|
|
"""
|
|
|
|
def __init__(self, open_ephys_path: pathlib.Path, relacs_nix_path: str, nix_file: str):
|
|
self.relacs_nix_file = nixio.File.open(relacs_nix_path, nixio.FileMode.ReadOnly)
|
|
self.dataset = rlx.Dataset(relacs_nix_path)
|
|
self.relacs_block = self.relacs_nix_file.blocks[0]
|
|
self.relacs_sections = self.relacs_nix_file.sections
|
|
self.neo_data = OpenEphysBinaryIO(open_ephys_path).read(lazy=True)
|
|
|
|
self.nix_file = nixio.File(nix_file, nixio.FileMode.ReadWrite)
|
|
self.nix_file.create_block("open-ephys.data", "open-ephys.sampled")
|
|
self.block = self.nix_file.blocks[0]
|
|
|
|
def append_section(self) -> None:
|
|
"""Append sections from relacs."""
|
|
sec = self.nix_file.create_section(
|
|
self.relacs_sections[0].name, self.relacs_sections[0].type
|
|
)
|
|
d = create_dict_from_section(self.relacs_sections[0])
|
|
create_metadata_from_dict(d, sec)
|
|
self.block.metadata = sec
|
|
|
|
def append_fish_lines(self) -> None:
|
|
"""Append fish lines from open-ephys."""
|
|
efishs = ["ttl-line", "global-eod", "stimulus", "local-eod", "sinus"]
|
|
|
|
efish_types = [
|
|
"open-ephys.data.sampled",
|
|
"open-ephys.data.sampled",
|
|
"open-ephys.data.sampled",
|
|
"open-ephys.data.sampled",
|
|
"open-ephys.data.sampled",
|
|
]
|
|
|
|
efish_group = self.block.create_group("efish", "open-ephys.sampled")
|
|
|
|
efish_neo_data = self._load_neo_object("Data_ADC")
|
|
|
|
for i in np.arange(len(efishs)):
|
|
log.debug(f"Appending efish traces {efishs[i]}")
|
|
efish_neo_data_array = efish_neo_data[:, i]
|
|
data_array = self.block.create_data_array(
|
|
f"{efishs[i]}",
|
|
f"{efish_types[i]}",
|
|
data=efish_neo_data_array.magnitude.flatten(),
|
|
label="voltage",
|
|
unit="V",
|
|
)
|
|
data_array.append_sampled_dimension(
|
|
1 / efish_neo_data.sampling_rate.magnitude, label="time", unit="s"
|
|
)
|
|
efish_group.data_arrays.append(data_array)
|
|
|
|
def _load_neo_object(self, name: str):
|
|
for sig in self.neo_data[0].segments[0].analogsignals:
|
|
if sig.name.endswith(name):
|
|
return sig.load()
|
|
|
|
log.error(f"No {name} found in open ephys data")
|
|
sys.exit(1)
|
|
|
|
def append_relacs_lines(self) -> None:
|
|
"""Append relacs lines."""
|
|
relacs = [
|
|
"V-1",
|
|
"EOD",
|
|
"LocalEOD-1",
|
|
"GlobalEFieldStimulus",
|
|
"Spikes-1",
|
|
"Chirps",
|
|
"LocalBeat-1-1",
|
|
]
|
|
|
|
relacs_types = [
|
|
"relacs.data.sampled.V-1",
|
|
"relacs.data.sampled.EOD",
|
|
"relacs.data.sampled.LocalEOD-1",
|
|
"relacs.data.sampled.GlobalEFieldStimulus",
|
|
"relacs.data.events.Spikes-1",
|
|
"relacs.data.events.Chirps",
|
|
"relacs.data.events.LocalBeat-1-1",
|
|
]
|
|
|
|
efish_group = self.block.create_group("relacs", "relacs.sampled")
|
|
|
|
for i, re in enumerate(relacs):
|
|
log.debug(f"Appending relacs efish traces {re}")
|
|
efish_relacs_data_array = self.relacs_block.data_arrays[re]
|
|
data_array = self.block.create_data_array(
|
|
f"{re}",
|
|
f"{relacs_types[i]}",
|
|
data=efish_relacs_data_array[:],
|
|
label=efish_relacs_data_array.label,
|
|
unit=efish_relacs_data_array.unit,
|
|
)
|
|
for d in efish_relacs_data_array.dimensions:
|
|
if d.dimension_type == nixio.DimensionType.Sample:
|
|
data_array.append_sampled_dimension(
|
|
efish_relacs_data_array.dimensions[0].sampling_interval,
|
|
label="time",
|
|
unit="s",
|
|
)
|
|
elif d.dimension_type == nixio.DimensionType.Range:
|
|
data_array.append_range_dimension(
|
|
np.sort(efish_relacs_data_array.dimensions[0].ticks),
|
|
label="time",
|
|
unit="s",
|
|
)
|
|
else:
|
|
data_array.append_set_dimension(
|
|
label="time",
|
|
unit="s",
|
|
)
|
|
|
|
efish_group.data_arrays.append(data_array)
|
|
|
|
def append_raw_data(self) -> None:
|
|
"""Append Open-Ephys Raw data."""
|
|
gr = self.block.create_group("neuronal-data", "open-epyhs.sampled")
|
|
raw_neo_data = self._load_neo_object("Data")
|
|
|
|
log.debug("Appending raw data")
|
|
nix_data_array = self.block.create_data_array(
|
|
name="data",
|
|
array_type="open-ephys.data.sampled",
|
|
dtype=nixio.DataType.Int16,
|
|
data=raw_neo_data,
|
|
unit="uV",
|
|
)
|
|
nix_data_array.append_sampled_dimension(
|
|
1 / raw_neo_data.sampling_rate.magnitude, label="time", unit="s"
|
|
)
|
|
gr.data_arrays.append(nix_data_array)
|
|
|
|
def close(self) -> None:
|
|
"""Close all nix files."""
|
|
self.nix_file.close()
|
|
self.relacs_nix_file.close()
|