Add various improvements from HA core

custom_components/bang_olufsen/__init__.py

@@ -5,6 +5,8 @@
 import asyncio
 from dataclasses import dataclass
 
+from aiohttp import ClientConnectorError, ClientOSError, ServerTimeoutError
+from mozart_api.exceptions import ApiException
 from mozart_api.mozart_client import MozartClient
 
 from homeassistant.config_entries import ConfigEntry
@@ -54,6 +56,7 @@ async def _start_websocket_listener(data: BangOlufsenData) -> None:
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up from a config entry."""
+
     # Remove casts to str
     assert entry.unique_id
 
@@ -72,7 +75,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     # Check API and WebSocket connection
     try:
         await client.check_device_connection(True)
-    except ExceptionGroup as error:
+    except* (
+        ClientConnectorError,
+        ClientOSError,
+        ServerTimeoutError,
+        ApiException,
+        TimeoutError,
+    ) as error:
         await client.close_api_client()
         raise ConfigEntryNotReady(f"Unable to connect to {entry.title}") from error
 

custom_components/bang_olufsen/const.py

@@ -70,6 +70,7 @@ class BangOlufsenProximity(Enum):
 class BangOlufsenModel(StrEnum):
     """Enum for compatible model names."""
 
+    BEOCONNECT_CORE = "Beoconnect Core"
     BEOLAB_8 = "BeoLab 8"
     BEOLAB_28 = "BeoLab 28"
     BEOSOUND_2 = "Beosound 2 3rd Gen"
@@ -85,40 +86,40 @@ class BangOlufsenModel(StrEnum):
 class WebsocketNotification(StrEnum):
     """Enum for WebSocket notification types."""
 
-    ACTIVE_LISTENING_MODE: Final[str] = "active_listening_mode"
-    ACTIVE_SPEAKER_GROUP: Final[str] = "active_speaker_group"
-    ALARM_TRIGGERED: Final[str] = "alarm_triggered"
-    BATTERY: Final[str] = "battery"
-    BEOLINK_EXPERIENCES_RESULT: Final[str] = "beolink_experiences_result"
-    BEOLINK_JOIN_RESULT: Final[str] = "beolink_join_result"
-    BEO_REMOTE_BUTTON: Final[str] = "beo_remote_button"
-    BUTTON: Final[str] = "button"
-    CURTAINS: Final[str] = "curtains"
-    PLAYBACK_ERROR: Final[str] = "playback_error"
-    PLAYBACK_METADATA: Final[str] = "playback_metadata"
-    PLAYBACK_PROGRESS: Final[str] = "playback_progress"
-    PLAYBACK_SOURCE: Final[str] = "playback_source"
-    PLAYBACK_STATE: Final[str] = "playback_state"
-    POWER_STATE: Final[str] = "power_state"
-    ROLE: Final[str] = "role"
-    SOFTWARE_UPDATE_STATE: Final[str] = "software_update_state"
-    SOUND_SETTINGS: Final[str] = "sound_settings"
-    SOURCE_CHANGE: Final[str] = "source_change"
-    VOLUME: Final[str] = "volume"
+    ACTIVE_LISTENING_MODE = "active_listening_mode"
+    ACTIVE_SPEAKER_GROUP = "active_speaker_group"
+    ALARM_TRIGGERED = "alarm_triggered"
+    BATTERY = "battery"
+    BEOLINK_EXPERIENCES_RESULT = "beolink_experiences_result"
+    BEOLINK_JOIN_RESULT = "beolink_join_result"
+    BEO_REMOTE_BUTTON = "beo_remote_button"
+    BUTTON = "button"
+    CURTAINS = "curtains"
+    PLAYBACK_ERROR = "playback_error"
+    PLAYBACK_METADATA = "playback_metadata"
+    PLAYBACK_PROGRESS = "playback_progress"
+    PLAYBACK_SOURCE = "playback_source"
+    PLAYBACK_STATE = "playback_state"
+    POWER_STATE = "power_state"
+    ROLE = "role"
+    SOFTWARE_UPDATE_STATE = "software_update_state"
+    SOUND_SETTINGS = "sound_settings"
+    SOURCE_CHANGE = "source_change"
+    VOLUME = "volume"
 
     # Sub-notifications
-    BEOLINK_AVAILABLE_LISTENERS: Final[str] = "beolinkAvailableListeners"
-    BEOLINK_LISTENERS: Final[str] = "beolinkListeners"
-    BEOLINK_PEERS: Final[str] = "beolinkPeers"
-    BEOLINK: Final[str] = "beolink"
-    BLUETOOTH_DEVICES: Final[str] = "bluetooth"
-    CONFIGURATION: Final[str] = "configuration"
-    NOTIFICATION: Final[str] = "notification"
-    PROXIMITY: Final[str] = "proximity"
-    REMOTE_CONTROL_DEVICES: Final[str] = "remoteControlDevices"
-    REMOTE_MENU_CHANGED: Final[str] = "remoteMenuChanged"
+    BEOLINK_AVAILABLE_LISTENERS = "beolinkAvailableListeners"
+    BEOLINK_LISTENERS = "beolinkListeners"
+    BEOLINK_PEERS = "beolinkPeers"
+    BEOLINK = "beolink"
+    BLUETOOTH_DEVICES = "bluetooth"
+    CONFIGURATION = "configuration"
+    NOTIFICATION = "notification"
+    PROXIMITY = "proximity"
+    REMOTE_CONTROL_DEVICES = "remoteControlDevices"
+    REMOTE_MENU_CHANGED = "remoteMenuChanged"
 
-    ALL: Final[str] = "all"
+    ALL = "all"
 
 
 class BangOlufsenModelSupport(Enum):
@@ -268,15 +269,15 @@ class BangOlufsenModelSupport(Enum):
 
 
 # Valid commands and their expected parameter type for beolink_command service
-FLOAT_PARAMETERS: Final[tuple] = (
+FLOAT_PARAMETERS: Final[tuple[str, str, str, type[float]]] = (
     "set_volume_level",
     "media_seek",
     "set_relative_volume_level",
     float,
 )
-BOOL_PARAMETERS: Final[tuple] = ("mute_volume", bool)
-STR_PARAMETERS: Final[tuple] = ("select_source", str)
-NONE_PARAMETERS: Final[tuple] = (
+BOOL_PARAMETERS: Final[tuple[str, type[bool]]] = ("mute_volume", bool)
+STR_PARAMETERS: Final[tuple[str, type[str]]] = ("select_source", str)
+NONE_PARAMETERS: Final[tuple[str, str, str, str, str, str, str, str, None]] = (
     "volume_up",
     "volume_down",
     "media_play_pause",
@@ -289,15 +290,20 @@ class BangOlufsenModelSupport(Enum):
 )
 
 # Tuple of accepted commands for input validation
-ACCEPTED_COMMANDS: Final[tuple] = (
-    FLOAT_PARAMETERS[:-1]
+ACCEPTED_COMMANDS: Final[tuple[tuple[str]]] = (
+    FLOAT_PARAMETERS[:-1]  # type: ignore[assignment]
     + BOOL_PARAMETERS[:-1]
     + STR_PARAMETERS[:-1]
     + NONE_PARAMETERS[:-1]
 )
 
 # Tuple of all commands and their types for executing commands.
-ACCEPTED_COMMANDS_LISTS: Final[tuple] = (
+ACCEPTED_COMMANDS_LISTS: tuple[
+    tuple[str, str, str, type[float]],
+    tuple[str, type[bool]],
+    tuple[str, type[str]],
+    tuple[str, str, str, str, str, str, str, str, None],
+] = (
     FLOAT_PARAMETERS,
     BOOL_PARAMETERS,
     STR_PARAMETERS,

custom_components/bang_olufsen/coordinator.py

@@ -30,9 +30,9 @@
 from homeassistant.const import CONF_DEVICE_ID, CONF_TYPE
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import device_registry as dr
-from homeassistant.helpers.device_registry import DeviceEntry
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from homeassistant.util.enum import try_parse_enum
 
 from .const import (
     BANG_OLUFSEN_EVENT,
@@ -63,7 +63,7 @@ def __init__(
         )
         BangOlufsenBase.__init__(self, entry, client)
 
-        self._device = self._get_device()
+        self._device = self.get_device()
 
         # WebSocket callbacks
         self._client.get_active_listening_mode_notifications(
@@ -104,6 +104,14 @@ def __init__(
         # Used for firing events and debugging
         self._client.get_all_notifications_raw(self.on_all_notifications_raw)
 
+    def get_device(self) -> dr.DeviceEntry:
+        """Get the device."""
+        device_registry = dr.async_get(self.hass)
+        device = device_registry.async_get_device({(DOMAIN, self._unique_id)})
+        assert device
+
+        return device
+
     async def _async_update_data(self) -> dict[str, Any]:
         """Get all information needed by the polling entities."""
         # Try to update coordinator_data.
@@ -122,17 +130,6 @@ def _update_connection_status(self) -> None:
             self._client.websocket_connected,
         )
 
-    def _get_device(self) -> DeviceEntry | None:
-        """Get the Home Assistant device."""
-        if not self.hass:
-            return None
-
-        device_registry = dr.async_get(self.hass)
-        device = device_registry.async_get_device({(DOMAIN, self._unique_id)})
-        assert device
-
-        return device
-
     def on_connection(self) -> None:
         """Handle WebSocket connection made."""
         _LOGGER.debug("Connected to the %s notification channel", self.entry.title)
@@ -169,11 +166,6 @@ def on_battery_notification(self, notification: BatteryState) -> None:
 
     def on_beo_remote_button_notification(self, notification: BeoRemoteButton) -> None:
         """Send beo_remote_button dispatch."""
-        if not self._device:
-            self._device = self._get_device()
-
-        assert self._device
-
         if notification.type == "KeyPress":
             # Trigger the device trigger
             self.hass.bus.async_fire(
@@ -186,11 +178,6 @@ def on_beo_remote_button_notification(self, notification: BeoRemoteButton) -> No
 
     def on_button_notification(self, notification: ButtonEvent) -> None:
         """Send button dispatch."""
-        if not self._device:
-            self._device = self._get_device()
-
-        assert self._device
-
         # Trigger the device trigger
         self.hass.bus.async_fire(
             BANG_OLUFSEN_EVENT,
@@ -204,39 +191,45 @@ def on_notification_notification(
         self, notification: WebsocketNotificationTag
     ) -> None:
         """Send notification dispatch."""
-        if notification.value:
-            if notification.value is WebsocketNotification.PROXIMITY:
-                async_dispatcher_send(
-                    self.hass,
-                    f"{self._unique_id}_{WebsocketNotification.PROXIMITY}",
-                    notification,
-                )
-
-            elif notification.value is WebsocketNotification.REMOTE_MENU_CHANGED.value:
-                async_dispatcher_send(
-                    self.hass,
-                    f"{self._unique_id}_{WebsocketNotification.REMOTE_MENU_CHANGED}",
-                )
-
-            elif notification.value is WebsocketNotification.CONFIGURATION.value:
-                async_dispatcher_send(
-                    self.hass,
-                    f"{self._unique_id}_{WebsocketNotification.CONFIGURATION}",
-                )
-
-            elif notification.value in (
-                WebsocketNotification.BLUETOOTH_DEVICES.value,
-                WebsocketNotification.REMOTE_CONTROL_DEVICES.value,
-            ):
-                async_dispatcher_send(
-                    self.hass,
-                    f"{self._unique_id}_{WebsocketNotification.BLUETOOTH_DEVICES}",
-                )
-            elif WebsocketNotification.BEOLINK.value in notification.value:
-                async_dispatcher_send(
-                    self.hass,
-                    f"{self._unique_id}_{WebsocketNotification.BEOLINK}",
-                )
+        # Try to match the notification type with available WebsocketNotification members
+        notification_type = try_parse_enum(WebsocketNotification, notification.value)
+
+        if notification_type is WebsocketNotification.PROXIMITY:
+            async_dispatcher_send(
+                self.hass,
+                f"{self._unique_id}_{WebsocketNotification.PROXIMITY}",
+                notification,
+            )
+
+        elif notification_type is WebsocketNotification.REMOTE_MENU_CHANGED:
+            async_dispatcher_send(
+                self.hass,
+                f"{self._unique_id}_{WebsocketNotification.REMOTE_MENU_CHANGED}",
+            )
+
+        elif notification_type is WebsocketNotification.CONFIGURATION:
+            async_dispatcher_send(
+                self.hass,
+                f"{self._unique_id}_{WebsocketNotification.CONFIGURATION}",
+            )
+
+        elif notification_type in (
+            WebsocketNotification.BLUETOOTH_DEVICES,
+            WebsocketNotification.REMOTE_CONTROL_DEVICES,
+        ):
+            async_dispatcher_send(
+                self.hass,
+                f"{self._unique_id}_{WebsocketNotification.BLUETOOTH_DEVICES}",
+            )
+        elif notification_type in (
+            WebsocketNotification.BEOLINK_PEERS,
+            WebsocketNotification.BEOLINK_LISTENERS,
+            WebsocketNotification.BEOLINK_AVAILABLE_LISTENERS,
+        ):
+            async_dispatcher_send(
+                self.hass,
+                f"{self._unique_id}_{WebsocketNotification.BEOLINK}",
+            )
 
     def on_playback_error_notification(self, notification: PlaybackError) -> None:
         """Send playback_error dispatch."""
@@ -301,11 +294,6 @@ async def on_software_update_state(self, _: SoftwareUpdateState) -> None:
         software_status = await self._client.get_softwareupdate_status()
 
         # Update the HA device if the sw version does not match
-        if not self._device:
-            self._device = self._get_device()
-
-        assert self._device
-
         if software_status.software_version != self._device.sw_version:
             device_registry = dr.async_get(self.hass)
 
@@ -316,10 +304,6 @@ async def on_software_update_state(self, _: SoftwareUpdateState) -> None:
 
     def on_all_notifications_raw(self, notification: dict) -> None:
         """Receive all notifications."""
-        if not self._device:
-            self._device = self._get_device()
-
-        assert self._device
 
         # Add the device_id and serial_number to the notification
         notification["device_id"] = self._device.id

custom_components/bang_olufsen/manifest.json

@@ -9,6 +9,6 @@
 	"iot_class": "local_push",
 	"issue_tracker": "https://github.com/bang-olufsen/bang_olufsen-hacs/issues",
 	"requirements": ["mozart-api==3.4.1.8.6"],
-	"version": "2.3.4",
+	"version": "2.3.5",
 	"zeroconf": ["_bangolufsen._tcp.local."]
 }