Source code for allensdk.brain_observatory.vbn_2022.input_json_writer.utils
from typing import List, Optional, Dict, Union
import pandas as pd
import numpy as np
import numbers
from allensdk.internal.api.queries.wkf_lims_queries import (
wkf_path_from_attachable)
from allensdk.internal.api.queries.equipment_lims_queries import (
experiment_configs_from_equipment_id_and_type)
from allensdk.internal.api import PostgresQueryMixin
from allensdk import OneResultExpectedError
from allensdk.internal.api.queries.utils import (
build_in_list_selector_query)
from allensdk.brain_observatory.behavior.data_objects.\
metadata.subject_metadata.reporter_line import ReporterLine
from allensdk.brain_observatory.behavior.data_objects.\
metadata.subject_metadata.driver_line import DriverLine
from allensdk.brain_observatory.vbn_2022.metadata_writer.lims_queries import (
_ecephys_summary_table_from_ecephys_session_id_list,
probes_table_from_ecephys_session_id_list,
channels_table_from_ecephys_session_id_list,
units_table_from_ecephys_session_id_list,
get_list_of_bad_probe_ids)
from allensdk.brain_observatory.vbn_2022.metadata_writer.\
dataframe_manipulations import (
_add_age_in_days,
_patch_date_and_stage_from_pickle_file)
from allensdk.core.auth_config import (
LIMS_DB_CREDENTIAL_MAP)
from allensdk.internal.api import db_connection_creator
[docs]
class NwbConfigErrorLog(object):
"""
This is a class meant to keep track of all of the non-fatal
data irregularities encountered during input json generation
for the VBN NWB writer.
"""
def __init__(self):
self._messages = dict()
[docs]
def log(self,
ecephys_session_id: Union[int, str],
msg: str) -> None:
"""
Log an irregularity associated with a specifiec
ecephys session
Parameters
----------
ecephys_session_id: Union[int, str]
Will get cast to int before actual logging
msg: str
The specific message that you want attached to
that ecephys_session_id
"""
ecephys_session_id = int(ecephys_session_id)
if ecephys_session_id not in self._messages:
self._messages[ecephys_session_id] = []
self._messages[ecephys_session_id].append(msg)
[docs]
def write(self) -> str:
"""
Returns a string summarizing all of the logged
message. The string will look like
ecephys_session: 1111
first message associated with ecephys_session 1111
second message assocated with ecephys_session 1111
ecephys_session: 2222
first message associated with ecephys_session 2222
...
"""
k_list = list(self._messages.keys())
k_list.sort()
msg = ""
for k in k_list:
msg += f"ecephys_session: {k}\n"
for m in self._messages[k]:
msg += f" {m}\n"
return msg
[docs]
def vbn_nwb_config_from_ecephys_session_id_list(
ecephys_session_id_list: List[int],
probes_to_skip: Optional[List[dict]]
) -> dict:
"""
Return a list of dicts. Each dict the specification for
an NWB writer job, suitable for serialization with
json.dumps
Parameters
----------
ecephys_session_id_list: List[int]
The ecephys_session_ids for which you want to create
NWB writer specifications
probes_to_skip: List[dict]
A list of dicts, each specifying a probe to ignore
when generating the NWB writer specifications.
Each dict should look like
{
"session": 12345 # an ecephys_session_id
"probe": "probeB" # the probe's name
}
Returns
-------
A dict
{
"specfications": the list of dicts representing
the NWB writer specifications
"log": a string with a summary of all non-fatal
irregularities encountered in the data
}
"""
error_log = NwbConfigErrorLog()
lims_connection = db_connection_creator(
fallback_credentials=LIMS_DB_CREDENTIAL_MAP)
# convert probes_to_skip into a list of ecephys_probe_ids
if probes_to_skip is not None:
probe_ids_to_skip = get_list_of_bad_probe_ids(
lims_connection=lims_connection,
probes_to_skip=probes_to_skip)
else:
probe_ids_to_skip = None
# get a list of basic session configuresions (i.e. the
# data for each session excluding the lists of probes,
# channels, and units)
session_list = session_input_from_ecephys_session_id_list(
ecephys_session_id_list=ecephys_session_id_list,
lims_connection=lims_connection,
error_log=error_log)
# iterate over each session, adding the probes, channels,
# and units as appropriate
for session in session_list:
session_id = session['ecephys_session_id']
probe_list = probe_input_from_ecephys_session_id(
ecephys_session_id=session_id,
probe_ids_to_skip=probe_ids_to_skip,
lims_connection=lims_connection,
error_log=error_log
)
session['probes'] = probe_list
channel_input = channel_input_from_ecephys_session_id(
ecephys_session_id=session_id,
probe_ids_to_skip=probe_ids_to_skip,
lims_connection=lims_connection,
error_log=error_log)
# bad_probe_list keeps track of any probes that did not have
# channels attached to it; these probes will be excluded
# from the final configuration, and a message will be logged
bad_probe_list = []
for idx, probe in enumerate(session['probes']):
probe_id = probe['id']
if probe_id in channel_input:
channels = channel_input[probe_id]
probe['channels'] = channels
else:
bad_probe_list.append(idx)
msg = (f"could not find channels for probe {probe_id}; "
"not listing in the input.json")
error_log.log(ecephys_session_id=session_id,
msg=msg)
bad_probe_list.reverse()
for idx in bad_probe_list:
this_probe = session['probes'].pop(idx)
assert 'channels' not in this_probe
unit_input = unit_input_from_ecephys_session_id(
ecephys_session_id=session_id,
probe_ids_to_skip=probe_ids_to_skip,
lims_connection=lims_connection,
error_log=error_log)
for probe in session['probes']:
probe_id = probe['id']
if probe_id in unit_input:
units = unit_input[probe_id]
probe['units'] = units
else:
msg = f"could not find units for probe {probe_id}"
error_log.log(ecephys_session_id=session_id,
msg=msg)
return {'sessions': session_list,
'log': error_log.write()}
[docs]
def session_input_from_ecephys_session_id_list(
ecephys_session_id_list: List[int],
lims_connection: PostgresQueryMixin,
error_log: NwbConfigErrorLog) -> List[dict]:
"""
Return a list of dicts, each dict representing the configuration
data necessary for writing an NWB file for a session, excluding
the lists of probes, channels, and units associated with that
session.
Parameters
----------
ecephys_session_id_list: List[int]
List of ecephys_session_ids for which we are generating
the configurations
lims_connection: PostgresQueryMixin
error_log: NwbConfigErrorLog
An object for keeping track of all of the non-fatal
irregularities encountered in the data.
Returns
-------
result: List[dict]
Each dict represents a single session's configuration
"""
session_table = _ecephys_summary_table_from_ecephys_session_id_list(
lims_connection=lims_connection,
ecephys_session_id_list=ecephys_session_id_list,
failed_ecephys_session_id_list=None)
# get date_of_acquisition from the pickle file by nulling out the
# dates of acqusition from any sessions with behavior_session_ids,
# then filling the values back in from the pickle file.
session_table.loc[
np.logical_not(session_table.behavior_session_id.isna()),
'date_of_acquisition'] = None
session_table = _patch_date_and_stage_from_pickle_file(
lims_connection=lims_connection,
behavior_df=session_table,
flag_columns=['date_of_acquisition'],
columns_to_patch=['date_of_acquisition'])
# clip fractions of a second off the date of acquisition
# (DateTime data object will fail deserialization if you do not)
session_table.date_of_acquisition = \
session_table.date_of_acquisition.dt.floor('S')
session_table = _add_age_in_days(
df=session_table,
index_column='ecephys_session_id')
session_table.age_in_days = session_table.age_in_days.apply(
lambda x: f'P{int(x):d}')
# apply naming conventions from the NWB writer's schema
session_table.rename(
columns={'equipment_name': 'rig_name',
'mouse_id': 'external_specimen_name',
'genotype': 'full_genotype',
'age_in_days': 'age'},
inplace=True)
session_table.external_specimen_name = \
session_table.external_specimen_name.astype(int)
session_table.drop(
labels=['session_type', 'project_code'],
axis='columns',
inplace=True)
session_table = session_table.set_index(
'ecephys_session_id')
session_table = session_table.to_dict(orient='index')
# get lookup tables mapping ecephys_session_id to the
# ecephys_analysis_run_ids for the optotagging and
# stimulus table files
optotagging_run_lookup = _analysis_run_from_session_id(
lims_connection=lims_connection,
ecephys_session_id_list=ecephys_session_id_list,
strategy_class='EcephysOptotaggingTableStrategy')
stim_table_run_lookup = _analysis_run_from_session_id(
lims_connection=lims_connection,
ecephys_session_id_list=ecephys_session_id_list,
strategy_class='VbnCreateStimTableStrategy')
# A list of tuples associating fields in the final specification
# returned by this method with the names of files in the
# well_known_file_types table on LIMS. The zeroth element in each
# tuple is the field in the returned specification; the first
# element is the associated well_known_file_types.name
input_from_wkf_session = [
('behavior_stimulus_file', 'StimulusPickle'),
('mapping_stimulus_file', 'MappingPickle'),
('replay_stimulus_file', 'EcephysReplayStimulus'),
('raw_eye_tracking_video_meta_data',
'RawEyeTrackingVideoMetadata'),
('eye_dlc_file', 'EyeDlcOutputFile'),
('face_dlc_file', 'FaceDlcOutputFile'),
('side_dlc_file', 'SideDlcOutputFile'),
('eye_tracking_filepath', 'EyeTracking Ellipses'),
('sync_file', 'EcephysRigSync')]
# cast to a SQL-safe string
wkf_types_to_query_session = [f"'{el[1]}'"
for el in input_from_wkf_session]
result = []
for session_id in ecephys_session_id_list:
session_id = int(session_id)
if session_id not in session_table:
error_log.log(ecephys_session_id=session_id,
msg="No session data was found at all; skipping")
continue
data = session_table[session_id]
data['ecephys_session_id'] = session_id
# lookup all of the well known files we need for this
# specification
wkf_path_lookup = wkf_path_from_attachable(
lims_connection=lims_connection,
wkf_type_name=wkf_types_to_query_session,
attachable_type='EcephysSession',
attachable_id=session_id)
for key_pair in input_from_wkf_session:
this_path = wkf_path_lookup.get(key_pair[1], None)
if this_path is None:
msg = (f"Could not find {key_pair[1]} "
f"for ecephys_session {session_id}")
error_log.log(ecephys_session_id=session_id,
msg=msg)
data[key_pair[0]] = this_path
# get stimulus_table
stim_path_lookup = wkf_path_from_attachable(
lims_connection=lims_connection,
wkf_type_name=["'EcephysStimulusTable'", ],
attachable_type="EcephysAnalysisRun",
attachable_id=stim_table_run_lookup[session_id])
data['stim_table_file'] = stim_path_lookup['EcephysStimulusTable']
# get optotagging_table
optotagging_path_lookup = wkf_path_from_attachable(
lims_connection=lims_connection,
wkf_type_name=["'EcephysOptotaggingTable'", ],
attachable_type="EcephysAnalysisRun",
attachable_id=optotagging_run_lookup[session_id])
data['optotagging_table_path'] = optotagging_path_lookup[
"EcephysOptotaggingTable"]
driver_line = DriverLine.from_lims(
lims_db=lims_connection,
behavior_session_id=data['behavior_session_id'],
allow_none=True).value
if driver_line is not None:
if isinstance(driver_line, list):
data['driver_line'] = driver_line
else:
data['driver_line'] = [driver_line, ]
else:
data['driver_line'] = []
reporter_line = ReporterLine.from_lims(
lims_db=lims_connection,
behavior_session_id=data['behavior_session_id'],
allow_none=True).value
if reporter_line is not None:
if isinstance(reporter_line, list):
data['reporter_line'] = reporter_line
else:
data['reporter_line'] = [reporter_line, ]
else:
data['reporter_line'] = []
eye_geometry = eye_tracking_geometry_from_equipment_id(
equipment_id=data.pop('equipment_id'),
date_of_acquisition=data['date_of_acquisition'],
lims_connection=lims_connection)
eye_geometry['equipment'] = data['rig_name']
data['eye_tracking_rig_geometry'] = eye_geometry
for k in ('date_of_acquisition',
'date_of_birth'):
data[k] = str(data[k])
result.append(data)
return result
def _get_probe_analysis_run_from_probe_id(
lims_connection: PostgresQueryMixin,
probe_id: int,
lims_strategy: str
):
query = f'''
SELECT earp.id
FROM ecephys_analysis_run_probes earp
JOIN ecephys_analysis_runs ear on ear.id = earp.ecephys_analysis_run_id
WHERE earp.ecephys_probe_id = {probe_id} and
job_strategy_class = '{lims_strategy}' and
ear.current
'''
res = lims_connection.select_one(query)
if not res:
raise OneResultExpectedError(
f'Expected to find one analysis probe run for probe '
f'{probe_id}')
return res['id']
def _get_probe_lfp_meta(
lims_connection: PostgresQueryMixin,
probe_id: int
):
"""Gets filepaths for files needed to build LFP data
Parameters
----------
lims_connection
probe_id
"""
lfp_subsampling_run_well_known_files = [
'EcephysSubsampledLfpContinuous',
'EcephysSubsampledLfpTimestamps',
'EcephysSubsampledChannelStates'
]
current_source_density_well_known_files = [
'EcephysCurrentSourceDensity'
]
probe_lfp_subsampling_run_id = _get_probe_analysis_run_from_probe_id(
lims_connection=lims_connection,
probe_id=probe_id,
lims_strategy='EcephysLfpSubsamplingStrategy'
)
probe_current_source_density_run_id = \
_get_probe_analysis_run_from_probe_id(
lims_connection=lims_connection,
probe_id=probe_id,
lims_strategy='EcephysCurrentSourceDensityStrategy'
)
probe_lfp_well_known_files = wkf_path_from_attachable(
lims_connection=lims_connection,
wkf_type_name=lfp_subsampling_run_well_known_files,
attachable_type='EcephysAnalysisRunProbe',
attachable_id=probe_lfp_subsampling_run_id)
probe_csd_well_known_files = wkf_path_from_attachable(
lims_connection=lims_connection,
wkf_type_name=current_source_density_well_known_files,
attachable_type='EcephysAnalysisRunProbe',
attachable_id=probe_current_source_density_run_id)
lfp = {
'input_data_path':
probe_lfp_well_known_files.get(
'EcephysSubsampledLfpContinuous'),
'input_timestamps_path':
probe_lfp_well_known_files.get(
'EcephysSubsampledLfpTimestamps'),
'input_channels_path':
probe_lfp_well_known_files.get(
'EcephysSubsampledChannelStates'),
'csd_path': probe_csd_well_known_files.get(
'EcephysCurrentSourceDensity')
}
return lfp
[docs]
def probe_input_from_ecephys_session_id(
ecephys_session_id: int,
probe_ids_to_skip: Optional[List[int]],
lims_connection: PostgresQueryMixin,
error_log: NwbConfigErrorLog,
) -> List[dict]:
"""
Get the list of probe specifications, excluding the lists
of channels and units, for a given ecephys_session_id
Parameters
----------
ecephys_session_id: int
probe_ids_to_skip: Optional[List[int]]:
List of probes not to return because we already know they
are "bad" in some way.
lims_connection: PostgresQueryMixin
error_log: NwbConfigErrorLog
object to store all of the non-fatal irregularities
encountered in the data
Returns
-------
probe_list: List[dict]
Each dict represents the specifications of a probe
that needs to be written to the input.json.
These dicts will not include the channels or
units data. Those are added at a later step in
processing.
"""
probes_table = probes_table_from_ecephys_session_id_list(
lims_connection=lims_connection,
ecephys_session_id_list=[ecephys_session_id, ],
probe_ids_to_skip=probe_ids_to_skip)
probes_table = probes_table.set_index('ecephys_probe_id')
probes_table.drop(
labels=['ecephys_session_id',
'phase',
'unit_count',
'channel_count',
'structure_acronyms'],
axis='columns',
inplace=True)
probes_table = probes_table.to_dict(orient='index')
# A list of tuples associating fields in the final specification
# returned by this method with the names of files in the
# well_known_file_types table on LIMS. The zeroth element in each
# tuple is the field in the returned specification; the first
# element is the associated well_known_file_types.name
input_from_wkf_probe = [
('inverse_whitening_matrix_path', 'EcephysSortedWhiteningMatInv'),
('mean_waveforms_path', 'EcephysSortedMeanWaveforms'),
('spike_amplitudes_path', 'EcephysSortedAmplitudes'),
('spike_clusters_file', 'EcephysSortedSpikeClusters'),
('spike_templates_path', 'EcephysSortedSpikeTemplates'),
('templates_path', 'EcephysSortedTemplates')]
wkf_to_query = [f"'{el[1]}'"
for el in input_from_wkf_probe]
probe_list = []
probe_id_list = list(probes_table.keys())
probe_id_list.sort()
for probe_id in probe_id_list:
data = probes_table[probe_id]
has_lfp = data.pop('has_lfp_data')
data['id'] = probe_id
wkf_path_lookup = wkf_path_from_attachable(
lims_connection=lims_connection,
wkf_type_name=wkf_to_query,
attachable_type='EcephysProbe',
attachable_id=probe_id)
for key_pair in input_from_wkf_probe:
data[key_pair[0]] = wkf_path_lookup.get(key_pair[1], None)
if has_lfp:
lfp_meta = _get_probe_lfp_meta(
lims_connection=lims_connection,
probe_id=probe_id
)
data['csd_path'] = lfp_meta.pop('csd_path')
data['lfp'] = lfp_meta
else:
data['lfp'] = None
probe_list.append(_nan_to_none(data))
probe_list = _add_spike_times_path(
probe_list=probe_list,
ecephys_session_id=ecephys_session_id,
lims_connection=lims_connection,
error_log=error_log)
return probe_list
[docs]
def channel_input_from_ecephys_session_id(
ecephys_session_id: int,
probe_ids_to_skip: Optional[List[int]],
lims_connection: PostgresQueryMixin,
error_log: NwbConfigErrorLog) -> Dict[int, list]:
"""
Get a dict mapping probe_id to the list of channel
specifications for a given ecephys_session_id
Parameters
----------
ecephys_session_id: int
probe_ids_to_skip: Optional[List[int]]:
List of probes not to return because we already know they
are "bad" in some way.
lims_connection: PostgresQueryMixin
error_log: NwbConfigErrorLog
object to store all of the non-fatal irregularities
encountered in the data
Returns
-------
probe_id_to_channels: Dict[int, List[dict]]
A dict mapping probe_id to a list of dicts, each
of which represents the specifications of a channel
that needs to be written to the input.json.
"""
raw_channels_table = channels_table_from_ecephys_session_id_list(
ecephys_session_id_list=[ecephys_session_id, ],
probe_ids_to_skip=probe_ids_to_skip,
lims_connection=lims_connection)
raw_channels_table.rename(
columns={'ecephys_channel_id': 'id',
'ecephys_probe_id': 'probe_id'},
inplace=True)
raw_channels_table = raw_channels_table[[
'id',
'probe_id',
'probe_channel_number',
'structure_id',
'structure_acronym',
'anterior_posterior_ccf_coordinate',
'dorsal_ventral_ccf_coordinate',
'left_right_ccf_coordinate',
'probe_horizontal_position',
'probe_vertical_position',
'valid_data']]
raw_channels_table = raw_channels_table.set_index('id')
raw_channels_table = raw_channels_table.to_dict(orient='index')
probe_id_to_channels = dict()
for channel_id in raw_channels_table.keys():
this_channel = raw_channels_table[channel_id]
probe_id = this_channel['probe_id']
if probe_id not in probe_id_to_channels:
probe_id_to_channels[probe_id] = []
this_channel['id'] = channel_id
probe_id_to_channels[probe_id].append(_nan_to_none(this_channel))
return probe_id_to_channels
[docs]
def unit_input_from_ecephys_session_id(
ecephys_session_id: int,
probe_ids_to_skip: Optional[List[int]],
lims_connection: PostgresQueryMixin,
error_log: NwbConfigErrorLog) -> Dict[int, list]:
"""
Get a dict mapping probe_id to the list of unit
specifications for a given ecephys_session_id
Parameters
----------
ecephys_session_id: int
probe_ids_to_skip: Optional[List[int]]:
List of probes not to return because we already know they
are "bad" in some way.
lims_connection: PostgresQueryMixin
error_log: NwbConfigErrorLog
object to store all of the non-fatal irregularities
encountered in the data
Returns
-------
probe_id_to_units: Dict[int, List[dict]]
A dict mapping probe_id to a list of dicts, each
of which represents the specifications of a unit
that needs to be written to the input.json.
"""
raw_unit_table = units_table_from_ecephys_session_id_list(
ecephys_session_id_list=[ecephys_session_id, ],
probe_ids_to_skip=probe_ids_to_skip,
lims_connection=lims_connection)
raw_unit_table.rename(
columns={'unit_id': 'id',
'ecephys_channel_id': 'peak_channel_id'},
inplace=True)
if len(raw_unit_table) == 0:
msg = f"could not find units for session {ecephys_session_id}"
error_log.log(ecephys_session_id=ecephys_session_id,
msg=msg)
return dict()
raw_unit_table.drop(
labels=['ecephys_session_id',
'probe_vertical_position',
'probe_horizontal_position',
'anterior_posterior_ccf_coordinate',
'dorsal_ventral_ccf_coordinate',
'left_right_ccf_coordinate',
'structure_id',
'structure_acronym',
'valid_data'],
axis='columns',
inplace=True)
raw_unit_table = raw_unit_table.set_index('id')
raw_unit_table = raw_unit_table.to_dict(orient='index')
probe_id_to_units = dict()
for unit_id in raw_unit_table.keys():
this_unit = raw_unit_table[unit_id]
this_unit['id'] = unit_id
probe_id = this_unit.pop('ecephys_probe_id')
if probe_id not in probe_id_to_units:
probe_id_to_units[probe_id] = []
probe_id_to_units[probe_id].append(_nan_to_none(this_unit))
return probe_id_to_units
[docs]
def eye_tracking_geometry_from_equipment_id(
equipment_id: int,
date_of_acquisition: pd.Timestamp,
lims_connection: PostgresQueryMixin) -> dict:
"""
Return eye_tracking_rig_geometry given a specified
equipment_id and date_of_acquisition
Parameters
----------
equipment_id: int
date_of_acqisition: pd.Timestamp
lims_connection: PostgresQueryMixin
Returns
--------
eye_geometry: dict
The eye_tracking_geometry dict to be written to
the input.json
This dict conforms to the eye_traking_rig_geometry
schema specified in the NWB writer schema (except
that it will not list 'equipment'; that must be
added later)
Notes
-----
They eye tracking geometry that is specified is the latest
(as determined by LIMS' active_date column) that occured
before date_of_acquisition.
"""
raw_eye_geometry = _raw_eye_tracking_geometry_from_equipment_id(
equipment_id=equipment_id,
date_of_acquisition=date_of_acquisition,
lims_connection=lims_connection)
eye_geometry = dict()
eye_geometry['led_position'] = [
raw_eye_geometry['led position']['center_x_mm'],
raw_eye_geometry['led position']['center_y_mm'],
raw_eye_geometry['led position']['center_z_mm']]
eye_geometry['monitor_position_mm'] = [
raw_eye_geometry['screen position']['center_x_mm'],
raw_eye_geometry['screen position']['center_y_mm'],
raw_eye_geometry['screen position']['center_z_mm']]
eye_geometry['monitor_rotation_deg'] = [
raw_eye_geometry['screen position']['rotation_x_deg'],
raw_eye_geometry['screen position']['rotation_y_deg'],
raw_eye_geometry['screen position']['rotation_z_deg']]
eye_geometry['camera_position_mm'] = [
raw_eye_geometry['eye camera position']['center_x_mm'],
raw_eye_geometry['eye camera position']['center_y_mm'],
raw_eye_geometry['eye camera position']['center_z_mm']]
eye_geometry['camera_rotation_deg'] = [
raw_eye_geometry['eye camera position']['rotation_x_deg'],
raw_eye_geometry['eye camera position']['rotation_y_deg'],
raw_eye_geometry['eye camera position']['rotation_z_deg']]
return eye_geometry
def _raw_eye_tracking_geometry_from_equipment_id(
equipment_id: int,
date_of_acquisition: pd.Timestamp,
lims_connection: PostgresQueryMixin) -> dict:
"""
Return eye_tracking_rig_geometry given a specified
equipment_id and date_of_acquisition
Parameters
----------
equipment_id: int
date_of_acquisition: pd.Timestamp
lims_connection: PostgresQueryMixin
Returns
-------
config: dict
A dict listing the configuration of the
eye tracking rig geometry
This dict will contain entries for
'led position'
'behavior camera position'
'eye camera position'
'screen position'
Each of these keys maps to a dict containing
'center_x_mm'
'center_y_mm'
'center_z_mm'
'rotation_x_deg'
'rotation_y_deg'
'rotation_z_deg'
Notes
-----
Will return the configuration with the latest
active_date that is before date_of_acquisition.
"""
config = dict()
for name in ('led position', 'behavior camera position',
'eye camera position', 'screen position'):
this_df = experiment_configs_from_equipment_id_and_type(
equipment_id=equipment_id,
config_type=name,
lims_connection=lims_connection)
this_df = this_df.loc[
this_df.active_date.dt.date <= date_of_acquisition]
this_df = this_df.iloc[this_df.active_date.idxmax()]
this_config = dict()
this_config['center_x_mm'] = this_df.center_x_mm
this_config['center_y_mm'] = this_df.center_y_mm
this_config['center_z_mm'] = this_df.center_z_mm
this_config['rotation_x_deg'] = this_df.rotation_x_deg
this_config['rotation_y_deg'] = this_df.rotation_y_deg
this_config['rotation_z_deg'] = this_df.rotation_z_deg
config[name] = this_config
return config
def _analysis_run_from_session_id(
lims_connection: PostgresQueryMixin,
ecephys_session_id_list: List[int],
strategy_class: str) -> Dict[int, int]:
"""
Get a dict mapping ecephys_session_id to ecephys_analysis_runs.id
for a specific job strategy class ('VbnCreateStimTableStrategy',
'EcephysOptotaggingTableStrategy', etc.). Will only select the
instance of the run marked as 'current' in the LIMS database.
Parameters
----------
lims_connection: PostgresQueryMixin
ecephys_session_id_list: List[int]
strategy_class: str
Returns
-------
analysis_run_map: Dict[int, int]
A dict mapping ecephys_session_id to the
ecephys_analysis_runs.id associated with the specified
job strategy. Only returns the row marked as 'current'
Notes
-----
Raises a OneResultExpectedError if more than one run_id is returned
for the same session_id.
"""
query = """
SELECT
ecephys_session_id
,id as ecephys_analysis_run_id
FROM
ecephys_analysis_runs
WHERE
ecephys_analysis_runs.current
"""
query += build_in_list_selector_query(
col="ecephys_analysis_runs.ecephys_session_id",
valid_list=ecephys_session_id_list,
operator="AND",
valid=True)
query += build_in_list_selector_query(
col="ecephys_analysis_runs.job_strategy_class",
valid_list=[f"'{strategy_class}'"],
operator="AND",
valid=True)
query_result = lims_connection.select(query)
analysis_run_map = dict()
msg = ""
for session_id, run_id in zip(query_result.ecephys_session_id,
query_result.ecephys_analysis_run_id):
if session_id in analysis_run_map:
msg += ("More than one analysis run returned for "
f"ecephys_session_id={session_id}\n")
analysis_run_map[session_id] = run_id
if len(msg) > 0:
raise OneResultExpectedError(msg)
return analysis_run_map
def _add_spike_times_path(
probe_list: List[dict],
ecephys_session_id: int,
lims_connection: PostgresQueryMixin,
error_log: NwbConfigErrorLog) -> List[dict]:
"""
Add the 'spike_times_path' entry to a list of probe specifications.
Parameters
----------
probe_list: List[dict]
The list of probe specifications to be modified
ecephys_session_id:int
lims_connection: PostgresQueryMixin
Returns
-------
probe_list: List[dict]
Same as input with 'spike_times_path' added.
Notes
-----
Will alter probe_list in place
"""
probe_id_list = [this_probe['id'] for this_probe in probe_list]
timestamp_run_lookup = _analysis_run_from_session_id(
lims_connection=lims_connection,
ecephys_session_id_list=[ecephys_session_id, ],
strategy_class='EcephysAlignTimestampsStrategy')
# get mapping from probe_id to
# ecephys_analysis_run_probes.id
query = """
SELECT
ecephys_analysis_run_probes.ecephys_probe_id as probe_id
,ecephys_analysis_run_probes.id as ecephys_analysis_run_probe_id
FROM
ecephys_analysis_run_probes
"""
query += build_in_list_selector_query(
col="ecephys_analysis_run_probes.ecephys_probe_id",
valid_list=probe_id_list,
operator="WHERE",
valid=True)
query += build_in_list_selector_query(
col="ecephys_analysis_run_probes.ecephys_analysis_run_id",
valid_list=[
timestamp_run_lookup[ecephys_session_id], ],
operator="AND",
valid=True)
query_result = lims_connection.select(query)
probe_run_lookup = dict()
for p_id, r_id in zip(query_result.probe_id,
query_result.ecephys_analysis_run_probe_id):
probe_run_lookup[int(p_id)] = int(r_id)
for this_probe in probe_list:
probe_id = this_probe['id']
if probe_id in probe_run_lookup:
timestamp_lookup = wkf_path_from_attachable(
lims_connection=lims_connection,
wkf_type_name=[
"'EcephysAlignedEventTimestamps'", ],
attachable_type='EcephysAnalysisRunProbe',
attachable_id=probe_run_lookup[probe_id])
this_probe['spike_times_path'] = \
timestamp_lookup['EcephysAlignedEventTimestamps']
else:
msg = ("could not find EcephysAlignedEventTimestamps for "
f"probe {probe_id}")
error_log.log(ecephys_session_id=ecephys_session_id,
msg=msg)
this_probe['spike_times_path'] = None
return probe_list
def _nan_to_none(input_dict: dict) -> dict:
"""
Scan through a dict transforming any NaNs into
Nones (argschema does not like NaNs appearing in
float fields). Return the same dict after alteration.
Note
----
Alters the dict in place.
"""
for k in input_dict:
val = input_dict[k]
if isinstance(val, numbers.Number):
if np.isnan(val):
input_dict[k] = None
return input_dict