fix(sensors): normalize Victron power precision and dependency setup

This commit is contained in:
yellowcooln committed 2026-10-12 15:02:09 -04:00
1 parent 1f2fb86c78
commit 127b0a4df1
5 files changed
+123 -5

No files matched your search

+1 -1
View File
@@ -268,7 +268,7 @@ sensors:
enabled: true
# Read-only Victron VE.Direct USB telemetry; see docs/victron_vedirect.md.
# Install pyserial in the service environment; no runtime pip for this sensor.
# Standard dependency checks at initialization; auto-install disabled below.
# Stop cat/other serial readers before enabling. Unique name per device.
# - type: victron_vedirect
# name: solar-charger
+2 -2
View File
@@ -12,7 +12,7 @@ Install into the **same Python environment used by the service**, not an unrelat
sudo /opt/openhop_repeater/venv/bin/python -m pip install pyserial
```
The standard package already declares pyserial; this command is useful for minimal/custom installations. Import and type discovery remain safe without it: reads report unavailable and never invoke runtime pip, even if global sensor auto-install is enabled.
The standard package already declares pyserial; this command is useful for minimal/custom installations. The sensor now uses `SensorBase.ensure_python_modules` during initialization, respecting the per-sensor/global `auto_install_packages` setting. If disabled, missing pyserial leaves the sensor unavailable; if enabled, the standard helper may install it during initialization. Type discovery remains safe without pyserial and never installs packages. Polling never runs pip or repeats dependency checks. Restart/reload the backend after resolving an initialization-time dependency failure.
On Raspberry Pi OS/Debian, check the device's ownership and grant the actual service account access to its group (normally `dialout`):
@@ -100,7 +100,7 @@ Fields depend on model, firmware and configured auxiliary input. This table desc
### Calculated load power
VE.Direct does not provide a direct load-power field here. When both battery voltage and load current are valid in the current poll, the sensor calculates `load_power_w = battery_voltage_v * load_current_a` after merging that poll's validated blocks. This is an estimate using battery voltage, not a separately measured power value. Zero is a valid result and the current's sign is preserved. Missing or invalid inputs never reuse a previous value; a previously discovered calculated card becomes unavailable with the corresponding reason. Its descriptor uses `/derived/load_power_w` to identify a calculated local field, not a wire-protocol label. No UI or configuration change is required, and the existing V1 grouping remains intact.
VE.Direct does not provide a direct load-power field here. When both battery voltage and load current are valid in the current poll, the sensor calculates `load_power_w = battery_voltage_v * load_current_a` after merging that poll's validated blocks. This is an estimate using battery voltage, not a separately measured power value. Zero is a valid result and the current's sign is preserved. The calculated product is rounded to six decimal places, matching millivolt × milliamp input resolution and removing binary floating-point tails without adding trailing zeros. Missing or invalid inputs never reuse a previous value; a previously discovered calculated card becomes unavailable with the corresponding reason. Its descriptor uses `/derived/load_power_w` to identify a calculated local field, not a wire-protocol label. No UI or configuration change is required, and the existing V1 grouping remains intact.
### Readable status values — sensor only
+13 -2
View File
@@ -304,6 +304,14 @@ class VictronVEDirectSensor(SensorBase):
self._seen = set()
self._data = {}
self._load_power_seen = False
self._serial = None
if self.ensure_python_modules([("serial", "pyserial")]):
try:
self._serial = importlib.import_module("serial")
except ImportError:
self.log.warning(
"pyserial unavailable; install pyserial in the service environment"
)
def _reading_metrics(self, ok):
metrics = []
@@ -360,6 +368,8 @@ class VictronVEDirectSensor(SensorBase):
def _read(self):
self._data = {}
if self._serial is None:
raise RuntimeError("pyserial unavailable; install pyserial in the service environment")
try:
self._data = self._read_serial()
except ImportError:
@@ -375,13 +385,14 @@ class VictronVEDirectSensor(SensorBase):
self._load_power_seen = True
voltage = self._data["battery_voltage_v"]
current = self._data["load_current_a"]
# mV × mA resolves to micro-watts; discard only binary float noise.
self._data["load_power_w"] = (
voltage * current if voltage is not None and current is not None else None
round(voltage * current, 6) if voltage is not None and current is not None else None
)
return self._data
def _read_serial(self):
serial = importlib.import_module("serial")
serial = self._serial
deadline = time.monotonic() + self.read_timeout_seconds
parser = VEDirectParser()
data = {}
+97
View File
@@ -0,0 +1,97 @@
"""Initialization-only dependency handling; serial ports remain poll-scoped."""
from unittest.mock import MagicMock
import pytest
from repeater.sensors.base import SensorBase
from tests.test_victron_vedirect import manager
def test_dependency_check_runs_once_at_initialization_and_blocks_reads_when_missing(monkeypatch):
import serial
check = MagicMock(return_value=False)
monkeypatch.setattr(SensorBase, "ensure_python_modules", check)
open_port = MagicMock()
monkeypatch.setattr(serial, "Serial", open_port)
sm = manager()
assert len(sm.sensors) == 1
check.assert_called_once_with([("serial", "pyserial")])
for _ in range(2):
reading = sm.read_all()[0]
assert not reading["ok"] and reading["data"] == {}
assert "pyserial unavailable" in reading["error"]
assert check.call_count == 1
open_port.assert_not_called()
@pytest.mark.parametrize(
"auto_install,install_succeeds", [(False, False), (True, False), (True, True)]
)
def test_standard_helper_honors_auto_install_only_during_initialization(
monkeypatch, auto_install, install_succeeds
):
import sys
import serial
from repeater.sensors import base
from repeater.sensors.manager import SensorManager
from tests.test_victron_vedirect import frame
real_find_spec = base.importlib.util.find_spec
installed = False
def find_spec(name):
if name == "serial" and not installed:
return None
return real_find_spec(name)
def install(*args, **kwargs):
nonlocal installed
installed = install_succeeds
return MagicMock(
returncode=0 if install_succeeds else 1, stderr="synthetic pip failure", stdout=""
)
monkeypatch.setattr(base.importlib.util, "find_spec", find_spec)
pip = MagicMock(side_effect=install)
monkeypatch.setattr(base.subprocess, "run", pip)
port = MagicMock()
port.__enter__.return_value = port
port.read.return_value = frame({"V": 13490, "IL": 200})
open_port = MagicMock(return_value=port)
monkeypatch.setattr(serial, "Serial", open_port)
sm = SensorManager(
{
"sensors": {
"definitions": [
{
"type": "victron_vedirect",
"name": "solar",
"auto_install_packages": auto_install,
"settings": {
"serial_device": "/dev/synthetic",
"read_timeout_seconds": 0.05,
},
}
]
}
}
)
assert len(sm.sensors) == 1
open_port.assert_not_called()
expected_pip_calls = int(auto_install)
assert pip.call_count == expected_pip_calls
if auto_install:
assert pip.call_args.args[0] == [sys.executable, "-m", "pip", "install", "pyserial"]
for _ in range(2):
reading = sm.read_all()[0]
assert reading["ok"] == (auto_install and install_succeeds)
assert pip.call_count == expected_pip_calls
if auto_install and install_succeeds:
assert reading["data"]["load_power_w"] == 2.698
assert open_port.call_count == 2
assert port.__exit__.call_count == 2
else:
assert reading["data"] == {}
open_port.assert_not_called()
+10
View File
@@ -110,6 +110,16 @@ def test_load_power_never_combines_inputs_from_different_polls(monkeypatch):
assert all(metric["data_key"] != "load_power_w" for metric in result["metrics"])
@pytest.mark.parametrize("current,expected", [(200, 2.698), (-200, -2.698), (1, 0.01349)])
def test_load_power_removes_float_noise_without_losing_input_precision(
monkeypatch, current, expected
):
serial_port(monkeypatch, frame({"V": 13490, "IL": current}))
result = manager().read_all()[0]
# Exact comparison: approximate comparisons hide the reported rendering bug.
assert result["data"]["load_power_w"] == expected
def test_load_power_does_not_use_value_invalidated_later_in_same_poll(monkeypatch):
serial_port(monkeypatch, frame({"V": 12800, "IL": 1500}) + frame({"IL": "---"}))
result = manager().read_all()[0]