From ad4e7a0876d36acdb4ac98778e1e449a039546ce Mon Sep 17 00:00:00 2001 From: frederik Date: Tue, 2 Jan 2024 15:49:18 +0100 Subject: [PATCH] rework analysis --- scripts/birdnet_analysis.py | 145 +++++++++++++++++++++++++++ scripts/utils/helpers.py | 110 +++++++++++++++++++++ scripts/utils/reporting.py | 189 ++++++++++++++++++++++++++++++++++++ 3 files changed, 444 insertions(+) create mode 100644 scripts/birdnet_analysis.py create mode 100644 scripts/utils/helpers.py create mode 100644 scripts/utils/reporting.py diff --git a/scripts/birdnet_analysis.py b/scripts/birdnet_analysis.py new file mode 100644 index 0000000..fd7153c --- /dev/null +++ b/scripts/birdnet_analysis.py @@ -0,0 +1,145 @@ +import logging +import os +import os.path +import re +import signal +import sys +import threading +from queue import Queue + +import inotify.adapters +from inotify.constants import IN_CLOSE_WRITE + +from utils.reporting import extract_detection, summary, write_to_file, write_to_db, apprise, bird_weather, heartbeat +from server import load_global_model, run_analysis +from utils.helpers import get_settings, ParseFileName, get_wav_files, write_settings, ANALYZING_NOW + +shutdown = False + +log = logging.getLogger(__name__) + + +def sig_handler(sig_num, curr_stack_frame): + global shutdown + log.info('Caught shutdown signal %d', sig_num) + shutdown = True + + +def main(): + write_settings() + load_global_model() + conf = get_settings() + i = inotify.adapters.Inotify() + i.add_watch(os.path.join(conf['RECS_DIR'], 'StreamData'), mask=IN_CLOSE_WRITE) + + backlog = get_wav_files() + + report_queue = Queue() + thread = threading.Thread(target=handle_reporting_queue, args=(report_queue, )) + thread.start() + + log.info('backlog is %d', len(backlog)) + for file_name in backlog: + process_file(file_name, report_queue) + if shutdown: + break + log.info('backlog done') + + empty_count = 0 + for event in i.event_gen(): + if shutdown: + break + + if event is None: + if empty_count > (conf.getint('RECORDING_LENGTH') * 2): + log.error('no more notifications: restarting...') + break + empty_count += 1 + continue + + (_, type_names, path, file_name) = event + if re.search('.wav$', file_name) is None: + continue + log.debug("PATH=[%s] FILENAME=[%s] EVENT_TYPES=%s", path, file_name, type_names) + + file_path = os.path.join(path, file_name) + if file_path in backlog: + # if we're very lucky, the first event could be for the file in the backlog that finished + # while running get_wav_files() + backlog = [] + continue + + process_file(file_path, report_queue) + empty_count = 0 + + # we're all done + report_queue.put(None) + thread.join() + report_queue.join() + + +def process_file(file_name, report_queue): + try: + if os.path.getsize(file_name) == 0: + os.remove(file_name) + return + log.info('Analyzing %s', file_name) + with open(ANALYZING_NOW, 'w') as analyzing: + analyzing.write(file_name) + file = ParseFileName(file_name) + detections = run_analysis(file) + # we join() to make sure te reporting queue does not get behind + if not report_queue.empty(): + log.warning('reporting queue not yet empty') + report_queue.join() + report_queue.put((file, detections)) + except BaseException as e: + log.exception('Unexpected error:', exc_info=e) + + +def handle_reporting_queue(queue): + while True: + msg = queue.get() + # check for signal that we are done + if msg is None: + break + + file, detections = msg + try: + for detection in detections: + detection.file_name_extr = extract_detection(file, detection) + log.info('%s;%s', summary(file, detection), os.path.basename(detection.file_name_extr)) + write_to_file(file, detection) + write_to_db(file, detection) + apprise(file, detections) + bird_weather(file, detections) + heartbeat() + os.remove(file.file_name) + except BaseException as e: + log.exception('Unexpected error:', exc_info=e) + + queue.task_done() + + # mark the 'None' signal as processed + queue.task_done() + log.info('handle_reporting_queue done') + + +def setup_logging(): + logger = logging.getLogger() + formatter = logging.Formatter("[%(name)s][%(levelname)s] %(message)s") + handler = logging.StreamHandler(stream=sys.stdout) + handler.setFormatter(formatter) + logger.addHandler(handler) + logger.setLevel(logging.INFO) + global log + log = logging.getLogger('birdnet_analysis') + + +if __name__ == '__main__': + signal.signal(signal.SIGINT, sig_handler) + signal.signal(signal.SIGTERM, sig_handler) + + setup_logging() + + main() diff --git a/scripts/utils/helpers.py b/scripts/utils/helpers.py new file mode 100644 index 0000000..ac004a6 --- /dev/null +++ b/scripts/utils/helpers.py @@ -0,0 +1,110 @@ +import configparser +import datetime +import glob +import os +import stat +import re +import subprocess +from itertools import chain + +from tzlocal import get_localzone + +_settings = None + +DB_PATH = os.path.expanduser('~/BirdNET-Pi/scripts/birds.db') +THISRUN = os.path.expanduser('~/BirdNET-Pi/scripts/thisrun.txt') +ANALYZING_NOW = os.path.expanduser('~/BirdNET-Pi/analyzing_now.txt') + + +def _load_settings(settings_path='/etc/birdnet/birdnet.conf', force_reload=False): + global _settings + if _settings is None or force_reload: + with open(settings_path) as f: + parser = configparser.ConfigParser() + # preserve case + parser.optionxform = lambda option: option + lines = chain(("[top]",), f) + parser.read_file(lines) + _settings = parser['top'] + return _settings + + +def get_settings(settings_path='/etc/birdnet/birdnet.conf', force_reload=False): + settings = _load_settings(settings_path, force_reload) + return settings + + +def write_settings(file_name=THISRUN): + settings = _load_settings() + with open(file_name, 'w') as configfile: + for key, value in settings.items(): + configfile.write(f'{key}={value}\n') + os.chmod(file_name, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IWGRP) + + +class Detection: + def __init__(self, start_time, stop_time, species, confidence): + self.start = float(start_time) + self.stop = float(stop_time) + self.confidence = float(confidence) + self.confidence_pct = round(self.confidence * 100) + self.species = species + self.scientific_name = species.split('_')[0] + self.common_name = species.split('_')[1] + self.common_name_safe = self.common_name.replace("'", "").replace(" ", "_") + self.file_name_extr = None + + +class ParseFileName: + def __init__(self, file_name): + self.file_name = file_name + name = os.path.splitext(os.path.basename(file_name))[0] + date_created = re.search('^[0-9]+-[0-9]+-[0-9]+', name).group() + time_created = re.search('[0-9]+:[0-9]+:[0-9]+$', name).group() + self.file_date = datetime.datetime.strptime(f'{date_created}T{time_created}', "%Y-%m-%dT%H:%M:%S") + self.root = name + + ident_match = re.search("RTSP_[0-9]+-", file_name) + self.RTSP_id = ident_match.group() if ident_match is not None else "" + + @property + def date(self): + current_date = self.file_date.strftime("%Y-%m-%d") + return current_date + + @property + def time(self): + current_time = self.file_date.strftime("%H:%M:%S") + return current_time + + @property + def iso8601(self): + current_iso8601 = self.file_date.astimezone(get_localzone()).isoformat() + return current_iso8601 + + @property + def week(self): + week = self.file_date.isocalendar()[1] + return week + + +def get_open_files_in_dir(dir_name): + result = subprocess.run(['lsof', '-Fn', '+D', f'{dir_name}'], check=False, capture_output=True) + ret = result.stdout.decode('utf-8') + err = result.stderr.decode('utf-8') + if err: + raise RuntimeError(f'{ret}:\n {err}') + names = [line.lstrip('n') for line in ret.splitlines() if line.startswith('n')] + return names + + +def get_wav_files(): + conf = get_settings() + files = (glob.glob(os.path.join(conf['RECS_DIR'], '*/*/*.wav')) + + glob.glob(os.path.join(conf['RECS_DIR'], 'StreamData/*.wav'))) + files.sort() + files = [os.path.join(conf['RECS_DIR'], file) for file in files] + rec_dir = os.path.join(conf['RECS_DIR'], 'StreamData') + open_recs = get_open_files_in_dir(rec_dir) + files = [file for file in files if file not in open_recs] + return files diff --git a/scripts/utils/reporting.py b/scripts/utils/reporting.py new file mode 100644 index 0000000..8e3dd74 --- /dev/null +++ b/scripts/utils/reporting.py @@ -0,0 +1,189 @@ +import datetime +import gzip +import logging +import os +import sqlite3 +import subprocess +from time import sleep + +import requests +from tzlocal import get_localzone + +from .helpers import get_settings, ParseFileName, Detection, DB_PATH +from .notifications import sendAppriseNotifications + +log = logging.getLogger(__name__) + + +def get_safe_title(title): + result = subprocess.run(['iconv', '-f', 'utf8', '-t', 'ascii//TRANSLIT'], + check=True, input=title.encode('utf-8'), capture_output=True) + ret = result.stdout.decode('utf-8') + return ret + + +def extract(in_file, out_file, start, stop): + result = subprocess.run(['sox', '-V1', f'{in_file}', f'{out_file}', 'trim', f'={start}', f'={stop}'], + check=True, capture_output=True) + ret = result.stdout.decode('utf-8') + err = result.stderr.decode('utf-8') + if err: + raise RuntimeError(f'{ret}:\n {err}') + return ret + + +def extract_safe(in_file, out_file, start, stop): + conf = get_settings() + # This section sets the SPACER that will be used to pad the audio clip with + # context. If EXTRACTION_LENGTH is 10, for instance, 3 seconds are removed + # from that value and divided by 2, so that the 3 seconds of the call are + # within 3.5 seconds of audio context before and after. + try: + ex_len = conf.getint('EXTRACTION_LENGTH') + except ValueError: + ex_len = 6 + spacer = (ex_len - 3) / 2 + safe_start = max(0, start - spacer) + safe_stop = min(conf.getint('RECORDING_LENGTH'), stop + spacer) + + extract(in_file, out_file, safe_start, safe_stop) + + +def spectrogram(in_file, title, comment, raw=False): + args = ['sox', '-V1', f'{in_file}', '-n', 'remix', '1', 'rate', '24k', 'spectrogram', + '-t', f'{get_safe_title(title)}', '-c', f'{comment}', '-o', f'{in_file}.png'] + args += ['-r'] if raw else [] + result = subprocess.run(args, check=True, capture_output=True) + ret = result.stdout.decode('utf-8') + err = result.stderr.decode('utf-8') + if err: + raise RuntimeError(f'{ret}:\n {err}') + return ret + + +def extract_detection(file: ParseFileName, detection: Detection): + conf = get_settings() + new_file_name = f'{detection.common_name_safe}-{detection.confidence_pct}-{file.root}.{conf["AUDIOFMT"]}' + new_dir = os.path.join(conf['EXTRACTED'], 'By_Date', f'{file.date}', f'{detection.common_name_safe}') + new_file = os.path.join(new_dir, new_file_name) + if os.path.isfile(new_file): + log.warning('Extraction exists. Moving on: %s', new_file) + else: + os.makedirs(new_dir, exist_ok=True) + extract_safe(file.file_name, new_file, detection.start, detection.stop) + spectrogram(new_file, detection.common_name, new_file.replace(os.path.expanduser('~/'), '')) + return new_file + + +def write_to_db(file: ParseFileName, detection: Detection): + conf = get_settings() + # Connect to SQLite Database + for attempt_number in range(3): + try: + con = sqlite3.connect(DB_PATH) + cur = con.cursor() + cur.execute("INSERT INTO detections VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (file.date, file.time, detection.scientific_name, detection.common_name, detection.confidence, + conf['LATITUDE'], conf['LONGITUDE'], conf['CONFIDENCE'], str(file.week), conf['SENSITIVITY'], + conf['OVERLAP'], os.path.basename(detection.file_name_extr))) + # (Date, Time, Sci_Name, Com_Name, str(score), + # Lat, Lon, Cutoff, Week, Sens, + # Overlap, File_Name)) + + con.commit() + con.close() + break + except BaseException as e: + log.warning("Database busy: %s", e) + sleep(2) + + +def summary(file: ParseFileName, detection: Detection): + # Date;Time;Sci_Name;Com_Name;Confidence;Lat;Lon;Cutoff;Week;Sens;Overlap + # 2023-03-03;12:48:01;Phleocryptes melanops;Wren-like Rushbird;0.76950216;-1;-1;0.7;9;1.25;0.0 + conf = get_settings() + s = (f'{file.date};{file.time};{detection.scientific_name};{detection.common_name};' + f'{detection.confidence};' + f'{conf["LATITUDE"]};{conf["LONGITUDE"]};{conf["CONFIDENCE"]};{file.week};{conf["SENSITIVITY"]};' + f'{conf["OVERLAP"]}') + return s + + +def write_to_file(file: ParseFileName, detection: Detection): + with open(os.path.expanduser('~/BirdNET-Pi/BirdDB.txt'), 'a') as rfile: + rfile.write(f'{summary(file, detection)}\n') + + +def apprise(file: ParseFileName, detections: [Detection]): + species_apprised_this_run = [] + conf = get_settings() + + for detection in detections: + # Apprise of detection if not already alerted this run. + if detection.species not in species_apprised_this_run: + try: + sendAppriseNotifications(detection.species, str(detection.confidence), str(detection.confidence_pct), + os.path.basename(detection.file_name_extr), file.date, file.time, str(file.week), + conf['LATITUDE'], conf['LONGITUDE'], conf['CONFIDENCE'], conf['SENSITIVITY'], + conf['OVERLAP'], dict(conf), DB_PATH) + except BaseException as e: + log.exception('Error during Apprise:', exc_info=e) + + species_apprised_this_run.append(detection.species) + + +def bird_weather(file: ParseFileName, detections: [Detection]): + conf = get_settings() + if conf['BIRDWEATHER_ID'] == "": + return + if detections: + # POST soundscape to server + soundscape_url = (f'https://app.birdweather.com/api/v1/stations/' + f'{conf["BIRDWEATHER_ID"]}/soundscapes?timestamp={file.iso8601}') + + with open(file.file_name, 'rb') as f: + wav_data = f.read() + gzip_wav_data = gzip.compress(wav_data) + try: + response = requests.post(url=soundscape_url, data=gzip_wav_data, timeout=30, + headers={'Content-Type': 'application/octet-stream', 'Content-Encoding': 'gzip'}) + log.info("Soundscape POST Response Status - %d", response.status_code) + sdata = response.json() + except BaseException as e: + log.error("Cannot POST soundscape: %s", e) + return + if not sdata.get('success'): + log.error(sdata.get('message')) + return + soundscape_id = sdata['soundscape']['id'] + + for detection in detections: + # POST detection to server + detection_url = f'https://app.birdweather.com/api/v1/stations/{conf["BIRDWEATHER_ID"]}/detections' + + start = file.file_date + datetime.timedelta(seconds=detection.start) + current_iso8601 = start.astimezone(get_localzone()).isoformat() + + data = {'timestamp': current_iso8601, 'lat': conf['LATITUDE'], 'lon': conf['LONGITUDE'], + 'soundscapeId': soundscape_id, + 'soundscapeStartTime': detection.start, 'soundscapeEndTime': detection.stop, + 'commonName': detection.common_name, 'scientificName': detection.scientific_name, + 'algorithm': '2p4' if conf['MODEL'] == 'BirdNET_GLOBAL_6K_V2.4_Model_FP16' else 'alpha', + 'confidence': detection.confidence} + + log.debug(data) + try: + response = requests.post(detection_url, json=data, timeout=20) + log.info("Detection POST Response Status - %d", response.status_code) + except BaseException as e: + log.error("Cannot POST detection: %s", e) + + +def heartbeat(): + conf = get_settings() + if conf['HEARTBEAT_URL']: + try: + result = requests.get(url=conf['HEARTBEAT_URL'], timeout=10) + log.info('Heartbeat: %s', result.text) + except BaseException as e: + log.error('Error during heartbeat: %s', e)