srne-mqtt/consumers/mqtt.py

300 lines
9.5 KiB
Python
Raw Normal View History

2021-11-14 00:55:43 +00:00
# -*- coding: utf-8 -*-
2021-11-14 03:46:17 +00:00
import json
2022-01-26 19:41:59 +00:00
from time import sleep
2021-11-14 00:55:43 +00:00
from typing import Any, Dict, List, Optional
from uuid import uuid4
import paho.mqtt.client as mqtt
from solar_types import DataName
2021-11-14 00:55:43 +00:00
from . import BaseConsumer
MAP_VALUES: Dict[DataName, Dict[str, Any]] = {
# DataName.BATTERY_VOLTAGE_MIN: {},
# DataName.BATTERY_VOLTAGE_MAX: {},
# DataName.CHARGE_MAX_CURRENT: {},
# DataName._DISCHARGE_MAX_CURRENT: {},
# DataName.CHARGE_MAX_POWER: {},
# DataName.DISCHARGE_MAX_POWER: {},
# DataName.CHARGE_AMP_HOUR: {},
# DataName.DISCHARGE_AMP_HOUR: {},
DataName.PRODUCTION_ENERGY: {
2021-11-14 02:32:30 +00:00
"unit": "Wh",
"type": "energy",
"state_class": "total_increasing",
},
DataName.CONSUMPTION_ENERGY: {
2021-11-14 02:32:30 +00:00
"unit": "Wh",
"type": "energy",
"state_class": "total_increasing",
},
# DataName.RUN_DAYS: {},
# DataName.DISCHARGE_COUNT: {},
# DataName.FULL_CHARGE_COUNT: {},
# DataName.TOTAL_CHARGE_AMP_HOURS: {},
# DataName.TOTAL_DISCHARGE_AMP_HOURS: {},
DataName.TOTAL_PRODUCTION_ENERGY: {
2021-11-14 02:32:30 +00:00
"unit": "Wh",
"type": "energy",
"state_class": "total_increasing",
2021-11-14 03:46:17 +00:00
"expiry": 180,
2021-11-14 02:32:30 +00:00
},
DataName.TOTAL_CONSUMPTION_ENERGY: {
2021-11-14 02:32:30 +00:00
"unit": "Wh",
"type": "energy",
"state_class": "total_increasing",
2021-11-14 03:46:17 +00:00
"expiry": 180,
2021-11-14 02:32:30 +00:00
},
#
DataName.BATTERY_CHARGE: {
"unit": "%",
"type": "battery",
"state_class": "measurement",
},
DataName.BATTERY_VOLTAGE: {
"unit": "V",
"type": "voltage",
"state_class": "measurement",
},
DataName.BATTERY_CURRENT: {
"unit": "A",
"type": "current",
"state_class": "measurement",
},
DataName.INTERNAL_TEMPERATURE: {
2021-11-14 02:32:30 +00:00
"unit": "°C",
"type": "temperature",
"state_class": "measurement",
},
DataName.BATTERY_TEMPERATURE: {
2021-11-14 02:32:30 +00:00
"unit": "°C",
"type": "temperature",
"state_class": "measurement",
},
DataName.LOAD_VOLTAGE: {
"unit": "V",
"type": "voltage",
"state_class": "measurement",
},
DataName.LOAD_CURRENT: {
"unit": "A",
"type": "current",
"state_class": "measurement",
},
DataName.LOAD_POWER: {"unit": "W", "type": "power", "state_class": "measurement"},
DataName.PANEL_VOLTAGE: {
"unit": "V",
"type": "voltage",
"state_class": "measurement",
},
DataName.PANEL_CURRENT: {
"unit": "A",
"type": "current",
"state_class": "measurement",
},
DataName.PANEL_POWER: {"unit": "W", "type": "power", "state_class": "measurement"},
# DataName.LOAD_ENABLED: {},
DataName.CALCULATED_BATTERY_POWER: {
"unit": "W",
"type": "power",
"state_class": "measurement",
},
DataName.CALCULATED_PANEL_POWER: {
"unit": "W",
"type": "power",
"state_class": "measurement",
},
DataName.CALCULATED_LOAD_POWER: {
"unit": "W",
"type": "power",
"state_class": "measurement",
},
2021-11-14 02:32:30 +00:00
}
2021-11-14 00:55:43 +00:00
class MqttConsumer(BaseConsumer):
client: mqtt.Client
initialized: List[str]
def __init__(self, settings: Dict[str, Any]) -> None:
self.initialized = []
super().__init__(settings)
self.client = mqtt.Client(client_id=settings["client"]["id"], userdata=self)
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
2022-01-26 19:41:59 +00:00
self.client.on_disconnect = self.on_disconnect
self.client.on_connect_fail = self.on_connect_fail
2021-11-14 00:55:43 +00:00
# Will must be set before connecting!!
self.client.will_set(
f"{self.topic_prefix}/available", payload="offline", retain=True
)
while True:
try:
self.client.connect(
settings["client"]["host"],
settings["client"]["port"],
settings["client"]["keepalive"],
)
break
except OSError as err:
# Network is unreachable
if err.errno == 101:
pass
# Temporary failure in name resolution
elif err.errno == -3:
pass
else:
raise
print(err)
sleep(0.1)
2021-11-14 00:55:43 +00:00
def config(self, settings: Dict[str, Any]):
super().config(settings)
settings.setdefault("client", {})
settings["client"].setdefault("id", None)
settings["client"].setdefault("host", "")
settings["client"].setdefault("port", 1883)
settings["client"].setdefault("keepalive", 60)
if not settings.get("device_id"):
settings["device_id"] = str(uuid4())
settings.setdefault("prefix", "solarmppt")
settings.setdefault("discovery_prefix", "homeassistant")
@property
def topic_prefix(self):
return f"{self.settings['prefix']}/{self.settings['device_id']}"
def get_ha_config(
self,
id,
name,
unit: Optional[str] = None,
type: Optional[str] = None,
expiry: int = 90,
state_class: Optional[str] = None,
):
assert state_class in [None, "measurement", "total", "total_increasing"]
res = {
"~": f"{self.topic_prefix}",
"unique_id": f"{self.settings['device_id']}_{id}",
"availability_topic": "~/available",
"state_topic": f"~/{id}",
2021-11-14 02:32:30 +00:00
"name": name,
2021-11-14 00:55:43 +00:00
"device": {
"identifiers": [
self.settings["device_id"],
],
# TODO: Get charger serial and use for identifier instead
# See: https://www.home-assistant.io/integrations/sensor.mqtt/#device
# "via_device": self.settings["device_id"],
2021-11-14 03:46:17 +00:00
"suggested_area": "Solar panel",
2021-11-14 00:55:43 +00:00
},
"force_update": True,
"expire_after": expiry,
}
if unit:
res["unit_of_meas"] = unit
if type:
res["dev_cla"] = type
if state_class:
res["state_class"] = state_class
2021-11-14 03:46:17 +00:00
return res
2021-11-14 00:55:43 +00:00
# The callback for when the client receives a CONNACK response from the server.
@staticmethod
def on_connect(client: mqtt.Client, userdata: "MqttConsumer", flags, rc):
print("Connected with result code " + str(rc))
# Subscribing in on_connect() means that if we lose the connection and
# reconnect then subscriptions will be renewed.
# client.subscribe("$SYS/#")
client.publish(
f"{userdata.topic_prefix}/available", payload="online", retain=True
)
2022-01-26 19:41:59 +00:00
@staticmethod
def on_connect_fail(client: mqtt.Client, userdata: "MqttConsumer"):
print(userdata.__class__.__name__, "on_connect_fail")
2021-11-14 00:55:43 +00:00
# The callback for when a PUBLISH message is received from the server.
@staticmethod
def on_message(client, userdata, msg):
print(msg.topic + " " + str(msg.payload))
2022-01-26 19:41:59 +00:00
@staticmethod
def on_disconnect(client: mqtt.Client, userdata: "MqttConsumer", rc, prop=None):
print(userdata.__class__.__name__, "on_disconnect", rc)
2021-11-14 00:55:43 +00:00
def poll(self):
2022-01-26 19:41:59 +00:00
res = self.client.loop(timeout=0.1, max_packets=5)
if res != mqtt.MQTT_ERR_SUCCESS:
print(self.__class__.__name__, "loop returned non-success:", res)
try:
sleep(1)
res = self.client.reconnect()
if res != mqtt.MQTT_ERR_SUCCESS:
print(self.__class__.__name__, "Reconnect failed:", res)
except (OSError, mqtt.WebsocketConnectionError) as err:
print(self.__class__.__name__, "Reconnect failed:", err)
2021-11-14 00:55:43 +00:00
return super().poll()
def write(self, data: Dict[str, Any]):
2021-11-14 03:46:17 +00:00
self.client.publish(f"{self.topic_prefix}/raw", payload=json.dumps(data))
2021-11-14 02:32:30 +00:00
for k, v in data.items():
if k in MAP_VALUES:
if k not in self.initialized:
km = MAP_VALUES[DataName(k)]
2021-11-14 02:32:30 +00:00
pretty_name = k.replace("_", " ").capitalize()
disc_prefix = self.settings["discovery_prefix"]
device_id = self.settings["device_id"]
2021-11-14 02:32:30 +00:00
self.client.publish(
f"{disc_prefix}/sensor/{device_id}_{k}/config",
payload=json.dumps(self.get_ha_config(k, pretty_name, **km)),
2021-11-14 02:32:30 +00:00
retain=True,
)
self.initialized.append(k)
self.client.publish(f"{self.topic_prefix}/{k}", v, retain=True)
2021-11-14 00:55:43 +00:00
def exit(self):
self.client.publish(
f"{self.topic_prefix}/available", payload="offline", retain=True
)
while self.client.want_write():
self.client.loop_write(10)
self.client.disconnect()
return super().exit()
# Client(client_id="", clean_session=True, userdata=None,
# protocol=MQTTv311, transport="tcp")
# connect_srv(domain, keepalive=60, bind_address="")
# Connect to a broker using an SRV DNS lookup to obtain the broker address.
# Takes the following arguments:
# domain
# the DNS domain to search for SRV records.
# If None, try to determine the local domain name.
# client.will_set(topic, payload=None, qos=0, retain=False)
# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
# Other loop*() functions are available that give a threaded interface and a
# manual interface.
# client.loop_forever()