Compare commits

...

5 Commits

Author SHA1 Message Date
vaku
93c1f8d3bc Fix entity names and add units per the product brief
VOUT_MAX is the safety limit above which the UPS refuses to boot an
externally-powered Pi, not an output voltage setting. Also clarify
PI_RUNNING, WATCHDOG_GRACE and TOUCH_STATE names, and add documented
units: timeouts and watchdog values are CLI-scaled seconds,
AUTO_SHDN_TIME is minutes; SHDN_DELAY stays raw ticks.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-19 00:53:01 +03:00
vaku
c1c96d3cfb Rename VIN_THRESHOLD entity to Input voltage threshold
It is a voltage level, not a power quantity.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-19 00:50:00 +03:00
vaku
3d8deb1176 Disable Docker init so s6-overlay v3 can run as PID 1
The latest homeassistant base images use s6-overlay v3, whose /init
refuses to run unless it is PID 1. Set init:false so the Supervisor does
not inject docker-init in front of it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-18 13:53:42 +03:00
vaku
952fc24e90 Make lifepo4wered-daemon the primary process in run.sh
The daemon must contact the UPS within PI_BOOT_TO seconds of power-on
or the UPS cuts power. Start it immediately via exec as the container's
main process; run the MQTT monitor in a background loop with 30s restart
backoff so Supervisor API delays or monitor crashes can never delay or
kill the daemon.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-18 01:18:46 +03:00
vaku
25b3e2dcc4 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>
2026-07-18 01:15:03 +03:00
15 changed files with 603 additions and 9 deletions

View File

@@ -1,5 +1,37 @@
# hassio-addon-lifepo4wered # 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 https://www.linkedin.com/pulse/creating-your-first-home-assistant-add-on-issac-goldstand

View File

@@ -0,0 +1,69 @@
# 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): `lifepo4wered-daemon -f` is the exec'd main process,
started immediately — it must contact the UPS within PI_BOOT_TO seconds of
power-on or the UPS cuts power, so nothing (Supervisor API calls included)
may delay it and monitor failures must never take it down. The monitor runs
in a background subshell that pulls MQTT host/port/user/password from the
Supervisor services API (works out of the box with the Mosquitto addon,
addon options can override) and restarts the monitor with 30 s backoff if
it exits.
- `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.

View File

@@ -9,11 +9,13 @@ FROM homeassistant/aarch64-base:latest
ENV LANG C.UTF-8 ENV LANG C.UTF-8
WORKDIR / 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/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-cli /usr/bin/lifepo4wered-cli
COPY --from=build /tmp/LiFePO4wered-Pi/build/lifepo4wered-daemon /usr/sbin/lifepo4wered-daemon COPY --from=build /tmp/LiFePO4wered-Pi/build/lifepo4wered-daemon /usr/sbin/lifepo4wered-daemon
# COPY start.sh /app/start.sh COPY lifepo4wered_monitor.py /usr/bin/lifepo4wered_monitor.py
# ENTRYPOINT ["/app/start.sh"] COPY run.sh /run.sh
CMD lifepo4wered-daemon -f RUN chmod a+x /run.sh
CMD ["/run.sh"]
LABEL io.hass.version="VERSION" io.hass.type="addon" io.hass.arch="aarch64" LABEL io.hass.version="VERSION" io.hass.type="addon" io.hass.arch="aarch64"

View File

@@ -1,14 +1,23 @@
{ {
"name": "LiFePO4wered Addon", "name": "LiFePO4wered Addon",
"version": "0.0.5", "version": "0.1.3",
"init": false,
"slug": "lifepo4wered", "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", "url": "https://github.com/vakius/hassio-addon-lifepo4wered/tree/master/hassio-addon-lifepo4wered",
"startup": "application", "startup": "application",
"boot": "auto", "boot": "auto",
"options": {}, "services": ["mqtt:want"],
"schema": {}, "options": {
"poll_interval": 30
},
"schema": {
"poll_interval": "int(5,3600)",
"mqtt_host": "str?",
"mqtt_port": "port?",
"mqtt_user": "str?",
"mqtt_password": "password?"
},
"arch": ["aarch64"], "arch": ["aarch64"],
"devices": [ "/dev/i2c-1" ] "devices": [ "/dev/i2c-1" ]
} }

View File

@@ -0,0 +1,289 @@
#!/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", "Max output voltage to allow boot"),
("VIN_THRESHOLD", "Input voltage threshold"),
]
# raw-value diagnostic registers; units per the product brief (the CLI
# scales the timeouts to s and AUTO_SHDN_TIME to min; SHDN_DELAY stays
# raw ~8 Hz ticks)
DIAG_RAW = [
("AUTO_BOOT", "Auto boot mode", None),
("SHDN_DELAY", "Shutdown delay", None),
("AUTO_SHDN_TIME", "Auto shutdown time", "min"),
("PI_BOOT_TO", "Pi boot timeout", "s"),
("PI_SHDN_TO", "Pi shutdown timeout", "s"),
("WATCHDOG_CFG", "Watchdog mode", None),
("WATCHDOG_GRACE", "Watchdog grace period", "s"),
("WATCHDOG_TIMER", "Watchdog timer", "s"),
("PI_RUNNING", "Pi running", None),
("LED_STATE", "LED state", None),
("TOUCH_STATE", "Touch button state", None),
]
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, unit in DIAG_RAW:
extra = {"unit_of_measurement": unit} if unit else {}
yield entity(
"sensor", key, name,
entity_category="diagnostic",
value_template="{{ value_json.%s }}" % key,
**extra,
)
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())

View File

@@ -1,3 +1,47 @@
#!/usr/bin/with-contenv bashio #!/usr/bin/with-contenv bashio
echo Hello world! # The daemon must contact the UPS within PI_BOOT_TO seconds of power-on,
# otherwise the UPS assumes the boot failed and cuts power. It is therefore
# the main process, started before any MQTT/Supervisor API calls, and the
# monitor runs on the side where its failures can never take the daemon down.
setup_mqtt_env() {
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
}
(
setup_mqtt_env
bashio::log.info "Starting MQTT monitor (poll interval: ${POLL_INTERVAL}s)"
while true; do
python3 /usr/bin/lifepo4wered_monitor.py || true
bashio::log.warning "MQTT monitor exited; restarting in 30s"
sleep 30
done
) &
bashio::log.info "Starting lifepo4wered-daemon"
exec lifepo4wered-daemon -f

Binary file not shown.

View File

Binary file not shown.

View File

View 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()

108
tests/test_monitor.py Normal file
View File

@@ -0,0 +1,108 @@
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_timer_registers_have_documented_units(self):
for key, unit in (("pi_boot_to", "s"), ("pi_shdn_to", "s"),
("watchdog_grace", "s"), ("watchdog_timer", "s"),
("auto_shdn_time", "min")):
cfg = self.messages[
"homeassistant/sensor/lifepo4wered/%s/config" % key]
self.assertEqual(cfg["unit_of_measurement"], unit, key)
self.assertNotIn(
"unit_of_measurement",
self.messages["homeassistant/sensor/lifepo4wered/shdn_delay/config"])
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()