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 from fishbook.backend.util import progress 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 %i matches" % total) 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: """Repro class represents an entry in the repro table. This is a run of a certain relacs "Research Protocol". """ 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=[], quality=None, test=False): """ Finds Repro runs in the database. When called without arguments, all RePro runs are found and returned. Search can be narrowed by providing further search hints. **Note:** Cell type, quality, and species are ignored, if cell_id is provided **Note:** If there are many results fetching and creating objects may take a while. Consider running as test. Args: name (str, optional): The RePro name, or a part of it. Defaults to None. cell_id (str, optional): The cell_id. If given type, quality and species are ignored. Defaults to None. cell_type (str, optional): type of cell. Defaults to None. species (str, optional): The species name, or parts of it. Defaults to None. settings (list of string, optional): List of repro settings e.g. ["contrast: 20", "am: false"]. An AND connection is assumed. Defaults to None. quality (str, optional): The quality assessment. Defaults to None. test (bool, optional): defines whether or not the database matches should be fetched from the database. Returns: list of fishbook.RePro: list of results or empty list if test == True int: number of matches """ repros = Repros & True if name: repros = repros & "repro_name like '%{0:s}%'".format(name) if len(settings) > 0: settings_pattern = " AND ".join(["settings like '%{0:s}%'".format(s) for s in settings]) repros = repros & settings_pattern 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 results = [] total = len(repros) if not test: for i, r in enumerate(repros): results.append(RePro(tuple=r)) progress(i+1, total, "fetching %i matches" % total) return results, total @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=[], test=False): """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 []]. test (bool, optional): whether or not this is a test run. When true, results will not be fetched from the database. Defaults to False. Returns: List: of matching Stimulus presentations, empty if test==True int: count of matches """ 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 len(settings) > 0: settings_pattern = " AND ".join(["settings like '%{0:s}%'".format(s) for s in settings]) stimuli = stimuli & settings_pattern results = [] total = len(stimuli) if not test: for i, s in enumerate(stimuli): results.append(Stimulus(tuple=s)) progress(i+1, total, "fetching %i matches" % total) return results, total 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, test=False): subjs = Subjects & True if species: subjs = (Subjects & "species like '%{0:s}%'".format(species)) results = [] total = len(subjs) if not test: for i, s in enumerate(subjs): results.append(Subject(tuple=s)) progress(i+1, total, "fetching %i matches" % total) return results, total @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()