refactor: enum classes, state handling

intg-denonavr/avr.py

@@ -19,7 +19,7 @@
 LOG = logging.getLogger(__name__)
 
 
-class EVENTS(IntEnum):
+class Events(IntEnum):
     """Internal driver events."""
 
     CONNECTING = 0
@@ -30,13 +30,15 @@ class EVENTS(IntEnum):
     UPDATE = 5
 
 
-class STATES(IntEnum):
+class States(IntEnum):
     """State of a connected AVR."""
 
-    OFF = 0
-    ON = 1
-    PLAYING = 2
-    PAUSED = 3
+    UNKNOWN = 0
+    UNAVAILABLE = 1
+    OFF = 2
+    ON = 3
+    PLAYING = 4
+    PAUSED = 5
 
 
 async def discover_denon_avrs():
@@ -72,8 +74,7 @@ def __init__(self, loop: AbstractEventLoop, ipaddress: str):
         self.ipaddress: str = ipaddress
         self.getting_data: bool = False
 
-        # TODO shouldn't default state be UNKNOWN from the entity base class?
-        self.state: STATES = STATES.OFF
+        self.state: States = States.UNKNOWN
         self.volume: float = 0
         self.input: str | None = None
         self.input_list: list[str] = []
@@ -140,18 +141,11 @@ async def connect(self):
 
         await self._subscribe_events()
         if self.id:
-            self.events.emit(EVENTS.CONNECTED, self.id)
+            self.events.emit(Events.CONNECTED, self.id)
         else:
             LOG.error("Device communication error: no serial number retrieved from AVR!")
 
-        if self._avr.state == "on":
-            self.state = STATES.ON
-        elif self._avr.state == "off":
-            self.state = STATES.OFF
-        elif self._avr.state == "playing":
-            self.state = STATES.PLAYING
-        elif self._avr.state == "paused":
-            self.state = STATES.PAUSED
+        self.state = self._map_denonavr_state(self._avr.state)
 
         self.input_list = self._avr.input_func_list
         self.input = self._avr.input_func
@@ -167,7 +161,21 @@ async def disconnect(self):
         await self._unsubscribe_events()
         self._avr = None
         if self.id:
-            self.events.emit(EVENTS.DISCONNECTED, self.id)
+            self.events.emit(Events.DISCONNECTED, self.id)
+
+    @staticmethod
+    def _map_denonavr_state(avr_state: str | None) -> States:
+        """Map the DenonAVR library state to our state."""
+        state = States.UNKNOWN
+        if avr_state == "on":
+            state = States.ON
+        elif avr_state == "off":
+            state = States.OFF
+        elif avr_state == "playing":
+            state = States.PLAYING
+        elif avr_state == "paused":
+            state = States.PAUSED
+        return state
 
     async def _get_data(self):
         if self._avr is None:
@@ -185,19 +193,12 @@ async def _get_data(self):
             self.artwork = self._avr.image_url
 
             if self._avr.power == "OFF":
-                self.state = STATES.OFF
+                self.state = States.OFF
             else:
-                if self._avr.state == "on":
-                    self.state = STATES.ON
-                elif self._avr.state == "off":
-                    self.state = STATES.OFF
-                elif self._avr.state == "playing":
-                    self.state = STATES.PLAYING
-                elif self._avr.state == "paused":
-                    self.state = STATES.PAUSED
+                self.state = self._map_denonavr_state(self._avr.state)
 
             self.events.emit(
-                EVENTS.UPDATE,
+                Events.UPDATE,
                 {
                     "state": self.state,
                     "artist": self.artist,
@@ -223,7 +224,7 @@ async def _update_callback(self, zone, event, parameter):
 
         if event == "MV":
             self.volume = self._convert_volume_to_percent(self._avr.volume)
-            self.events.emit(EVENTS.UPDATE, {"volume": self.volume})
+            self.events.emit(Events.UPDATE, {"volume": self.volume})
         else:
             _ = asyncio.ensure_future(self._get_data())
             # if self.state == STATES.OFF:

intg-denonavr/driver.py

@@ -180,19 +180,19 @@ async def _on_subscribe_entities(entity_ids):
             LOG.debug("We have a match, start listening to events")
             a = configuredAVRs[entity_id]
 
-            @a.events.on(avr.EVENTS.CONNECTED)
+            @a.events.on(avr.Events.CONNECTED)
             async def on_connected(identifier):
                 await _handle_connected(identifier)
 
-            @a.events.on(avr.EVENTS.DISCONNECTED)
+            @a.events.on(avr.Events.DISCONNECTED)
             async def on_disconnected(identifier):
                 await _handle_disconnected(identifier)
 
-            @a.events.on(avr.EVENTS.ERROR)
+            @a.events.on(avr.Events.ERROR)
             async def on_error(identifier, message):
                 await _handle_connection_error(identifier, message)
 
-            @a.events.on(avr.EVENTS.UPDATE)
+            @a.events.on(avr.Events.UPDATE)
             async def on_update(update):
                 # FIXME W0640: Cell variable entity_id defined in loop (cell-var-from-loop)
                 #       This is most likely the WRONG entity_id if we have MULTIPLE configuredAVRs
@@ -203,9 +203,7 @@ async def on_update(update):
             api.configuredEntities.updateEntityAttributes(
                 entity_id,
                 {
-                    entities.media_player.ATTRIBUTES.STATE: entities.media_player.STATES.ON
-                    if a.state == avr.STATES.ON
-                    else entities.media_player.STATES.OFF,
+                    entities.media_player.ATTRIBUTES.STATE: _media_player_state_from_avr(a.state),
                     entities.media_player.ATTRIBUTES.SOURCE_LIST: a.input_list,
                     entities.media_player.ATTRIBUTES.SOURCE: a.input,
                     entities.media_player.ATTRIBUTES.VOLUME: a.volume,
@@ -216,6 +214,22 @@ async def on_update(update):
             )
 
 
+def _media_player_state_from_avr(avr_state: avr.States) -> entities.media_player.STATES:
+    """Convert the AVR device state to a media-player entity state."""
+    state = entities.media_player.STATES.UNKNOWN
+    if avr_state == avr.States.ON:
+        state = entities.media_player.STATES.ON
+    elif avr_state == avr.States.OFF:
+        state = entities.media_player.STATES.OFF
+    elif avr_state == avr.States.PLAYING:
+        state = entities.media_player.STATES.PLAYING
+    elif avr_state == avr.States.PAUSED:
+        state = entities.media_player.STATES.PAUSED
+    elif avr_state == avr.States.UNAVAILABLE:
+        state = entities.media_player.STATES.UNAVAILABLE
+    return state
+
+
 # On unsubscribe, we disconnect the objects and remove listeners for events
 @api.events.on(uc.uc.EVENTS.UNSUBSCRIBE_ENTITIES)
 async def _on_unsubscribe_entities(entity_ids):
@@ -378,13 +392,13 @@ def _get_media_player_state(avr_state) -> entities.media_player.STATES:
     """
     state = entities.media_player.STATES.UNKNOWN
 
-    if avr_state == avr.STATES.ON:
+    if avr_state == avr.States.ON:
         state = entities.media_player.STATES.ON
-    elif avr_state == avr.STATES.PLAYING:
+    elif avr_state == avr.States.PLAYING:
         state = entities.media_player.STATES.PLAYING
-    elif avr_state == avr.STATES.PAUSED:
+    elif avr_state == avr.States.PAUSED:
         state = entities.media_player.STATES.PAUSED
-    elif avr_state == avr.STATES.OFF:
+    elif avr_state == avr.States.OFF:
         state = entities.media_player.STATES.OFF
 
     return state