743 lines
26 KiB
Python
743 lines
26 KiB
Python
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
|
|
import datetime as dt
|
|
import yaml
|
|
|
|
class Cell:
|
|
"""The Cell class represents a recorded cell. It is characterized by *id*, the cell *type*, the *firing_rate*, and the recording *location*.
|
|
"""
|
|
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):
|
|
"""The id of the cell. Unsually the same as the dataset name.
|
|
|
|
Returns:
|
|
str: the id
|
|
"""
|
|
return self.__tuple["cell_id"] if "cell_id" in self.__tuple.keys() else ""
|
|
|
|
@property
|
|
def type(self):
|
|
"""The cell type.
|
|
|
|
Returns:
|
|
str: the cell type
|
|
"""
|
|
return self.__tuple["cell_type"] if "cell_type" in self.__tuple.keys() else ""
|
|
|
|
@property
|
|
def firing_rate(self):
|
|
"""The firing rate of the neuron, if applicable, else 0.0.
|
|
|
|
Returns:
|
|
float: the firing rate.
|
|
"""
|
|
return self.__tuple["firing_rate"] if "firing_rate" in self.__tuple.keys() else 0.0
|
|
|
|
@property
|
|
def location(self):
|
|
"""The recording location which consists of the structure (e.g. brain), the region (e.g. ELL), the subregion (e.g. lateral segment),
|
|
the depth below surface, the lateral position and the transversal section if applicable. The latter two are given according to the
|
|
[Apteronotus brain atlas](http://nelson.beckman.illinois.edu/atlas.html).
|
|
|
|
Returns:
|
|
dict: the recording location
|
|
"""
|
|
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):
|
|
"""The subject in which this cell was recorded.
|
|
|
|
Returns:
|
|
fishbook.Subject: the subject
|
|
"""
|
|
return Subject(tuple=(Subjects & {"subject_id": self.__tuple["subject_id"]}).fetch(limit=1, as_dict=True)[0])
|
|
|
|
@property
|
|
def datasets(self):
|
|
"""All datasets that have been collected from this cell.
|
|
|
|
Returns:
|
|
list of fishbook.Dataset: the linked datasets
|
|
"""
|
|
return [Dataset(tuple=d) for d in (Datasets & (CellDatasetMap & {"cell_id": self.id})).fetch(as_dict=True)]
|
|
|
|
@property
|
|
def repro_runs(self):
|
|
"""All relacs RePros that have been run while recording this cell.
|
|
|
|
Returns:
|
|
list of fishbook.RePro: the repros
|
|
"""
|
|
repros = (Repros & "cell_id = '%s'" % self.id)
|
|
return [RePro(tuple=r) for r in repros]
|
|
|
|
@staticmethod
|
|
def unique_cell_types():
|
|
"""List of unique cell types stored in the database.
|
|
|
|
Returns:
|
|
np.array of str: the cell types
|
|
"""
|
|
return np.unique(Cells.fetch("cell_type"))
|
|
|
|
@staticmethod
|
|
def find(cell_type=None, species=None, quality="good", test=False):
|
|
"""Find cells that match with respect to cell type, subject species, and recording quality.
|
|
|
|
Args:
|
|
cell_type (str, optional): The desired cell type, case insensitive. Defaults to None.
|
|
species (str, optional): The species name, or parts of it. Defaults to None.
|
|
quality (str, optional): The desired qulality. Defaults to "good".
|
|
test (bool, optional): Whether or not this query is a test run. Defaults to False.
|
|
|
|
Returns:
|
|
list of fishbook.Cell: the matching cells, empty it test==True.
|
|
int : the number of matches
|
|
"""
|
|
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)
|
|
results = []
|
|
total = len(cs)
|
|
if not test:
|
|
for i, c in enumerate(cs):
|
|
progress(i+1, total, "fetching %i matches" % total)
|
|
results.append(Cell(tuple=c))
|
|
return results, total
|
|
|
|
def __str__(self):
|
|
str = ""
|
|
str += "Cell: %s \t type: %s\n" % (self.id, self.type)
|
|
str += "Baseline firing: %.3f\n" % self.firing_rate
|
|
str += "structure: %s \t region: %s\t subregion: %s\n" % (self.__tuple["structure"], self.__tuple["region"], self.__tuple["subregion"])
|
|
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 data_host(self):
|
|
"""Returns the fully qualified name of the host from which the data was imported. That is, where it
|
|
should be available.
|
|
|
|
Returns:
|
|
str: the fully qualified domain.
|
|
"""
|
|
return self.__tuple["data_host"]
|
|
|
|
@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 repro_runs(self):
|
|
"""The repros that have been run in this dataset.
|
|
|
|
Returns:
|
|
list of fishbook.RePro: list of repro runs
|
|
"""
|
|
repros = (Repros * Cells * (CellDatasetMap & self.__tuple))
|
|
return [RePro(tuple=r) for r in repros]
|
|
|
|
@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(dataset_id=None, min_duration=None, experimenter=None, quality=None, min_date=None, max_date=None, test=False):
|
|
"""Find dataset entries in the database. You may restrict the search by providing the following arguments. All restrictions are connected
|
|
with a logical AND.
|
|
|
|
Args:
|
|
dataset_id (str, optional): the id of the given Dataset, if unique, all other restrictions will be ignored. Defaults to None
|
|
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.
|
|
min_date (datetime, optional): the minimum recording date. Dates may be given as datetime objects or string of the format "2010.01.01" or "2010-01-01". Defaults to None.
|
|
max_date (datetime, optional): the maximum recording date. 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 dataset_id:
|
|
dataset_list = dataset_list & "dataset_id like '%{0:s}%'".format(dataset_id)
|
|
if len(dataset_list) == 1:
|
|
return [Dataset(tuple=dataset_list.fetch(as_dict=True)[0])], 1
|
|
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)
|
|
if min_date and isinstance(min_date, (dt.date, str)):
|
|
dataset_list = dataset_list & "recording_date >= '%s'" % min_date
|
|
if max_date and isinstance(min_date, (dt.date, str)):
|
|
dataset_list = dataset_list & "recording_date < '%s'" % max_date
|
|
|
|
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()
|
|
|
|
@property
|
|
def yaml(self):
|
|
settings = yaml.dump({"dataset id": self.id, "recording date": self.recording_date,
|
|
"duration": self.recording_duration, "comment": self.comment,
|
|
"experimenter": self.experimenter, "quality": self.quality,
|
|
"data_source": self.data_source, "host": self.data_host,
|
|
"setup": self.setup, "nixed": self.has_nix})
|
|
return settings
|
|
|
|
|
|
def __str__(self):
|
|
str = "dataset id: %s\n" % self.id
|
|
str += "recorded: %s \t by: %s\n" % (self.recording_date, self.experimenter)
|
|
str += "duration: %s min \t quality: %s\n" % (self.recording_duration, self.quality)
|
|
str += "location: %s\t host:%s\n" % (self.data_source, self.data_host)
|
|
str += "comment: %s" % self.comment
|
|
return str
|
|
|
|
def __repr__(self):
|
|
return self.__str__()
|
|
|
|
class RePro:
|
|
"""The 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', 'data_host', '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,
|
|
min_date=None, max_date=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.
|
|
min_date (datetime, optional): the minimum recording date. Dates may be given as datetime objects or string of the format "2010.01.01" or "2010-01-01". Defaults to None.
|
|
max_date (datetime, optional): the maximum recording date. 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 min_date and isinstance(min_date, (dt.date, str)):
|
|
cells = cells & "recording_date >= '%s'" % min_date
|
|
if max_date and isinstance(min_date, (dt.date, str)):
|
|
cells = cells & "recording_date < '%s'" % max_date
|
|
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()
|
|
|
|
def __str__(self):
|
|
str = "RePro id: %s\t repro name: %s\n" % (self.id, self.name)
|
|
str += "run: %i\t on cell: %s\n" %(self.run, self.cell_id)
|
|
str += "start time: %s\t duration: %s\n" % (self.start, self.duration)
|
|
return str
|
|
|
|
def __repr__(self):
|
|
return self.__str__()
|
|
|
|
@property
|
|
def to_dict(self):
|
|
r_settings = yaml.safe_load(self.settings.replace("\t", " "))
|
|
settings = {"repro id": self.id, "run": self.run, "cell": self.cell_id,
|
|
"name": self.name, "start": self.start, "duration": self.duration,
|
|
"settings":r_settings}
|
|
return settings
|
|
|
|
|
|
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):
|
|
embed() # FIXME
|
|
return Cells & ("cell_id = %s" % self.__tuple["cell_id"])
|
|
|
|
@property
|
|
def repro(self):
|
|
embed() # FIXME
|
|
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
|
|
|
|
def __repr__(self):
|
|
return self.__str__()
|
|
|
|
@property
|
|
def to_dict(self):
|
|
s_settings = yaml.safe_load(self.settings.replace("\t", " "))
|
|
settings = {"id": self.id, "run": self.run, "stimulus name" : self.name,
|
|
"stimulus index": self.index, "duration": self.duration, "start time": self.start_time,
|
|
"name": self.name, "settings": s_settings}
|
|
return settings
|
|
|
|
|
|
class Subject:
|
|
"""Representation of the recorded subject's properties.
|
|
A subject is defined by the its *id*, and its *species*.
|
|
|
|
The
|
|
"""
|
|
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):
|
|
"""Find all subjects matching the given species.
|
|
|
|
Args:
|
|
species ([type], optional): [description]. Defaults to None.
|
|
test (bool, optional): [description]. Defaults to False.
|
|
|
|
Returns:
|
|
[type]: [description]
|
|
"""
|
|
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():
|
|
"""Unique list of species names.
|
|
|
|
Returns:
|
|
np.array: unique names
|
|
"""
|
|
all_species = (Subjects & True).fetch("species")
|
|
return np.unique(all_species)
|
|
|
|
def __str__(self):
|
|
str = "Subject: %s\n" % self.id
|
|
str += "Species: %s\n" % self.species
|
|
return str
|
|
|
|
|
|
if __name__ == "__main__":
|
|
from IPython import embed
|
|
cell = Cell("2010-04-16-ak")
|
|
embed()
|