feat(api): support values2.xml API used by firmware >= 4.4.171

custom_components/wibeee/api.py

@@ -1,13 +1,33 @@
 import asyncio
 import logging
 from datetime import timedelta
+from typing import NamedTuple, Optional
 from urllib.parse import quote_plus
 
 import aiohttp
 import xmltodict
+from homeassistant.helpers.typing import StateType
+from packaging import version
 
 _LOGGER = logging.getLogger(__name__)
 
+StatusResponse = dict[str, StateType]
+
+
+class DeviceInfo(NamedTuple):
+    id: str
+    "Device ID (default is 'WIBEEE')"
+    macAddr: str
+    "MAC address (formatted for use in HA)"
+    softVersion: str
+    "Firmware version"
+    model: str
+    "Wibeee Model (single or 3-phase, etc)"
+    ipAddr: str
+    "IP address"
+    use_values2: bool
+    "Whether to use values2.xml format"
+
 
 class WibeeeAPI(object):
     """Gets the latest data from Wibeee device."""
@@ -21,12 +41,20 @@ 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, retries: int = 0):
+    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"""
-        status = await self.async_fetch_url(f'http://{self.host}/en/status.xml', retries)
-        return status["response"]
+        if device.use_values2:
+            device_id = quote_plus(device.id)
+            # Temporarily disabled: request the specific vars we need, otherwise Wibeee will send down everything including WiFi keys.
+            # var_ids = [f"{device_id}.{name}" for name in var_names]
+            # values2_response_ = await self.async_fetch_url(f'http://{self.host}/services/user/values2.xml?var={"&".join(var_ids)}', retries)
+            values2_response = await self.async_fetch_url(f'http://{self.host}/services/user/values2.xml?id={device_id}', retries)
+            return values2_response['values']
+        else:
+            status_response = await self.async_fetch_url(f'http://{self.host}/en/status.xml', retries)
+            return status_response['response']
 
-    async def async_fetch_device_info(self, retries: int):
+    async def async_fetch_device_info(self, retries: int) -> 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']
@@ -38,11 +66,14 @@ async def async_fetch_device_info(self, retries: int):
         # <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']}
 
-        return {
-            **device_vars,
-            'macAddr': device_vars['macAddr'].replace(':', ''),
-            'id': device_id,
-        } if len(device_vars) == len(var_names) else None
+        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 len(device_vars) == len(var_names) else None
 
     async def async_fetch_url(self, url, retries: int = 0):
         async def fetch_with_retries(try_n):

custom_components/wibeee/config_flow.py

@@ -29,7 +29,7 @@ async def validate_input(hass: HomeAssistant, user_input: dict) -> [str, str, di
     except Exception as e:
         raise NoDeviceInfo from e
 
-    mac_addr = format_mac(device['macAddr'])
+    mac_addr = format_mac(device.macAddr)
     unique_id = mac_addr
     name = f"Wibeee {short_mac(mac_addr)}"
 

custom_components/wibeee/sensor.py

@@ -1,16 +1,16 @@
 """
 Support for Energy consumption Sensors from Circutor via local Web API
 
