Initial commit

python/mqtt-systray.py

@@ -0,0 +1,234 @@
+import gi
+import tempfile
+import time
+import os
+import signal
+
+import paho.mqtt.client as mqtt
+
+gi.require_version('Gtk', '3.0')
+from gi.repository import Gtk
+
+gi.require_version('AppIndicator3', '0.1')
+from gi.repository import AppIndicator3 as AppIndicator
+
+from PIL import Image, ImageDraw, ImageFont
+import yaml
+
+mqttprefix = "devices"
+mqtthost = '192.168.67.1'
+
+oldtmpfile = ''
+oldalertcount = 0
+menu = Gtk.Menu()
+
+fp = open("mqtt-systray.yaml")
+conf = yaml.safe_load(fp)
+fp.close()
+
+appindicator = AppIndicator.Indicator.new(
+            'mqttalert',
+            '',
+            AppIndicator.IndicatorCategory.APPLICATION_STATUS)
+
+def create_image(width, height, count):
+    # Generate an image and draw a pattern
+    image = Image.new('RGB', (width, height), 'white')
+    dc = ImageDraw.Draw(image)
+    color_circle = 'green'
+    if count > 0: color_circle = 'red'
+    #dc.circle((width // 2, height // 2), height // 2, fill=color2)
+    radius = height // 2
+    ellipse_xy = (0, 0, radius*2, radius*2)
+    dc.ellipse(ellipse_xy, color_circle)
+
+    if count > 0:
+      #font = ImageFont.truetype("OpenSans-Regular.ttf", int(width))
+      font = ImageFont.load_default(float(width))
+      dc.text((12, -12),str(count),(0,0,0),font=font)
+
+    return image
+
+def quit():
+    os.remove(oldtmpfile)
+    Gtk.main_quit()
+
+def signal_handler(sig, frame):
+    print('You pressed Ctrl+C!')
+    quit()
+
+def gtkquit(source):
+  quit()
+
+def get_alert(devicea):
+  for sensor in conf[devicea]['values'].keys():
+    if 'value' in conf[devicea]['values'][sensor] and (conf[devicea]['values'][sensor]['value'] > conf[devicea]['values'][sensor]['max'] or conf[devicea]['values'][sensor]['value'] < conf[devicea]['values'][sensor]['min']):
+      return True
+  return False
+
+def get_alert2(devicea, sensora):
+    alert_text = ''
+    if 'value' in conf[devicea]['values'][sensora]:
+      if conf[devicea]['values'][sensora]['value'] > conf[devicea]['values'][sensora]['max']:
+        alert_text = '!!HIGH!! '
+      elif conf[devicea]['values'][sensora]['value'] < conf[devicea]['values'][sensora]['min']:
+        alert_text = '!!LOW!! '
+    return alert_text
+
+def alert_count():
+  count = 0
+  for device in conf:
+    for sensor in conf[device]['values'].keys():
+      if get_alert2(device, sensor) != '': count = count + 1
+  return count
+  
+def get_icon():
+  global oldtmpfile
+  if oldtmpfile!='': os.remove(oldtmpfile)
+  icon_path = tempfile.mktemp()
+  oldtmpfile=icon_path
+  with open(icon_path, 'wb') as f:
+     create_image(64, 64, alert_count()).save(f, 'PNG')
+  return icon_path
+
+#def menun2(deviceg):
+#  itemn2 = []
+#  submenu = Gtk.Menu()
+#  for sensor in conf[deviceg]['values'].keys():
+#    text=''
+#    if 'value' in conf[deviceg]['values'][sensor]:
+#      alert_text = get_alert2(deviceg, sensor)
+#      text='{}{}: {}'.format(alert_text, sensor, conf[deviceg]['values'][sensor]['value'])
+#    else:
+#      text=sensor
+#    submenuitem = Gtk.MenuItem(text)
+#    submenu.append(submenuitem)
+#  return submenu
+
+def build_menu():
+   menu = Gtk.Menu()
+   for device1 in conf:
+      devname = conf[device1]['name']
+      conf[device1]['menuitem'] = Gtk.MenuItem(devname)
+      submenu = Gtk.Menu()
+      for sensor in conf[device1]['values'].keys():
+         conf[device1]['values'][sensor]['menuitem'] = Gtk.MenuItem(sensor)
+         submenu.append(conf[device1]['values'][sensor]['menuitem'])
+      conf[device1]['menuitem'].set_submenu(submenu)
+      menu.append(conf[device1]['menuitem'])
+
+   item_quit = Gtk.MenuItem('Quit')
+   item_quit.connect('activate', gtkquit)
+   menu.append(item_quit)
+      
+   menu.show_all()
+   return menu
+      
+
+#def build_menu_old():
+#    global menu
+#    global quitcount
+#    global item_quit
+#    #menu = Gtk.Menu()
+#    menu.connect('popup-menu', menu_show)
+#    #menu.connect('destroy', menu_hide)
+#    for device1 in conf:
+#      devname = conf[device1]['name']
+#      if get_alert(device1): devname = "!!ALERT!! " + devname
+#      menuitem = Gtk.MenuItem(devname)
+#      menuitem.set_submenu(menun2(device1))
+#      menu.append(menuitem)
+#      
+#    if quitcount == 0:
+#      item_quit = Gtk.MenuItem('Quit')
+#      item_quit.connect('select', menu_show)
+#      print('Quit0')
+#      menu.append(item_quit)
+#      quitcount = quitcount + 1
+#    else:
+#      #item_quit = Gtk.MenuItem('Quit')
+#      item_quit.set_label(str(quitcount))
+#      print('QuitX')
+#      #menu.append(item_quit)
+#      #item_quit.show()
+#      quitcount = quitcount + 1
+#
+#    print("Alert Count: {}".format(alert_count()))
+#
+#    menu.show_all()
+#    #menu.show_now()
+#    return menu
+
+def update_menu():
+  appindicator.set_menu(build_menu())
+
+def set_value(topic, value):
+  global oldalertcount
+  topicds = topic.rsplit("/", 1)
+  devices=topicds[0]
+  devices = devices.removeprefix(mqttprefix+'/')
+  conf[devices]['values'][topicds[1]]['value'] = float(value)
+  print("menu")
+  #update_menu()
+  
+  alert_text = get_alert2(devices, topicds[1])
+  text='{}{}: {}'.format(alert_text, topicds[1], conf[devices]['values'][topicds[1]]['value'])
+  conf[devices]['values'][topicds[1]]['menuitem'].set_label(text)
+  #conf[devices]['values'][topicds[1]]['menuitem'].set_label("{}".format(conf[devices]['values'][topicds[1]]['value']))
+  devname = conf[devices]['name']
+  #print(get_alert(devices))
+  #if get_alert(devices): devname = "! " + devname
+  #if get_alert(devices): devname = "!!ALERT!! " + devname
+  #conf[devices]['menuitem'].set_label(devname) 
+  
+  if oldalertcount != alert_count():
+    oldalertcount = alert_count()
+    appindicator.set_icon(get_icon())
+    os.system("notify-send '{}' '{}'".format(conf[devices]['name'], text))
+
+def subscribe_mqtt(mqttc):
+  for device in conf:
+    #print(conf[device])
+    for sensor in conf[device]['values'].keys():
+      #print(sensor)
+      tosubscribe = "{}/{}/{}".format(mqttprefix, device, sensor)
+      mqttc.subscribe(tosubscribe, 0)
+      print(tosubscribe)
+
+def on_connect(mqttc, obj, flags, rc):
+    print("rc: " + str(rc))
+#def on_subscribe(mqttc, obj, mid, granted_qos):
+#    print("Subscribed: " + str(mid) + " " + str(granted_qos))
+def on_message(client, userdata, msg):
+  print(msg.topic+" "+str(msg.payload))
+  set_value(msg.topic, msg.payload)
+
+
+signal.signal(signal.SIGINT, signal_handler)
+
+appindicator.set_icon(get_icon())
+appindicator.set_title('MQTT Alert')
+appindicator.set_status(AppIndicator.IndicatorStatus.ACTIVE)
+appindicator.set_menu(build_menu())
+
+set_value("model1/1/1/temperature1C", "10.1")
+print(conf)
+
+mqttc = mqtt.Client()
+mqttc.on_message = on_message
+mqttc.on_connect = on_connect
+#mqttc.on_subscribe = on_subscribe
+connected = False
+while connected == False:
+  try:
+    mqttc.connect(mqtthost, 1883, 60)
+  except:
+    time.sleep(5)
+    connected=False
+  else:
+    connected=True
+mqttc.loop_start()
+
+subscribe_mqtt(mqttc)
+
+Gtk.main()

