diff --git a/scripts/send_test_notification.py b/scripts/send_test_notification.py index 9974b18..d28a8f9 100755 --- a/scripts/send_test_notification.py +++ b/scripts/send_test_notification.py @@ -38,6 +38,5 @@ if __name__ == "__main__": 'Lat': conf.getfloat('LATITUDE'), 'Lon': conf.getfloat('LONGITUDE'), 'Cutoff': conf.getfloat('CONFIDENCE'), 'Sens': conf.getfloat('SENSITIVITY'), 'Overlap': conf.getfloat('OVERLAP')} - notifications.sendAppriseNotifications(d['Sci_Name'], d['Com_Name'], d['Confidence'], round(d['Confidence'] * 100), d['File_Name'], d['Date'], - d['Time'], d['Week'], d['Lat'], d['Lon'], d['Cutoff'], d['Sens'], d['Overlap'], dict(conf)) - + notifications.sendAppriseNotifications(d['Sci_Name'], d['Com_Name'], d['Confidence'], round(d['Confidence'] * 100), d['File_Name'], d['Date'], d['Time'], + d['Week'], d['Lat'], d['Lon'], d['Cutoff'], d['Sens'], d['Overlap']) diff --git a/scripts/utils/notifications.py b/scripts/utils/notifications.py index 4bd8182..98aab55 100644 --- a/scripts/utils/notifications.py +++ b/scripts/utils/notifications.py @@ -3,14 +3,14 @@ import os import socket import requests import html -import time as timeim +import time from .db import get_todays_count_for, get_this_weeks_count_for +from .helpers import get_settings userDir = os.path.expanduser('~') APPRISE_CONFIG = userDir + '/BirdNET-Pi/apprise.txt' APPRISE_BODY = userDir + '/BirdNET-Pi/body.txt' -DB_PATH = userDir + '/BirdNET-Pi/scripts/birds.db' apobj = None images = {} @@ -44,7 +44,7 @@ def notify(body, title, attached=""): ) -def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path, date, time, week, latitude, longitude, cutoff, sens, overlap, settings_dict): +def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path, date, time_of_day, week, latitude, longitude, cutoff, sens, overlap): def render_template(template, reason=""): ret = template.replace("$sciname", sci_name) \ .replace("$comname", com_name) \ @@ -53,7 +53,7 @@ def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path .replace("$listenurl", listenurl) \ .replace("$friendlyurl", friendlyurl) \ .replace("$date", str(date)) \ - .replace("$time", str(time)) \ + .replace("$time", str(time_of_day)) \ .replace("$week", str(week)) \ .replace("$latitude", str(latitude)) \ .replace("$longitude", str(longitude)) \ @@ -64,80 +64,89 @@ def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path .replace("$overlap", str(overlap)) \ .replace("$reason", reason) return ret - # print(sendAppriseNotifications) - # print(settings_dict) - if os.path.exists(APPRISE_CONFIG) and os.path.getsize(APPRISE_CONFIG) > 0: - title = html.unescape(settings_dict.get('APPRISE_NOTIFICATION_TITLE')) - f = open(APPRISE_BODY, 'r') - body = f.read() + if not should_notify(com_name): + return - APPRISE_ONLY_NOTIFY_SPECIES_NAMES = settings_dict.get('APPRISE_ONLY_NOTIFY_SPECIES_NAMES') - if APPRISE_ONLY_NOTIFY_SPECIES_NAMES is not None and APPRISE_ONLY_NOTIFY_SPECIES_NAMES.strip() != "": - if any(bird.lower().replace(" ", "") in com_name.lower().replace(" ", "") for bird in APPRISE_ONLY_NOTIFY_SPECIES_NAMES.split(",")): - return + settings_dict = get_settings() + title = html.unescape(settings_dict.get('APPRISE_NOTIFICATION_TITLE')) + f = open(APPRISE_BODY, 'r') + body = f.read() - APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2 = settings_dict.get('APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2') - if APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2 is not None and APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2.strip() != "": - if not any(bird.lower().replace(" ", "") in com_name.lower().replace(" ", "") for bird in APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2.split(",")): - return + websiteurl = settings_dict.get('BIRDNETPI_URL') + if websiteurl is None or len(websiteurl) == 0: + websiteurl = f"http://{socket.gethostname()}.local" - APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES = settings_dict.get('APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES') - if APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES != "0": - if species_last_notified.get(com_name) is not None: - try: - if int(timeim.time()) - species_last_notified[com_name] < int(APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES): - return - except Exception as e: - print("APPRISE NOTIFICATION EXCEPTION: "+str(e)) - return + listenurl = websiteurl+"?filename="+path + friendlyurl = "[Listen here]("+listenurl+")" + image_url = "" - # TODO: this all needs to be changed, we changed the caddy default to allow direct IP access, so birdnetpi.local shouldn't be relied on anymore - try: - websiteurl = settings_dict.get('BIRDNETPI_URL') - if len(websiteurl) == 0: - raise ValueError('Blank URL') - except Exception: - websiteurl = "http://"+socket.gethostname()+".local" + if "$flickrimage" in body or "$image" in body: + if com_name not in images: + try: + url = f"http://localhost/api/v1/image/{sci_name}" + 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)) + image_url = images.get(com_name, "") - listenurl = websiteurl+"?filename="+path - friendlyurl = "[Listen here]("+listenurl+")" - image_url = "" + if settings_dict.get('APPRISE_NOTIFY_EACH_DETECTION') == "1": + reason = "detection" + notify_body = render_template(body, reason) + notify_title = render_template(title, reason) + notify(notify_body, notify_title, image_url) + species_last_notified[com_name] = int(time.time()) - if "$flickrimage" in body or "$image" in body: - if com_name not in images: - try: - url = f"http://localhost/api/v1/image/{sci_name}" - 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)) - image_url = images.get(com_name, "") - - if settings_dict.get('APPRISE_NOTIFY_EACH_DETECTION') == "1": - notify_body = render_template(body, "detection") - notify_title = render_template(title, "detection") + 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": + numberDetections = get_todays_count_for(sci_name) + if 0 < numberDetections <= APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT: + reason = "first time today" + notify_body = render_template(body, reason) + notify_title = render_template(title, reason) notify(notify_body, notify_title, image_url) - species_last_notified[com_name] = int(timeim.time()) + species_last_notified[com_name] = int(time.time()) - 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": - numberDetections = get_todays_count_for(sci_name) - if 0 < numberDetections <= APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT: - print("send the notification") - notify_body = render_template(body, "first time today") - notify_title = render_template(title, "first time today") - notify(notify_body, notify_title, image_url) - species_last_notified[com_name] = int(timeim.time()) + if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES') == "1": + numberDetections = get_this_weeks_count_for(sci_name) + if 0 < numberDetections <= 5: + reason = f"only seen {numberDetections} times in last 7d" + notify_body = render_template(body, reason) + notify_title = render_template(title, reason) + notify(notify_body, notify_title, image_url) + species_last_notified[com_name] = int(time.time()) - if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES') == "1": - numberDetections = get_this_weeks_count_for(sci_name) - if 0 < numberDetections <= 5: - reason = f"only seen {numberDetections} times in last 7d" - notify_body = render_template(body, reason) - notify_title = render_template(title, reason) - notify(notify_body, notify_title, image_url) - species_last_notified[com_name] = int(timeim.time()) + +def should_notify(com_name): + settings_dict = get_settings() + if not (os.path.exists(APPRISE_CONFIG) and os.path.getsize(APPRISE_CONFIG) > 0): + return False + + # check if this is an excluded species + APPRISE_ONLY_NOTIFY_SPECIES_NAMES = settings_dict.get('APPRISE_ONLY_NOTIFY_SPECIES_NAMES') + if APPRISE_ONLY_NOTIFY_SPECIES_NAMES is not None and APPRISE_ONLY_NOTIFY_SPECIES_NAMES.strip() != "": + if any(bird.lower().replace(" ", "") in com_name.lower().replace(" ", "") for bird in APPRISE_ONLY_NOTIFY_SPECIES_NAMES.split(",")): + return False + + # check if this is an included species + APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2 = settings_dict.get('APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2') + if APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2 is not None and APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2.strip() != "": + if not any(bird.lower().replace(" ", "") in com_name.lower().replace(" ", "") for bird in APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2.split(",")): + return False + + # is it still too soon? + APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES = settings_dict.get('APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES') + if APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES != "0": + if species_last_notified.get(com_name) is not None: + try: + if int(time.time()) - species_last_notified[com_name] < int(APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES): + return False + except Exception as e: + print("APPRISE NOTIFICATION EXCEPTION: " + str(e)) + return False + + return True if __name__ == "__main__": diff --git a/scripts/utils/reporting.py b/scripts/utils/reporting.py index 60f2372..dfa8a66 100644 --- a/scripts/utils/reporting.py +++ b/scripts/utils/reporting.py @@ -159,8 +159,7 @@ def apprise(file: ParseFileName, detections: [Detection]): try: sendAppriseNotifications(detection.scientific_name, detection.common_name, str(detection.confidence), str(detection.confidence_pct), os.path.basename(detection.file_name_extr), detection.date, detection.time, str(detection.week), - conf['LATITUDE'], conf['LONGITUDE'], conf['CONFIDENCE'], conf['SENSITIVITY'], - conf['OVERLAP'], dict(conf)) + conf['LATITUDE'], conf['LONGITUDE'], conf['CONFIDENCE'], conf['SENSITIVITY'], conf['OVERLAP']) except BaseException as e: log.exception('Error during Apprise:', exc_info=e) diff --git a/tests/helpers.py b/tests/helpers.py index 679abb4..3cb8557 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -13,16 +13,23 @@ class Settings(dict): @classmethod def with_defaults(cls): settings = { - 'OVERLAP': 0.0, - 'LATITUDE': 50, - 'LONGITUDE': 5, - 'CONFIDENCE': 0.7, - 'DATABASE_LANG': 'en', - 'PRIVACY_THRESHOLD': 0, - 'EXTRACTION_LENGTH': 6, - 'MODEL': 'BirdNET_GLOBAL_6K_V2.4_Model_FP16', - 'DATA_MODEL_VERSION': 1, - 'SENSITIVITY': 1.25, - 'SF_THRESH': 0.003, + "OVERLAP": 0.0, + "LATITUDE": 50, + "LONGITUDE": 5, + "CONFIDENCE": 0.7, + "DATABASE_LANG": "en", + "PRIVACY_THRESHOLD": 0, + "EXTRACTION_LENGTH": 6, + "MODEL": "BirdNET_GLOBAL_6K_V2.4_Model_FP16", + "DATA_MODEL_VERSION": 1, + "SENSITIVITY": 1.25, + "SF_THRESH": 0.003, + "APPRISE_NOTIFICATION_TITLE": "New backyard bird!", + "APPRISE_NOTIFY_EACH_DETECTION": "0", + "APPRISE_NOTIFY_NEW_SPECIES": "0", + "APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY": "0", + "APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES": "0", + "APPRISE_ONLY_NOTIFY_SPECIES_NAMES": "", + "APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2": "" } return cls(settings) diff --git a/tests/test_apprise_notifications.py b/tests/test_apprise_notifications.py index 2b1e836..b2c3d89 100644 --- a/tests/test_apprise_notifications.py +++ b/tests/test_apprise_notifications.py @@ -1,15 +1,20 @@ import os import sqlite3 -from datetime import datetime import unittest +from datetime import datetime from unittest.mock import patch +from scripts.utils import db from scripts.utils import notifications from scripts.utils.notifications import sendAppriseNotifications +from tests.helpers import Settings class TestAppriseNotifications(unittest.TestCase): + def setUp(self): + db.DB_PATH = self.db_file + @classmethod def setUpClass(cls): cls.db_file = "test.db" @@ -59,32 +64,33 @@ class TestAppriseNotifications(unittest.TestCase): if os.path.exists(self.apprise_config_file): os.remove(self.apprise_config_file) + def get_default_params(self): + return { + "sci_name": "Myiarchus crinitus", + "com_name": "Great Crested Flycatcher", + "confidence": "0.91", + "confidencepct": "91", + "path": "filename", + "date": "1666-06-06", + "time_of_day": "06:06:06", + "week": "06", + "latitude": "-1", + "longitude": "-1", + "cutoff": "0.7", + "sens": "1.25", + "overlap": "0.0" + } + + @patch('scripts.utils.helpers._load_settings') @patch('scripts.utils.notifications.notify') - def test_notifications(self, mock_notify): + def test_notifications(self, mock_notify, mock_load_settings): self.create_test_db() self.create_apprise_config() - settings_dict = { - "APPRISE_NOTIFICATION_TITLE": "New backyard bird!", - "APPRISE_NOTIFY_EACH_DETECTION": "0", - "APPRISE_NOTIFY_NEW_SPECIES": "0", - "APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY": "0", - "APPRISE_MINIMUM_SECONDS_BETWEEN_NOTIFICATIONS_PER_SPECIES": "0" - } - sendAppriseNotifications("Myiarchus crinitus", - "Great Crested Flycatcher", - "0.91", - "91", - "filename", - "1666-06-06", - "06:06:06", - "06", - "-1", - "-1", - "0.7", - "1.25", - "0.0", - settings_dict, - self.db_file) + notifications.DB_PATH = self.db_file + settings_dict = Settings.with_defaults() + + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) # No active apprise notifications configured. Confirm no notifications. self.assertEqual(mock_notify.call_count, 0) @@ -92,21 +98,8 @@ class TestAppriseNotifications(unittest.TestCase): # Add daily notification. mock_notify.reset_mock() settings_dict["APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY"] = "1" - sendAppriseNotifications("Myiarchus crinitus", - "Great Crested Flycatcher", - "0.91", - "91", - "filename", - "1666-06-06", - "06:06:06", - "06", - "-1", - "-1", - "0.7", - "1.25", - "0.0", - settings_dict, - self.db_file) + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) self.assertEqual(mock_notify.call_count, 1) self.assertEqual( @@ -117,21 +110,8 @@ class TestAppriseNotifications(unittest.TestCase): # Add new species notification. mock_notify.reset_mock() settings_dict["APPRISE_NOTIFY_NEW_SPECIES"] = "1" - sendAppriseNotifications("Myiarchus crinitus", - "Great Crested Flycatcher", - "0.91", - "91", - "filename", - "1666-06-06", - "06:06:06", - "06", - "-1", - "-1", - "0.7", - "1.25", - "0.0", - settings_dict, - self.db_file) + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) self.assertEqual(mock_notify.call_count, 2) self.assertEqual( @@ -146,24 +126,57 @@ class TestAppriseNotifications(unittest.TestCase): # Add each species notification. mock_notify.reset_mock() settings_dict["APPRISE_NOTIFY_EACH_DETECTION"] = "1" - sendAppriseNotifications("Myiarchus crinitus", - "Great Crested Flycatcher", - "0.91", - "91", - "filename", - "1666-06-06", - "06:06:06", - "06", - "-1", - "-1", - "0.7", - "1.25", - "0.0", - settings_dict, - self.db_file) + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) self.assertEqual(mock_notify.call_count, 3) + @patch('scripts.utils.helpers._load_settings') + @patch('scripts.utils.notifications.notify') + def test_notifications_excluded(self, mock_notify, mock_load_settings): + self.create_test_db() + self.create_apprise_config() + notifications.DB_PATH = self.db_file + settings_dict = Settings.with_defaults() + settings_dict["APPRISE_NOTIFY_EACH_DETECTION"] = "1" + + settings_dict['APPRISE_ONLY_NOTIFY_SPECIES_NAMES'] = 'Quailfinch' + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) + + # Not excluded. Confirm notifications. + self.assertEqual(mock_notify.call_count, 1) + + mock_notify.reset_mock() + settings_dict['APPRISE_ONLY_NOTIFY_SPECIES_NAMES'] = 'Quailfinch,Great Crested Flycatcher' + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) + + self.assertEqual(mock_notify.call_count, 0) + + @patch('scripts.utils.helpers._load_settings') + @patch('scripts.utils.notifications.notify') + def test_notifications_included(self, mock_notify, mock_load_settings): + self.create_test_db() + self.create_apprise_config() + notifications.DB_PATH = self.db_file + settings_dict = Settings.with_defaults() + settings_dict["APPRISE_NOTIFY_EACH_DETECTION"] = "1" + + settings_dict['APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2'] = 'Quailfinch' + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) + + # No wanted species. Confirm no notifications. + self.assertEqual(mock_notify.call_count, 0) + + mock_notify.reset_mock() + settings_dict['APPRISE_ONLY_NOTIFY_SPECIES_NAMES_2'] = 'Quailfinch,Great Crested Flycatcher' + mock_load_settings.return_value = settings_dict + sendAppriseNotifications(**self.get_default_params()) + + self.assertEqual(mock_notify.call_count, 1) + if __name__ == '__main__': unittest.main()