Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
waveform_list = []
if grouping_property is not None:
if grouping_property not in recording.get_channel_property_names():
raise ValueError("'grouping_property' should be a property of recording extractors")
if compute_property_from_recording:
compute_sorting_group = True
elif grouping_property not in sorting.get_unit_property_names():
print(grouping_property, ' not in sorting extractor. Computing it from the recording extractor')
compute_sorting_group = True
else:
compute_sorting_group = False
print("Waveforms by property: ", grouping_property)
if not compute_sorting_group:
rec_list, rec_props = se.get_sub_extractors_by_property(recording, grouping_property,
return_property_list=True)
sort_list, sort_props = se.get_sub_extractors_by_property(sorting, grouping_property,
return_property_list=True)
if len(rec_props) != len(sort_props):
print('Different' + grouping_property + ' numbers: using largest number of ' + grouping_property)
if len(rec_props) > len(sort_props):
for i_r, rec in enumerate(rec_props):
if rec not in sort_props:
print('Inserting None for property ', rec)
sort_list.insert(i_r, None)
else:
for i_s, sort in enumerate(sort_props):
if sort not in rec_props:
rec_list.insert(i_s, None)
else:
assert len(rec_list) == len(sort_list)
if grouping_property is not None:
if grouping_property not in recording.get_channel_property_names():
raise ValueError("'grouping_property' should be a property of recording extractors")
if compute_property_from_recording:
compute_sorting_group = True
elif grouping_property not in sorting.get_unit_property_names():
print(grouping_property, ' not in sorting extractor. Computing it from the recording extractor')
compute_sorting_group = True
else:
compute_sorting_group = False
print("Waveforms by property: ", grouping_property)
if not compute_sorting_group:
rec_list, rec_props = se.get_sub_extractors_by_property(recording, grouping_property,
return_property_list=True)
sort_list, sort_props = se.get_sub_extractors_by_property(sorting, grouping_property,
return_property_list=True)
if len(rec_props) != len(sort_props):
print('Different' + grouping_property + ' numbers: using largest number of ' + grouping_property)
if len(rec_props) > len(sort_props):
for i_r, rec in enumerate(rec_props):
if rec not in sort_props:
print('Inserting None for property ', rec)
sort_list.insert(i_r, None)
else:
for i_s, sort in enumerate(sort_props):
if sort not in rec_props:
rec_list.insert(i_s, None)
else:
assert len(rec_list) == len(sort_list)
for i_list, (rec, sort) in enumerate(zip(rec_list, sort_list)):
self.debug = debug
self.grouping_property = grouping_property
self.parallel = parallel
if output_folder is None:
output_folder = 'test_' + self.sorter_name
output_folder = Path(output_folder).absolute()
if grouping_property is None:
# only one groups
self.recording_list = [recording]
self.output_folders = [output_folder]
else:
# several groups
self.recording_list = se.get_sub_extractors_by_property(recording, grouping_property)
n_group = len(self.recording_list)
if n_group>1:
self.output_folders = [Path(str(output_folder) + '_'+str(i)) for i in range(n_group)]
else:
self.output_folders = [output_folder]
# make folders
for output_folder in self.output_folders:
if not output_folder.is_dir():
os.makedirs(str(output_folder))
self.delete_folders = delete_output_folder
assert sorter_name in sorter_dict, '{} is not in sorter list'.format(sorter_name)
if isinstance(recording_dict_or_list, list):
# in case of list
recording_dict = { 'recording_{}'.format(i): rec for i, rec in enumerate(recording_dict_or_list) }
elif isinstance(recording_dict_or_list, dict):
recording_dict = recording_dict_or_list
else:
raise(ValueError('bad recording dict'))
# when grouping_property is not None : split in subrecording
# but the subrecording must have len=1 because otherwise it break
# the internal organisation of folder name.
if grouping_property is not None:
for rec_name, recording in recording_dict.items():
recording_list = se.get_sub_extractors_by_property(recording, grouping_property)
n_group = len(recording_list)
assert n_group == 1, 'run_sorters() work only if grouping_property=None or if it split into one subrecording'
recording_dict[rec_name] = recording_list[0]
grouping_property = None
task_list = []
for rec_name, recording in recording_dict.items():
for sorter_name in sorter_list:
output_folder = working_folder / rec_name / sorter_name
if is_log_ok(output_folder):
# check is output_folders exists
if mode == 'raise':
raise(Exception('output folder already exists for {} {}'.format(rec_name, sorter_name)))