avm/hassio-addons
1
0
Fork 0
mirror of https://github.com/alexbelgium/hassio-addons.git synced 2026-01-11 10:21:02 +01:00
Code Issues Packages Projects Releases Wiki Activity

Update birdnet_to_mqtt.py

Browse Source
This commit is contained in:
Alexandre
2024-08-27 17:21:27 +02:00
committed by GitHub
parent 9a448996e4
commit f153d6fabc
1 changed files with 69 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
battybirdnet-pi/rootfs/helpers/birdnet_to_mqtt.py
View File

@@ -1,124 +1,108 @@
#! /usr/bin/env python3
# birdnet_to_mqtt.py
#
# Adapted from : https://gist.github.com/deepcoder/c309087c456fc733435b47d83f4113ff
# Adapted from : https://gist.github.com/JuanMeeske/08b839246a62ff38778f701fc1da5554
#
# monitor the records in the syslog file for info from the birdnet system on birds that it detects
# publish this data to mqtt
#
import time
import re
import dateparser
import datetime
import json
import logging
import paho.mqtt.client as mqtt
import subprocess
import requests
import sys
import os
sys.path.append('/home/pi/BirdNET-Pi/scripts/utils')
from helpers import get_settings
# Setup basic configuration for logging
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
# this generator function monitors the requested file handle for new lines added at its end
# the newly added line is returned by the function
def file_row_generator(s):
while True :
line = s.readline()
if not line:
time.sleep(0.1)
continue
yield line
# Used in flickrimage
flickr_images = {}
conf = get_settings()
settings_dict = dict(conf)
# mqtt server
mqtt_server = "%%mqtt_server%%" # server for mqtt
mqtt_user = "%%mqtt_user%%" # Replace with your MQTT username
mqtt_pass = "%%mqtt_pass%%" # Replace with your MQTT password
mqtt_port = %%mqtt_port%% # port for mqtt
# mqtt topic for bird heard above threshold will be published
mqtt_topic_confident_birds = 'birdnet'
# url base for website that will be used to look up info about bird
# MQTT server configuration
mqtt_server = "%%mqtt_server%%"
mqtt_user = "%%mqtt_user%%"
mqtt_pass = "%%mqtt_pass%%"
mqtt_port = %%mqtt_port%%
mqtt_topic = 'birdnet'
bird_lookup_url_base = 'http://en.wikipedia.org/wiki/'
# regular expression patters used to decode the records from birdnet
re_high_clean = re.compile(r'(?<=^\[birdnet_analysis\]\[INFO\] ).*?(?=\.mp3$)')
syslog = open('/proc/1/fd/1', 'r')
def on_connect(client, userdata, flags, rc, properties=None):
def on_connect(client, userdata, flags, rc ): #, properties=None):
""" Callback for when the client receives a CONNACK response from the server. """
if rc == 0:
logging.info("Connected to MQTT Broker!")
log.info("Connected to MQTT Broker!")
else:
logging.error(f"Failed to connect, return code {rc}\n")
log.error(f"Failed to connect, return code {rc}\n")
def get_bird_code(scientific_name):
with open('/home/pi/BirdNET-Pi/scripts/ebird.php', 'r') as file:
data = file.read()
# Extract the array from the PHP file
array_str = re.search(r'\$ebirds = \[(.*?)\];', data, re.DOTALL).group(1)
# Convert the PHP array to a Python dictionary
bird_dict = {re.search(r'"(.*?)"', line).group(1): re.search(r'=> "(.*?)"', line).group(1)
for line in array_str.split('\n') if '=>' in line}
# Return the corresponding value for the given bird's scientific name
return bird_dict.get(scientific_name)
# this little hack is to make each received record for the all birds section unique
# the date and time that the log returns is only down to the 1 second accuracy, do
# you can get multiple records with same date and time, this will make Home Assistant not
# think there is a new reading so we add a incrementing tenth of second to each record received
ts_noise = 0.0
def automatic_mqtt_publish(file, detection, path):
bird = {}
bird['Date'] = file.date
bird['Time'] = file.time
bird['ScientificName'] = detection.scientific_name.replace('_', ' ')
bird['CommonName'] = detection.common_name
bird['Confidence'] = detection.confidence
bird['SpeciesCode'] = get_bird_code(detection.scientific_name)
bird['ClipName'] = path
bird['url'] = bird_lookup_url_base + detection.scientific_name.replace(' ', '_')
#try :
# connect to MQTT server
mqttc = mqtt.Client('birdnet_mqtt') # Create instance of client with client ID
mqttc.username_pw_set(mqtt_user, mqtt_pass) # Use credentials
mqttc.connect(mqtt_server, mqtt_port) # Connect to (broker, port, keepalive-time)
mqttc.on_connect = on_connect
mqttc.loop_start()
# Flickimage
image_url = ""
common_name = detection.common_name
if len(settings_dict.get('FLICKR_API_KEY')) > 0:
if common_name not in flickr_images:
try:
headers = {'User-Agent': 'Python_Flickr/1.0'}
url = ('https://www.flickr.com/services/rest/?method=flickr.photos.search&api_key=' + str(settings_dict.get('FLICKR_API_KEY')) +
'&text=' + str(common_name) + ' bird&sort=relevance&per_page=5&media=photos&format=json&license=2%2C3%2C4%2C5%2C6%2C9&nojsoncallback=1')
resp = requests.get(url=url, headers=headers, timeout=10)
# call the generator function and process each line that is returned
for row in file_row_generator(syslog):
# bird found above confidence level found, process it
if re_high_clean.search(row) :
resp.encoding = "utf-8"
data = resp.json()["photos"]["photo"][0]
# this slacker regular expression work, extracts the data about the bird found from the log line
# I do the parse in two passes, because I did not know the re to do it in one!
image_url = 'https://farm'+str(data["farm"])+'.static.flickr.com/'+str(data["server"])+'/'+str(data["id"])+'_'+str(data["secret"])+'_n.jpg'
flickr_images[common_name] = image_url
except Exception as e:
print("FLICKR API ERROR: "+str(e))
image_url = ""
else:
image_url = flickr_images[common_name]
raw_high_bird = re.search(re_high_clean, row)
raw_high_bird = raw_high_bird.group(0)
bird['FlickrImage'] = image_url
# the fields we want are separated by semicolons, so split
high_bird_fields = raw_high_bird.split(';')
# build a structure in python that will be converted to json
bird = {}
# human time in this record is in two fields, date and time. They are human format
# combine them together separated by a space and they turn the human data into a python
# timestamp
raw_ts = high_bird_fields[0] + ' ' + high_bird_fields[1]
#bird['ts'] = str(datetime.datetime.timestamp(dateparser.parse(raw_ts)))
bird['Date'] = high_bird_fields[0]
bird['Time'] = high_bird_fields[1]
bird['ScientificName'] = high_bird_fields[2]
bird['CommonName'] = high_bird_fields[3]
bird['Confidence'] = high_bird_fields[4]
bird['SpeciesCode'] = get_bird_code(high_bird_fields[2])
bird['ClipName'] = high_bird_fields[11]
# build a url from scientific name of bird that can be used to lookup info about bird
bird['url'] = bird_lookup_url_base + high_bird_fields[2].replace(' ', '_')
# convert to json string we can sent to mqtt
json_bird = json.dumps(bird)
mqttc.reconnect()
mqttc.publish(mqtt_topic, json_bird, 1)
log.info("Posted to MQTT: ok")
print('Posted to MQTT : ok')
mqttc = mqtt.Client('birdnet_mqtt')
mqttc.username_pw_set(mqtt_user, mqtt_pass)
mqttc.on_connect = on_connect
mqttc.publish(mqtt_topic_confident_birds, json_bird, 1)
try:
mqttc.connect(mqtt_server, mqtt_port)
mqttc.loop_start()
# Assuming `file` and `detections` are provided from somewhere
# automatic_mqtt_publish(file, detections)
except Exception as e:
log.error("Cannot post mqtt: %s", e)
finally:
mqttc.loop_stop()
mqttc.disconnect()