python/mqtt-systray.yaml

@@ -0,0 +1,22 @@
+model1/1/1:
+  name: "toto"
+  values:
+    temperature1C:
+      name: "Temperature"
+      min: 5
+      max: 20
+    humidity1:
+      name: "Humidity"
+      min: 30
+      max: 70
+model2/2/2:
+  name: "tata"
+  values:
+    temperature2C:
+      name: "Temperature"
+      min: 5
+      max: 20
+    humidity2:
+      name: "Humidity"
+      min: 30
+      max: 70

python/rtl_433_json2mqtt.py

@@ -0,0 +1,65 @@
+import json
+#import fileinput
+import serial
+
+import paho.mqtt.client as paho
+
+#input = '{"model":"Oregon-THGR810","id":226,"channel":10,"battery_ok":0,"temperature_C":19.7,"humidity":24,"protocol":"Oregon Scientific Weather Sensor"}'
+SKIP_KEYS = [ "type", "model", "subtype", "channel", "id", "mic", "mod", "freq", "sequence_num", "message_type", "exception", "raw_msg", "protocol", "duration", "sync", "flags", "status" ]
+TOPIC_KEYS = [ "type", "model", "subtype", "channel", "id" ]
+
+#default="devices[/type][/model][/subtype][/channel][/id]"
+prefix = "home/rtl_433"
+
+broker = "localhost"
+port = 1883
+
+serialdev = "/dev/ttyACM0"
+serialspeed = 115200
+
+
+def generate_topic(jsonin):
+  topic = prefix+'/devices'
+  for t in TOPIC_KEYS:
+    if t in jsonin:
+      topic += '/' + str(jsonin[t])
+  return topic
+
+def publish(jsonin, prefix_device):
+  for t in jsonin:
+    if t not in SKIP_KEYS:
+      topic = prefix_device + '/' + t
+      value = jsonin[t]
+      #print("{} {}".format(topic, value))
+      mqtt.publish(topic, value) 
+
+def on_connect(client, userdata, flags, reason_code, properties):
+  print(f"Connected to MQTT")
+  client.subscribe(prefix+"/cmd")
+
+def on_message(client, userdata, msg):
+  print(msg.topic+" "+str(msg.payload))
+  if(str(msg.topic) == prefix+"/cmd"):
+    ser.write(msg.payload)
+
+mqtt=paho.Client(paho.CallbackAPIVersion.VERSION2)
+mqtt.on_connect = on_connect
+mqtt.on_message = on_message
+mqtt.connect(broker,port)
+mqtt.loop_start()
+
+ser = serial.Serial(serialdev, serialspeed)
+
+#for input in fileinput.input():
+while True:
+  input = ser.readline()
+  try:
+    data = json.loads(input)
+  except json.decoder.JSONDecodeError:
+    print("Error JSON, received {}".format(input))
+    continue 
+  #print(data)
+  mqtt.publish(prefix+'/events', input.rstrip()) 
+  prefix_device = generate_topic(data)
+  publish(data, prefix_device)
+

