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

Add utility to parse thisrun format

Browse Source
This commit is contained in:
Joe Weiss
2022-06-18 14:52:42 -04:00
parent bd52aabaab
commit 6865274bd2
6 changed files with 110 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/notifications.py
+14 -26
View File
@@ -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)"
scripts/server.py
+3 -1
View File
@@ -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])
scripts/utils/__init__.py
View File
scripts/utils/parse_settings.py
+16
View File
@@ -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
tests/test_apprise_notifications.py
+33 -4
View File
@@ -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)
tests/test_settings_parse.py
+44
View File
@@ -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.