From 6865274bd253413b5555ca0a560db9c81f09ca9d Mon Sep 17 00:00:00 2001 From: Joe Weiss Date: Sat, 18 Jun 2022 14:52:42 -0400 Subject: [PATCH] Add utility to parse thisrun format --- scripts/notifications.py | 40 +++++++++----------------- scripts/server.py | 4 ++- scripts/utils/__init__.py | 0 scripts/utils/parse_settings.py | 16 +++++++++++ tests/test_apprise_notifications.py | 37 +++++++++++++++++++++--- tests/test_settings_parse.py | 44 +++++++++++++++++++++++++++++ 6 files changed, 110 insertions(+), 31 deletions(-) create mode 100644 scripts/utils/__init__.py create mode 100644 scripts/utils/parse_settings.py create mode 100644 tests/test_settings_parse.py diff --git a/scripts/notifications.py b/scripts/notifications.py index 8c1ef4a..312b8a8 100644 --- a/scripts/notifications.py +++ b/scripts/notifications.py @@ -21,19 +21,16 @@ def notify(body, title): title=title, ) -def get_setting(param): - with open(THIS_RUN_PATH, 'r') as f: - this_run = f.readlines() - return str(str(str([i for i in this_run if i.startswith(param)]).split('=')[1]).split('\\')[0]).replace('"', '') - -def sendAppriseNotifications(species, confidence, path, db_path=DB_PATH): +def sendAppriseNotifications(species, confidence, path, settings_dict, db_path=DB_PATH): + # print(settings_dict) if os.path.exists(APPRISE_CONFIG) and os.path.getsize(APPRISE_CONFIG) > 0: - title = get_setting('APPRISE_NOTIFICATION_TITLE') - body = get_setting('APPRISE_NOTIFICATION_BODY') + title = settings_dict.get('APPRISE_NOTIFICATION_TITLE') + body = settings_dict.get('APPRISE_NOTIFICATION_BODY') + sciName, comName = species.split("_") try: - websiteurl = get_setting('BIRDNETPI_URL') + websiteurl = settings_dict.get('BIRDNETPI_URL') if len(websiteurl) == 0: raise ValueError('Blank URL') except Exception as e: @@ -41,28 +38,20 @@ def sendAppriseNotifications(species, confidence, path, db_path=DB_PATH): listenurl = websiteurl+"?filename="+path - if get_setting('APPRISE_NOTIFY_EACH_DETECTION') == "1": - apobj = apprise.Apprise() - config = apprise.AppriseConfig() - config.add(APPRISE_CONFIG) - apobj.add(config) + if settings_dict.get('APPRISE_NOTIFY_EACH_DETECTION') == "1": + notify_body=body.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + notify_title=title.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + notify(notify_body, notify_title) - apobj.notify( - body=body.replace("$sciname", species.split("_")[0]).replace("$comname", species.split("_")[1]).replace("$confidence", confidence).replace("$listenurl", listenurl), - title=title.replace("$sciname", species.split("_")[0]).replace("$comname", species.split("_")[1]).replace("$confidence", confidence).replace("$listenurl", listenurl), - ) - - APPRISE_NOTIFICATION_NEW_SPECIES_DAILY = True - APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT = 2 # Notifies the first N per day. - if APPRISE_NOTIFICATION_NEW_SPECIES_DAILY: + APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT = 1 # Notifies the first N per day. + if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY') == "1": try: con = sqlite3.connect(db_path) cur = con.cursor() cur.execute("SELECT DISTINCT(Com_Name), count(Com_Name) FROM detections WHERE date > (SELECT DATETIME('now', '-1 day')) GROUP BY Com_Name") known_species = cur.fetchall() - sciName, comName = species.split("_") numberDetections = [d[1] for d in known_species if d[0] == comName.replace("'","")][0] - if numberDetections < APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT: + if numberDetections <= APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT: notify_body=body.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections))+" times today)", notify_title=title.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections))+" times today)", notify(notify_body, notify_title) @@ -72,13 +61,12 @@ def sendAppriseNotifications(species, confidence, path, db_path=DB_PATH): print("Database busy") time.sleep(2) - if get_setting('APPRISE_NOTIFY_NEW_SPECIES') == "1": + if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES') == "1": try: con = sqlite3.connect(db_path) cur = con.cursor() cur.execute("SELECT DISTINCT(Com_Name), count(Com_Name) FROM detections WHERE date > (SELECT DATETIME('now', '-7 day')) GROUP BY Com_Name") known_species = cur.fetchall() - sciName, comName = species.split("_") numberDetections = [d[1] for d in known_species if d[0] == comName.replace("'","")][0] if numberDetections <= 5: notify_body=body.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections))+" times in last 7d)" diff --git a/scripts/server.py b/scripts/server.py index fce5319..72b7cc3 100755 --- a/scripts/server.py +++ b/scripts/server.py @@ -15,6 +15,7 @@ import threading import os from notifications import sendAppriseNotifications +from utils.parse_settings import config_to_settings os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' os.environ['CUDA_VISIBLE_DEVICES'] = '' @@ -49,6 +50,7 @@ with open(userDir + '/BirdNET-Pi/scripts/thisrun.txt', 'r') as f: audiofmt = "." + str(str(str([i for i in this_run if i.startswith('AUDIOFMT')]).split('=')[1]).split('\\')[0]) priv_thresh = float("." + str(str(str([i for i in this_run if i.startswith('PRIVACY_THRESHOLD')]).split('=')[1]).split('\\')[0])) / 10 +settings_dict = config_to_settings(userDir + '/BirdNET-Pi/scripts/thisrun.txt') def loadModel(): @@ -401,7 +403,7 @@ def handle_client(conn, addr): # Apprise of detection if not already alerted this run. if not str(entry[0] in species_apprised_this_run: - sendAppriseNotifications(str(entry[0]), str(entry[1]), File_Name) + sendAppriseNotifications(str(entry[0]), str(entry[1]), File_Name, settings_dict) species_apprised_this_run.append(str(entry[0]) diff --git a/scripts/utils/__init__.py b/scripts/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scripts/utils/parse_settings.py b/scripts/utils/parse_settings.py new file mode 100644 index 0000000..d0e2166 --- /dev/null +++ b/scripts/utils/parse_settings.py @@ -0,0 +1,16 @@ +def config_to_settings(path): + # Returns settings dict from BirdNET-pi text format. + # Consider refactoring to use ConfigParser files, or another standardized format. + settings = {} + with open(path, 'r') as f: + this_run = f.readlines() + for i in this_run: + key = i.split("=")[0] + value = i.split("=")[1][:-1] + # Trim for strings, not ideal + if value.startswith('"') and value.endswith('"'): + value = value[1:-1] + if value.startswith("'") and value.endswith("'"): + value = value[1:-1] + settings[key] = value + return settings diff --git a/tests/test_apprise_notifications.py b/tests/test_apprise_notifications.py index deee959..5418113 100644 --- a/tests/test_apprise_notifications.py +++ b/tests/test_apprise_notifications.py @@ -37,15 +37,44 @@ def clean_up_after_each_test(): yield os.remove("test.db") -def test_daily_notifications(mocker): +def test_notifications(mocker): notify_call = mocker.patch('scripts.notifications.notify') create_test_db("test.db") - assert(0 == 0) - sendAppriseNotifications("Myiarchus crinitus_Great Crested Flycatcher", "91", "filename", "test.db") + settings_dict = { + "APPRISE_NOTIFICATION_TITLE": "New backyard bird!", + "APPRISE_NOTIFICATION_BODY": "A $comname ($sciname) was just detected with a confidence of $confidence", + "APPRISE_NOTIFY_EACH_DETECTION": "0", + "APPRISE_NOTIFY_NEW_SPECIES": "0", + "APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY": "0" + } + + # No active apprise notifcations configured. Confirm no notifications. + sendAppriseNotifications("Myiarchus crinitus_Great Crested Flycatcher", "91", "filename", settings_dict, "test.db") + assert(notify_call.call_count == 0) # No notification should be sent. + + # Add daily notification. + notify_call.reset_mock() + settings_dict["APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY"] = "1" + sendAppriseNotifications("Myiarchus crinitus_Great Crested Flycatcher", "91", "filename", settings_dict, "test.db") + assert(notify_call.call_count == 1) + assert( + notify_call.call_args_list[0][0][0][0] == "A Great Crested Flycatcher (Myiarchus crinitus) was just detected with a confidence of 91 (only seen 1 times today)" + ) + + # Add new species notification. + notify_call.reset_mock() + settings_dict["APPRISE_NOTIFY_NEW_SPECIES"] = "1" + sendAppriseNotifications("Myiarchus crinitus_Great Crested Flycatcher", "91", "filename", settings_dict, "test.db") assert(notify_call.call_count == 2) assert( notify_call.call_args_list[0][0][0][0] == "A Great Crested Flycatcher (Myiarchus crinitus) was just detected with a confidence of 91 (only seen 1 times today)" ) assert( notify_call.call_args_list[1][0][0] == "A Great Crested Flycatcher (Myiarchus crinitus) was just detected with a confidence of 91 (only seen 1 times in last 7d)" - ) + ) + + # Add each species notification. + notify_call.reset_mock() + settings_dict["APPRISE_NOTIFY_EACH_DETECTION"] = "1" + sendAppriseNotifications("Myiarchus crinitus_Great Crested Flycatcher", "91", "filename", settings_dict, "test.db") + assert(notify_call.call_count == 3) diff --git a/tests/test_settings_parse.py b/tests/test_settings_parse.py new file mode 100644 index 0000000..de68b9f --- /dev/null +++ b/tests/test_settings_parse.py @@ -0,0 +1,44 @@ +from scripts.utils.parse_settings import config_to_settings +import tempfile + +def test(): + text = """LATITUDE=32.0 +LONGITUDE=-73.0 +BIRDWEATHER_ID= +CADDY_PWD="nonsuchpass" +ICE_PWD=birdnetpi +BIRDNETPI_URL= +RTSP_STREAM= +APPRISE_NOTIFICATION_TITLE="Bird!" +APPRISE_NOTIFICATION_BODY="A $comname ($sciname) was just detected with a confidence of $confidence" +APPRISE_NOTIFY_EACH_DETECTION=0 +APPRISE_NOTIFY_NEW_SPECIES=1 +FLICKR_API_KEY= +FLICKR_FILTER_EMAIL= +RECS_DIR=/home/pi/BirdSongs +REC_CARD=default +PROCESSED=/home/pi/BirdSongs/Processed +EXTRACTED=/home/pi/BirdSongs/Extracted +OVERLAP=0.0 +CONFIDENCE=0.7 +SENSITIVITY=1.25 +CHANNELS=2 +FULL_DISK=purge +PRIVACY_THRESHOLD=0 +RECORDING_LENGTH=15 +EXTRACTION_LENGTH= +AUDIOFMT=mp3 +DATABASE_LANG=en +LAST_RUN= +THIS_RUN= +IDFILE=/home/pi/BirdNET-Pi/IdentifiedSoFar.txt""" + + filename = tempfile.NamedTemporaryFile(suffix='.txt', delete=False) + with open(filename.name, 'w', encoding='utf8', newline='') as f: + f.write(text) + + settings = config_to_settings(filename.name) + assert(settings["APPRISE_NOTIFICATION_TITLE"] == "Bird!") + assert(settings["FULL_DISK"] == "purge") + assert(settings["OVERLAP"] == "0.0") # Yes, it's a string at this point. +