feat: use values.xml api when polling sensor values (#83)

* remove status.xml support

* use values.xml instead of values2.xml

* ensure WiFi secrets are scrubbed from logs

* ensure correct requirements in manifest.json

* update HACS action

.github/workflows/hacs.yaml

@@ -11,6 +11,6 @@ jobs:
     runs-on: "ubuntu-latest"
     steps:
         - uses: "actions/checkout@v2"
-        - uses: hacs/action@21.12.1
+        - uses: hacs/action@main
           with:
             category: 'integration'

custom_components/wibeee/api.py

@@ -1,22 +1,22 @@
 import asyncio
 import logging
 from datetime import timedelta
-from typing import NamedTuple, Optional
+from typing import NamedTuple, Optional, Dict
 from urllib.parse import quote_plus
 
 import aiohttp
 import xmltodict
+from homeassistant.components.diagnostics import async_redact_data
 from homeassistant.helpers.typing import StateType
-from packaging import version
 
-from .util import scrub_xml_text_naively, scrub_dict_top_level
+from .util import scrub_values_xml
 
 _LOGGER = logging.getLogger(__name__)
 
 StatusResponse = dict[str, StateType]
 
-_VALUES2_SCRUB_KEYS = ['securKey', 'ssid']
-"""Values that we'll attempt to scrub from the values2.xml response."""
+_VALUES_SCRUB_KEYS = ['securKey', 'ssid']
+"""Values that we'll attempt to scrub from the values.xml response."""
 
 
 class DeviceInfo(NamedTuple):
@@ -30,8 +30,6 @@ class DeviceInfo(NamedTuple):
     "Wibeee Model (single or 3-phase, etc)"
     ipAddr: str
     "IP address"
-    use_values2: bool
-    "Whether to use values2.xml format"
 
 
 class WibeeeAPI(object):
@@ -46,38 +44,36 @@ def __init__(self, session: aiohttp.ClientSession, host: str, timeout: timedelta
         self.max_wait = min(timedelta(seconds=5), timeout)
         _LOGGER.info("Initializing WibeeeAPI with host: %s, timeout %s, max_wait: %s", host, self.timeout, self.max_wait)
 
-    async def async_fetch_status(self, device: DeviceInfo, var_names: list[str], retries: int = 0) -> dict[str, any]:
-        """Fetches the status XML from Wibeee as a dict, optionally retries"""
-        if device.use_values2:
-            url = f'http://{self.host}/services/user/values2.xml?id={quote_plus(device.id)}'
+    async def async_fetch_values(self, device_id: str, var_names: list[str] = None, retries: int = 0) -> Dict[str, any]:
+        """Fetches the values from Wibeee as a dict, optionally retries"""
+        if var_names:
+            var_ids = [f"{quote_plus(device_id)}.{quote_plus(var)}" for var in var_names]
+            query = f'var={"&".join(var_ids)}'
+        else:
+            query = f'id={quote_plus(device_id)}'
 
-            # attempt to scrub WiFi secrets before they make it into logs, etc.
-            values2_response = await self.async_fetch_url(url, retries, _VALUES2_SCRUB_KEYS)
-            return scrub_dict_top_level(_VALUES2_SCRUB_KEYS, values2_response['values'])
+        values = await self.async_fetch_url(f'http://{self.host}/services/user/values.xml?{query}', retries, scrub_keys=_VALUES_SCRUB_KEYS)
 
-        else:
-            status_response = await self.async_fetch_url(f'http://{self.host}/en/status.xml', retries)
-            return status_response['response']
+        # <values><variable><id>macAddr</id><value>11:11:11:11:11:11</value></variable></values>
+        values_vars = {var['id']: var['value'] for var in values['values']['variable']}
+
+        # attempt to scrub WiFi secrets before they make it into logs, etc.
+        return async_redact_data(values_vars, _VALUES_SCRUB_KEYS)
 
     async def async_fetch_device_info(self, retries: int = 0) -> Optional[DeviceInfo]:
         # <devices><id>WIBEEE</id></devices>
         devices = await self.async_fetch_url(f'http://{self.host}/services/user/devices.xml', retries)
         device_id = devices['devices']['id']
 
         var_names = ['macAddr', 'softVersion', 'model', 'ipAddr']
-        var_ids = [f"{quote_plus(device_id)}.{name}" for name in var_names]
-        values = await self.async_fetch_url(f'http://{self.host}/services/user/values.xml?var={"&".join(var_ids)}', retries)
-
-        # <values><variable><id>macAddr</id><value>11:11:11:11:11:11</value></variable></values>
-        device_vars = {var['id']: var['value'] for var in values['values']['variable']}
+        device_vars = await self.async_fetch_values(device_id, var_names, retries)
 
         return DeviceInfo(
             device_id,
             device_vars['macAddr'].replace(':', ''),
             device_vars['softVersion'],
             device_vars['model'],
             device_vars['ipAddr'],
-            version.parse(device_vars['softVersion']) >= version.parse('4.4.171')
         ) if set(var_names) <= set(device_vars.keys()) else None
 
     async def async_fetch_url(self, url: str, retries: int = 0, scrub_keys: list[str] = []):
@@ -100,7 +96,7 @@ async def fetch_with_retries(try_n):
 
                 xml_data = await resp.text()
                 if _LOGGER.isEnabledFor(logging.DEBUG):
-                    _LOGGER.debug("RAW Response from %s: %s)", url, scrub_xml_text_naively(scrub_keys, xml_data))
+                    _LOGGER.debug("RAW Response from %s: %s)", url, scrub_values_xml(scrub_keys, await resp.read()))
 
                 xml_as_dict = xmltodict.parse(xml_data)
                 return xml_as_dict

custom_components/wibeee/manifest.json

@@ -6,11 +6,15 @@
   ],
   "config_flow": true,
   "dependencies": [
-    "network"
+    "network",
+    "diagnostics"
   ],
   "documentation": "https://github.com/luuuis/hass_wibeee",
   "iot_class": "local_polling",
   "issue_tracker": "https://github.com/luuuis/hass_wibeee/issues",
-  "requirements": [],
+  "requirements": [
+    "xmltodict==0.13.0",
+    "lxml==5.1.0"
+  ],
   "version": "3.4.3"
 }

