Refactor notifications into its own module with tests

requirements.txt

@@ -10,4 +10,6 @@ pandas
 seaborn
 streamlit
 plotly
-apprise
+apprise
+pytest==7.1.2
+pytest-mock==3.7.0

scripts/notifications.py

@@ -0,0 +1,96 @@
+import apprise
+import os
+import socket
+import sqlite3
+from sqlite3 import Error
+import time
+from datetime import datetime, timedelta
+
+userDir = os.path.expanduser('~')
+APPRISE_CONFIG = userDir + '/BirdNET-Pi/apprise.txt'
+THIS_RUN_PATH = userDir + '/BirdNET-Pi/scripts/thisrun.txt'
+DB_PATH = userDir + '/BirdNET-Pi/scripts/birds.db'
+
+def notify(body, title):
+    apobj = apprise.Apprise()
+    config = apprise.AppriseConfig()
+    config.add(APPRISE_CONFIG)
+    apobj.add(config)
+    apobj.notify(
+        body=body,
+        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):
+    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')
+        
+        try:
+            websiteurl = get_setting('BIRDNETPI_URL')
+            if len(websiteurl) == 0:
+                raise ValueError('Blank URL')
+        except Exception as e:
+            websiteurl = "http://"+socket.gethostname()+".local"
+
+        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)
+
+            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:
+            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:
+                    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)
+                con.close()
+            except sqlite3.Error as e:
+                print(e)
+                print("Database busy")
+                time.sleep(2)
+
+        if get_setting('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)"
+                    notify_title=title.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections))+" times in last 7d)"
+                    notify(notify_body, notify_title)
+                con.close()
+            except sqlite3.Error:
+                print("Database busy")
+                time.sleep(2)
+
+
+if __name__ == "__main__":
+    print("notfications")
+
+    

scripts/server.py

@@ -13,6 +13,9 @@ import operator
 import socket
 import threading
 import os
+
+from notifications import sendAppriseNotifications
+
 os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
 os.environ['CUDA_VISIBLE_DEVICES'] = ''
 
@@ -228,59 +231,6 @@ def analyzeAudioData(chunks, lat, lon, week, sensitivity, overlap,):
     return detections
 
 
-def sendAppriseNotifications(species, confidence, path):
-    if os.path.exists(userDir + '/BirdNET-Pi/apprise.txt') and os.path.getsize(userDir + '/BirdNET-Pi/apprise.txt') > 0:
-        with open(userDir + '/BirdNET-Pi/scripts/thisrun.txt', 'r') as f:
-            this_run = f.readlines()
-            title = str(str(str([i for i in this_run if i.startswith('APPRISE_NOTIFICATION_TITLE')]).split('=')[1]).split('\\')[0]).replace('"', '')
-            body = str(str(str([i for i in this_run if i.startswith('APPRISE_NOTIFICATION_BODY')]).split('=')[1]).split('\\')[0]).replace('"', '')
-            
-            try:
-                websiteurl = str(str(str([i for i in this_run if i.startswith('BIRDNETPI_URL')]).split('=')[1]).split('\\')[0]).replace('"', '')
-                if len(websiteurl) == 0:
-                    raise ValueError('Blank URL')
-            except Exception as e:
-                websiteurl = "http://"+socket.gethostname()+".local"
-
-            listenurl = websiteurl+"?filename="+path
-
-        if str(str(str([i for i in this_run if i.startswith('APPRISE_NOTIFY_EACH_DETECTION')]).split('=')[1]).split('\\')[0]) == "1":
-
-            apobj = apprise.Apprise()
-            config = apprise.AppriseConfig()
-            config.add(userDir + '/BirdNET-Pi/apprise.txt')
-            apobj.add(config)
-
-            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),
-            )
-
-        if str(str(str([i for i in this_run if i.startswith('APPRISE_NOTIFY_NEW_SPECIES')]).split('=')[1]).split('\\')[0]) == "1":
-            try:
-                con = sqlite3.connect(userDir + '/BirdNET-Pi/scripts/birds.db')
-                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:
-                    apobj = apprise.Apprise()
-                    config = apprise.AppriseConfig()
-                    config.add(userDir + '/BirdNET-Pi/apprise.txt')
-                    apobj.add(config)
-                    apobj.notify(
-                        body=body.replace("$sciname", species.split("_")[0]).replace("$comname", species.split("_")[1]).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections)+1)+" times in last 7d)",
-                        title=title.replace("$sciname", species.split("_")[0]).replace("$comname", species.split("_")[1]).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections)+1)+" times in last 7d)",
-                    )
-
-                con.close()
-            except BaseException:
-                print("Database busy")
-                time.sleep(2)
-
-
 def writeResultsToFile(detections, min_conf, path):
 
     print('WRITING RESULTS TO', path, '...', end=' ')
@@ -408,13 +358,16 @@ def handle_client(conn, addr):
 
                 with open(userDir + '/BirdNET-Pi/BirdDB.txt', 'a') as rfile:
                     for d in detections:
+                        species_apprised_this_run = []
                         for entry in detections[d]:
                             if entry[1] >= min_conf and ((entry[0] in INCLUDE_LIST or len(INCLUDE_LIST) == 0)
                                                          and (entry[0] not in EXCLUDE_LIST or len(EXCLUDE_LIST) == 0)):
+                                # Write to text file.
                                 rfile.write(str(current_date) + ';' + str(current_time) + ';' + entry[0].replace('_', ';') + ';'
                                             + str(entry[1]) + ";" + str(args.lat) + ';' + str(args.lon) + ';' + str(min_conf) + ';' + str(week) + ';'
                                             + str(args.sensitivity) + ';' + str(args.overlap) + '\n')
 
+                                # Write to database
                                 Date = str(current_date)
                                 Time = str(current_time)
                                 species = entry[0]
@@ -445,8 +398,12 @@ def handle_client(conn, addr):
                                     except BaseException:
                                         print("Database busy")
                                         time.sleep(2)
-                                        
+                                
-                                sendAppriseNotifications(str(entry[0]), str(entry[1]), File_Name)
+                                # 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)
+                                
+                                species_apprised_this_run.append(str(entry[0])
 
                                 print(str(current_date) +
                                       ';' +

tests/test_apprise_notifications.py

@@ -0,0 +1,51 @@
+import os
+import sqlite3
+import pytest
+
+from scripts.notifications import sendAppriseNotifications
+from datetime import datetime, timedelta
+
+def create_test_db(db_file):
+    """ create a database connection to a SQLite database """
+    conn = None
+    try:
+        conn = sqlite3.connect(db_file)
+        sql_create_detections_table = """ CREATE TABLE IF NOT EXISTS detections (
+                                        id integer PRIMARY KEY,
+                                        Com_Name text NOT NULL,
+                                        Date date NULL,
+                                        Time time NULL
+                                    ); """
+        cur = conn.cursor()
+        cur.execute(sql_create_detections_table)
+        sql = ''' INSERT INTO detections(Com_Name, Date)
+              VALUES(?,?) '''
+        
+        cur = conn.cursor()
+        cur.execute(sql, ["Great Crested Flycatcher", datetime.now()])
+        conn.commit()
+
+    except Error as e:
+        print(e)
+    finally:
+        if conn:
+            conn.close()
+
+
+@pytest.fixture(autouse=True)
+def clean_up_after_each_test():
+    yield
+    os.remove("test.db")
+
+def test_daily_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")
+    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)"
+    )