claudecode/AvianVisitors
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cleanup: factor out passing in settings

Browse Source
This commit is contained in:
frederik
2025-11-15 16:38:45 +01:00
parent d154b5f6f1
commit 1e2edbfd6e
5 changed files with 180 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/send_test_notification.py
+2 -3
View File
@@ -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'])
scripts/utils/notifications.py
+77 -68
View File
@@ -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__":
scripts/utils/reporting.py
+1 -2
View File
@@ -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)
tests/helpers.py
+18 -11
View File
@@ -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)
tests/test_apprise_notifications.py
+82 -69
View File
@@ -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()