custom_components/wibeee/sensor.py

@@ -10,7 +10,7 @@
 
 import logging
 from datetime import timedelta
-from typing import NamedTuple, Optional, Callable
+from typing import NamedTuple, Optional
 
 import homeassistant.helpers.config_validation as cv
 import voluptuous as vol
@@ -77,15 +77,11 @@
 class SensorType(NamedTuple):
     """
     Wibeee supported sensor definition.
-
-    One of `status_xml_suffix` (for older firmwares) or `value2_xml_prefix` (for newer firmwares) is required.
     """
-    status_xml_suffix: Optional[str]
-    "optional suffix used for elements in `status.xml` output (e.g.: 'vrms')"
-    values2_xml_prefix: Optional[str]
-    "optional suffix used for elements in `values2.xml` output (e.g.: 'vrms')"
-    nest_push_prefix: Optional[str]
-    "optional prefix used in Wibeee Nest push requests such as receiverLeap (e.g.: 'v')"
+    poll_var_prefix: Optional[str]
+    "prefix used for elements in `values.xml` output (e.g.: 'vrms')"
+    push_var_prefix: Optional[str]
+    "prefix used in Wibeee Nest push requests such as receiverLeap (e.g.: 'v')"
     unique_name: str
     "used to build the sensor unique_id (e.g.: 'Vrms')"
     friendly_name: str