python/rtl_433_mqtt2influxdb.py

@@ -0,0 +1,133 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2010-2013 Roger Light <roger@atchoo.org>
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Eclipse Distribution License v1.0
+# which accompanies this distribution.
+#
+# The Eclipse Distribution License is available at
+#   http://www.eclipse.org/org/documents/edl-v10.php.
+#
+# Contributors:
+#    Roger Light - initial implementation
+# Copyright (c) 2010,2011 Roger Light <roger@atchoo.org>
+# All rights reserved.
+
+# This shows a simple example of an MQTT subscriber.
+
+import signal
+import sys
+import json
+import datetime
+import paho.mqtt.client as mqtt
+from influxdb import InfluxDBClient
+
+def signal_handler(sig, frame):
+    influxc.close()
+    mqttc.disconnect()
+    sys.exit(0)
+
+prefix = 'home/rtl_433'
+#prefix = 'rtl_433/fedora'
+TAGS_KEYS = ['type','model','subtype','channel','id']
+
+DBNAME = "rtl433"
+DELAYRECORD=300
+recorddb = {}
+
+def on_connect(mqttc, obj, flags, rc):
+    print("rc: " + str(rc))
+
+
+def on_message(mqttc, obj, msg):
+    #print(msg.topic + " " + str(msg.qos) + " " + str(msg.payload))
+    currentdt=datetime.datetime.now(datetime.UTC)
+    dateinflux=currentdt.strftime("%Y-%m-%dT%H:%M:%SZ")
+    jsonin=json.loads(msg.payload)
+    ref=""
+    if "model" in jsonin:
+        ref+=str(jsonin["model"])
+    if "channel" in jsonin:
+        ref+=str(jsonin["channel"])
+    if "id" in jsonin:
+        ref+=str(jsonin["id"])
+    if ref not in recorddb:
+        recorddb[ref] = {}
+        recorddb[ref]["lastsend"]=0
+    #print(jsonin)
+    tags = {}
+    for t in TAGS_KEYS:
+        if t in jsonin:
+            tags[t] = jsonin[t]
+    points = []
+    fields = {}
+
+    changed = False
+    if 'temperature_C' in jsonin:
+        fields = {'value': float(jsonin['temperature_C'])}
+        measurement = 'temperature_C'
+        points.append({'measurement': measurement, 'tags': tags, 'time': dateinflux, 'fields': fields})
+        if 'temperature_C' not in recorddb[ref] or recorddb[ref]['temperature_C'] != fields["value"]:
+            changed = True
+            recorddb[ref]['temperature_C'] = fields["value"]
+    if 'humidity' in jsonin:
+        fields = {'value': int(jsonin['humidity'])}
+        measurement = 'humidity'
+        points.append({'measurement': measurement, 'tags': tags, 'time': dateinflux, 'fields': fields})
+        if 'humidity' not in recorddb[ref] or recorddb[ref]['humidity'] != fields["value"]:
+            changed = True
+            recorddb[ref]['humidity'] = fields["value"]
+    if 'battery_ok' in jsonin:
+        fields = {'value': True if jsonin['battery_ok']==1 else False}
+        measurement = 'battery_ok'
+        points.append({'measurement': measurement, 'tags': tags, 'time': dateinflux, 'fields': fields})
+        if 'battery_ok' not in recorddb[ref] or recorddb[ref]['battery_ok'] != fields["value"]:
+            changed = True
+            recorddb[ref]['battery_ok'] = fields["value"]
+    
+    current_ts = currentdt.timestamp()
+    if changed == True and (current_ts < recorddb[ref]["lastsend"] or current_ts > (recorddb[ref]["lastsend"] + DELAYRECORD)):
+        recorddb[ref]["lastsend"] = current_ts
+        #print("Write")
+        #print(ref)
+        print(points)
+        influxc.write_points(points)
+    #else:
+    #    print("Not Write")
+    #    print(ref)
+    #    print(recorddb[ref]["lastsend"])
+    
+    
+
+def on_subscribe(mqttc, obj, mid, granted_qos):
+    print("Subscribed: " + str(mid) + " " + str(granted_qos))
+
+signal.signal(signal.SIGINT, signal_handler)
+
+# If you want to use a specific client id, use
+# mqttc = mqtt.Client("client-id")
+# but note that the client id must be unique on the broker. Leaving the client
+# id parameter empty will generate a random id for you.
+mqttc = mqtt.Client()
+mqttc.on_message = on_message
+mqttc.on_connect = on_connect
+mqttc.on_subscribe = on_subscribe
+# Uncomment to enable debug messages
+# mqttc.on_log = on_log
+mqttc.connect("localhost", 1883, 60)
+mqttc.subscribe(prefix+"/events", 0)
+
+influxc = InfluxDBClient(host="localhost", port=8428)
+
+dbs = influxc.get_list_database()
+dbs_list = []
+for db in dbs:
+    dbs_list.append(db.get("name"))
+
+if DBNAME not in dbs_list:
+    influxc.create_database(DBNAME)
+influxc.switch_database(DBNAME)
+
+mqttc.loop_forever()

