timeouts (in theory) added to Zeroconf listener and Dispatcher to suppress spurious "Removes". Virtual Household reverting to showing solar always

Author: bookman-dius

custom_components/powersensor/PowersensorDiscoveryService.py

@@ -1,5 +1,6 @@
 import asyncio
 from typing import Optional
+import logging
 
 from homeassistant.core import HomeAssistant
 from zeroconf import ServiceBrowser, ServiceListener, Zeroconf
@@ -8,11 +9,15 @@
 
 from .const import DOMAIN
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class PowersensorServiceListener(ServiceListener):
-    def __init__(self, hass: HomeAssistant):
+    def __init__(self, hass: HomeAssistant, debounce_timeout: float = 60):
         self._hass = hass
         self._plugs = {}
+        self._pending_removals = {}
+        self._debounce_seconds = debounce_timeout
 
     def add_service(self, zc, type_, name):
         info = self.__add_plug(zc, type_, name)
@@ -25,17 +30,35 @@ def add_service(self, zc, type_, name):
     async def _async_service_add(self, *args):
         async_dispatcher_send(self._hass, f"{DOMAIN}_zeroconf_add_plug", *args)
 
+    async def _async_delayed_remove(self, name):
+        """Actually process the removal after delay."""
+        try:
+            await asyncio.sleep(self._debounce_seconds)
+            _LOGGER.info(f"Request to remove service {name} still pending after timeout. Processing remove request...")
+            if name in self._plugs:
+                data = self._plugs[name].copy()
+                del self._plugs[name]
+            else:
+                data = None
+            asyncio.run_coroutine_threadsafe(
+                self._async_service_remove(name, data),
+                self._hass.loop
+            )
+        except asyncio.CancelledError:
+            # Task was cancelled because service came back
+            _LOGGER.info(f"Request to remove service {name} was canceled by request to update or add plug.")
+
+        # Either way were done with this task
+        self._pending_removals.pop(name, None)
+
     def remove_service(self, zc, type_, name):
-        if name in self._plugs:
-            data =  self._plugs[name].copy()
-            del self._plugs[name]
-        else:
-            data = None
+        if name in self._pending_removals:
+            # removal for this service is already pending
+            return
+
+        _LOGGER.info(f"Scheduling removal for {name}")
+        self._pending_removals[name] = asyncio.create_task(self._async_delayed_remove(name))
 
-        asyncio.run_coroutine_threadsafe(
-            self._async_service_remove(name, data),
-            self._hass.loop
-        )
 
     async def _async_service_remove(self, *args):
         async_dispatcher_send(self._hass, f"{DOMAIN}_zeroconf_remove_plug", *args)
@@ -52,8 +75,11 @@ async def _async_service_update(self, *args):
         async_dispatcher_send(self._hass, f"{DOMAIN}_zeroconf_update_plug", *args)
 
     def __add_plug(self, zc, type_, name):
-
         info = zc.get_service_info(type_, name)
