parallelize training

Author: bjarthur

configuration.py

@@ -194,10 +194,6 @@
 # how many points to use for the precision-recall, sensitivity-specificity, and congruence curves
 nprobabilities=20
 
-# used by freeze and classify to specify how many output time tics to process in parallel.
-# must be greater than one.
-classify_parallelize=64
-
 # used by train, generalize, xvalidate, and activations
 data_loader_maxprocs=0  # 0 = num CPU cores
 data_loader_queuesize=1  # 0 = infinite

src/accuracy

@@ -114,7 +114,7 @@ def plot_probability_density(test_ground_truth, test_logits, ratios, thresholds,
         igt = test_ground_truth==gt
       else:
         igt = test_ground_truth[:,ilabel]==gt
-      if sum(igt)==0:
+      if np.sum(igt)==0:
         continue
       xdata = test_logits[igt,ilabel]
       xdata = np.minimum(np.finfo(float).max, np.exp(xdata))
@@ -156,7 +156,6 @@ def doit(logdir, key_to_plot, ckpt, labels, nprobabilities, error_ratios, loss,
                            'specificity-sensitivity.ckpt-'+str(ckpt)+'.pdf'))
   plt.close()
 
-  already_written = set()
   predictions_path = os.path.join(logdir,key_to_plot,'predictions.ckpt-'+str(ckpt))
   if not os.path.isdir(predictions_path):
     os.mkdir(os.path.join(logdir,key_to_plot,'predictions.ckpt-'+str(ckpt)))
