From 8d08e17770daad889f4c12ec196fa740205b68a3 Mon Sep 17 00:00:00 2001 From: frederik Date: Sat, 22 Nov 2025 12:47:15 +0100 Subject: [PATCH] cleanup filter_humans a bit --- scripts/utils/analysis.py | 66 +++++++-------- scripts/utils/notifications.py | 8 +- tests/test_analysis.py | 147 +++++++++++++++++++++++++++++++++ 3 files changed, 179 insertions(+), 42 deletions(-) diff --git a/scripts/utils/analysis.py b/scripts/utils/analysis.py index 32ca89d..37869c6 100644 --- a/scripts/utils/analysis.py +++ b/scripts/utils/analysis.py @@ -87,52 +87,42 @@ def analyzeAudioData(chunks, overlap, lat, lon, week): return labeled, predicted_species_list -def filter_humans(detections): +def filter_humans(predictions): conf = get_settings() priv_thresh = conf.getfloat('PRIVACY_THRESHOLD') human_cutoff = max(10, int(6000 * priv_thresh / 100.0)) log.debug("HUMAN-CUTOFF AT: %d", human_cutoff) - - censored_detections = [] - for detection in detections: - p = detection[:human_cutoff] - human_detected = False - # Catch if Human is recognized in any of the predictions - for x in p: - if 'Human' in x[0]: - human_detected = True - - # If human detected set detection to human to make sure voices are not saved - if human_detected is True: - p = [('Human_Human', 0.0)] - else: - p = p[:10] - - censored_detections.append(p) - - # now overwrite detections that have a human neighbour too try: - extraction_length = conf.getint('EXTRACTION_LENGTH') + if conf.getint('EXTRACTION_LENGTH') > 9: + log.warning("EXTRACTION_LENGTH is set to %d. Privacy filter might miss human sound, " + "if you care about privacy, set EXTRACTION_LENGTH to below 9 or leave empty.", conf.getint('EXTRACTION_LENGTH')) except ValueError: - extraction_length = 6 - if extraction_length > 9: - log.warning("EXTRACTION_LENGTH is set to %d. Privacy filter might miss human sound, " - "if you care about privacy, set EXTRACTION_LENGTH to below 9 or leave empty.", extraction_length) - human_neighbour_mask = [False] * len(censored_detections) - for i, detection in enumerate(censored_detections): - if i != 0: - if censored_detections[i - 1][0][0] == 'Human_Human': - human_neighbour_mask[i] = True - if i != len(censored_detections) - 1: - if censored_detections[i + 1][0][0] == 'Human_Human': - human_neighbour_mask[i] = True + pass + + # mask for humans + human_mask = [False] * len(predictions) + for i, prediction in enumerate(predictions): + for p in prediction[:human_cutoff]: + if 'Human' in p[0]: + human_mask[i] = True + break + + # mask for predictions that have a human neighbour + human_neighbour_mask = [False] * len(predictions) + for i, _ in enumerate(human_mask): + if i != 0 and human_mask[i - 1]: + human_neighbour_mask[i] = True + if i != len(human_mask) - 1 and human_mask[i + 1]: + human_neighbour_mask[i] = True clean_detections = [] - for i, (has_human_neighbour, detection) in enumerate(zip(human_neighbour_mask, censored_detections)): - if has_human_neighbour and detection[0][0] != 'Human_Human': - log.debug('Overwriting detection %d %s - Has Human neighbour', i + 1, detection[0]) - detection = [('Human_Human', 0.0)] - clean_detections.append(detection) + for prediction, human, has_human_neighbour in zip(predictions, human_mask, human_neighbour_mask): + if human or has_human_neighbour: + log.debug('Overwriting prediction %s', prediction[0]) + prediction = [('Human_Human', 0.0)] + else: + prediction = prediction[:10] + clean_detections.append(prediction) return clean_detections diff --git a/scripts/utils/notifications.py b/scripts/utils/notifications.py index 98aab55..7df85dd 100644 --- a/scripts/utils/notifications.py +++ b/scripts/utils/notifications.py @@ -77,10 +77,10 @@ def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path if websiteurl is None or len(websiteurl) == 0: websiteurl = f"http://{socket.gethostname()}.local" - listenurl = websiteurl+"?filename="+path - friendlyurl = "[Listen here]("+listenurl+")" - image_url = "" + listenurl = f"{websiteurl}?filename= {path}" + friendlyurl = f"[Listen here]({listenurl})" + image_url = "" if "$flickrimage" in body or "$image" in body: if com_name not in images: try: @@ -88,7 +88,7 @@ def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path resp = requests.get(url=url, timeout=10).json() images[com_name] = resp['data']['image_url'] except Exception as e: - print("IMAGE API ERROR: " + str(e)) + print("IMAGE API ERROR:", e) image_url = images.get(com_name, "") if settings_dict.get('APPRISE_NOTIFY_EACH_DETECTION') == "1": diff --git a/tests/test_analysis.py b/tests/test_analysis.py index bcf8a99..a74abc5 100644 --- a/tests/test_analysis.py +++ b/tests/test_analysis.py @@ -5,6 +5,7 @@ from unittest.mock import patch from scripts.utils.analysis import run_analysis from scripts.utils.classes import ParseFileName from tests.helpers import TESTDATA, Settings +from scripts.utils.analysis import filter_humans class TestRunAnalysis(unittest.TestCase): @@ -36,5 +37,151 @@ class TestRunAnalysis(unittest.TestCase): self.assertEqual(det.scientific_name, expected['sci_name']) +class TestFilterHumans(unittest.TestCase): + + @patch('scripts.utils.helpers._load_settings') + def test_filter_humans_no_human(self, mock_load_settings): + mock_load_settings.return_value = Settings.with_defaults() + + # Input detections without humans + detections = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Bird_C', 0.7), ('Bird_D', 0.6)] + ] + + # Expected output + expected = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Bird_C', 0.7), ('Bird_D', 0.6)] + ] + + # Run filter_humans + result = filter_humans(detections) + + # Assertions + self.assertEqual(result, expected) + + @patch('scripts.utils.helpers._load_settings') + def test_filter_empty(self, mock_load_settings): + mock_load_settings.return_value = Settings.with_defaults() + + # Input detections without humans + detections = [] + + # Expected output + expected = [] + + # Run filter_humans + result = filter_humans(detections) + + # Assertions + self.assertEqual(result, expected) + + @patch('scripts.utils.helpers._load_settings') + def test_filter_humans_with_human(self, mock_load_settings): + mock_load_settings.return_value = Settings.with_defaults() + + # Input detections with humans + detections = [ + [('Human_Human', 0.95), ('Bird_A', 0.8)], + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Bird_C', 0.9), ('Bird_D', 0.8)], + [('Bird_B', 0.7), ('Human vocal_Human vocal', 0.9)] + ] + + # Expected output + expected = [ + [('Human_Human', 0.0)], + [('Human_Human', 0.0)], + [('Human_Human', 0.0)], + [('Human_Human', 0.0)] + ] + + # Run filter_humans + result = filter_humans(detections) + + # Assertions + self.assertEqual(result, expected) + + @patch('scripts.utils.helpers._load_settings') + def test_filter_humans_with_human_neighbour(self, mock_load_settings): + mock_load_settings.return_value = Settings.with_defaults() + + # Input detections with human neighbours + detections = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Bird_D', 0.9), ('Bird_E', 0.8)], + [('Human_Human', 0.95), ('Bird_C', 0.7)], + [('Bird_F', 0.6), ('Bird_G', 0.5)] + ] + + # Expected output + expected = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Human_Human', 0.0)], + [('Human_Human', 0.0)], + [('Human_Human', 0.0)] + ] + + # Run filter_humans + result = filter_humans(detections) + + # Assertions + self.assertEqual(result, expected) + + @patch('scripts.utils.helpers._load_settings') + def test_filter_humans_with_deep_human(self, mock_load_settings): + mock_load_settings.return_value = Settings.with_defaults() + + # Input detections with human neighbours + detections = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Bird_D', 0.9), ('Bird_E', 0.8)], + [('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.5), ('Bird_C', 0.7), ('Human_Human', 0.95)], + [('Bird_F', 0.6), ('Bird_G', 0.5)] + ] + + # Expected output + expected = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Bird_D', 0.9), ('Bird_E', 0.8)], + [('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.5)], + [('Bird_F', 0.6), ('Bird_G', 0.5)] + ] + + # Run filter_humans + result = filter_humans(detections) + + # Assertions + self.assertEqual(result, expected) + + @patch('scripts.utils.helpers._load_settings') + def test_filter_humans_with_human_deep(self, mock_load_settings): + settings = Settings.with_defaults() + settings['PRIVACY_THRESHOLD'] = 1 + mock_load_settings.return_value = settings + + # Input detections with human neighbours + detections = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Bird_D', 0.9), ('Bird_E', 0.8)], + [('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.7), ('Bird_C', 0.5), ('Bird_C', 0.7), ('Human_Human', 0.95)], + [('Bird_F', 0.6), ('Bird_G', 0.5)] + ] + + # Expected output + expected = [ + [('Bird_A', 0.9), ('Bird_B', 0.8)], + [('Human_Human', 0.0)], + [('Human_Human', 0.0)], + [('Human_Human', 0.0)] + ] + + # Run filter_humans + result = filter_humans(detections) + + # Assertions + self.assertEqual(result, expected) + if __name__ == '__main__': unittest.main()