+        task = self._pending_removals.pop(name, None)
+        if task:
+            task.cancel()
+            _LOGGER.info(f"Cancelled pending removal for {name}")
         if info:
             self._plugs[name] = {'type': type_,
                                  'name': name,

custom_components/powersensor/PowersensorMessageDispatacher.py

@@ -11,14 +11,16 @@
 
 _LOGGER = logging.getLogger(__name__)
 class PowersensorMessageDispatcher:
-    def __init__(self, hass: HomeAssistant, vhh: VirtualHousehold):
+    def __init__(self, hass: HomeAssistant, vhh: VirtualHousehold,debounce_timeout: float = 60):
         self._hass = hass
         self._vhh = vhh
         self.plugs = dict()
         self._known_plugs = set()
         self._known_plug_names = dict()
         self.sensors = dict()
         self.on_start_sensor_queue = dict()
+        self._pending_removals = {}
+        self._debounce_seconds = debounce_timeout
         self._unsubscribe_from_signals = [
             async_dispatcher_connect(self._hass,
                                      f"{DOMAIN}_sensor_added_to_homeassistant",
@@ -31,7 +33,7 @@ def __init__(self, hass: HomeAssistant, vhh: VirtualHousehold):
                                      self._plug_updated),
             async_dispatcher_connect(self._hass,
                                      f"{DOMAIN}_zeroconf_remove_plug",
-                                     self._plug_remove),
+                                     self._schedule_plug_removal),
             async_dispatcher_connect(self._hass,
                                      f"{DOMAIN}_plug_added_to_homeassistant",
                                      self._acknowledge_plug_added_to_homeassistant),
@@ -127,6 +129,11 @@ def add_api(self, network_info):
 
     async def handle_message(self, event: str, message: dict):
         mac = message['mac']
+        task = self._pending_removals.pop(mac, None)
+        if task:
+            task.cancel()
+            _LOGGER.info(f"Cancelled pending removal for {mac} by new message received from plug.")
+
         if mac not in self.plugs.keys():
             if mac not in self.sensors:
                 role = None
@@ -165,11 +172,17 @@ async def _acknowledge_plug_added_to_homeassistant(self, mac_address, host, port
     async def _plug_added(self, info):
         _LOGGER.debug(f" Request to add plug received: {info}")
         network_info = dict()
-        network_info['mac'] = info['properties'][b'id'].decode('utf-8')
+        mac = info['properties'][b'id'].decode('utf-8')
+        network_info['mac'] = mac
         network_info['host'] = info['addresses'][0]
         network_info['port'] = info['port']
         network_info['name'] = info['name']
 
+        task = self._pending_removals.pop(mac, None)
+        if task:
+            task.cancel()
+            _LOGGER.info(f"Cancelled pending removal for {mac} by request to add api.")
+
         if self._safe_to_process_plug_queue:
             await self.enqueue_plug_for_adding(network_info)
             await self.process_plug_queue()
@@ -182,7 +195,10 @@ async def _plug_updated(self, info):
         host = info['addresses'][0]
         port = info['port']
         name = info['name']
-
+        task = self._pending_removals.pop(mac, None)
+        if task:
+            task.cancel()
+            _LOGGER.info(f"Cancelled pending removal for {mac} by request to update api.")
         if mac in self.plugs:
             current_api: PlugApi = self.plugs[mac]
             if current_api._listener._ip == host and current_api._listener._port == port:
@@ -202,14 +218,32 @@ async def _plug_updated(self, info):
             await self.enqueue_plug_for_adding(network_info)
             await self.process_plug_queue()
 
-    async def _plug_remove(self,name, info):
+    async def _schedule_plug_removal(self, name, info):
         _LOGGER.debug(f" Request to delete plug received: {info}")
         if name in self._known_plug_names:
             mac = self._known_plug_names[name]
             if mac in self.plugs:
-                await self.plugs[mac].disconnect()
-                del self.plugs[mac]
-                del self._known_plug_names[name]
+                if mac in self._pending_removals:
+                    # removal for this service is already pending
+                    return
+
+                _LOGGER.info(f"Scheduling removal for {name}")
+                self._pending_removals[name] = asyncio.create_task(self._delayed_plug_remove(name,mac))
         else:
             _LOGGER.warning(f"Received request to delete api for gateway with name [{name}], but this name"
                             f"is not associated with an existing PlugAPI. Ignoring...")
+
+    async def _delayed_plug_remove(self, name, mac):
+        """Actually process the removal after delay."""
+        try:
+            await asyncio.sleep(self._debounce_seconds)
+            _LOGGER.info(f"Request to remove plug {mac} still pending after timeout. Processing remove request...")
+            await self.plugs[mac].disconnect()
+            del self.plugs[mac]
+            del self._known_plug_names[name]
+        except asyncio.CancelledError:
+            # Task was cancelled because service came back
+            _LOGGER.info(f"Request to remove plug {mac} was canceled by request to update, add plug or new message.")
+
+        # Either way were done with this task
+        self._pending_removals.pop(name, None)

custom_components/powersensor/sensor.py

@@ -54,19 +54,8 @@ async def handle_discovered_plug(plug_mac_address: str, host: str, port: int, na
 
     async def handle_discovered_sensor(sensor_mac: str, sensor_role: str):
         if sensor_role == 'solar':
-            add_solar_vhh_entities = False
-            if "with_solar" not in my_data.keys() or not my_data["with_solar"] :
-                add_solar_vhh_entities = True
-
             my_data["with_solar"] = True  # Remember for next time we start
 
-            if add_solar_vhh_entities:
-                solar_household_entities = []
-                for solar_measurement_type in ProductionMeasurements:
-                    household_entities.append(PowersensorHouseholdEntity(vhh, solar_measurement_type))
-
-                async_add_entities(solar_household_entities)
-
         new_sensors = [
             PowersensorSensorEntity(hass, sensor_mac, SensorMeasurements.Battery),
             PowersensorSensorEntity(hass, sensor_mac, SensorMeasurements.WATTS),
@@ -92,10 +81,9 @@ async def handle_discovered_sensor(sensor_mac: str, sensor_role: str):
     for measurement_type in ConsumptionMeasurements:
         household_entities.append(PowersensorHouseholdEntity(vhh, measurement_type))
 
-    if "with_solar" in my_data.keys():
-        if my_data["with_solar"]:
-            for measurement_type in ProductionMeasurements:
-                household_entities.append(PowersensorHouseholdEntity(vhh, measurement_type))
+
+    for measurement_type in ProductionMeasurements:
+        household_entities.append(PowersensorHouseholdEntity(vhh, measurement_type))
 
     async_add_entities(household_entities)