@@ -169,20 +168,17 @@ def doit(logdir, key_to_plot, ckpt, labels, nprobabilities, error_ratios, loss,
             for i in range(len(validation_sounds)):
                 if validation_sounds[i]['file'][0] != subdir:
                     continue
-                classified_as = np.argmax(validation_logits[i,:])
-                id = os.path.join(*validation_sounds[i]['file']) + \
-                     str(validation_sounds[i]['ticks']) + \
-                     validation_sounds[i]['label'] + \
-                     labels[classified_as]
-                if id in already_written:
-                    continue
-                already_written |= set([id])
-                csvwriter.writerow([validation_sounds[i]['file'][1],
-                        validation_sounds[i]['ticks'][0],
-                        validation_sounds[i]['ticks'][1],
-                        'correct' if classified_as == validation_ground_truth[i] else 'mistaken',
-                        labels[classified_as],
-                        validation_sounds[i]['label']])
+                classified_as = np.argmax(validation_logits[i,:], axis=1)
+                scores = 1+np.where(np.diff(classified_as == validation_ground_truth[i]))[0]
+                scores = [0, *scores, len(classified_as)]
+                for iscore in range(len(scores)-1):
+                    if validation_ground_truth[i][scores[iscore]] == -1: continue
+                    csvwriter.writerow([validation_sounds[i]['file'][1],
+                            validation_sounds[i]['offset_tic']-len(classified_as)//2+1+scores[iscore],
+                            validation_sounds[i]['offset_tic']-len(classified_as)//2+scores[iscore+1],
+                            'correct' if classified_as[scores[iscore]] == validation_ground_truth[i][scores[iscore]] else 'mistaken',
+                            labels[classified_as[scores[iscore]]],
+                            validation_sounds[i]['label']])
   else:
       for subdir in set([x[0]['file'][0] for x in validation_sounds]):
           with open(os.path.join(logdir, key_to_plot, 'predictions.ckpt-'+str(ckpt), \
@@ -195,13 +191,6 @@ def doit(logdir, key_to_plot, ckpt, labels, nprobabilities, error_ratios, loss,
                 for j in range(len(validation_sounds[i])):
                     k = labels.index(validation_sounds[i][j]['label'].removeprefix(overlapped_prefix))
                     classified_as = validation_logits[i,k]>0.5
-                    id = os.path.join(*validation_sounds[i][j]['file']) + \
-                         str(validation_sounds[i][j]['ticks']) + \
-                         validation_sounds[i][j]['label'] + \
-                         str(classified_as)
-                    if id in already_written:
-                        continue
-                    already_written |= set([id])
                     csvwriter.writerow([validation_sounds[i][j]['file'][1],
                             validation_sounds[i][j]['ticks'][0],
                             validation_sounds[i][j]['ticks'][1],

src/activations

@@ -120,12 +120,19 @@ def main():
                     'video_frame_width': FLAGS.video_frame_width,
                     'video_frame_height': FLAGS.video_frame_height,
                     'video_channels': [int(x)-1 for x in FLAGS.video_channels.split(',')],
-                    'parallelize': 1,
+                    'parallelize': FLAGS.parallelize,
                     'context': FLAGS.context}
 
+  thismodel = model.create_model(model_settings, FLAGS.model_parameters)
+  thismodel.summary(line_length=120, positions=[0.4,0.6,0.7,1])
+
+  input_shape = thismodel.input_shape
+  clip_window_samples = input_shape[0][1] if model.use_video else input_shape[1]
+
   D.init(
       FLAGS.data_dir,
       FLAGS.shiftby,
+      clip_window_samples,
       FLAGS.labels_touse.split(','), FLAGS.kinds_touse.split(','),
       FLAGS.validation_percentage, FLAGS.validation_offset_percentage,
       FLAGS.validation_files.split(','),
@@ -141,9 +148,6 @@ def main():
       FLAGS.audio_read_plugin, FLAGS.video_read_plugin,
       audio_read_plugin_kwargs, video_read_plugin_kwargs)
 
-  thismodel = model.create_model(model_settings, FLAGS.model_parameters)
-  thismodel.summary(line_length=120, positions=[0.4,0.6,0.7,1])
-
   checkpoint = tf.train.Checkpoint(thismodel=thismodel)
   checkpoint.read(FLAGS.start_checkpoint).expect_partial()
 
@@ -326,6 +330,11 @@ if __name__ == '__main__':
       type=float,
       default=1000,
       help='Expected duration in milliseconds of the wavs',)
+  parser.add_argument(
+      '--parallelize',
+      type=int,
+      default=64,
+      help='how many output time tics to simultaneously process',)
   parser.add_argument(
       '--loss',
       type=str,

src/classify

@@ -163,7 +163,7 @@ def main():
       exit()
 
   if FLAGS.parallelize==1:
-    print("WARNING: classify_parallelize in configuration.py is set to 1.  making predictions is faster if it is > 1")
+    print("WARNING: parallelize in configuration.py is set to 1.  making predictions is faster if it is > 1")
 
   input_shape = thismodel.get_input_shape()
   if use_audio:

src/data.py

@@ -108,7 +108,7 @@ def which_set(filename, validation_percentage, validation_offset_percentage, tes
     return result
 
 def init(_data_dir,
-         shiftby,
+         shiftby, _clip_window_samples,
          labels_touse, kinds_touse,
          validation_percentage, validation_offset_percentage, validation_files,
          testing_percentage, testing_files, subsample_skip, subsample_label,
@@ -121,9 +121,10 @@ def init(_data_dir,
          _audio_read_plugin, _video_read_plugin,
          _audio_read_plugin_kwargs, _video_read_plugin_kwargs):
 
-    global data_dir, np_rng, audio_read_plugin_kwargs, audio_read_plugin, video_read_plugin_kwargs, video_read_plugin, queue_size, max_procs
+    global data_dir, clip_window_samples, np_rng, audio_read_plugin_kwargs, audio_read_plugin, video_read_plugin_kwargs, video_read_plugin, queue_size, max_procs
 
     data_dir = _data_dir
+    clip_window_samples = _clip_window_samples
     np_rng = np.random.default_rng(None if random_seed_batch==-1 else random_seed_batch)
 
     audio_read_plugin = _audio_read_plugin
@@ -198,6 +199,8 @@ def prepare_data_index(shiftby,
     audio_tic_rate = model_settings['audio_tic_rate']
     time_scale = model_settings['time_scale']
     context_tics = int(audio_tic_rate * model_settings['context'] * time_scale)
+    parallelize = int(model_settings['parallelize'])
+    stride_x_downsample = (clip_window_samples - context_tics) // (parallelize-1)
     shiftby_tics = int(shiftby * audio_tic_rate * time_scale)
     audio_ntics = {}
     video_nframes = {}
@@ -266,19 +269,21 @@ def prepare_data_index(shiftby,
                           f'in configuration.py but is actually {audio_data_shape[1]} in {wav_path}')
                 audio_ntics[wav_path] = audio_data_shape[0]
             if use_audio:
-                if ticks[1] < context_tics//2 + shiftby_tics or \
-                   ticks[0] > (audio_ntics[wav_path] - context_tics//2 + shiftby_tics):
+                left_room = context_tics//2 + (parallelize//2+1)*stride_x_downsample + shiftby_tics
+                right_room = context_tics//2 + (parallelize//2)*stride_x_downsample - shiftby_tics
+                if ticks[1] < left_room or \
+                   ticks[0] > (audio_ntics[wav_path] - right_room):
                     print(f"WARNING: {str(annotation)} is too close to an edge of the recording.  "
                           f"not using at all")
                     continue
-                if ticks[0] < context_tics//2 + shiftby_tics:
+                if ticks[0] < left_room:
                     print(f"WARNING: {str(annotation)} is close to beginning of recording.  "
                           f"shortening interval to usable range")
-                    ticks[0] = context_tics//2 + shiftby_tics
-                if ticks[1] > audio_ntics[wav_path] - context_tics//2 + shiftby_tics:
+                    ticks[0] = left_room
+                if ticks[1] > audio_ntics[wav_path] - right_room:
                     print(f"WARNING: {str(annotation)} is close to end of recording.  "
                           f"shortening interval to usable range")
-                    ticks[1] = audio_ntics[wav_path] - context_tics//2 + shiftby_tics
+                    ticks[1] = audio_ntics[wav_path] - right_room
             if use_video and wav_path not in video_nframes:
                 sound_dirname = os.path.join(data_dir, wav_base2[0])
                 vidfile = video_findfile(sound_dirname, wavfile)
@@ -438,11 +443,14 @@ def augment(audio_slice, augmentation_parameters):
         video_channels = model_settings['video_channels']
         time_scale = model_settings['time_scale']
         context_tics = int(audio_tic_rate * model_settings['context'] * time_scale)
+        parallelize = int(model_settings['parallelize'])
+        ninput_tics = clip_window_samples
+        stride_x_downsample = (clip_window_samples - context_tics) // (parallelize-1)
         shiftby_tics = int(shiftby * audio_tic_rate * time_scale)
         if use_audio:
-            audio_slice = np.zeros((nsounds, context_tics, audio_nchannels), dtype=np.float32)
+            audio_slice = np.zeros((nsounds, ninput_tics, audio_nchannels), dtype=np.float32)
         if use_video:
-            nframes = round(model_settings['context'] * time_scale * video_frame_rate)
+            nframes = round(ninput_tics / audio_tic_rate * video_frame_rate)
             video_slice = np.zeros((nsounds,
                                     nframes,
                                     model_settings['video_frame_height'],
@@ -451,9 +459,9 @@ def augment(audio_slice, augmentation_parameters):
                                    dtype=np.float32)
             bkg = {}
         if loss=='exclusive':
-            labels = np.zeros(nsounds, dtype=np.int32)
+            labels = -1*np.ones((nsounds, parallelize), dtype=np.int32)
         elif loss=='overlapped':
-            labels = 2*np.ones((nsounds, len(labels_list)), dtype=np.float32)
+            labels = 2*np.ones((nsounds, parallelize, len(labels_list)), dtype=np.float32)
         # repeatedly to generate the final output sound data we'll use in training.
         for i in range(offset, offset + nsounds):
             # Pick which sound to use.
@@ -466,8 +474,8 @@ def augment(audio_slice, augmentation_parameters):
             offset_tic = (np_rng.integers(sound['ticks'][0], high=1+sound['ticks'][1]) \
                           if sound['ticks'][0] < sound['ticks'][1] \
                           else sound['ticks'][0])
-            start_tic = offset_tic - math.floor(context_tics/2) - shiftby_tics
-            stop_tic  = offset_tic + math.ceil(context_tics/2) - shiftby_tics
+            start_tic = offset_tic - math.ceil(ninput_tics/2) - shiftby_tics
+            stop_tic  = offset_tic + math.floor(ninput_tics/2) - shiftby_tics
             if use_audio:
                 wavpath = os.path.join(data_dir, *sound['file'])
                 _, _, audio_data = audio_read(wavpath, start_tic, stop_tic)
@@ -486,9 +494,18 @@ def augment(audio_slice, augmentation_parameters):
                     video_slice[i-offset,iframe,:,:,:] = \
                             frame[:,:,video_channels] - bkg[vidfile][:,:,video_channels]
             if loss=='exclusive':
-                labels[i - offset] = labels_list.index(sound['label'])
+                start_in_tic = max(sound['ticks'][0] - offset_tic,
+                                   (1 - parallelize / 2) * stride_x_downsample)
+                start_out_tic = math.ceil(start_in_tic / stride_x_downsample)
+                start_out_tic += parallelize // 2 - 1
+                stop_in_tic = min(sound['ticks'][1] - offset_tic + stride_x_downsample,
+                                  parallelize / 2 * stride_x_downsample)
+                stop_out_tic = math.floor(stop_in_tic / stride_x_downsample)
+                stop_out_tic += parallelize // 2 - 1
+                labels[i - offset, start_out_tic : 1+stop_out_tic] = labels_list.index(sound['label'])
+                sound['offset_tic'] = offset_tic
                 sounds.append({k: v for k,v in sound.items() if k!='overlaps'})
-            elif loss=='overlapped':
+            elif loss=='overlapped':  ### !!!
                 target = 0 if sound['label'].startswith(overlapped_prefix) else 1
                 root = sound['label'].removeprefix(overlapped_prefix)
                 labels[i - offset, labels_list.index(root)] = target

src/generalize

@@ -7,6 +7,7 @@
 #     --shiftby=0.0 \
 #     --optimizer=Adam \
 #     --loss=exclusive \
+#     --parallelize=64 \
 #     --overlapped_prefix=not_ \
 #     --learning_rate=0.0002 \
 #     --audio_read_plugin=load_wav \
@@ -102,6 +103,7 @@ def main():
             "--shiftby="+str(FLAGS.shiftby),
             "--optimizer="+FLAGS.optimizer,
             "--loss="+FLAGS.loss,
+            "--parallelize="+str(FLAGS.parallelize),
             "--overlapped_prefix="+FLAGS.overlapped_prefix,
             "--learning_rate="+str(FLAGS.learning_rate),
             "--audio_read_plugin="+FLAGS.audio_read_plugin,
@@ -232,6 +234,11 @@ if __name__ == '__main__':
       type=float,
       default=1000,
       help='Expected duration in milliseconds of the wavs',)
+  parser.add_argument(
+      '--parallelize',
+      type=int,
+      default=64,
+      help='how many output time tics to simultaneously process',)
   parser.add_argument(
       '--learning_rate',
       type=float,

src/gui/controller.py

@@ -113,7 +113,7 @@ def generic_parameters_callback(n):
     M.save_state_callback()
     V.buttons_update()
 
-def context_callback(n):
+def context_parallelize_callback(n):
     V.model_summary_update()
     generic_parameters_callback(n)
 
@@ -1190,6 +1190,7 @@ async def train_actuate():
                 "--shiftby="+V.shiftby.value, \
                 "--optimizer="+V.optimizer.value, \
                 "--loss="+V.loss.value, \
+                "--parallelize="+str(V.parallelize.value),
                 "--overlapped_prefix="+M.overlapped_prefix, \
                 "--learning_rate="+V.learning_rate.value, \
                 "--audio_read_plugin="+str(M.audio_read_plugin), \
@@ -1289,6 +1290,7 @@ async def leaveout_actuate(kind):
                 "--shiftby="+V.shiftby.value, \
                 "--optimizer="+V.optimizer.value, \
                 "--loss="+V.loss.value, \
+                "--parallelize="+str(V.parallelize.value),
                 "--overlapped_prefix="+M.overlapped_prefix, \
                 "--learning_rate="+V.learning_rate.value, \
                 "--audio_read_plugin="+str(M.audio_read_plugin), \
@@ -1362,6 +1364,7 @@ async def xvalidate_actuate():
                 "--shiftby="+V.shiftby.value, \
                 "--optimizer="+V.optimizer.value, \
                 "--loss="+V.loss.value, \
+                "--parallelize="+str(V.parallelize.value),
                 "--overlapped_prefix="+M.overlapped_prefix, \
                 "--learning_rate="+V.learning_rate.value, \
                 "--audio_read_plugin="+str(M.audio_read_plugin), \
@@ -1735,7 +1738,7 @@ async def _freeze_actuate(ckpts):
                             "--model_architecture="+M.architecture_plugin,
                             "--model_parameters="+json.dumps({k:v.value for k,v in V.model_parameters.items()}),
                             "--loss="+V.loss.value, \
-                            "--parallelize="+str(M.classify_parallelize),
+                            "--parallelize="+str(V.parallelize.value),
                             "--time_units="+str(M.time_units),
                             "--freq_units="+str(M.freq_units),
                             "--time_scale="+str(M.time_scale),
@@ -1821,7 +1824,7 @@ async def ensemble_actuate():
                             "--context="+V.context.value,
                             "--model_architecture="+M.architecture_plugin,
                             "--model_parameters="+json.dumps({k:v.value for k,v in V.model_parameters.items()}),
-                            "--parallelize="+str(M.classify_parallelize),
+                            "--parallelize="+str(V.parallelize.value),
                             "--time_units="+str(M.time_units),
                             "--freq_units="+str(M.freq_units),
                             "--time_scale="+str(M.time_scale),
@@ -1871,7 +1874,7 @@ async def _classify_actuate(wavfiles):
             "--model="+os.path.join(logdir,model,"frozen-graph.ckpt-"+check_point+".pb"),
             "--model_labels="+os.path.join(logdir,model,"labels.txt"),
             "--wav="+wavfile,
-            "--parallelize="+str(M.classify_parallelize),
+            "--parallelize="+str(V.parallelize.value),
             "--time_scale="+str(M.time_scale),
             "--audio_tic_rate="+str(M.audio_tic_rate),
             "--audio_nchannels="+str(M.audio_nchannels),
@@ -2230,6 +2233,9 @@ def _copy_callback():
             elif "context = " in line:
                 m=re.search('context = (.*)', line)
                 V.context.value = m.group(1)
+            elif "parallelize = " in line:
+                m=re.search('parallelize = (.*)', line)
+                V.parallelize.value = m.group(1)
             elif "time_shift_sec = " in line:
                 m=re.search('time_shift_sec = (.*)', line)
                 V.shiftby.value = m.group(1)

src/gui/main.py

@@ -109,7 +109,7 @@
                 V.prevalences_button, V.prevalences,
                 V.delete_ckpts, V.copy, width=M.gui_width_pix),
             row(V.nsteps, V.restore_from, V.weights_seed, V.optimizer, V.context,
-                V.mini_batch, V.nreplicates, V.activations_equalize_ratio,
+                V.parallelize, V.mini_batch, V.nreplicates, V.activations_equalize_ratio,
                 V.precision_recall_ratios, V.congruence_portion,
                 width=M.gui_width_pix),
             row(V.save_and_validate_period, V.validate_percentage, V.batch_seed,

src/gui/model.py

@@ -60,6 +60,7 @@ def save_state_callback():
                      'labels': str.join(',',[x.value for x in V.label_texts]),
                      'file_dialog_string': V.file_dialog_string.value,
                      'context': V.context.value,
+                     'parallelize': V.parallelize.value,
                      'shiftby': V.shiftby.value,
                      'optimizer': V.optimizer.value,
                      'loss': V.loss.value,
@@ -502,6 +503,7 @@ def is_local_server_or_cluster(varname, varvalue):
                           'labels':','*(nlabels-1), \
                           'file_dialog_string':os.getcwd(), \
                           'context':str(0.2048 / time_scale), \
+                          'parallelize':'64', \
                           'shiftby':'0.0', \
                           'optimizer':'Adam', \
                           'loss':'exclusive', \