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

add db

Browse Source
This commit is contained in:
frederik
2025-10-19 11:22:44 +02:00
parent f2a2214c6f
commit 771855812f
3 changed files with 89 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/utils/db.py
+82
View File
@@ -0,0 +1,82 @@
import sqlite3
import time as timeim
from datetime import datetime
from .helpers import DB_PATH
_DB = None
def get_db():
global _DB
if _DB is None:
con = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
con.row_factory = sqlite3.Row
_DB = con
return _DB
def get_records(select_sql):
con = get_db()
try:
cur = con.execute(select_sql)
records = cur.fetchall()
except sqlite3.Error as e:
print(e)
timeim.sleep(2)
records = []
return records
def get_record(select_sql):
records = get_records(select_sql)
return dict(records[0]) if records else None
def get_latest():
select_sql = "SELECT * FROM detections ORDER BY Date DESC, Time DESC LIMIT 1"
return get_record(select_sql)
def get_todays_count_for(sci_name):
today = datetime.now().strftime("%Y-%m-%d")
select_sql = f"SELECT COUNT(*) FROM detections WHERE Date = DATE('{today}') AND Sci_Name = '{sci_name}'"
records = get_records(select_sql)
return records[0][0] if records else 0
def get_this_weeks_count_for(sci_name):
today = datetime.now().strftime("%Y-%m-%d")
select_sql = f"SELECT COUNT(*) FROM detections WHERE Date >= DATE('{today}', '-7 day') AND Sci_Name = '{sci_name}'"
records = get_records(select_sql)
return records[0][0] if records else 0
def get_summary():
total_count = get_record("SELECT COUNT(*) as total_count FROM detections")
todays_count = get_record("SELECT COUNT(*) as todays_count FROM detections WHERE Date == DATE('now', 'localtime')")
hour_count = get_record("SELECT COUNT(*) as hour_count FROM detections "
"WHERE Date == Date('now', 'localtime') AND TIME >= TIME('now', 'localtime', '-1 hour')")
todays_species_tally = get_record("SELECT COUNT(DISTINCT(Sci_Name)) as todays_species_tally FROM detections WHERE Date == Date('now','localtime')")
species_tally = get_record("SELECT COUNT(DISTINCT(Sci_Name)) as species_tally FROM detections")
summary = {**total_count, **todays_count, **hour_count, **todays_species_tally, **species_tally}
return summary
def get_species_by(sort_by=None, date=None):
where = "" if date is None else f'WHERE Date == "{date}"'
if sort_by == "occurrences":
select_sql = (f"SELECT Date, Time, File_Name, Com_Name, Sci_Name, COUNT(*) as Count, MAX(Confidence) as MaxConfidence "
f"FROM detections {where} GROUP BY Sci_Name ORDER BY COUNT(*) DESC;")
elif sort_by == "confidence":
select_sql = (f"SELECT Date, Time, File_Name, Com_Name, Sci_Name, COUNT(*) as Count, MAX(Confidence) as MaxConfidence "
f"FROM detections {where} GROUP BY Sci_Name ORDER BY MAX(Confidence) DESC;")
elif sort_by == "date":
select_sql = (f"SELECT Date, Time, File_Name, Com_Name, Sci_Name, COUNT(*) as Count, MAX(Confidence) as MaxConfidence "
f"FROM detections {where} GROUP BY Sci_Name ORDER BY MIN(Date) DESC, Time DESC;")
else:
select_sql = (f"SELECT Date, Time, File_Name, Com_Name, Sci_Name, COUNT(*) as Count, MAX(Confidence) as MaxConfidence "
f"FROM detections {where} GROUP BY Sci_Name ORDER BY Com_Name ASC;")
records = get_records(select_sql)
return records
scripts/utils/notifications.py
+5 -35
View File
@@ -1,12 +1,12 @@
import apprise
import os
import socket
import sqlite3
from datetime import datetime
import requests
import html
import time as timeim
from .db import get_todays_count_for, get_this_weeks_count_for
userDir = os.path.expanduser('~')
APPRISE_CONFIG = userDir + '/BirdNET-Pi/apprise.txt'
APPRISE_BODY = userDir + '/BirdNET-Pi/body.txt'
@@ -44,9 +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, db_path=DB_PATH):
def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path, date, time, week, latitude, longitude, cutoff, sens, overlap, settings_dict):
def render_template(template, reason=""):
ret = template.replace("$sciname", sci_name) \
.replace("$comname", com_name) \
@@ -124,7 +122,7 @@ def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path
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(db_path, sci_name)
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")
@@ -133,7 +131,7 @@ def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path
species_last_notified[com_name] = int(timeim.time())
if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES') == "1":
numberDetections = get_this_weeks_count_for(db_path, sci_name)
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)
@@ -142,33 +140,5 @@ def sendAppriseNotifications(sci_name, com_name, confidence, confidencepct, path
species_last_notified[com_name] = int(timeim.time())
def get_todays_count_for(db_path, sci_name):
today = datetime.now().strftime("%Y-%m-%d")
select_sql = f"SELECT COUNT(*) FROM detections WHERE Date = DATE('{today}') AND Sci_Name = '{sci_name}'"
records = get_records(db_path, select_sql)
return records[0][0] if records else 0
def get_this_weeks_count_for(db_path, sci_name):
today = datetime.now().strftime("%Y-%m-%d")
select_sql = f"SELECT COUNT(*) FROM detections WHERE Date >= DATE('{today}', '-7 day') AND Sci_Name = '{sci_name}'"
records = get_records(db_path, select_sql)
return records[0][0] if records else 0
def get_records(db_path, select_sql):
try:
con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
cur = con.cursor()
cur.execute(select_sql)
records = cur.fetchall()
con.close()
except sqlite3.Error as e:
print(f"Database busy: {e}")
timeim.sleep(2)
records = []
return records
if __name__ == "__main__":
print("notfications")
scripts/utils/reporting.py
+2 -1
View File
@@ -160,7 +160,8 @@ def apprise(file: ParseFileName, detections: [Detection]):
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), DB_PATH)
conf['OVERLAP'], dict(conf))
except BaseException as e:
log.exception('Error during Apprise:', exc_info=e)