165 lines
6.2 KiB
Python
165 lines
6.2 KiB
Python
""" powermon / outputformats / hass.py
|
|
|
|
LOCAL PATCH (g) — see ../Install.md §5(g). Original behavior preserved
|
|
except for these changes (all aimed at cleaner HA-frontend rendering and
|
|
correct long-term-statistics behavior):
|
|
|
|
- Removed `last_reset` from discovery payload. That field is HA-spec
|
|
for `state_class: total` cumulative counters; setting it on
|
|
`measurement` sensors causes HA to warn and ignore. Upstream emits
|
|
`str(datetime.now())` on every discovery — meaningless timestamp
|
|
that fires a state-class warning per entity per startup.
|
|
|
|
- Removed `force_update: "true"` default. Upstream forces a recorder
|
|
write on every state publish even if the value hasn't changed,
|
|
bloating the HA history database with hundreds of duplicate rows
|
|
per day per slowly-changing entity. HA defaults are correct.
|
|
|
|
- Added `suggested_display_precision` based on unit. Without this,
|
|
HA's voltage/current/temperature device classes round to integers
|
|
in the frontend (e.g. `52.5 V` displays as `52 V`).
|
|
|
|
- Default `state_class: measurement` for numeric sensors when the
|
|
protocol definition didn't supply one. Without state_class, HA
|
|
excludes the sensor from long-term statistics, the Energy
|
|
dashboard, and the standard "show this as a trend" graph.
|
|
"""
|
|
import json as js
|
|
import logging
|
|
|
|
from powermon.commands.command import Command
|
|
from powermon.commands.reading import Reading
|
|
from powermon.commands.result import Result
|
|
from powermon.outputformats.abstractformat import AbstractFormat
|
|
|
|
log = logging.getLogger("hass")
|
|
|
|
|
|
# Suggested decimal places per unit for the HA frontend. Keys must match
|
|
# whatever powermon's reading_definitions emit as the unit string (the
|
|
# value the BMS / inverter parser sets as `data_unit`).
|
|
_PRECISION_BY_UNIT: dict[str, int] = {
|
|
"V": 1,
|
|
"mV": 0,
|
|
"A": 1,
|
|
"mA": 0,
|
|
"W": 0,
|
|
"VA": 0,
|
|
"Hz": 1,
|
|
"%": 0,
|
|
"°C": 0,
|
|
"Ah": 1,
|
|
"Wh": 0,
|
|
"kWh": 1,
|
|
"ms": 0,
|
|
"s": 0,
|
|
}
|
|
|
|
|
|
class Hass(AbstractFormat):
|
|
""" formatter to generate home assistant auto config mqtt messages """
|
|
def __init__(self, config):
|
|
super().__init__(config)
|
|
self.name = "hass"
|
|
self.discovery_prefix = config.get("discovery_prefix", "homeassistant")
|
|
self.entity_id_prefix = config.get("entity_id_prefix", None)
|
|
|
|
def __str__(self):
|
|
return f"{self.name}: generates Home Assistant auto config and update mqtt messages"
|
|
|
|
def get_options(self):
|
|
""" return a dict of all options and defaults """
|
|
extra_options = {"discovery_prefix": "homeassistant", "entity_id_prefix": None}
|
|
options = super().get_options()
|
|
options.update(extra_options)
|
|
return options
|
|
|
|
def format(self, command: Command, result: Result, device_info) -> list:
|
|
log.info("Using output formatter: %s", self.name)
|
|
|
|
config_msgs = []
|
|
value_msgs = []
|
|
|
|
_result = []
|
|
if result.readings is None:
|
|
return _result
|
|
display_data: list[Reading] = self.format_and_filter_data(result)
|
|
log.debug("displayData: %s", display_data)
|
|
|
|
for response in display_data:
|
|
data_name = self.format_key(response.data_name)
|
|
value = response.data_value
|
|
unit = response.data_unit
|
|
icon = response.icon
|
|
device_class = response.device_class
|
|
state_class = response.state_class
|
|
|
|
if unit == "bool" or value == "enabled" or value == "disabled":
|
|
component = "binary_sensor"
|
|
else:
|
|
component = "sensor"
|
|
|
|
if component == "binary_sensor":
|
|
if value == 0 or value == "0" or value == "disabled":
|
|
value = "OFF"
|
|
elif value == 1 or value == "1" or value == "enabled":
|
|
value = "ON"
|
|
|
|
if self.entity_id_prefix is None:
|
|
object_id = f"{data_name}".lower().replace(" ", "_")
|
|
name = f"{data_name}"
|
|
else:
|
|
object_id = f"{self.entity_id_prefix}_{data_name}".lower().replace(" ", "_")
|
|
name = f"{self.entity_id_prefix} {data_name}"
|
|
|
|
topic_base = f"{self.discovery_prefix}/{component}/{object_id}".replace(" ", "_")
|
|
topic = f"{topic_base}/config"
|
|
state_topic = f"{topic_base}/state"
|
|
|
|
payload = {
|
|
"name": f"{name}",
|
|
"state_topic": f"{state_topic}",
|
|
"unique_id": f"{object_id}_{device_info.serial_number}",
|
|
# PATCH (g): removed `force_update: "true"` and `last_reset`
|
|
}
|
|
|
|
payload["device"] = {
|
|
"name": device_info.name,
|
|
"identifiers": [device_info.serial_number],
|
|
"model": device_info.model,
|
|
"manufacturer": device_info.manufacturer,
|
|
}
|
|
|
|
if unit and unit != "bool":
|
|
payload["unit_of_measurement"] = f"{unit}"
|
|
|
|
if icon:
|
|
payload.update({"icon": icon})
|
|
|
|
if device_class:
|
|
payload["device_class"] = device_class
|
|
|
|
# PATCH (g): default state_class to "measurement" for numeric
|
|
# sensors that don't have one set in the reading definition.
|
|
# Required for HA long-term statistics / Energy dashboard wiring.
|
|
if not state_class and component == "sensor" and isinstance(value, (int, float)):
|
|
state_class = "measurement"
|
|
if state_class:
|
|
payload["state_class"] = state_class
|
|
|
|
# PATCH (g): emit suggested_display_precision so HA frontend
|
|
# renders the correct decimal count instead of rounding voltages
|
|
# / currents / etc. to integers per device_class default.
|
|
if component == "sensor" and unit in _PRECISION_BY_UNIT:
|
|
payload["suggested_display_precision"] = _PRECISION_BY_UNIT[unit]
|
|
|
|
payloads = js.dumps(payload)
|
|
msg = {"topic": topic, "payload": payloads}
|
|
config_msgs.append(msg)
|
|
|
|
msg = {"topic": state_topic, "payload": value}
|
|
value_msgs.append(msg)
|
|
|
|
# order value msgs after config to allow HA time to build entity before state data arrives
|
|
return config_msgs + value_msgs
|