from fishbook.backend.database import Cells, Datasets, CellDatasetMap, Subjects, SubjectProperties, SubjectDatasetMap, Stimuli, Repros
from .util import safe_get_val, results_check
import nixio as nix
import os
import numpy as np
from IPython import embed
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:
""" The Dataset class represents an entry in the "datasets" table in the database. In the relacs
context a Dataset represents all that has been recorded between pressing enter. For example
all that is stored in the folder 2020-01-01-aa.
Dataset contains several basic properties and allows access to associated cells and subjects.
"""
def __init__(self, dataset_id=None, tuple=None):
"""Constructor of a Dataset entity.
Args:
dataset_id (str, optional): the id. Defaults to None.
tuple (tuple, optional): The properties used to identify the entity in the database. Usually only for internal use. 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 did record the dataset.
Returns:
str: The experimenter
"""
return self.__tuple["experimenter"]
@property
def recording_date(self):
"""The Dataset's recording date.
Returns:
str: the recording date.
"""
return self.__tuple["recording_date"]
@property
def recording_duration(self):
"""Recording duration.
Returns:
float: recprding duration given in seconds.
"""
return self.__tuple["duration"]
@property
def quality(self):
"""The quality assessment as assigned by the experimenter.
Returns:
str: the quality
"""
return self.__tuple["quality"]
@property
def has_nix(self):
"""Whether or not a nix file was recorded.
Returns:
bool: true or false
"""
return self.__tuple["has_nix"]
@property
def comment(self):
"""The comment entered by the experimenter.
Returns:
str: the comment
"""
return self.__tuple["comment"]
@property
def data_source(self):
"""Where the data is stored (at least at the time of import).
Returns:
str: the path to the data.
"""
return self.__tuple["data_source"]
@property
def setup(self):
"""The recording setup.
Returns:
str: the setup
"""
return self.__tuple["setup"]
@property
def cells(self):
"""The list of cells that are associated with this dataset.
Returns:
list of fishbook.Cell: the cells.
"""
cell_list = (Cells * (CellDatasetMap & self.__tuple))
return [Cell(tuple=c) for c in cell_list]
@property
def subjects(self):
"""The subjects that are associated with this dataset.
Returns:
list of fishbook.Subject: list of subjects
"""
subject_list = (Subjects * (SubjectDatasetMap & self.__tuple))
return [Subject(tuple=s) for s in subject_list]
@property
def samplerate(self):
"""Get the samplerate of the data.
Returns:
float: the sample rate
"""
return self.__samplerate
@staticmethod
def datasetCount():
"""Returns the total number of datasets defined in the database.
Returns:
int: the count
"""
return len(Datasets())
@staticmethod
def find(min_duration=None, experimenter=None, quality=None, test=False):
"""Find dataset entries in the database. You may restrict the search by providing the following arguments.
Args:
min_duration (float, optional): minimum duration of the recording session, if not given, any length datasets will be returned. Defaults to None.
experimenter (str, optional): the name of the one who did the recording. The name does not need to be the full name. Defaults to None.
quality (str, optional): the quality assigned to the dataset during recording (e.g. good, fair, poor). Defaults to None.
test (bool, optional): defines whether this is a test run and thus whether or not the search results should be fetched, which can take a while. Defaults to False.
Returns:
list: list of Dataset object matching the search criteria, empty if test==True
int: Count of matching results
"""
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)
results = []
total = len(dataset_list)
if not test:
for i, d in enumerate(dataset_list):
progress(i+1, total, "fetching data")
results.append(Dataset(tuple=d))
return results, total
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()
def __str__(self):
str = "id: %s\n" % self.id
str += "recorded: %s \t by:%s\n" % (self.recording_date, self.experimenter)
str += "duration: %ss \t quality:%s\n" % (self.recording_duration, self.quality)
str += "comment: %s" % self.comment
return str
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()