mirror of
https://github.com/vakius/hassio-addon-lifepo4wered.git
synced 2026-08-01 20:44:12 +03:00
Add MQTT monitor with Home Assistant autodiscovery
Poll lifepo4wered-cli every 30s (configurable), publish all registers as one JSON state payload, and announce sensors/binary sensor via MQTT discovery. run.sh pulls broker credentials from the Supervisor mqtt service (Mosquitto addon) with manual overrides in addon options. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
32
README.md
32
README.md
@@ -1,5 +1,37 @@
|
||||
# hassio-addon-lifepo4wered
|
||||
|
||||
Home Assistant OS addon for the [LiFePO4wered/Pi+](https://lifepo4wered.com/lifepo4wered-pi+.html)
|
||||
UPS: runs `lifepo4wered-daemon` and publishes UPS stats to MQTT with Home
|
||||
Assistant autodiscovery.
|
||||
|
||||
## MQTT monitor
|
||||
|
||||
`lifepo4wered_monitor.py` polls `lifepo4wered-cli get` (default every 30 s)
|
||||
and publishes all registers as one JSON payload to `lifepo4wered/state`.
|
||||
Entities appear automatically under one device, "LiFePO4wered/Pi+":
|
||||
|
||||
- **Sensors**: input/battery/output voltage (V), output current (A)
|
||||
- **Binary sensor** "External power": ON when `VIN >= VIN_THRESHOLD`
|
||||
(i.e. OFF means running on battery)
|
||||
- **Diagnostics**: thresholds, timers, watchdog and state registers
|
||||
- **Availability**: `lifepo4wered/availability` with MQTT Last Will, so
|
||||
entities go *unavailable* if the addon stops
|
||||
|
||||
MQTT connection details are taken automatically from the Supervisor when the
|
||||
Mosquitto addon is installed. Addon options: `poll_interval` (seconds) and
|
||||
optional `mqtt_host`/`mqtt_port`/`mqtt_user`/`mqtt_password` overrides.
|
||||
|
||||
### Testing off the Pi
|
||||
|
||||
```
|
||||
python3 -m unittest discover -s tests # unit tests, no deps needed
|
||||
LIFEPO4WERED_FAKE=1 MQTT_HOST=... python3 hassio-addon-lifepo4wered/lifepo4wered_monitor.py
|
||||
```
|
||||
|
||||
`LIFEPO4WERED_FAKE=1` replaces the CLI with embedded sample data.
|
||||
|
||||
## Setup notes
|
||||
|
||||
|
||||
https://www.linkedin.com/pulse/creating-your-first-home-assistant-add-on-issac-goldstand
|
||||
|
||||
|
||||
65
docs/superpowers/specs/2026-07-18-mqtt-monitor-design.md
Normal file
65
docs/superpowers/specs/2026-07-18-mqtt-monitor-design.md
Normal file
@@ -0,0 +1,65 @@
|
||||
# LiFePO4wered MQTT Monitor — Design
|
||||
|
||||
Date: 2026-07-18. Approved by vaku in conversation.
|
||||
|
||||
## Goal
|
||||
|
||||
Publish LiFePO4wered/Pi+ UPS stats from the Home Assistant OS addon to MQTT,
|
||||
with Home Assistant autodiscovery, so the UPS shows up as a device with sensors.
|
||||
|
||||
## Approach
|
||||
|
||||
A single Python script (`lifepo4wered_monitor.py`) inside the existing addon
|
||||
container. It shells out to `lifepo4wered-cli get` (already built into the
|
||||
image), parses the `KEY = value` output, and publishes to MQTT via `paho-mqtt`.
|
||||
Bindings via ctypes/PyPI were considered and rejected: the CLI is already
|
||||
present and one subprocess call reads everything at once.
|
||||
|
||||
## Behavior
|
||||
|
||||
- Poll every `POLL_INTERVAL` seconds (default 30).
|
||||
- One retained JSON state payload per cycle on `lifepo4wered/state`; all
|
||||
entities read it with `value_template` — one MQTT message per cycle.
|
||||
- Discovery configs (retained) on `<DISCOVERY_PREFIX>/<component>/lifepo4wered/<key>/config`,
|
||||
republished on every (re)connect. All entities belong to one HA device
|
||||
("LiFePO4wered/Pi+", manufacturer Silicognition LLC).
|
||||
- Main sensors: VIN, VBAT, VOUT (mV → V, device_class `voltage`),
|
||||
IOUT (mA → A, device_class `current`).
|
||||
- Binary sensor "External power" (device_class `power`):
|
||||
`VIN >= VIN_THRESHOLD` → ON.
|
||||
- Diagnostic entities (`entity_category: diagnostic`): threshold/config
|
||||
registers (VBAT_MIN, VBAT_SHDN, VBAT_BOOT, VOUT_MAX, VIN_THRESHOLD scaled to
|
||||
V; timers, watchdog, AUTO_BOOT, LED_STATE, TOUCH_STATE, PI_RUNNING raw;
|
||||
RTC_TIME as timestamp).
|
||||
- Skipped: identity/calibration registers (I2C_REG_VER, I2C_ADDRESS, DCO_*,
|
||||
TOUCH_CAP_CYCLES/THRESHOLD/HYSTERESIS, *_OFFSET, CFG_WRITE, RTC_WAKE_TIME,
|
||||
WAKE_TIME).
|
||||
- Availability: MQTT LWT on `lifepo4wered/availability` (`online`/`offline`).
|
||||
CLI failure (I2C hiccup): log, publish `offline`, retry next cycle.
|
||||
- Config via env vars: MQTT_HOST, MQTT_PORT, MQTT_USER, MQTT_PASSWORD,
|
||||
POLL_INTERVAL, DISCOVERY_PREFIX, MQTT_TOPIC_PREFIX.
|
||||
- `LIFEPO4WERED_FAKE=1` uses embedded sample output instead of the CLI, for
|
||||
testing off the Pi.
|
||||
|
||||
## Addon integration
|
||||
|
||||
- Dockerfile: add `python3` + `py3-paho-mqtt` (Alpine packages), copy script
|
||||
and `run.sh`, `CMD ["/run.sh"]`.
|
||||
- `run.sh` (bashio): pull MQTT host/port/user/password from Supervisor
|
||||
services API (works out of the box with the Mosquitto addon), allow
|
||||
overrides from addon options, start `lifepo4wered-daemon -f` in the
|
||||
background, run the monitor in the foreground.
|
||||
- `config.json`: add `"services": ["mqtt:need"]`, addon options
|
||||
(`poll_interval`, optional MQTT overrides), bump version.
|
||||
|
||||
## Error handling
|
||||
|
||||
- paho built-in reconnect for broker drops; discovery republished on connect.
|
||||
- 10 s subprocess timeout on the CLI; malformed lines skipped with a warning.
|
||||
- Script compatible with paho-mqtt 1.x and 2.x callback APIs.
|
||||
|
||||
## Testing
|
||||
|
||||
- Unit tests (stdlib `unittest`) for output parsing and discovery payload
|
||||
generation, runnable on any machine without paho or a broker.
|
||||
- Fake mode for end-to-end runs against a real broker off the Pi.
|
||||
@@ -9,11 +9,13 @@ FROM homeassistant/aarch64-base:latest
|
||||
ENV LANG C.UTF-8
|
||||
|
||||
WORKDIR /
|
||||
RUN apk add --no-cache python3 py3-paho-mqtt
|
||||
COPY --from=build /tmp/LiFePO4wered-Pi/build/liblifepo4wered.so /usr/lib/liblifepo4wered.so
|
||||
COPY --from=build /tmp/LiFePO4wered-Pi/build/lifepo4wered-cli /usr/bin/lifepo4wered-cli
|
||||
COPY --from=build /tmp/LiFePO4wered-Pi/build/lifepo4wered-daemon /usr/sbin/lifepo4wered-daemon
|
||||
# COPY start.sh /app/start.sh
|
||||
# ENTRYPOINT ["/app/start.sh"]
|
||||
CMD lifepo4wered-daemon -f
|
||||
COPY lifepo4wered_monitor.py /usr/bin/lifepo4wered_monitor.py
|
||||
COPY run.sh /run.sh
|
||||
RUN chmod a+x /run.sh
|
||||
CMD ["/run.sh"]
|
||||
|
||||
LABEL io.hass.version="VERSION" io.hass.type="addon" io.hass.arch="aarch64"
|
||||
|
||||
Binary file not shown.
@@ -1,14 +1,22 @@
|
||||
{
|
||||
"name": "LiFePO4wered Addon",
|
||||
"version": "0.0.5",
|
||||
"version": "0.1.0",
|
||||
"slug": "lifepo4wered",
|
||||
"description": "Add-on with service for LiFePO4Wered Pi+ UPS",
|
||||
"description": "Add-on with service and MQTT monitor for LiFePO4wered/Pi+ UPS",
|
||||
"url": "https://github.com/vakius/hassio-addon-lifepo4wered/tree/master/hassio-addon-lifepo4wered",
|
||||
"startup": "application",
|
||||
"boot": "auto",
|
||||
"options": {},
|
||||
"schema": {},
|
||||
"services": ["mqtt:want"],
|
||||
"options": {
|
||||
"poll_interval": 30
|
||||
},
|
||||
"schema": {
|
||||
"poll_interval": "int(5,3600)",
|
||||
"mqtt_host": "str?",
|
||||
"mqtt_port": "port?",
|
||||
"mqtt_user": "str?",
|
||||
"mqtt_password": "password?"
|
||||
},
|
||||
"arch": ["aarch64"],
|
||||
"devices": [ "/dev/i2c-1" ]
|
||||
|
||||
}
|
||||
|
||||
285
hassio-addon-lifepo4wered/lifepo4wered_monitor.py
Normal file
285
hassio-addon-lifepo4wered/lifepo4wered_monitor.py
Normal file
@@ -0,0 +1,285 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Publish LiFePO4wered/Pi+ UPS stats to MQTT with Home Assistant autodiscovery.
|
||||
|
||||
Polls `lifepo4wered-cli get`, publishes the parsed registers as one JSON
|
||||
payload, and announces entities via MQTT discovery. Configured entirely
|
||||
through environment variables (see main()).
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
try:
|
||||
import paho.mqtt.client as mqtt
|
||||
except ImportError: # allow unit-testing parse/discovery without paho
|
||||
mqtt = None
|
||||
|
||||
LOG = logging.getLogger("lifepo4wered_monitor")
|
||||
|
||||
CLI_COMMAND = ["lifepo4wered-cli", "get"]
|
||||
CLI_TIMEOUT = 10
|
||||
|
||||
# Sample of real `lifepo4wered-cli get` output; used when LIFEPO4WERED_FAKE=1.
|
||||
SAMPLE_OUTPUT = """\
|
||||
I2C_REG_VER = 7
|
||||
I2C_ADDRESS = 67
|
||||
LED_STATE = 1
|
||||
TOUCH_STATE = 0
|
||||
TOUCH_CAP_CYCLES = 0
|
||||
TOUCH_THRESHOLD = 12
|
||||
TOUCH_HYSTERESIS = 2
|
||||
DCO_RSEL = 14
|
||||
DCO_DCOMOD = 149
|
||||
VIN = 4709
|
||||
VBAT = 3233
|
||||
VOUT = 4993
|
||||
IOUT = 1355
|
||||
VBAT_MIN = 2850
|
||||
VBAT_SHDN = 2950
|
||||
VBAT_BOOT = 3150
|
||||
VOUT_MAX = 3500
|
||||
VIN_THRESHOLD = 4498
|
||||
IOUT_SHDN_THRESHOLD = 0
|
||||
VBAT_OFFSET = 103
|
||||
VOUT_OFFSET = 108
|
||||
VIN_OFFSET = 99
|
||||
IOUT_OFFSET = 0
|
||||
AUTO_BOOT = 4
|
||||
WAKE_TIME = 0
|
||||
SHDN_DELAY = 40
|
||||
AUTO_SHDN_TIME = 15
|
||||
PI_BOOT_TO = 300
|
||||
PI_SHDN_TO = 120
|
||||
RTC_TIME = 1784323638
|
||||
RTC_WAKE_TIME = 0
|
||||
WATCHDOG_CFG = 0
|
||||
WATCHDOG_GRACE = 20
|
||||
WATCHDOG_TIMER = 20
|
||||
PI_RUNNING = 1
|
||||
CFG_WRITE = 0
|
||||
"""
|
||||
|
||||
DEVICE = {
|
||||
"identifiers": ["lifepo4wered_pi"],
|
||||
"name": "LiFePO4wered/Pi+",
|
||||
"manufacturer": "Silicognition LLC",
|
||||
"model": "LiFePO4wered/Pi+",
|
||||
}
|
||||
|
||||
# (register, entity name) — millivolt registers shown as V main sensors
|
||||
MAIN_VOLTAGE = [
|
||||
("VIN", "Input voltage"),
|
||||
("VBAT", "Battery voltage"),
|
||||
("VOUT", "Output voltage"),
|
||||
]
|
||||
|
||||
# millivolt threshold registers, shown as diagnostic V sensors
|
||||
DIAG_VOLTAGE = [
|
||||
("VBAT_MIN", "Battery minimum voltage"),
|
||||
("VBAT_SHDN", "Battery shutdown voltage"),
|
||||
("VBAT_BOOT", "Battery boot voltage"),
|
||||
("VOUT_MAX", "Output maximum voltage"),
|
||||
("VIN_THRESHOLD", "Input power threshold"),
|
||||
]
|
||||
|
||||
# raw-value diagnostic registers (units vary or are unitless flags/timers)
|
||||
DIAG_RAW = [
|
||||
("AUTO_BOOT", "Auto boot mode"),
|
||||
("SHDN_DELAY", "Shutdown delay"),
|
||||
("AUTO_SHDN_TIME", "Auto shutdown time"),
|
||||
("PI_BOOT_TO", "Pi boot timeout"),
|
||||
("PI_SHDN_TO", "Pi shutdown timeout"),
|
||||
("WATCHDOG_CFG", "Watchdog mode"),
|
||||
("WATCHDOG_GRACE", "Watchdog grace"),
|
||||
("WATCHDOG_TIMER", "Watchdog timer"),
|
||||
("PI_RUNNING", "Pi running flag"),
|
||||
("LED_STATE", "LED state"),
|
||||
("TOUCH_STATE", "Touch state"),
|
||||
]
|
||||
|
||||
|
||||
def parse_cli_output(text):
|
||||
"""Parse `KEY = value` lines into a dict; skip anything malformed."""
|
||||
values = {}
|
||||
for line in text.splitlines():
|
||||
key, sep, value = line.partition("=")
|
||||
if not sep:
|
||||
continue
|
||||
try:
|
||||
values[key.strip()] = int(value.strip())
|
||||
except ValueError:
|
||||
LOG.warning("Skipping malformed line: %r", line)
|
||||
return values
|
||||
|
||||
|
||||
def read_values():
|
||||
"""Run the CLI (or use the embedded sample) and return parsed registers."""
|
||||
if os.environ.get("LIFEPO4WERED_FAKE"):
|
||||
return parse_cli_output(SAMPLE_OUTPUT)
|
||||
output = subprocess.run(
|
||||
CLI_COMMAND, capture_output=True, text=True,
|
||||
timeout=CLI_TIMEOUT, check=True,
|
||||
).stdout
|
||||
return parse_cli_output(output)
|
||||
|
||||
|
||||
def discovery_messages(discovery_prefix, state_topic, availability_topic):
|
||||
"""Yield (topic, config-dict) pairs for every Home Assistant entity."""
|
||||
|
||||
def entity(component, key, name, **extra):
|
||||
cfg = {
|
||||
"name": name,
|
||||
"unique_id": "lifepo4wered_" + key.lower(),
|
||||
"state_topic": state_topic,
|
||||
"availability_topic": availability_topic,
|
||||
"device": DEVICE,
|
||||
}
|
||||
cfg.update(extra)
|
||||
topic = "%s/%s/lifepo4wered/%s/config" % (
|
||||
discovery_prefix, component, key.lower())
|
||||
return topic, cfg
|
||||
|
||||
def millivolts(key):
|
||||
return "{{ (value_json.%s / 1000) | round(3) }}" % key
|
||||
|
||||
for key, name in MAIN_VOLTAGE:
|
||||
yield entity(
|
||||
"sensor", key, name,
|
||||
device_class="voltage",
|
||||
unit_of_measurement="V",
|
||||
state_class="measurement",
|
||||
suggested_display_precision=2,
|
||||
value_template=millivolts(key),
|
||||
)
|
||||
|
||||
yield entity(
|
||||
"sensor", "IOUT", "Output current",
|
||||
device_class="current",
|
||||
unit_of_measurement="A",
|
||||
state_class="measurement",
|
||||
suggested_display_precision=2,
|
||||
value_template="{{ (value_json.IOUT / 1000) | round(3) }}",
|
||||
)
|
||||
|
||||
yield entity(
|
||||
"binary_sensor", "EXTERNAL_POWER", "External power",
|
||||
device_class="power",
|
||||
value_template=(
|
||||
"{{ 'ON' if value_json.VIN >= value_json.VIN_THRESHOLD"
|
||||
" else 'OFF' }}"),
|
||||
)
|
||||
|
||||
for key, name in DIAG_VOLTAGE:
|
||||
yield entity(
|
||||
"sensor", key, name,
|
||||
device_class="voltage",
|
||||
unit_of_measurement="V",
|
||||
entity_category="diagnostic",
|
||||
suggested_display_precision=2,
|
||||
value_template=millivolts(key),
|
||||
)
|
||||
|
||||
for key, name in DIAG_RAW:
|
||||
yield entity(
|
||||
"sensor", key, name,
|
||||
entity_category="diagnostic",
|
||||
value_template="{{ value_json.%s }}" % key,
|
||||
)
|
||||
|
||||
yield entity(
|
||||
"sensor", "RTC_TIME", "UPS clock",
|
||||
device_class="timestamp",
|
||||
entity_category="diagnostic",
|
||||
value_template="{{ value_json.RTC_TIME | as_datetime }}",
|
||||
)
|
||||
|
||||
|
||||
def make_client(host, port, username, password, availability_topic):
|
||||
client_kwargs = {"client_id": "lifepo4wered_monitor"}
|
||||
try:
|
||||
client = mqtt.Client(
|
||||
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
|
||||
**client_kwargs)
|
||||
except AttributeError: # paho-mqtt 1.x
|
||||
client = mqtt.Client(**client_kwargs)
|
||||
if username:
|
||||
client.username_pw_set(username, password or None)
|
||||
client.will_set(availability_topic, "offline", retain=True)
|
||||
client.connect(host, port, keepalive=60)
|
||||
return client
|
||||
|
||||
|
||||
def main():
|
||||
logging.basicConfig(
|
||||
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
|
||||
format="%(asctime)s %(levelname)s %(message)s")
|
||||
|
||||
host = os.environ.get("MQTT_HOST", "localhost")
|
||||
port = int(os.environ.get("MQTT_PORT", "1883"))
|
||||
username = os.environ.get("MQTT_USER", "")
|
||||
password = os.environ.get("MQTT_PASSWORD", "")
|
||||
interval = float(os.environ.get("POLL_INTERVAL", "30"))
|
||||
discovery_prefix = os.environ.get("DISCOVERY_PREFIX", "homeassistant")
|
||||
topic_prefix = os.environ.get("MQTT_TOPIC_PREFIX", "lifepo4wered")
|
||||
|
||||
state_topic = topic_prefix + "/state"
|
||||
availability_topic = topic_prefix + "/availability"
|
||||
|
||||
if mqtt is None:
|
||||
LOG.error("paho-mqtt is not installed")
|
||||
return 1
|
||||
|
||||
def on_connect(client, userdata, flags, reason_code, properties=None):
|
||||
LOG.info("Connected to MQTT broker %s:%d", host, port)
|
||||
for topic, cfg in discovery_messages(
|
||||
discovery_prefix, state_topic, availability_topic):
|
||||
client.publish(topic, json.dumps(cfg), retain=True)
|
||||
client.publish(availability_topic, "online", retain=True)
|
||||
|
||||
client = make_client(host, port, username, password, availability_topic)
|
||||
client.on_connect = on_connect
|
||||
client.loop_start()
|
||||
|
||||
running = True
|
||||
|
||||
def stop(signum, frame):
|
||||
nonlocal running
|
||||
running = False
|
||||
|
||||
signal.signal(signal.SIGTERM, stop)
|
||||
signal.signal(signal.SIGINT, stop)
|
||||
|
||||
online = True
|
||||
while running:
|
||||
try:
|
||||
values = read_values()
|
||||
except (subprocess.SubprocessError, OSError) as exc:
|
||||
LOG.error("Failed to read UPS state: %s", exc)
|
||||
if online:
|
||||
client.publish(availability_topic, "offline", retain=True)
|
||||
online = False
|
||||
else:
|
||||
if not online:
|
||||
client.publish(availability_topic, "online", retain=True)
|
||||
online = True
|
||||
client.publish(state_topic, json.dumps(values), retain=True)
|
||||
LOG.debug("Published %d values", len(values))
|
||||
# sleep in small steps so SIGTERM is handled promptly
|
||||
deadline = time.monotonic() + interval
|
||||
while running and time.monotonic() < deadline:
|
||||
time.sleep(1)
|
||||
|
||||
LOG.info("Shutting down")
|
||||
client.publish(availability_topic, "offline", retain=True)
|
||||
client.loop_stop()
|
||||
client.disconnect()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,3 +1,33 @@
|
||||
#!/usr/bin/with-contenv bashio
|
||||
|
||||
echo Hello world!
|
||||
export POLL_INTERVAL="$(bashio::config 'poll_interval')"
|
||||
|
||||
# Manual MQTT settings from addon options win; otherwise use the broker
|
||||
# provided by the Supervisor (e.g. the Mosquitto addon).
|
||||
if bashio::config.has_value 'mqtt_host'; then
|
||||
export MQTT_HOST="$(bashio::config 'mqtt_host')"
|
||||
if bashio::config.has_value 'mqtt_port'; then
|
||||
export MQTT_PORT="$(bashio::config 'mqtt_port')"
|
||||
fi
|
||||
if bashio::config.has_value 'mqtt_user'; then
|
||||
export MQTT_USER="$(bashio::config 'mqtt_user')"
|
||||
fi
|
||||
if bashio::config.has_value 'mqtt_password'; then
|
||||
export MQTT_PASSWORD="$(bashio::config 'mqtt_password')"
|
||||
fi
|
||||
elif bashio::services.available "mqtt"; then
|
||||
export MQTT_HOST="$(bashio::services mqtt 'host')"
|
||||
export MQTT_PORT="$(bashio::services mqtt 'port')"
|
||||
export MQTT_USER="$(bashio::services mqtt 'username')"
|
||||
export MQTT_PASSWORD="$(bashio::services mqtt 'password')"
|
||||
else
|
||||
bashio::log.warning \
|
||||
"No MQTT broker configured and no mqtt service available;" \
|
||||
"trying localhost:1883"
|
||||
fi
|
||||
|
||||
bashio::log.info "Starting lifepo4wered-daemon"
|
||||
lifepo4wered-daemon -f &
|
||||
|
||||
bashio::log.info "Starting MQTT monitor (poll interval: ${POLL_INTERVAL}s)"
|
||||
exec python3 /usr/bin/lifepo4wered_monitor.py
|
||||
|
||||
BIN
tests/__pycache__/test_monitor.cpython-314.pyc
Normal file
BIN
tests/__pycache__/test_monitor.cpython-314.pyc
Normal file
Binary file not shown.
0
tests/stubs/paho/__init__.py
Normal file
0
tests/stubs/paho/__init__.py
Normal file
BIN
tests/stubs/paho/__pycache__/__init__.cpython-314.pyc
Normal file
BIN
tests/stubs/paho/__pycache__/__init__.cpython-314.pyc
Normal file
Binary file not shown.
0
tests/stubs/paho/mqtt/__init__.py
Normal file
0
tests/stubs/paho/mqtt/__init__.py
Normal file
BIN
tests/stubs/paho/mqtt/__pycache__/__init__.cpython-314.pyc
Normal file
BIN
tests/stubs/paho/mqtt/__pycache__/__init__.cpython-314.pyc
Normal file
Binary file not shown.
BIN
tests/stubs/paho/mqtt/__pycache__/client.cpython-314.pyc
Normal file
BIN
tests/stubs/paho/mqtt/__pycache__/client.cpython-314.pyc
Normal file
Binary file not shown.
41
tests/stubs/paho/mqtt/client.py
Normal file
41
tests/stubs/paho/mqtt/client.py
Normal file
@@ -0,0 +1,41 @@
|
||||
"""Minimal paho-mqtt stand-in for end-to-end smoke tests without a broker.
|
||||
|
||||
Every publish is appended to the file named by $STUB_MQTT_LOG as
|
||||
"<topic>\t<payload>".
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
|
||||
class CallbackAPIVersion:
|
||||
VERSION1 = 1
|
||||
VERSION2 = 2
|
||||
|
||||
|
||||
class Client:
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.on_connect = None
|
||||
self._log = open(os.environ["STUB_MQTT_LOG"], "a")
|
||||
|
||||
def username_pw_set(self, username, password=None):
|
||||
pass
|
||||
|
||||
def will_set(self, topic, payload=None, qos=0, retain=False):
|
||||
pass
|
||||
|
||||
def connect(self, host, port=1883, keepalive=60):
|
||||
pass
|
||||
|
||||
def loop_start(self):
|
||||
if self.on_connect:
|
||||
self.on_connect(self, None, {}, 0, None)
|
||||
|
||||
def loop_stop(self):
|
||||
pass
|
||||
|
||||
def disconnect(self):
|
||||
self._log.close()
|
||||
|
||||
def publish(self, topic, payload=None, qos=0, retain=False):
|
||||
self._log.write("%s\t%s\n" % (topic, payload))
|
||||
self._log.flush()
|
||||
97
tests/test_monitor.py
Normal file
97
tests/test_monitor.py
Normal file
@@ -0,0 +1,97 @@
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..",
|
||||
"hassio-addon-lifepo4wered"))
|
||||
|
||||
import lifepo4wered_monitor as mon
|
||||
|
||||
|
||||
class ParseOutputTest(unittest.TestCase):
|
||||
def test_parses_sample_output(self):
|
||||
values = mon.parse_cli_output(mon.SAMPLE_OUTPUT)
|
||||
self.assertEqual(values["VIN"], 4709)
|
||||
self.assertEqual(values["VBAT"], 3233)
|
||||
self.assertEqual(values["IOUT"], 1355)
|
||||
self.assertEqual(values["PI_RUNNING"], 1)
|
||||
self.assertEqual(len(values), 36)
|
||||
|
||||
def test_skips_malformed_lines(self):
|
||||
values = mon.parse_cli_output("VIN = 5000\ngarbage line\nVBAT = abc\n")
|
||||
self.assertEqual(values, {"VIN": 5000})
|
||||
|
||||
def test_handles_negative_values(self):
|
||||
values = mon.parse_cli_output("IOUT_OFFSET = -12\n")
|
||||
self.assertEqual(values["IOUT_OFFSET"], -12)
|
||||
|
||||
|
||||
class DiscoveryTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.messages = dict(mon.discovery_messages(
|
||||
discovery_prefix="homeassistant",
|
||||
state_topic="lifepo4wered/state",
|
||||
availability_topic="lifepo4wered/availability",
|
||||
))
|
||||
|
||||
def test_main_voltage_sensor(self):
|
||||
topic = "homeassistant/sensor/lifepo4wered/vin/config"
|
||||
self.assertIn(topic, self.messages)
|
||||
cfg = self.messages[topic]
|
||||
self.assertEqual(cfg["device_class"], "voltage")
|
||||
self.assertEqual(cfg["unit_of_measurement"], "V")
|
||||
self.assertEqual(cfg["state_class"], "measurement")
|
||||
self.assertIn("value_json.VIN", cfg["value_template"])
|
||||
self.assertNotIn("entity_category", cfg)
|
||||
|
||||
def test_current_sensor(self):
|
||||
cfg = self.messages["homeassistant/sensor/lifepo4wered/iout/config"]
|
||||
self.assertEqual(cfg["device_class"], "current")
|
||||
self.assertEqual(cfg["unit_of_measurement"], "A")
|
||||
|
||||
def test_external_power_binary_sensor(self):
|
||||
topic = "homeassistant/binary_sensor/lifepo4wered/external_power/config"
|
||||
self.assertIn(topic, self.messages)
|
||||
cfg = self.messages[topic]
|
||||
self.assertEqual(cfg["device_class"], "power")
|
||||
self.assertIn("VIN_THRESHOLD", cfg["value_template"])
|
||||
|
||||
def test_diagnostic_entities_are_categorized(self):
|
||||
cfg = self.messages["homeassistant/sensor/lifepo4wered/vbat_shdn/config"]
|
||||
self.assertEqual(cfg["entity_category"], "diagnostic")
|
||||
self.assertEqual(cfg["unit_of_measurement"], "V")
|
||||
|
||||
def test_skipped_registers_have_no_entity(self):
|
||||
for key in ("i2c_reg_ver", "dco_rsel", "vbat_offset", "cfg_write"):
|
||||
self.assertNotIn(
|
||||
"homeassistant/sensor/lifepo4wered/%s/config" % key,
|
||||
self.messages)
|
||||
|
||||
def test_common_fields(self):
|
||||
for topic, cfg in self.messages.items():
|
||||
self.assertEqual(cfg["state_topic"], "lifepo4wered/state")
|
||||
self.assertEqual(cfg["availability_topic"],
|
||||
"lifepo4wered/availability")
|
||||
self.assertTrue(cfg["unique_id"].startswith("lifepo4wered_"),
|
||||
topic)
|
||||
self.assertEqual(cfg["device"]["identifiers"], ["lifepo4wered_pi"])
|
||||
json.dumps(cfg) # must be serializable
|
||||
|
||||
def test_unique_ids_are_unique(self):
|
||||
ids = [cfg["unique_id"] for cfg in self.messages.values()]
|
||||
self.assertEqual(len(ids), len(set(ids)))
|
||||
|
||||
|
||||
class ReadValuesTest(unittest.TestCase):
|
||||
def test_fake_mode(self):
|
||||
os.environ["LIFEPO4WERED_FAKE"] = "1"
|
||||
try:
|
||||
values = mon.read_values()
|
||||
finally:
|
||||
del os.environ["LIFEPO4WERED_FAKE"]
|
||||
self.assertEqual(values["VOUT"], 4993)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user