-Device's website: http://wibeee.circutor.com/
+Vendor docs: https://support.wibeee.com/space/CH/184025089/XML
 Documentation: https://github.com/luuuis/hass_wibeee/
 
 """
 
 REQUIREMENTS = ["xmltodict"]
 
 import logging
-from collections import namedtuple
 from datetime import timedelta
+from typing import NamedTuple, Optional, Callable
 
 import homeassistant.helpers.config_validation as cv
 import voluptuous as vol
@@ -34,24 +34,22 @@
     CONF_TIMEOUT,
     CONF_UNIQUE_ID,
     STATE_UNAVAILABLE,
-    PERCENTAGE,
 )
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import PlatformNotReady
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
-from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity import DeviceInfo as HassDeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.event import async_track_time_interval
 from homeassistant.helpers.typing import StateType
 from homeassistant.util import slugify
 
-from .api import WibeeeAPI
+from .api import WibeeeAPI, DeviceInfo
 from .const import (
     DOMAIN,
     DEFAULT_SCAN_INTERVAL,
     DEFAULT_TIMEOUT,
     CONF_NEST_UPSTREAM,
-    NEST_DEFAULT_UPSTREAM,
     NEST_PROXY_DISABLED,
 )
 from .nest import get_nest_proxy
@@ -76,38 +74,41 @@
 })
 
 
-class SensorType(namedtuple('SensorType', [
-    'status_xml_suffix',
-    'nest_push_prefix',
-    'unique_name',
-    'friendly_name',
-    'unit',
-    'device_class',
-])):
-    """\
-    SensorType: Wibeee supported sensor definition.
-
-    status_xml_suffix - the suffix used for elements in `status.xml` output (e.g.: "vrms")
-    nest_push_prefix  - optional prefix used in Wibeee Nest push requests such as receiverLeap (e.g.: "v")
-    friendly_name     - used to build the sensor name and entity id (e.g.: "Phase Voltage")
-    unique_name       - used to build the sensor unique_id (e.g.: "Vrms")
-    unit              - unit to use for the sensor (e.g.: "V")
-    device_class      - optional device class to use for the sensor (e.g.: "voltage")
+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')"
+    unique_name: str
+    "used to build the sensor unique_id (e.g.: 'Vrms')"
+    friendly_name: str
+    "used to build the sensor name and entity id (e.g.: 'Phase Voltage')"
+    unit: Optional[str]
+    "unit to use for the sensor (e.g.: 'V')"
+    device_class: Optional[str]
+    "optional device class to use for the sensor (e.g.: 'voltage')"
 
 
 KNOWN_SENSORS = [
-    SensorType('vrms', 'v', 'Vrms', 'Phase Voltage', ELECTRIC_POTENTIAL_VOLT, SensorDeviceClass.VOLTAGE),
-    SensorType('irms', 'i', 'Irms', 'Current', ELECTRIC_CURRENT_AMPERE, SensorDeviceClass.CURRENT),
-    SensorType('frecuencia', 'q', 'Frequency', 'Frequency', FREQUENCY_HERTZ, device_class=None),
-    SensorType('p_activa', 'a', 'Active_Power', 'Active Power', POWER_WATT, SensorDeviceClass.POWER),
-    SensorType('p_reactiva_ind', 'r', 'Inductive_Reactive_Power', 'Inductive Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
-    SensorType('p_reactiva_cap', None, 'Capacitive_Reactive_Power', 'Capacitive Reactive Power', POWER_VOLT_AMPERE_REACTIVE, SensorDeviceClass.REACTIVE_POWER),
-    SensorType('p_aparent', 'p', 'Apparent_Power', 'Apparent Power', POWER_VOLT_AMPERE, SensorDeviceClass.APPARENT_POWER),
-    SensorType('factor_potencia', 'f', 'Power_Factor', 'Power Factor', None, SensorDeviceClass.POWER_FACTOR),
-    SensorType('energia_activa', 'e', 'Active_Energy', 'Active Energy', ENERGY_WATT_HOUR, SensorDeviceClass.ENERGY),
-    SensorType('energia_reactiva_ind', 'o', 'Inductive_Reactive_Energy', 'Inductive Reactive Energy', ENERGY_VOLT_AMPERE_REACTIVE_HOUR, SensorDeviceClass.ENERGY),
-    SensorType('energia_reactiva_cap', None, 'Capacitive_Reactive_Energy', 'Capacitive Reactive Energy', ENERGY_VOLT_AMPERE_REACTIVE_HOUR, SensorDeviceClass.ENERGY),
+    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),
 ]
 
 KNOWN_MODELS = {
@@ -137,12 +138,33 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
     ))
 
 
-def create_sensors(device, status) -> list['WibeeeSensor']:
-    xml_suffixes = {sensor_type.status_xml_suffix: sensor_type for sensor_type in KNOWN_SENSORS}
-    phase_xml_suffixes = [(key[4:].split("_", 1), value) for key, value in status.items() if key.startswith('fase')]
-    known_sensors = [(phase, xml_suffixes[suffix], value) for ((phase, suffix), value) in phase_xml_suffixes if suffix in xml_suffixes]
+class StatusElement(NamedTuple):
+    phase: str
+    xml_name: str
+    sensor_type: SensorType
+
 
-    return [WibeeeSensor(device, phase, sensor_type, initial_value) for (phase, sensor_type, initial_value) in known_sensors]
+def get_status_elements(device: DeviceInfo) -> 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']],
+    )
+
+    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)
+    ]
 
 
 def update_sensors(sensors, update_source, lookup_key, data):
@@ -153,14 +175,14 @@ def update_sensors(sensors, update_source, lookup_key, data):
         s.update_value(value, update_source)
 
 
