sonoff POW into influxdb

etc_mosquitto_conf.d_sonoff.conf

#listener 1883 192.168.178.25
listener 1883
log_dest syslog
log_dest stdout
log_dest topic
log_type error
log_type warning
log_type notice
log_type information
connection_messages true
log_timestamp true
allow_anonymous false
password_file /home/lee/mosquitto/pwfile

mqtt2influxdb.service

[Unit]
Description=mqtt to inFluxDB
After=network.target

[Service]
User=lee
ExecStart=/home/lee/mosquitto/venv/bin/mqtt2influxdb -c /home/lee/mosquitto/mqtt2influxdb.yaml

[Install]
WantedBy=default.target

mqtt2influxdb.yaml

mqtt:
  host: 127.0.0.1
  port: 1883
  username: "x"
  password: "xx"

influxdb:
  host: 127.0.0.1
  port: 8086
  database: testsonoff

points:
# {"Time":"2021-03-14T02:12:52","ENERGY":{"TotalStartTime":"2018-11-19T21:24:13","Total":557.551,"Yesterday":0.582,"Today":0.073,"Period":1,"Power":37,"ApparentPower":45,"ReactivePower":26,"Factor":0.82,"Voltage":231,"Current":0.194}}
  - measurement: sensor
    topic: tele/+/SENSOR
    fields:
      voltage:
        value: $.payload.ENERGY.Voltage
        type: "float"
      power:
        value: $.payload.ENERGY.Power
        type: "float"
      current:
        value: $.payload.ENERGY.Current
        type: "float"
      factor:
        value: $.payload.ENERGY.Factor
        type: "float"
      apparentPower:
        value: $.payload.ENERGY.ApparentPower
        type: "float"
      reactivePower:
        value: $.payload.ENERGY.ReactivePower
        type: "float"
      # {"Time":"2021-03-14T02:07:46","AHT2X-0x38":{"Temperature":23.7,"Humidity":29.0,"DewPoint":4.5},"TempUnit":"C"}
      temperature_c:
        value: $.payload.AHT2X-0x38.Temperature
        type: "float"
      humidity:
        value: $.payload.AHT2X-0x38.Humidity
        type: "float"
      # {"Time":"2021-03-14T15:35:08","CCS811":{"eCO2":832,"TVOC":65},"AHT2X-0x38":{"Temperature":25.9,"Humidity":25.9,"DewPoint":4.9},"TempUnit":"C"}
      eCO2:
        value: $.payload.CCS811.eCO2
        type: int
      tvoc:
        value: $.payload.CCS811.TVOC
        type: int
    tags:
      id: $.topic[1]
  - measurement: sensor
    topic: v1/bob/ambient
    fields:
       temperature_c:
         value: $.payload.temperature
         type: "float"
       humidity:
         value: $.payload.humidity
         type: "float"
    tags:
      id: bob-AHT20
#  - measurement: ambient
#    topic: tele/tasmota_9F9586/SENSOR
#    fields:
#    tags:
#      id: $.topic[1]
  - measurement: state
    topic: tele/+/STATE
    fields:
      uptime:
        value: $.payload.UptimeSec
        type: "int"
      wifi_channel:
        value: $.payload.Wifi.Channel
        type: "int"
      wifi_rssi:
        value: $.payload.Wifi.RSSI
        type: "int"
      wifi_signal:
        value: $.payload.Wifi.Signal
        type: "int"
      wifi_LinkCount:
        value: $.payload.Wifi.LinkCount
        type: "int"
    tags:
      id: $.topic[1]
  # {"clientId":"pixel5","timeStamp":1617268849154,"deviceId":"65089929","hr":84}
  - measurement: polar
    topic: psl/hr
    tags:
      deviceId: $.payload.deviceId
    fields:
      hr: $.payload.hr
      type: "int"
  # {"clientId":"pixel5","timeStamp":1617268813280,"deviceId":"65089929","sensorTimeStamp":670591211689608319,"ppg":[[227213,354758,227831,-225],[227106,354769,227889,-229],[227058,354825,228032,-184],[227033,354752,228058,-258],[226973,354770,227981,-239],[226979,354854,228025,-204],[227064,354781,227974,-255],[227070,354871,228081,-157],[227093,354863,228100,-250],[227040,354835,228101,-219],[227053,354893,228127,-230],[227026,354836,228040,-222],[227161,354923,228038,-200],[227096,354966,228102,-224],[227116,355070,228136,-301],[227156,355125,228201,-163],[227275,355092,228140,-158],[227153,355239,228202,-245]]}

sonoff2influx.service

[Unit]
Description=sonoff to inFluxDB
After=network.target

[Service]
User=lee
ExecStart=/home/x/mosquitto/venv/bin/python /home/x/mosquitto/sonoff2influxs2.py

