Added ROC Curve Visualizations, Confidence, and annotation chunker updates

PyHa/IsoAutio.py

@@ -9,8 +9,9 @@
 import pandas as pd
 import scipy.signal as scipy_signal
 import numpy as np
-from math import ceil
+from math import ceil, floor
 from copy import deepcopy
+from tqdm import tqdm
 
 
 def checkVerbose(
@@ -773,8 +774,10 @@ def generate_automated_labels_birdnet(audio_dir, isolation_parameters):
         Dataframe of automated labels for the audio clip(s) in audio_dir.
     """
     annotations = analyze(audio_path=audio_dir, **isolation_parameters)
-    return annotations
-
+    local_scores_dir  = {}
+    for file in np.unique(annotations["IN FILE"]):
+        local_scores_dir[file] = annotations["CONFIDENCE"].to_list()
+    return annotations, local_scores_dir
 def generate_automated_labels_microfaune(
         audio_dir,
         isolation_parameters,
@@ -825,8 +828,10 @@ def generate_automated_labels_microfaune(
 
     # init labels dataframe
     annotations = pd.DataFrame()
+    local_score_dir = {}
+
     # generate local scores for every bird file in chosen directory
-    for audio_file in os.listdir(audio_dir):
+    for audio_file in tqdm(os.listdir(audio_dir), desc=f"Processing labels for {audio_dir}", colour='green'):
         # skip directories
         if os.path.isdir(audio_dir + audio_file):
             continue
@@ -871,6 +876,8 @@ def generate_automated_labels_microfaune(
         # get duration of clip
         duration = len(SIGNAL) / SAMPLE_RATE
 
+        local_score_dir[audio_file] = local_scores[0]
+        #print(local_scores[0])
         try:
             # Running moment to moment algorithm and appending to a master
             # dataframe.
@@ -883,7 +890,13 @@ def generate_automated_labels_microfaune(
                 isolation_parameters,
                 manual_id=manual_id,
                 normalize_local_scores=normalize_local_scores)
-            # print(new_entry)
+            #print(new_entry)
+
+            #determine confidence of annotations
+            new_entry = add_confidence_to_annotations(new_entry, local_scores[0])
+
+            #print(new_entry)
+
             if annotations.empty:
                 annotations = new_entry
             else:
@@ -893,7 +906,7 @@ def generate_automated_labels_microfaune(
             continue
     # Quick fix to indexing
     annotations.reset_index(inplace=True, drop=True)
-    return annotations
+    return annotations, local_score_dir
 
 def generate_automated_labels_tweetynet(
         audio_dir,
@@ -941,8 +954,9 @@ def generate_automated_labels_tweetynet(
 
     # init labels dataframe
     annotations = pd.DataFrame()
+    local_score_dir = {}
     # generate local scores for every bird file in chosen directory
-    for audio_file in os.listdir(audio_dir):
+    for audio_file in tqdm(os.listdir(audio_dir), desc=f"Processing labels for {audio_dir}", colour='green'):
         # skip directories
         if os.path.isdir(audio_dir + audio_file):
             continue
@@ -1003,16 +1017,21 @@ def generate_automated_labels_tweetynet(
                     manual_id=manual_id,
                     normalize_local_scores=normalize_local_scores)
             # print(new_entry)
+
+            new_entry = add_confidence_to_annotations(new_entry, local_scores[0])
+
+
             if annotations.empty:
                 annotations = new_entry
             else:
                 annotations = annotations.append(new_entry)
         except BaseException as e:
             checkVerbose("Error in isolating bird calls from" + audio_file, isolation_parameters)
             continue
+        local_score_dir[audio_file] = local_scores[0]
     # Quick fix to indexing
     annotations.reset_index(inplace=True, drop=True)
-    return annotations
+    return annotations,local_score_dir
 
 
 def generate_automated_labels(
@@ -1021,7 +1040,8 @@ def generate_automated_labels(
         manual_id="bird",
         weight_path=None,
         normalized_sample_rate=44100,
-        normalize_local_scores=False):
+        normalize_local_scores=False,
+        include_local_scores=False):
     """
     Function that generates the bird labels across a folder of audio clips
     given the isolation_parameters
@@ -1048,13 +1068,18 @@ def generate_automated_labels(
         normalize_local_scores (bool)
             - Set whether or not to normalize the local scores.
 
+    include_local_scores (bool)
+            - Set whether or not to also output local_scores for ROC curve generation
+
     Returns:
         Dataframe of automated labels for the audio clips in audio_dir.
+        If include_local_scores = True, it outputs a tuple containing the dataframe of automated labels
+        and a directory containing the local scores of each file as used by the generate ROC curve functions
     """
 
     #try:
     if(isolation_parameters["model"] == 'microfaune'):
-        annotations = generate_automated_labels_microfaune(
+        annotations, local_scores = generate_automated_labels_microfaune(
                         audio_dir=audio_dir,
                         isolation_parameters=isolation_parameters,
                         manual_id=manual_id,
@@ -1069,10 +1094,10 @@ def generate_automated_labels(
             'threshold_const', 'chunk_size']
         for key in keys_to_delete:
             birdnet_parameters.pop(key, None)
-        annotations = generate_automated_labels_birdnet(
+        annotations, local_scores = generate_automated_labels_birdnet(
                         audio_dir, birdnet_parameters)
     elif(isolation_parameters['model'] == 'tweetynet'):
-        annotations = generate_automated_labels_tweetynet(
+        annotations, local_scores = generate_automated_labels_tweetynet(
                         audio_dir=audio_dir,
                         isolation_parameters=isolation_parameters,
                         manual_id=manual_id,
@@ -1088,7 +1113,10 @@ def generate_automated_labels(
     # except:
     #     print("Error. Check your isolation_parameters")
     #     return None
-    return annotations
+    if (include_local_scores) : 
+        return annotations, local_scores
+    else: 
+        return annotations
 
 def kaleidoscope_conversion(df):
     """
@@ -1111,6 +1139,76 @@ def kaleidoscope_conversion(df):
     kaleidoscope_df = pd.concat(kaleidoscope_df, axis=1, keys=headers)
     return kaleidoscope_df
 
+
+def add_confidence_to_annotations(clip_df, local_score_array):
+    """
+        Adds confidence of each annotation from a local_score array.
+        Takes the maximum normalized local score value within each
+        annotation as the confidence
+
+        Args:
+            clip_df: (Dataframe)
+                - Dataframe containing the automated annotations of a single
+                clip 
+
+            local_score_array: (Numpy Array)
+                - array of local_scores from predict functions
+
+        Returns:
+            clip_df with an extra column containing the confidence of each annotation
+    """
+    #data prep for processing
+    local_score_array = normalize(local_score_array, 0, 1)
+    clip_df["CONFIDENCE"] = 0
+    confidence_array = []
+
+    for i in range(clip_df.shape[0]):
+        annotation_data = clip_df.iloc[i]
+        # now iterate through the local_score array for each chunk
+        clip_length = annotation_data["CLIP LENGTH"]
+        #seconds_per_index = clip_length/len(local_score_clip)
+        index_per_seconds = len(local_score_array)/clip_length
+
+        #Get the starting and ending index of that chunk as respective to
+        #the local score array
+        start_time =  annotation_data["OFFSET"]
+        end_time = annotation_data["OFFSET"] + annotation_data["DURATION"]
+        start_index = floor(start_time * index_per_seconds)
+        end_index = floor((end_time * index_per_seconds))
+        max_index = floor((clip_length * index_per_seconds))
+
+        #Compute the local maximum in this chunk in the local scores
+        max_score = max(local_score_array[start_index: min(end_index, max_index)])
+        confidence_array.append(max_score)
+    clip_df["CONFIDENCE"] = confidence_array
+    return clip_df
+
+def normalize(arr, t_min, t_max):
+    """
+        normalize local_score for better confidence value
+
+        Args:
+            arr: (Numpy Array)
+                - Local Score array.
+
+            t_min: (int)
+                - minimum value to set.
+
+            t_max: (int)
+                - maximum value to set.
+
+        Returns:
+            Numpy array of the normalized local score array.
+        """
+    norm_arr = []
+    diff = t_max - t_min
+    arr_min = min(arr)
+    diff_arr = max(arr) - arr_min
+    for i in arr:
+        temp = (((i - arr_min)*diff)/diff_arr) + t_min
+        norm_arr.append(temp)
+    return norm_arr
+
 # def annotation_combiner(df):
 #    # Initializing the output Pandas dataframe
 #    combined_annotation_df = pd.DataFrame()