cleanup filter_humans a bit

This commit is contained in:
frederik
2025-11-22 12:47:15 +01:00
parent 1e2edbfd6e
commit 8d08e17770
3 changed files with 179 additions and 42 deletions
+26 -36
View File
@@ -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')
except ValueError:
extraction_length = 6
if extraction_length > 9:
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.", 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':
"if you care about privacy, set EXTRACTION_LENGTH to below 9 or leave empty.", conf.getint('EXTRACTION_LENGTH'))
except ValueError:
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(censored_detections) - 1:
if censored_detections[i + 1][0][0] == 'Human_Human':
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
+4 -4
View File
@@ -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":
+147
View File
@@ -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()