[Install]
WantedBy=default.target

sonoff2influx2.py

#!/usr/bin/env python
import paho.mqtt.client as mqtt
import json
import time
import requests
#from influxdb import InfluxDBClient

db_name="sonoff"
db_user="sonoff"
db_password="xx"
nodes=["sonoff","sonoff2"]
#nodename="sonoff2"


def send_to_influx(power,voltage,current,factor,aPower,rPower,devid,timestamp):
    import requests
    import json
#"ApparentPower":35,"ReactivePower"
    # <measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]
    data=[]
    data.append("line_voltage,instance=%s value=%s %s"%(devid,voltage,timestamp))
    values=["value=%s"%power]
    if aPower:
        values.append("apparentPower=%s"%aPower)
    if rPower:
        values.append("reactivePower=%s"%rPower)
    data.append("power_watts,instance=%s %s %s"%(devid,",".join(values),timestamp))
    data.append("current_ampere,instance=%s value=%s %s"%(devid,current,timestamp))
    data.append("factor,instance=%s value=%f %s"%(devid,factor,timestamp))
    
    try:
        print("Data: %s"%data)
        #ret = requests.post("http://127.0.0.1:8086/write?db=%s&p=%s&u=%s&precision=s"%(db_name,db_password,db_user), timeout=2, data="\n".join(data))
        return ret
    except requests.exceptions.Timeout:
        print("Request timeout... skipping",flush=True)
    except Exception as e:
        print("Error:",e,flush=True)
    return "Error"

def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    for nodename in nodes:
        client.subscribe("tele/%s/#"%nodename)
        client.publish("cmnd/%s/TelePeriod"%nodename, 60)

#TODO: parse nodename from topic
def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))
           
    try:
        topic = msg.topic.split('/')
        dat = json.loads(msg.payload.decode())
        ts = time.strptime(dat["Time"],"%Y-%m-%dT%H:%M:%S") 
        uts= time.mktime(ts) #+3600.0
        if topic[2]=="SENSOR":
            # {"Time":"2018-11-26T17:55:06","ENERGY":{"TotalStartTime":"2018-11-19T21:24:13","Total":0.210,"Yesterday":0.018,"Today":0.192,"Period":0,"Power":24,"ApparentPower":35,"ReactivePower":25,"Factor":0.69,"Voltage":233,"Current":0.150}}
            power=dat["ENERGY"]["Power"]
            apparentPower=dat["ENERGY"]["ApparentPower"]
            reactivePower=dat["ENERGY"]["ReactivePower"]
            voltage=dat["ENERGY"]["Voltage"]
            current=dat["ENERGY"]["Current"]
            factor=dat["ENERGY"]["Factor"]
            r = send_to_influx(power, voltage, current, factor, apparentPower, reactivePower, devid=topic[1], timestamp=int(uts))
            print("inFluxDB > %s"%r,flush=True)
        elif topic[2]=="STATE":
            # {"Time":"2018-11-26T19:45:06","Uptime":"0T03:00:34","Vcc":3.186,"POWER":"ON","Wifi":{"AP":1,"SSId":"WLAN","BSSId":"xx:xx:BF:xx:xx:xx","Channel":6,"RSSI":100}}
            # mqtt_on_message tele/sonoff/STATE b'{"Time":"2020-05-22T22:34:16","Uptime":"48T03:41:50","UptimeSec":4160510,"Heap":27,"SleepMode":"Dynamic","Sleep":50,"LoadAvg":19,"MqttCount":91,"POWER1":"ON","Wifi":{"AP":1,"SSId":"WLAN","BSSId":"xx:xx:BF:xx:xx:xx","Channel":13,"RSSI":100,"Signal":-48,"LinkCount":30,"Downtime":"0T00:02:15"}}'
            uptime=dat.get("Uptime",None)
            wifi=dat.get("Wifi")
            if wifi:
                channel=wifi.get("Channel",None)
                rssi=get("RSSI",None)
            vcc=dat.get("Vcc"),None)
            
            pass
#            send_misc(timestamp=int(uts),rssi=dat["Wifi"]["RSSI"])
        elif topic[2]=="UPTIME":
            #tele/sonoff/UPTIME b'{"Time":"2018-05-29T21:02:00","Uptime":"9T00:15:38"}'
            pass
    except Exception as e:
        print("Error:",e)

client = mqtt.Client()
#ifcli = InfluxDBClient('localhost', 8086, db_user, db_password, db_name)
client.on_connect = on_connect
client.on_message = on_message
client.username_pw_set("reader", password="xxx")
client.connect("localhost", 1883, 60)

client.loop_forever()

sonoff2iotplotter.py

#!/usr/bin/env python
import paho.mqtt.client as mqtt
import json
import time
import requests

