rework analysis

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()

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

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)