@@ -97,18 +93,18 @@ class SensorType(NamedTuple):
 
 
 KNOWN_SENSORS = [
-    SensorType('vrms', 'vrms', 'v', 'Vrms', 'Phase Voltage', ELECTRIC_POTENTIAL_VOLT, SensorDeviceClass.VOLTAGE),
-    SensorType('irms', 'irms', 'i', 'Irms', 'Current', ELECTRIC_CURRENT_AMPERE, SensorDeviceClass.CURRENT),
-    SensorType('frecuencia', 'freq', 'q', 'Frequency', 'Frequency', FREQUENCY_HERTZ, device_class=None),
-    SensorType('p_activa', 'pac', 'a', 'Active_Power', 'Active Power', POWER_WATT, SensorDeviceClass.POWER),
-    SensorType(None, 'preac', 'r', 'Reactive_Power', 'Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
-    SensorType('p_reactiva_ind', None, 'r', 'Inductive_Reactive_Power', 'Inductive Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
-    SensorType('p_reactiva_cap', None, None, 'Capacitive_Reactive_Power', 'Capacitive Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
-    SensorType('p_aparent', 'pap', 'p', 'Apparent_Power', 'Apparent Power', POWER_VOLT_AMPERE, SensorDeviceClass.APPARENT_POWER),
-    SensorType('factor_potencia', 'fpot', 'f', 'Power_Factor', 'Power Factor', None, SensorDeviceClass.POWER_FACTOR),
-    SensorType('energia_activa', 'eac', 'e', 'Active_Energy', 'Active Energy', ENERGY_WATT_HOUR, SensorDeviceClass.ENERGY),
-    SensorType('energia_reactiva_ind', 'ereact', 'o', 'Inductive_Reactive_Energy', 'Inductive Reactive Energy', ENERGY_VOLT_AMPERE_REACTIVE_HOUR, SensorDeviceClass.ENERGY),
-    SensorType('energia_reactiva_cap', 'ereactc', None, 'Capacitive_Reactive_Energy', 'Capacitive Reactive Energy', ENERGY_VOLT_AMPERE_REACTIVE_HOUR, SensorDeviceClass.ENERGY),
+    SensorType('vrms', 'v', 'Vrms', 'Phase Voltage', ELECTRIC_POTENTIAL_VOLT, SensorDeviceClass.VOLTAGE),
+    SensorType('irms', 'i', 'Irms', 'Current', ELECTRIC_CURRENT_AMPERE, SensorDeviceClass.CURRENT),
+    SensorType('freq', 'q', 'Frequency', 'Frequency', FREQUENCY_HERTZ, device_class=None),
+    SensorType('pac', 'a', 'Active_Power', 'Active Power', POWER_WATT, SensorDeviceClass.POWER),
+    SensorType('preac', 'r', 'Reactive_Power', 'Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
+    SensorType(None, 'r', 'Inductive_Reactive_Power', 'Inductive Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
+    SensorType(None, None, 'Capacitive_Reactive_Power', 'Capacitive Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
+    SensorType('pap', 'p', 'Apparent_Power', 'Apparent Power', POWER_VOLT_AMPERE, SensorDeviceClass.APPARENT_POWER),
+    SensorType('fpot', 'f', 'Power_Factor', 'Power Factor', None, SensorDeviceClass.POWER_FACTOR),
+    SensorType('eac', 'e', 'Active_Energy', 'Active Energy', ENERGY_WATT_HOUR, SensorDeviceClass.ENERGY),
+    SensorType('ereact', 'o', 'Inductive_Reactive_Energy', 'Inductive Reactive Energy', ENERGY_VOLT_AMPERE_REACTIVE_HOUR, SensorDeviceClass.ENERGY),
+    SensorType('ereactc', None, 'Capacitive_Reactive_Energy', 'Capacitive Reactive Energy', ENERGY_VOLT_AMPERE_REACTIVE_HOUR, SensorDeviceClass.ENERGY),
 ]
 
 KNOWN_MODELS = {
@@ -144,26 +140,16 @@ class StatusElement(NamedTuple):
     sensor_type: SensorType
 
 
-def get_status_elements(device: DeviceInfo) -> list[StatusElement]:
+def get_status_elements() -> list[StatusElement]:
     """Returns the expected elements in the status XML response for this device."""
 
-    class StatusLookup(NamedTuple):
-        """Strategy for handling `status.xml` or `values2.xml` response lookups."""
-        is_expected: Callable[[SensorType], bool]
-        xml_names: Callable[[SensorType], list[(str, str)]]
-
-    lookup = StatusLookup(
-        lambda s: s.values2_xml_prefix is not None,
-        lambda s: [('4' if phase == 't' else phase, f"{s.values2_xml_prefix}{phase}") for phase in ['1', '2', '3', 't']],
-    ) if device.use_values2 else StatusLookup(
-        lambda s: s.status_xml_suffix is not None,
-        lambda s: [(phase, f"fase{phase}_{s.status_xml_suffix}") for phase in ['1', '2', '3', '4']],
-    )
+    def get_xml_names(s: SensorType) -> list[(str, str)]:
+        return [('4' if ph == 't' else ph, f"{s.poll_var_prefix}{ph}") for ph in ['1', '2', '3', 't']]
 
     return [
         StatusElement(phase, xml_name, sensor_type)
-        for sensor_type in KNOWN_SENSORS if lookup.is_expected(sensor_type)
-        for phase, xml_name in lookup.xml_names(sensor_type)
+        for sensor_type in KNOWN_SENSORS if sensor_type.poll_var_prefix is not None
+        for phase, xml_name in get_xml_names(sensor_type)
     ]
 
 
@@ -178,17 +164,18 @@ def update_sensors(sensors, update_source, lookup_key, data):
 
 
 def setup_local_polling(hass: HomeAssistant, api: WibeeeAPI, device: DeviceInfo, sensors: list['WibeeeSensor'], scan_interval: timedelta):
-    source = 'values2.xml' if device.use_values2 else 'status.xml'
+    def poll_xml_param(sensor: WibeeeSensor) -> str:
+        return sensor.status_xml_param
 
     async def fetching_data(now=None):
         fetched = {}
         try:
-            fetched = await api.async_fetch_status(device, [s.status_xml_param for s in sensors], retries=3)
+            fetched = await api.async_fetch_values(device.id, retries=3)
         except Exception as err:
             if now is None:
                 raise PlatformNotReady from err
 
-        update_sensors(sensors, source, lambda s: s.status_xml_param, fetched)
+        update_sensors(sensors, 'values.xml', poll_xml_param, fetched)
 
     return async_track_time_interval(hass, fetching_data, scan_interval)
 
@@ -226,9 +213,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e
 
     api = WibeeeAPI(session, host, min(timeout, scan_interval))
     device = await api.async_fetch_device_info(retries=5)
-    status_elements = get_status_elements(device)
+    status_elements = get_status_elements()
 
-    initial_status = await api.async_fetch_status(device, [e.xml_name for e in status_elements], retries=10)
+    initial_status = await api.async_fetch_values(device.id, retries=10)
     sensors = [
         WibeeeSensor(device, e.phase, e.sensor_type, e.xml_name, initial_status.get(e.xml_name))
         for e in status_elements if e.xml_name in initial_status
@@ -269,7 +256,7 @@ def __init__(self, device: DeviceInfo, sensor_phase: str, sensor_type: SensorTyp
         self._attr_device_info = _make_device_info(device, sensor_phase)
         self.entity_id = f"sensor.{entity_id}"  # we don't want this derived from the name
         self.status_xml_param = status_xml_param
-        self.nest_push_param = f"{sensor_type.nest_push_prefix}{'t' if sensor_phase == '4' else sensor_phase}"
+        self.nest_push_param = f"{sensor_type.push_var_prefix}{'t' if sensor_phase == '4' else sensor_phase}"
 
     @callback
     def update_value(self, value: StateType, update_source: str = '') -> None:

custom_components/wibeee/util.py

@@ -1,22 +1,21 @@
-import re
+from io import BytesIO
+
+from lxml import etree
 
 
 def short_mac(mac_addr):
     """Returns the last 6 chars of the MAC address for showing in UI."""
     return mac_addr.replace(':', '')[-6:].upper()
 
 
-def scrub_xml_text_naively(keys: list[str], xml_text: str) -> str:
-    """Naively use a RegEx (facepalm) to scrub values from XML text."""
-    scrubbed_text = re.sub(f'<({"|".join(keys)})>.*?</\\1>', f'<\\1>*MASKED*</\\1>', xml_text)
-    return scrubbed_text
-
+def scrub_values_xml(keys: list[str], xml_text: bytes) -> str:
+    """Scrubs sensitive data from the values.xml response."""
+    tree = etree.parse(BytesIO(xml_text))
 
-def scrub_dict_top_level(keys: list[str], values: dict) -> dict:
-    """Scrubs values from """
-    scrubbed_values = dict(values)
-    for scrub_key in keys:
-        if scrub_key in values:
-            scrubbed_values.update({scrub_key: '*MASKED*'})
+    # <values><variable><id>ssid</id><value>MY_SSID</value></variable></values>
+    for key in keys:
+        values = tree.xpath(f"/values/variable[id/text()='{key}']/value")
+        for v in values:
+            v.text = '*MASKED*'
 
-    return scrubbed_values
+    return etree.tostring(tree)