python/rtl_433_mqtt2vm.py

@@ -0,0 +1,119 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2010-2013 Roger Light <roger@atchoo.org>
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Eclipse Distribution License v1.0
+# which accompanies this distribution.
+#
+# The Eclipse Distribution License is available at
+#   http://www.eclipse.org/org/documents/edl-v10.php.
+#
+# Contributors:
+#    Roger Light - initial implementation
+# Copyright (c) 2010,2011 Roger Light <roger@atchoo.org>
+# All rights reserved.
+
+# This shows a simple example of an MQTT subscriber.
+
+import signal
+import sys
+import json
+import datetime
+import paho.mqtt.client as mqtt
+import requests
+
+def signal_handler(sig, frame):
+    mqttc.disconnect()
+    sys.exit(0)
+
+prefix = 'home/rtl_433'
+vmhost = 'localhost'
+mqtthost = 'localhost'
+#prefix = 'rtl_433/fedora'
+TAGS_KEYS = ['type','model','subtype','channel','id']
+
+DELAYRECORD = 300
+MINTS = 1735686000 # 01/01/2025 00:00:00
+recorddb = {}
+
+def on_connect(mqttc, obj, flags, rc):
+    print("rc: " + str(rc))
+
+
+def create_measure(name, timestamp, tags, value):
+    #jsonmeasure = { "metric": { "__name__": name, "instance": "rtl433", "job": "rtl433tovm" }, "values": [value], "timestamps":[timestamp*1000] }
+    jsonmeasure = { "metric": { "__name__": name }, "values": [value], "timestamps":[timestamp*1000] }
+    jsonmeasure["metric"].update(tags)
+    return json.dumps(jsonmeasure) + '\n'
+
+def on_message(mqttc, obj, msg):
+    #print(msg.topic + " " + str(msg.qos) + " " + str(msg.payload))
+    currentts=int(datetime.datetime.now(datetime.UTC).timestamp())
+    jsonin=json.loads(msg.payload)
+    ref=""
+    if "model" in jsonin:
+        ref+=str(jsonin["model"])
+    if "channel" in jsonin:
+        ref+=str(jsonin["channel"])
+    if "id" in jsonin:
+        ref+=str(jsonin["id"])
+    if ref not in recorddb:
+        recorddb[ref] = {}
+        recorddb[ref]["lastsend"]=0
+
+    
+    if currentts > MINTS and (currentts < recorddb[ref]["lastsend"] or currentts > (recorddb[ref]["lastsend"] + DELAYRECORD)):
+        tags = {}
+        for t in TAGS_KEYS:
+            if t in jsonin:
+                tags[t] = str(jsonin[t])
+        
+        measures = ""
+
+        if 'temperature_C' in jsonin:
+            value = float(jsonin['temperature_C'])
+            if 'temperature' not in recorddb[ref] or recorddb[ref]['temperature'] != value:
+                measures += create_measure("temperature", currentts, tags, value)
+                recorddb[ref]['temperature'] = value
+        if 'humidity' in jsonin:
+            value = int(jsonin['humidity'])
+            if 'humidity' not in recorddb[ref] or recorddb[ref]['humidity'] != value:
+                measures += create_measure("humidity", currentts, tags, value)
+                recorddb[ref]['humidity'] = value
+        if 'battery_ok' in jsonin:
+            value = int(jsonin['battery_ok'])
+            if 'battery_ok' not in recorddb[ref] or recorddb[ref]['battery_ok'] != value:
+                measures += create_measure("battery_ok", currentts, tags, value)
+                recorddb[ref]['battery_ok'] = value
+    
+        if len(measures) > 0:
+            recorddb[ref]["lastsend"] = currentts
+            #print(measures)
+            try:
+                requests.post("http://{}:8428/api/v1/import".format(vmhost), data=measures)
+            except:
+                print("VictoriaMetrics communication error")
+    
+    
+
+def on_subscribe(mqttc, obj, mid, granted_qos):
+    print("Subscribed: " + str(mid) + " " + str(granted_qos))
+
+signal.signal(signal.SIGINT, signal_handler)
+
+# If you want to use a specific client id, use
+# mqttc = mqtt.Client("client-id")
+# but note that the client id must be unique on the broker. Leaving the client
+# id parameter empty will generate a random id for you.
+mqttc = mqtt.Client()
+mqttc.on_message = on_message
+mqttc.on_connect = on_connect
+mqttc.on_subscribe = on_subscribe
+# Uncomment to enable debug messages
+# mqttc.on_log = on_log
+mqttc.connect(mqtthost, 1883, 60)
+mqttc.subscribe(prefix+"/events", 0)
+
+mqttc.loop_forever()