-def setup_local_polling(hass: HomeAssistant, api: WibeeeAPI, sensors: list['WibeeeSensor'], scan_interval: timedelta):
+def setup_local_polling(hass: HomeAssistant, api: WibeeeAPI, device: DeviceInfo, sensors: list['WibeeeSensor'], scan_interval: timedelta):
     def status_xml_param(sensor: WibeeeSensor) -> str:
         return sensor.status_xml_param
 
     async def fetching_data(now=None):
         try:
-            fetched = await api.async_fetch_status(retries=3)
-            update_sensors(sensors, 'status.xml', status_xml_param, fetched)
+            fetched = await api.async_fetch_status(device, [s.status_xml_param for s in sensors], retries=3)
+            update_sensors(sensors, 'values2.xml' if device.use_values2 else 'status.xml', status_xml_param, fetched)
         except Exception as err:
             if now is None:
                 raise PlatformNotReady from err
@@ -204,16 +226,21 @@ 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)
-    initial_status = await api.async_fetch_status(retries=10)
+    status_elements = get_status_elements(device)
+
+    initial_status = await api.async_fetch_status(device, [e.xml_name for e in status_elements], 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
+    ]
 
-    sensors = create_sensors(device, initial_status)
     for sensor in sensors:
         _LOGGER.debug("Adding '%s' (unique_id=%s)", sensor, sensor.unique_id)
     async_add_entities(sensors, True)
 
     disposers = hass.data[DOMAIN][entry.entry_id]['disposers']
 
-    remove_fetch_listener = setup_local_polling(hass, api, sensors, scan_interval)
+    remove_fetch_listener = setup_local_polling(hass, api, device, sensors, scan_interval)
     disposers.update(fetch_status=remove_fetch_listener)
 
     if use_nest_proxy:
@@ -227,9 +254,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e
 class WibeeeSensor(SensorEntity):
     """Implementation of Wibeee sensor."""
 
-    def __init__(self, device, sensor_phase: str, sensor_type: SensorType, initial_value: StateType):
+    def __init__(self, device: DeviceInfo, sensor_phase: str, sensor_type: SensorType, status_xml_param: str, initial_value: StateType):
         """Initialize the sensor."""
-        [device_name, mac_addr] = [device['id'], device['macAddr']]
+        [device_name, mac_addr] = [device.id, device.macAddr]
         entity_id = slugify(f"{DOMAIN} {mac_addr} {sensor_type.friendly_name} L{sensor_phase}")
         self._attr_native_unit_of_measurement = sensor_type.unit
         self._attr_native_value = initial_value
@@ -241,7 +268,7 @@ def __init__(self, device, sensor_phase: str, sensor_type: SensorType, initial_v
         self._attr_should_poll = False
         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 = f"fase{sensor_phase}_{sensor_type.status_xml_suffix}"
+        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}"
 
     @callback
@@ -254,14 +281,14 @@ def update_value(self, value, update_source='') -> None:
             _LOGGER.debug("Updating from %s: %s", update_source, self)
 
 
-def _make_device_info(device, sensor_phase) -> DeviceInfo:
-    mac_addr = device['macAddr']
+def _make_device_info(device: DeviceInfo, sensor_phase) -> HassDeviceInfo:
+    mac_addr = device.macAddr
     is_clamp = sensor_phase != '4'
 
     device_name = f'Wibeee {short_mac(mac_addr)}'
-    device_model = KNOWN_MODELS.get(device['model'], 'Wibeee Energy Meter')
+    device_model = KNOWN_MODELS.get(device.model, 'Wibeee Energy Meter')
 
-    return DeviceInfo(
+    return HassDeviceInfo(
         # identifiers and links
         identifiers={(DOMAIN, f'{mac_addr}_L{sensor_phase}' if is_clamp else mac_addr)},
         via_device=(DOMAIN, f'{mac_addr}') if is_clamp else None,
@@ -270,6 +297,6 @@ def _make_device_info(device, sensor_phase) -> DeviceInfo:
         name=device_name if not is_clamp else f"{device_name} Line {sensor_phase}",
         model=device_model if not is_clamp else f'{device_model} Clamp',
         manufacturer='Smilics',
-        configuration_url=f"http://{device['ipAddr']}/" if not is_clamp else None,
-        sw_version=f"{device['softVersion']}" if not is_clamp else None,
-    )
+        configuration_url=f"http://{device.ipAddr}/" if not is_clamp else None,
+        sw_version=f"{device.softVersion}" if not is_clamp else None,
+    )