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

use models.py

Browse Source
This commit is contained in:
frederik
2025-11-08 15:07:15 +01:00
committed by Nachtzuster
parent 7c6caa376a
commit d2f4c6c6c0
1 changed files with 43 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/server.py
+43 -220
View File
@@ -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)