Merge pull request #234 from loopj/expose-status-type

Expose what status type a controller is using for state updates

Author: loopj

src/aiovantage/_controllers/base.py

@@ -1,6 +1,7 @@
 import asyncio
 from collections.abc import Callable
 from dataclasses import fields
+from enum import Enum
 from typing import TYPE_CHECKING, TypeVar, cast
 
 from aiovantage._logger import logger
@@ -25,13 +26,23 @@
 T = TypeVar("T", bound=SystemObject)
 
 
+class StatusType(Enum):
+    """The type of status that the controller is using for state monitoring."""
+
+    OBJECT = 1
+    """Object status events from the Enhanced Log."""
+
+    CATEGORY = 2
+    """Category status events, eg: S:LOAD, S:BLIND, etc."""
+
+
 class Controller(QuerySet[T], EventDispatcher):
     """Base controller for managing collections of Vantage objects."""
 
     vantage_types: tuple[str, ...]
     """The Vantage object types that this controller will fetch."""
 
-    category_status: bool = False
+    force_category_status: bool = False
     """Whether to force the controller to handle 'STATUS' categories."""
 
     def __init__(self, vantage: "Vantage") -> None:
@@ -43,8 +54,8 @@ def __init__(self, vantage: "Vantage") -> None:
         self._vantage = vantage
         self._objects: dict[int, T] = {}
         self._initialized = False
-        self._state_monitoring_enabled = False
-        self._state_monitoring_unsubs: list[Callable[[], None]] = []
+        self._status_type: StatusType | None = None
+        self._status_unsubs: list[Callable[[], None]] = []
         self._lock = asyncio.Lock()
 
         QuerySet[T].__init__(self, self._objects, self._lazy_initialize)
@@ -58,6 +69,11 @@ def __contains__(self, vid: int) -> bool:
         """Return True if the object with the given Vantage ID exists."""
         return vid in self._objects
 
+    @property
+    def status_type(self) -> StatusType | None:
+        """Return the type of status event that the controller is monitoring."""
+        return self._status_type
+
     async def initialize(
         self, *, fetch_state: bool = True, enable_state_monitoring: bool = True
     ) -> None:
@@ -141,7 +157,7 @@ async def fetch_state(self) -> None:
 
     async def enable_state_monitoring(self) -> None:
         """Monitor for state changes on objects managed by this controller."""
-        if self._state_monitoring_enabled:
+        if self._status_type is not None:
             return
 
         # Start the event stream if it isn't already running
@@ -152,39 +168,42 @@ async def enable_state_monitoring(self) -> None:
         # If these are not supported—either due to older firmware, or if the
         # controller explicitly requesting category statuses, we'll fall back to
         # "category" status events.
-        if event_conn.supports_enhanced_log and not self.category_status:
+        if event_conn.supports_enhanced_log and not self.force_category_status:
             # Subscribe to "object status" events from the Enhanced Log.
             status_unsub = self._vantage.event_stream.subscribe_enhanced_log(
                 self._handle_enhanced_log_event, "STATUS", "STATUSEX"
             )
+
+            self._status_type = StatusType.OBJECT
         else:
             # Subscribe to "STATUS {category}" updates
             status_unsub = self._vantage.event_stream.subscribe_status(
                 self._handle_status_event
             )
 
+            self._status_type = StatusType.CATEGORY
+
         # Subscribe to reconnect events from the event stream
         reconnect_unsub = self._vantage.event_stream.subscribe(
             Reconnected, self._handle_reconnect_event
         )
 
         # Keep track of the subscriptions so we can unsubscribe later
-        self._state_monitoring_unsubs.extend([status_unsub, reconnect_unsub])
-        self._state_monitoring_enabled = True
+        self._status_unsubs.extend([status_unsub, reconnect_unsub])
 
         logger.info("%s subscribed to state changes", type(self).__name__)
 
     async def disable_state_monitoring(self) -> None:
         """Stop monitoring for state changes on objects managed by this controller."""
-        if not self._state_monitoring_enabled:
+        if self._status_type is None:
             return
 
         # Unsubscribe status and reconnect events
-        while self._state_monitoring_unsubs:
-            unsub = self._state_monitoring_unsubs.pop()
+        while self._status_unsubs:
+            unsub = self._status_unsubs.pop()
             unsub()
 
-        self._state_monitoring_enabled = False
+        self._status_type = None
 
         logger.info("%s unsubscribed from state changes", type(self).__name__)
 

src/aiovantage/_controllers/gmem.py

@@ -7,4 +7,4 @@ class GMemController(Controller[GMem]):
     """GMem (variables) controller."""
 
     vantage_types = ("GMem",)
-    category_status = True
+    force_category_status = True

src/aiovantage/controllers.py

@@ -30,7 +30,7 @@
 from ._controllers.anemo_sensors import AnemoSensorsController
 from ._controllers.areas import AreasController
 from ._controllers.back_boxes import BackBoxesController
-from ._controllers.base import Controller
+from ._controllers.base import Controller, StatusType
 from ._controllers.blind_groups import BlindGroupsController, BlindGroupTypes
 from ._controllers.blinds import BlindsController, BlindTypes
 from ._controllers.buttons import ButtonsController
@@ -75,6 +75,7 @@
     "RGBLoadsController",
     "RGBLoadTypes",
     "StationsController",
+    "StatusType",
     "TasksController",
     "TemperaturesController",
     "ThermostatsController",