forked from jgrewe/fishbook
452 lines
14 KiB
Python
452 lines
14 KiB
Python
from fishbook.database.database import Cells, Datasets, CellDatasetMap, Subjects, SubjectProperties, SubjectDatasetMap, Stimuli, Repros
|
|
import nixio as nix
|
|
import os
|
|
import numpy as np
|
|
from IPython import embed
|
|
|
|
|
|
def _safe_get_val(dictionary:dict, key, default=None):
|
|
return dictionary[key] if key in dictionary.keys() else default
|
|
|
|
|
|
def _results_check(results, id, text="ID"):
|
|
if len(results) == 0:
|
|
raise ValueError("%s %s does not exist!" % (text, id))
|
|
elif len(results) > 1:
|
|
raise ValueError("%s %s is not unique!" % (text, id))
|
|
|
|
|
|
class Cell:
|
|
def __init__(self, cell_id=None, tuple=None):
|
|
if tuple:
|
|
self.__tuple = tuple
|
|
elif cell_id:
|
|
pattern = "cell_id like '{0:s}'".format(cell_id)
|
|
cells = (Cells & pattern)
|
|
_results_check(cells, cell_id, "Cell ID")
|
|
self.__tuple = cells.fetch(as_dict=True)[0]
|
|
else:
|
|
print("Empty Cell, not linked to any database entry!")
|
|
|
|
@property
|
|
def id(self):
|
|
return self.__tuple["cell_id"] if "cell_id" in self.__tuple.keys() else ""
|
|
|
|
@property
|
|
def type(self):
|
|
return self.__tuple["cell_type"] if "cell_type" in self.__tuple.keys() else ""
|
|
|
|
@property
|
|
def firing_rate(self):
|
|
return self.__tuple["firing_rate"] if "firing_rate" in self.__tuple.keys() else 0.0
|
|
|
|
@property
|
|
def location(self):
|
|
keys = ["structure", "region", "subregion", "depth", "lateral_pos", "transversal_section"]
|
|
loc = {}
|
|
for k in keys:
|
|
if k in self.__tuple.keys():
|
|
loc[k] = self.__tuple[k]
|
|
else:
|
|
loc[k] = ""
|
|
return loc
|
|
|
|
@property
|
|
def subject(self):
|
|
return Subject(tuple=(Subjects & {"subject_id": self.__tuple["subject_id"]}).fetch(limit=1, as_dict=True)[0])
|
|
|
|
@property
|
|
def datasets(self):
|
|
return [Dataset(tuple=d) for d in (Datasets & (CellDatasetMap & {"cell_id": self.id})).fetch(as_dict=True)]
|
|
|
|
@property
|
|
def repro_runs(self):
|
|
repros = (Repros & "cell_id = '%s'" % self.id)
|
|
return [RePro(tuple=r) for r in repros]
|
|
|
|
@staticmethod
|
|
def unique_cell_types():
|
|
return np.unique(Cells.fetch("cell_type"))
|
|
|
|
@staticmethod
|
|
def find(cell_type=None, species=None, quality="good"):
|
|
cs = Cells * CellDatasetMap * Datasets * Subjects
|
|
if cell_type:
|
|
cs = cs & "cell_type like '{0:s}'".format(cell_type)
|
|
if species:
|
|
cs = cs & "species like '%{0:s}%'".format(species)
|
|
if quality:
|
|
cs = cs & "quality like '{0:s}'".format(quality)
|
|
return [Cell(tuple=c) for c in cs]
|
|
|
|
def __str__(self):
|
|
str = ""
|
|
str += "Cell: %s \t type: %s\n"%(self.id, self.type)
|
|
return str
|
|
|
|
@property
|
|
def _tuple(self):
|
|
return self.__tuple.copy()
|
|
|
|
|
|
class Dataset:
|
|
"""[summary]
|
|
"""
|
|
def __init__(self, dataset_id=None, tuple=None):
|
|
"""[summary]
|
|
|
|
Args:
|
|
dataset_id ([type], optional): [description]. Defaults to None.
|
|
tuple ([type], optional): [description]. Defaults to None.
|
|
"""
|
|
self.__samplerate = 0.0
|
|
if tuple:
|
|
self.__tuple = tuple
|
|
elif dataset_id:
|
|
pattern = "dataset_id like '{0:s}'".format(dataset_id)
|
|
dsets = (Datasets & pattern)
|
|
_results_check(dsets, dataset_id, "Dataset ID")
|
|
self.__tuple = dsets.fetch(limit=1, as_dict=True)[0]
|
|
else:
|
|
print("Empty dataset, not linked to any database entry!")
|
|
if len(self.__tuple.keys()) > 0:
|
|
self.__find_samplerate()
|
|
|
|
@property
|
|
def id(self):
|
|
"""The id of the dataset.
|
|
|
|
Returns:
|
|
str: the id
|
|
"""
|
|
return self.__tuple["dataset_id"]
|
|
|
|
@property
|
|
def experimenter(self):
|
|
"""The person who recording the dataset.
|
|
|
|
Returns:
|
|
str: The experimenter
|
|
"""
|
|
return self.__tuple["experimenter"]
|
|
|
|
@property
|
|
def recording_date(self):
|
|
return self.__tuple["recording_date"]
|
|
|
|
@property
|
|
def recording_duration(self):
|
|
return self.__tuple["duration"]
|
|
|
|
@property
|
|
def quality(self):
|
|
return self.__tuple["quality"]
|
|
|
|
@property
|
|
def has_nix(self):
|
|
return self.__tuple["has_nix"]
|
|
|
|
@property
|
|
def comment(self):
|
|
return self.__tuple["comment"]
|
|
|
|
@property
|
|
def data_source(self):
|
|
return self.__tuple["data_source"]
|
|
|
|
@property
|
|
def setup(self):
|
|
return self.__tuple["setup"]
|
|
|
|
@property
|
|
def cells(self):
|
|
cell_list = (Cells * (CellDatasetMap & self.__tuple))
|
|
return [Cell(tuple=c) for c in cell_list]
|
|
|
|
@property
|
|
def subjects(self):
|
|
subject_list = (Subjects * (SubjectDatasetMap & self.__tuple))
|
|
return [Subject(tuple=s) for s in subject_list]
|
|
|
|
@property
|
|
def samplerate(self):
|
|
return self.__samplerate
|
|
|
|
@staticmethod
|
|
def find(min_duration=None, experimenter=None, quality=None):
|
|
dataset_list = Datasets
|
|
if min_duration:
|
|
dataset_list = dataset_list & "duration > %.2f" % min_duration
|
|
if experimenter:
|
|
dataset_list = dataset_list & "experimenter like '%{0:s}%'".format(experimenter)
|
|
if quality:
|
|
dataset_list = dataset_list & "quality like '{0:s}'".format(quality)
|
|
return [Dataset(tuple=d) for d in dataset_list]
|
|
|
|
def __find_samplerate(self, trace_name="V-1"):
|
|
if self.has_nix and os.path.exists(os.path.join(self.data_source, self.id, '.nix')):
|
|
f = nix.File.open(os.path.join(self.data_source, self.id, '.nix'), nix.FileMode.ReadOnly)
|
|
b = f.blocks[0]
|
|
if trace_name in b.data_arrays:
|
|
trace = b.data_arrays[trace_name]
|
|
if trace.dimensions[0].dimension_type == nix.DimensionType.Sample:
|
|
self.__samplerate = 1./trace.dimensions[0].sampling_interval
|
|
else:
|
|
print("Requested trace %s has no sampled dimension!" % trace_name)
|
|
else:
|
|
print("Requested trace %s was not found!" % trace_name)
|
|
f.close()
|
|
else:
|
|
stimulus_file = os.path.join(self.data_source , 'stimuli.dat')
|
|
if not os.path.exists(stimulus_file):
|
|
return
|
|
lines = None
|
|
with open(stimulus_file, 'r') as f:
|
|
lines = f.readlines()
|
|
for l in lines:
|
|
if "sample interval1" in l:
|
|
si = l.strip().split(":")[-1][:-2]
|
|
break
|
|
self.__samplerate = 1000. / float(si)
|
|
|
|
@property
|
|
def _tuple(self):
|
|
return self.__tuple.copy()
|
|
|
|
|
|
class RePro:
|
|
def __init__(self, repro_id=None, tuple=None):
|
|
if tuple:
|
|
self.__tuple = tuple
|
|
elif repro_id:
|
|
repros = (RePro & "repro_id like '{0:s}'".format(repro_id))
|
|
_results_check(repros, repro_id, "RePro ID")
|
|
self.__tuple = repros.fetch(limit=1, as_dict=True)[0]
|
|
else:
|
|
self.__tuple = {}
|
|
print("Empty RePro, not linked to any database entry!")
|
|
|
|
@property
|
|
def id(self):
|
|
return _safe_get_val(self.__tuple, "repro_id", "")
|
|
|
|
@property
|
|
def run(self):
|
|
return _safe_get_val(self.__tuple, "run", -1)
|
|
|
|
@property
|
|
def cell_id(self):
|
|
return _safe_get_val(self.__tuple, "cell_id", "")
|
|
|
|
@property
|
|
def cell(self):
|
|
return Cell(self.cell_id)
|
|
|
|
@property
|
|
def dataset(self):
|
|
datasets = (Cells & "cell_id = '%s'" % self.cell_id) * CellDatasetMap * Datasets
|
|
d = datasets.proj('dataset_id', 'data_source', 'experimenter', 'setup', 'recording_date',
|
|
'quality', 'comment', 'duration', 'has_nix').fetch(limit=1, as_dict=True)[0]
|
|
del d["cell_id"]
|
|
return Dataset(tuple=d)
|
|
|
|
@property
|
|
def name(self):
|
|
return _safe_get_val(self.__tuple, "repro_name", "")
|
|
|
|
@property
|
|
def settings(self):
|
|
return _safe_get_val(self.__tuple, "settings", "")
|
|
|
|
@property
|
|
def start(self):
|
|
return _safe_get_val(self.__tuple, "start", 0.0)
|
|
|
|
@property
|
|
def duration(self):
|
|
return _safe_get_val(self.__tuple, "duration", 0.0)
|
|
|
|
@property
|
|
def stimuli(self):
|
|
stims = Stimuli & "repro_id = '%s'" % self.id & "cell_id = '%s'" % self.cell_id
|
|
return [Stimulus(tuple=s) for s in stims]
|
|
|
|
@staticmethod
|
|
def find(name=None, cell_id=None, cell_type=None, species=None, settings=None, quality=None):
|
|
"""
|
|
Cell type, quality, and species are ignored, if cell_id is provided
|
|
:param repro_name:
|
|
:param cell_id:
|
|
:param cell_type:
|
|
:param species:
|
|
:param settings:
|
|
:param quality:
|
|
:return:
|
|
"""
|
|
repros = Repros & True
|
|
if name:
|
|
repros = repros & "repro_name like '%{0:s}%'".format(name)
|
|
if settings:
|
|
repros = repros & "settings like '%{0:s}%'".format(settings)
|
|
if cell_id:
|
|
repros = repros & "cell_id = '{0:s}'".format(cell_id)
|
|
if not cell_id and (cell_type or species or quality):
|
|
cells = (Cells * CellDatasetMap * Datasets) * Subjects
|
|
if cell_type:
|
|
cells = cells & "cell_type like '%{0:s}%'".format(cell_type)
|
|
if species:
|
|
cells = cells & "species like '%{0:s}%'".format(species)
|
|
if quality:
|
|
cells = cells & "quality like '{0:s}'".format(quality)
|
|
p = cells.proj("quality", "experimenter")
|
|
repros = repros & p
|
|
return [RePro(tuple=r) for r in repros]
|
|
|
|
@property
|
|
def _tuple(self):
|
|
return self.__tuple.copy()
|
|
|
|
|
|
class Stimulus:
|
|
"""The stimulus class represents a Stimulus that was presented. A Stimulus has several properties
|
|
such as the start time, the duarion
|
|
"""
|
|
def __init__(self, stimulus_id=None, tuple=None):
|
|
if tuple:
|
|
self.__tuple = tuple
|
|
elif stimulus_id:
|
|
stimuli = Stimuli & "stimulus_id = '%s'" %stimulus_id
|
|
_results_check(stimuli, stimulus_id, "Stimulus ID")
|
|
self.__tuple = stimuli.fetch(limit=1, as_dict=True)[0]
|
|
else:
|
|
print("Empty RePro, not linked to any database entry!")
|
|
|
|
def __str__(self):
|
|
str = "Stimulus %s: " % _safe_get_val(self.__tuple, "stimulus_id", "")
|
|
str += "\nStart time/index: %0.4f/%i, duration: %.3f" % (_safe_get_val(self.__tuple, "start_time", 0.0),
|
|
_safe_get_val(self.__tuple, "start_index", -1),
|
|
_safe_get_val(self.__tuple, "duration", 0.0))
|
|
return str
|
|
|
|
@property
|
|
def id(self):
|
|
return self.__tuple["stimulus_id"]
|
|
|
|
@property
|
|
def name(self):
|
|
return self.__tuple["stimulus_name"]
|
|
|
|
@property
|
|
def cell(self):
|
|
return Cells & "cell_id = %s" % self.__tuple["cell_id"]
|
|
|
|
@property
|
|
def repro(self):
|
|
return Repros & "repro_id = %s" % self.__tuple["repro_id"]
|
|
|
|
@property
|
|
def start_time(self):
|
|
return self.__tuple["start_time"]
|
|
|
|
@property
|
|
def start_index(self):
|
|
return self.__tuple["start_index"]
|
|
|
|
@property
|
|
def duration(self):
|
|
return self.__tuple["duration"]
|
|
|
|
@property
|
|
def multi_tag_id(self):
|
|
return self.__tuple["mtag_id"]
|
|
|
|
@property
|
|
def run(self):
|
|
return self.__tuple["run"]
|
|
|
|
@property
|
|
def index(self):
|
|
return self.__tuple["stimulus_index"]
|
|
|
|
@property
|
|
def settings(self):
|
|
return _safe_get_val(self.__tuple, "settings", "")
|
|
|
|
@property
|
|
def _tuple(self):
|
|
return self.__tuple.copy()
|
|
|
|
@staticmethod
|
|
def find(cell_id=None, repro_id=None, settings=None):
|
|
"""Find stimulus presentations that match in certain properties.
|
|
Examples:
|
|
|
|
Args:
|
|
cell_id (str, optional): [description]. Defaults to None.
|
|
repro_id (str, optional): [description]. Defaults to None.
|
|
settings (str, optional): [description]. Defaults to None.
|
|
|
|
Returns:
|
|
List: of matching Stimulus presentations
|
|
"""
|
|
stimuli = Stimuli & True
|
|
if cell_id:
|
|
stimuli = stimuli & "cell_id = '{0:s}'".format(cell_id)
|
|
if repro_id:
|
|
stimuli = stimuli & "repro_id = '{0:s}'".format(repro_id)
|
|
if settings:
|
|
stimuli = stimuli & "settings like '%{0:s}%'".format(settings)
|
|
return [Stimulus(tuple=s) for s in stimuli]
|
|
|
|
|
|
class Subject:
|
|
def __init__(self, subject_id=None, tuple=None):
|
|
if tuple:
|
|
self.__tuple = tuple
|
|
elif subject_id:
|
|
subjects = Subjects & "subject_id like '{0:s}'".format(subject_id)
|
|
_results_check(subjects, subject_id, "Subject ID")
|
|
self.__tuple = subjects.fetch()[0]
|
|
else:
|
|
self.__tuple = {}
|
|
print("Empty Subject, not linked to any database entry!")
|
|
|
|
@property
|
|
def id(self):
|
|
return self.__tuple["subject_id"]
|
|
|
|
@property
|
|
def species(self):
|
|
return self.__tuple["species"]
|
|
|
|
@property
|
|
def cells(self):
|
|
cs = Cells & self.__tuple
|
|
return [Cell(tuple=c) for c in cs]
|
|
|
|
@property
|
|
def properties(self):
|
|
return (SubjectProperties & self.__tuple).fetch(as_dict=True)
|
|
|
|
@property
|
|
def _tuple(self):
|
|
return self.__tuple.copy()
|
|
|
|
@staticmethod
|
|
def find(species=None):
|
|
subjs = Subjects & True
|
|
if species:
|
|
subjs = (Subjects & "species like '%{0:s}%'".format(species))
|
|
return [Subject(tuple=s) for s in subjs]
|
|
|
|
@staticmethod
|
|
def unique_species():
|
|
all_species = (Subjects & True).fetch("species")
|
|
return np.unique(all_species)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
from IPython import embed
|
|
cell = Cell("2010-04-16-ak")
|
|
embed()
|