123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123 |
- import re
- import pandas as pd
- import tools.pyrelacs as pr
- from tools.definitions import METADATA_FOLDER, RECORDINGS_FOLDER, OUTPUT_FOLDER, HELPER_TABLE_FOLDER, \
- STIMULATION_METADATA_RAW
- def extract_value_unit(value_unit_string):
- value_unit_pattern = re.compile('(\d+[e][+]\d*|[-+]?\d*\.\d+|\d+)\s*(\w+)')
- value, unit = value_unit_pattern.match(value_unit_string).groups()
- return float(value), unit
- def convert_to_sec(time_value, time_unit):
- if time_unit == "ms":
- time_value *= 0.001
- return time_value
- overview_of_recordings = METADATA_FOLDER+"cells.csv"
- test_protocol_for_pa = "SingleStimulus"
- print "# Extract stimulation protocols"
- print "Search the recordings listed in {}".format(overview_of_recordings)
- print "Collect all instances of the stimulation protocol {} used to test for persistent activity".format(
- test_protocol_for_pa)
- cell_recordings = pd.read_csv(overview_of_recordings, header=0)
- cell_recordings.columns = cell_recordings.columns.str.strip()
- for column in cell_recordings.columns:
- cell_recordings[column] = cell_recordings[column].str.strip()
- cells = cell_recordings["CellId"].unique()
- stimulation_id = []
- cell_ids = []
- recordings = []
- before_protocols = []
- after_protocols = []
- stimulus_lengths = []
- pulse_periods = []
- pulse_lengths = []
- pulse_amplitudes = []
- pulse_offsets = []
- dts = []
- start_indices = []
- end_indices = []
- parameter_dict={}
- before_stimulation_interval = 2
- after_stimulation_interval = 8
- for cell in cells:
- recording_counter = 0
- for _, recording in cell_recordings[cell_recordings["CellId"] == cell].iterrows():
- cell_id = cell
- recording_dir = RECORDINGS_FOLDER+recording["Recording"]
- for info, key, dat in pr.iload('%s/stimuli.dat' % (recording_dir,)):
- if info[1]['RePro'] != test_protocol_for_pa:
- continue
- repro_starting_idx = dat[1, 0]
- left_appendum, right_appendum = pr.get_trace_indices(info, before_stimulation_interval,
- after_stimulation_interval)
- dt = pr.get_sampling_interval(info)
- stimulation_id.append("{:s}_R{:03d}".format(cell_id, recording_counter))
- cell_ids.append(cell_id)
- recordings.append(recording_dir)
- available_model_parameters = map(lambda key: info[0][key], filter(lambda key: key.startswith("identifier") and key != "identifier1", info[0].keys()))
- parameter_values = [dat[0][key[1].index(param)+1] for param in available_model_parameters]
- for param, value in zip(available_model_parameters, parameter_values):
- try:
- parameter_dict[param].append(value)
- except KeyError as err:
- parameter_dict[param] = [value]
- dts.append(dt)
- stimulus_length, unit = extract_value_unit(info[1]["duration"])
- stimulus_length = convert_to_sec(stimulus_length, unit)
- pulse_period, unit = extract_value_unit(info[1]["period"])
- pulse_period = convert_to_sec(pulse_period, unit)
- pulse_length, unit = extract_value_unit(info[1]["pulseduration"])
- pulse_length = convert_to_sec(pulse_length, unit)
- pulse_amplitude, _ = extract_value_unit(info[1]["amplitude"])
- pulse_offset, _ = extract_value_unit(info[1]["offset"])
- before_protocols.append(before_stimulation_interval)
- after_protocols.append(after_stimulation_interval)
- stimulus_lengths.append(stimulus_length)
- pulse_periods.append(pulse_period)
- pulse_lengths.append(pulse_length)
- pulse_amplitudes.append(pulse_amplitude)
- pulse_offsets.append(pulse_offset)
- start_indices.append(repro_starting_idx - left_appendum)
- end_indices.append(repro_starting_idx + right_appendum)
- recording_counter += 1
- dictionary_of_stimulations = {
- "stimulation_id": stimulation_id,
- "cell_id": cell_ids,
- "recording": recordings,
- "dt": dts,
- "before_protocol": before_protocols,
- "after_protocol": after_protocols,
- "stimulus_length": stimulus_lengths,
- "pulse_period": pulse_periods,
- "pulse_length": pulse_lengths,
- "pulse_amplitude": pulse_amplitudes,
- "pulse_offset": pulse_offsets,
- "start_index": start_indices,
- "end_index": end_indices
- }
- dictionary_of_stimulations.update(parameter_dict)
- stimulus_table_saving_path = OUTPUT_FOLDER+HELPER_TABLE_FOLDER+STIMULATION_METADATA_RAW
- pd.DataFrame(data=dictionary_of_stimulations).set_index("stimulation_id").to_csv(path_or_buf=stimulus_table_saving_path)
- print "Gathered meta data for each stimulation and saved to {}".format(stimulus_table_saving_path)
- print ""
|