From d2f4c6c6c03fbf40fd79f6ce60f67e3dc384390f Mon Sep 17 00:00:00 2001 From: frederik Date: Sat, 8 Nov 2025 15:07:15 +0100 Subject: [PATCH] use models.py --- scripts/server.py | 263 ++++++++-------------------------------------- 1 file changed, 43 insertions(+), 220 deletions(-) diff --git a/scripts/server.py b/scripts/server.py index a72067e..a2883a0 100644 --- a/scripts/server.py +++ b/scripts/server.py @@ -1,133 +1,17 @@ import logging -import math -import operator import os import time import librosa import numpy as np -from utils.helpers import get_settings, Detection +from utils.helpers import get_settings, Detection, get_language +from utils.models import get_model -os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' -os.environ['CUDA_VISIBLE_DEVICES'] = '' -np.set_printoptions(legacy="1.21") - -try: - import tflite_runtime.interpreter as tflite -except BaseException: - from tensorflow import lite as tflite log = logging.getLogger(__name__) - -userDir = os.path.expanduser('~') -INTERPRETER, M_INTERPRETER, INCLUDE_LIST, EXCLUDE_LIST = (None, None, None, None) -PREDICTED_SPECIES_LIST = [] -WEEK = None -model, sf_thresh = (None, None) - -mdata, mdata_params = (None, None) - - -def loadModel(): - - global INPUT_LAYER_INDEX - global OUTPUT_LAYER_INDEX - global MDATA_INPUT_INDEX - global CLASSES - - log.info('LOADING TF LITE MODEL...') - - # Load TFLite model and allocate tensors. - # model will either be BirdNET_GLOBAL_6K_V2.4_Model_FP16 (new) or BirdNET_6K_GLOBAL_MODEL (old) - modelpath = userDir + '/BirdNET-Pi/model/'+model+'.tflite' - myinterpreter = tflite.Interpreter(model_path=modelpath, num_threads=2) - myinterpreter.allocate_tensors() - - # Get input and output tensors. - input_details = myinterpreter.get_input_details() - output_details = myinterpreter.get_output_details() - - # Get input tensor index - INPUT_LAYER_INDEX = input_details[0]['index'] - if model == "BirdNET_6K_GLOBAL_MODEL": - MDATA_INPUT_INDEX = input_details[1]['index'] - OUTPUT_LAYER_INDEX = output_details[0]['index'] - - # Load labels - CLASSES = [] - labelspath = userDir + '/BirdNET-Pi/model/labels.txt' - with open(labelspath, 'r') as lfile: - CLASSES = [line.strip() for line in lfile.readlines()] - - log.info('LOADING DONE!') - - return myinterpreter - - -def loadMetaModel(): - - global M_INTERPRETER - global M_INPUT_LAYER_INDEX - global M_OUTPUT_LAYER_INDEX - - if get_settings().getint('DATA_MODEL_VERSION') == 2: - data_model = 'BirdNET_GLOBAL_6K_V2.4_MData_Model_V2_FP16.tflite' - else: - data_model = 'BirdNET_GLOBAL_6K_V2.4_MData_Model_FP16.tflite' - - # Load TFLite model and allocate tensors. - M_INTERPRETER = tflite.Interpreter(model_path=os.path.join(userDir, 'BirdNET-Pi/model', data_model)) - M_INTERPRETER.allocate_tensors() - - # Get input and output tensors. - input_details = M_INTERPRETER.get_input_details() - output_details = M_INTERPRETER.get_output_details() - - # Get input tensor index - M_INPUT_LAYER_INDEX = input_details[0]['index'] - M_OUTPUT_LAYER_INDEX = output_details[0]['index'] - - log.info("loaded META model") - - -def predictFilter(lat, lon, week): - # Does interpreter exist? - if M_INTERPRETER is None: - loadMetaModel() - - # Prepare mdata as sample - sample = np.expand_dims(np.array([lat, lon, week], dtype='float32'), 0) - - # Run inference - M_INTERPRETER.set_tensor(M_INPUT_LAYER_INDEX, sample) - M_INTERPRETER.invoke() - - return M_INTERPRETER.get_tensor(M_OUTPUT_LAYER_INDEX)[0] - - -def explore(lat, lon, week): - - # Make filter prediction - l_filter = predictFilter(lat, lon, week) - - # Apply threshold - l_filter = np.where(l_filter >= float(sf_thresh), l_filter, 0) - - # Zip with labels - l_filter = list(zip(l_filter, CLASSES)) - - # Sort by filter value - l_filter = sorted(l_filter, key=lambda x: x[0], reverse=True) - - return l_filter - - -def set_predicted_species_list(lat, lon, week): - global PREDICTED_SPECIES_LIST - l_filter = explore(lat, lon, week) - PREDICTED_SPECIES_LIST = [s[1].split('_')[0] for s in l_filter if s[0] >= sf_thresh] +MODEL = None def loadCustomSpeciesList(path): @@ -140,7 +24,6 @@ def loadCustomSpeciesList(path): def splitSignal(sig, rate, overlap, seconds=3.0, minlen=1.5): - # Split signal with overlap sig_splits = [] for i in range(0, len(sig), int((seconds - overlap) * rate)): @@ -161,86 +44,33 @@ def splitSignal(sig, rate, overlap, seconds=3.0, minlen=1.5): return sig_splits -def readAudioData(path, overlap, sample_rate=48000): - +def readAudioData(path, overlap, sample_rate, chunk_duration): log.info('READING AUDIO DATA...') # Open file with librosa (uses ffmpeg or libav) sig, rate = librosa.load(path, sr=sample_rate, mono=True, res_type='kaiser_fast') - # Split audio into 3-second chunks - chunks = splitSignal(sig, rate, overlap) + # Split audio into chunks + chunks = splitSignal(sig, rate, overlap, seconds=chunk_duration) log.info('READING DONE! READ %d CHUNKS.', len(chunks)) return chunks -def convertMetadata(m): - - # Convert week to cosine - if m[2] >= 1 and m[2] <= 48: - m[2] = math.cos(math.radians(m[2] * 7.5)) + 1 - else: - m[2] = -1 - - # Add binary mask - mask = np.ones((3,)) - if m[0] == -1 or m[1] == -1: - mask = np.zeros((3,)) - if m[2] == -1: - mask[2] = 0.0 - - return np.concatenate([m, mask]) - - -def custom_sigmoid(x, sensitivity=1.0): - return 1 / (1.0 + np.exp(-sensitivity * x)) - - -def predict(sample, sensitivity): - # Make a prediction - INTERPRETER.set_tensor(INPUT_LAYER_INDEX, np.array(sample[0], dtype='float32')) - if model == "BirdNET_6K_GLOBAL_MODEL": - INTERPRETER.set_tensor(MDATA_INPUT_INDEX, np.array(sample[1], dtype='float32')) - INTERPRETER.invoke() - prediction = INTERPRETER.get_tensor(OUTPUT_LAYER_INDEX)[0] - - # Apply custom sigmoid - p_sigmoid = custom_sigmoid(prediction, sensitivity) - - # Get label and scores for pooled predictions - p_labels = dict(zip(CLASSES, p_sigmoid)) - - # Sort by score - p_sorted = sorted(p_labels.items(), key=operator.itemgetter(1), reverse=True) - - return p_sorted - - -def analyzeAudioData(chunks, lat, lon, week, sens, overlap,): - global WEEK - - sensitivity = max(0.5, min(1.0 - (sens - 1.0), 1.5)) - +def analyzeAudioData(chunks, overlap, lat, lon, week): detections = [] + model = load_global_model() + start = time.time() log.info('ANALYZING AUDIO...') - if model == "BirdNET_GLOBAL_6K_V2.4_Model_FP16": - if week != WEEK or len(INCLUDE_LIST) != 0: - WEEK = week - set_predicted_species_list(lat, lon, week) - - mdata = get_metadata(lat, lon, week) + model.set_meta_data(lat, lon, week) + predicted_species_list = model.get_species_list() # Parse every chunk - for c in chunks: - # Prepare as input signal - sig = np.expand_dims(c, 0) - - # Make prediction - p = predict([sig, mdata], sensitivity) + for chunk in chunks: + p = model.predict(chunk) log.debug("PPPPP: %s", p) detections.append(p) @@ -248,13 +78,13 @@ def analyzeAudioData(chunks, lat, lon, week, sens, overlap,): pred_start = 0.0 for p in filter_humans(detections): # Save timestamp and result - pred_end = pred_start + 3.0 + pred_end = pred_start + model.chunk_duration labeled[str(pred_start) + ';' + str(pred_end)] = p pred_start = pred_end - overlap log.info('DONE! Time %.2f SECONDS', time.time() - start) - return labeled + return labeled, predicted_species_list def filter_humans(detections): @@ -307,62 +137,55 @@ def filter_humans(detections): return clean_detections -def get_metadata(lat, lon, week): - global mdata, mdata_params - if mdata_params != [lat, lon, week]: - mdata_params = [lat, lon, week] - # Convert and prepare metadata - mdata = convertMetadata(np.array([lat, lon, week])) - mdata = np.expand_dims(mdata, 0) - - return mdata - - def load_global_model(): - global INTERPRETER - global model, sf_thresh - conf = get_settings() - model = conf['MODEL'] - sf_thresh = conf.getfloat('SF_THRESH') - INTERPRETER = loadModel() + global MODEL + if MODEL is None: + log.info('LOADING TF LITE MODEL...') + MODEL = get_model() + log.info('LOADING DONE!') + + return MODEL def run_analysis(file): - global INCLUDE_LIST, EXCLUDE_LIST, WHITELIST_LIST - INCLUDE_LIST = loadCustomSpeciesList(os.path.expanduser("~/BirdNET-Pi/include_species_list.txt")) - EXCLUDE_LIST = loadCustomSpeciesList(os.path.expanduser("~/BirdNET-Pi/exclude_species_list.txt")) - WHITELIST_LIST = loadCustomSpeciesList(os.path.expanduser("~/BirdNET-Pi/whitelist_species_list.txt")) + include_list = loadCustomSpeciesList(os.path.expanduser("~/BirdNET-Pi/include_species_list.txt")) + exclude_list = loadCustomSpeciesList(os.path.expanduser("~/BirdNET-Pi/exclude_species_list.txt")) + whitelist_list = loadCustomSpeciesList(os.path.expanduser("~/BirdNET-Pi/whitelist_species_list.txt")) conf = get_settings() + model = load_global_model() + names = get_language(conf['DATABASE_LANG']) # Read audio data & handle errors try: - audio_data = readAudioData(file.file_name, conf.getfloat('OVERLAP')) + audio_data = readAudioData(file.file_name, conf.getfloat('OVERLAP'), model.sample_rate, model.chunk_duration) except (NameError, TypeError) as e: log.error("Error with the following info: %s", e) return [] # Process audio data and get detections - raw_detections = analyzeAudioData(audio_data, conf.getfloat('LATITUDE'), conf.getfloat('LONGITUDE'), file.week, - conf.getfloat('SENSITIVITY'), conf.getfloat('OVERLAP')) + raw_detections, predicted_species_list = analyzeAudioData(audio_data, conf.getfloat('OVERLAP'), conf.getfloat('LATITUDE'), + conf.getfloat('LONGITUDE'), file.week) confident_detections = [] for time_slot, entries in raw_detections.items(): - log.info('%s-%s', time_slot, entries[0]) - for species, confidence in entries: + sci_name, confidence = entries[0] + log.info('%s-(%s_%s, %s)', time_slot, sci_name, names.get(sci_name, sci_name), confidence) + for sci_name, confidence in entries: if confidence >= conf.getfloat('CONFIDENCE'): - sci_name = species.split('_')[0] - if sci_name not in INCLUDE_LIST and len(INCLUDE_LIST) != 0: - log.warning("Excluded as INCLUDE_LIST is active but this species is not in it: %s", species) - elif sci_name in EXCLUDE_LIST and len(EXCLUDE_LIST) != 0: - log.warning("Excluded as species in EXCLUDE_LIST: %s", species) - elif sci_name not in PREDICTED_SPECIES_LIST and len(PREDICTED_SPECIES_LIST) != 0 and sci_name not in WHITELIST_LIST: - log.warning("Excluded as below Species Occurrence Frequency Threshold: %s", species) + com_name = names.get(sci_name, sci_name) + if sci_name not in include_list and len(include_list) != 0: + log.warning("Excluded as INCLUDE_LIST is active but this species is not in it: %s %s", sci_name, com_name) + elif sci_name in exclude_list and len(exclude_list) != 0: + log.warning("Excluded as species in EXCLUDE_LIST: %s %s", sci_name, com_name) + elif sci_name not in predicted_species_list and len(predicted_species_list) != 0 and sci_name not in whitelist_list: + log.warning("Excluded as below Species Occurrence Frequency Threshold: %s %s", sci_name, com_name) else: d = Detection( file.file_date, time_slot.split(';')[0], time_slot.split(';')[1], - species, + sci_name, + com_name, confidence, ) confident_detections.append(d) @@ -392,7 +215,7 @@ if __name__ == '__main__': {"confidence": 0.9317, 'sci_name': 'Pica pica'}, {"confidence": 0.8861, 'sci_name': 'Pica pica'}], }] - load_global_model() + for sample, expected in zip(test_files, results): file = ParseFileName(os.path.expanduser(sample)) detections = run_analysis(file)