feed_id="12345"
apikey=".-xxxx-."
nodename="sonoff"

def send_to_iotplotter(power,voltage,current,factor,timestamp):
    import requests
    import json
    
    headers = {'api-key': apikey}
    payload = {} #The variable to send to IoTPlotter
    payload["data"] = {} #A dictionary containing the data to send to IoTPlotter
    payload["data"]["Voltage"] = [] #Creates a list for sending one or more values to IoTPlotter for the 'GRAPH_NAME' graph.
    payload["data"]["Current"] = [] #Creates a list for sending one or more values to IoTPlotter for the 'GRAPH_NAME' graph.
    payload["data"]["Power"] = [] #Creates a list for sending one or more values to IoTPlotter for the 'GRAPH_NAME' graph.
    payload["data"]["Factor"] = [] #Creates a list for sending one or more values to IoTPlotter for the 'GRAPH_NAME' graph.
    
    #Appends three values for the graph 'GRAPH_NAME': 
    payload["data"]["Voltage"].append({"value":voltage,"epoch":timestamp}) #'GRAPH_NAME' is set to 25.05
    payload["data"]["Current"].append({"value":current,"epoch":timestamp}) #'GRAPH_NAME' is set to 25.05
    payload["data"]["Power"].append({"value":power,"epoch":timestamp}) #'GRAPH_NAME' is set to 25.05
    payload["data"]["Factor"].append({"value":factor,"epoch":timestamp}) #'GRAPH_NAME' is set to 25.05
    #Add some historical data to the graph:
    #payload["data"]["GRAPH_NAME"].append({"value":25.99, "epoch":1516195980})
    #payload["data"]["GRAPH_NAME"].append({"value":24.99, "epoch":1516195280})
    
    try:
        ret = requests.post("http://iotplotter.com/api/v2/feed/%s"%feed_id, timeout=2, headers=headers, data=json.dumps(payload))
        return ret
    except requests.exceptions.Timeout:
        print("Request timeout... skipping",flush=True)
    except Exception as e:
        print("Error:",e)
    return "Error"


def send_misc(timestamp,rssi=None,uptime=None):
#tele/sonoff/UPTIME b'{"Time":"2018-05-29T21:02:00","Uptime":"9T00:15:38"}'
    import requests
    import json
    
    headers = {'api-key': apikey}
    payload = {} #The variable to send to IoTPlotter
    payload["data"] = {} #A dictionary containing the data to send to IoTPlotter
    if rssi:
        payload["data"]["RSSI"] = [dict(value=rssi,epoch=timestamp)]
    if uptime:
        payload["data"]["Uptime"] = [dict(value=uptime,epoch=timestamp)]
    
    if len(payload["data"])>0:
        try:
            ret = requests.post("http://iotplotter.com/api/v2/feed/%s"%feed_id, headers=headers, data=json.dumps(payload))
            return ret
        except Exception as e:
            print("Error:",e)
            return "Error"
    return None

def on_connect(client, userdata, flags, rc):
    global nodename
    print("Connected with result code " + str(rc))
    client.subscribe("tele/%s/#"%nodename)
    client.publish("cmnd/%s/TelePeriod"%nodename, 60)

def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))
    try:
        dat = json.loads(msg.payload.decode())
        ts = time.strptime(dat["Time"],"%Y-%m-%dT%H:%M:%S") 
        uts= time.mktime(ts)+3600.0
        if msg.topic.endswith("SENSOR"):
            # {"Time":"2018-05-19T12:05:02","ENERGY":{"Total":0.264,"Yesterday":0.046,"Today":0.045,"Power":2,"Factor":0.12,"Voltage":221,"Current":0.080}}
            r = send_to_iotplotter(dat["ENERGY"]["Power"],dat["ENERGY"]["Voltage"],dat["ENERGY"]["Current"],dat["ENERGY"]["Factor"],timestamp=int(uts))
            print("IoTplotter > %s"%r,flush=True)
        elif msg.topic.endswith("STATE"):
            #tele/sonoff/STATE b'{"Time":"2018-05-29T21:02:20","Uptime":"9T00:15:58","Vcc":3.209,"POWER1":"ON","Wifi":{"AP":1,"SSId":"WLAN","RSSI":26,"APMac":"xx"}}'
            send_misc(timestamp=int(uts),rssi=dat["Wifi"]["RSSI"])
        elif msg.topic.endswith("UPTIME"):
            #tele/sonoff/UPTIME b'{"Time":"2018-05-29T21:02:00","Uptime":"9T00:15:38"}'
            pass
    except Exception as e:
        print("Error:",e)

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.username_pw_set("reader", password="xx")
client.connect("localhost", 1883, 60)

client.loop_forever()