home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-08-05 13:45:12 +02:00
Code Issues Packages Projects Releases Wiki Activity

Update integration with feedback from emontnemery

Browse Source
This commit is contained in:
mj23000
2023-12-01 17:03:25 +01:00
parent 3b3a77b5d0
commit 6061171bb4
14 changed files with 309 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@@ -114,6 +114,7 @@ omit =
homeassistant/components/bangolufsen/const.py
homeassistant/components/bangolufsen/entity.py
homeassistant/components/bangolufsen/media_player.py
homeassistant/components/bangolufsen/util.py
homeassistant/components/bangolufsen/websocket.py
homeassistant/components/bbox/device_tracker.py
homeassistant/components/bbox/sensor.py

74
homeassistant/components/bangolufsen/__init__.py
View File

@@ -1,42 +1,53 @@
"""The Bang & Olufsen integration."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from multiprocessing.pool import ApplyResult
from typing import cast
from mozart_api.exceptions import ServiceException
from mozart_api.models import BatteryState
from mozart_api.mozart_client import MozartClient
from urllib3.exceptions import MaxRetryError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_NAME, Platform
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.event import async_call_later
from .const import DOMAIN, ENTITY_ENUM, WEBSOCKET_CONNECTION_DELAY
from .media_player import BangOlufsenMediaPlayer
from .const import DOMAIN, WEBSOCKET_CONNECTION_DELAY
from .websocket import BangOlufsenWebsocket
PLATFORMS = [Platform.MEDIA_PLAYER]
_LOGGER = logging.getLogger(__name__)
@dataclass
class BangOlufsenData:
"""Dataclass for storing entities."""
websocket: BangOlufsenWebsocket
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up from a config entry."""
hass.data.setdefault(DOMAIN, {})
# Ensure that a unique id is available
if not entry.unique_id:
raise ConfigEntryError("Can't retrieve unique id from config entry. Aborting")
client = MozartClient(
host=entry.data[CONF_HOST], urllib3_logging_level=logging.DEBUG
)
# If connection can't be made abort.
if not await init_entities(hass, entry):
raise ConfigEntryNotReady(f"Unable to connect to {entry.title}")
# Check connection and try to initialize it.
try:
await client.get_battery_state(_request_timeout=3)
except (MaxRetryError, ServiceException, Exception) as error:
raise ConfigEntryNotReady(f"Unable to connect to {entry.title}") from error
websocket = BangOlufsenWebsocket(hass, entry)
# Add the websocket
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = BangOlufsenData(websocket)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
async_call_later(hass, WEBSOCKET_CONNECTION_DELAY, websocket.connect_websocket)
return True
@@ -46,39 +57,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.unique_id)
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
async def init_entities(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Initialise the supported entities of the device."""
client = MozartClient(
host=entry.data[CONF_HOST], urllib3_logging_level=logging.ERROR
)
# Check connection and try to initialize it.
try:
cast(
ApplyResult[BatteryState],
client.get_battery_state(async_req=True, _request_timeout=3),
).get()
except (MaxRetryError, ServiceException):
_LOGGER.error("Unable to connect to %s", entry.data[CONF_NAME])
return False
websocket = BangOlufsenWebsocket(hass, entry)
# Create the Media Player entity.
media_player = BangOlufsenMediaPlayer(entry)
# Add the created entities
hass.data[DOMAIN][entry.unique_id] = {
ENTITY_ENUM.WEBSOCKET: websocket,
ENTITY_ENUM.MEDIA_PLAYER: media_player,
}
# Start the WebSocket listener with a delay to allow for entity and dispatcher listener creation
async_call_later(hass, WEBSOCKET_CONNECTION_DELAY, websocket.connect_websocket)
return True

45
homeassistant/components/bangolufsen/config_flow.py
View File

@@ -3,11 +3,9 @@ from __future__ import annotations
import ipaddress
import logging
from multiprocessing.pool import ApplyResult
from typing import Any, TypedDict, cast
from typing import Any, TypedDict
from mozart_api.exceptions import ApiException, NotFoundException
from mozart_api.models import BeolinkPeer, VolumeSettings
from mozart_api.mozart_client import MozartClient
from urllib3.exceptions import MaxRetryError, NewConnectionError
import voluptuous as vol
@@ -28,11 +26,9 @@ from .const import (
CONF_DEFAULT_VOLUME,
CONF_MAX_VOLUME,
CONF_SERIAL_NUMBER,
CONF_VOLUME_STEP,
DEFAULT_DEFAULT_VOLUME,
DEFAULT_MAX_VOLUME,
DEFAULT_MODEL,
DEFAULT_VOLUME_STEP,
DOMAIN,
)
@@ -40,40 +36,33 @@ from .const import (
class UserInput(TypedDict, total=False):
"""TypedDict for user_input."""
name: str
volume_step: int
default_volume: int
max_volume: int
host: str
model: str
jid: str
max_volume: int
model: str
name: str
class BangOlufsenConfigFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
_beolink_jid = ""
_client: MozartClient
_host = ""
_model = ""
_name = ""
_serial_number = ""
def __init__(self) -> None:
"""Init the config flow."""
self._host: str = ""
self._name: str = ""
self._model: str = ""
self._serial_number: str = ""
self._beolink_jid: str = ""
self._client: MozartClient | None = None
VERSION = 1
async def _compile_data(self) -> UserInput:
"""Compile data for entry creation."""
if not self._client:
self._client = MozartClient(self._host, urllib3_logging_level=logging.ERROR)
# Get current volume settings
volume_settings = cast(
ApplyResult[VolumeSettings],
self._client.get_volume_settings(async_req=True),
).get()
volume_settings = await self._client.get_volume_settings()
# Create a dict containing all necessary information for setup
data = UserInput()
@@ -81,7 +70,6 @@ class BangOlufsenConfigFlowHandler(ConfigFlow, domain=DOMAIN):
data[CONF_HOST] = self._host
data[CONF_MODEL] = self._model
data[CONF_BEOLINK_JID] = self._beolink_jid
data[CONF_VOLUME_STEP] = DEFAULT_VOLUME_STEP
data[CONF_DEFAULT_VOLUME] = (
volume_settings.default.level
if volume_settings.default and volume_settings.default.level
@@ -114,10 +102,7 @@ class BangOlufsenConfigFlowHandler(ConfigFlow, domain=DOMAIN):
# Try to get information from Beolink self method.
try:
beolink_self = cast(
ApplyResult[BeolinkPeer],
self._client.get_beolink_self(async_req=True, _request_timeout=3),
).get()
beolink_self = await self._client.get_beolink_self(_request_timeout=3)
except (
ApiException,
@@ -173,7 +158,9 @@ class BangOlufsenConfigFlowHandler(ConfigFlow, domain=DOMAIN):
}
await self.async_set_unique_id(self._serial_number)
self._abort_if_unique_id_configured()
self._abort_if_unique_id_configured(updates={CONF_HOST: self._host})
self._client = MozartClient(self._host, urllib3_logging_level=logging.ERROR)
return await self.async_step_confirm()

123
homeassistant/components/bangolufsen/const.py
View File

@@ -2,37 +2,12 @@
from __future__ import annotations
from enum import Enum, StrEnum
import logging
from typing import Final, cast
from enum import StrEnum
from typing import Final
from mozart_api.models import (
PlaybackContentMetadata,
PlaybackProgress,
RenderingState,
Source,
SourceArray,
SourceTypeEnum,
VolumeLevel,
VolumeMute,
VolumeState,
)
from mozart_api.mozart_client import MozartClient
from mozart_api.models import Source, SourceArray, SourceTypeEnum
from homeassistant.components.media_player import MediaPlayerState, MediaType
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceEntry
class ART_SIZE_ENUM(Enum):
"""Enum used for sorting images that have size defined by a string."""
small = 1
medium = 2
large = 3
class SOURCE_ENUM(StrEnum):
@@ -58,14 +33,6 @@ class SOURCE_ENUM(StrEnum):
tidalConnect = "Tidal Connect" # noqa: N815
class REPEAT_ENUM(StrEnum):
"""Enum used for translating device repeat settings to Home Assistant settings."""
all = "all"
one = "track"
off = "none"
BANGOLUFSEN_STATES: dict[str, MediaPlayerState] = {
# Dict used for translating device states to Home Assistant states.
"started": MediaPlayerState.PLAYING,
@@ -75,9 +42,8 @@ BANGOLUFSEN_STATES: dict[str, MediaPlayerState] = {
"stopped": MediaPlayerState.PAUSED,
"ended": MediaPlayerState.PAUSED,
"error": MediaPlayerState.IDLE,
# A devices initial state is "unknown" and should be treated as "idle"
# A device's initial state is "unknown" and should be treated as "idle"
"unknown": MediaPlayerState.IDLE,
# Power states
}
@@ -105,13 +71,6 @@ class MODEL_ENUM(StrEnum):
BEOSOUND_THEATRE = "Beosound Theatre"
class ENTITY_ENUM(StrEnum):
"""Enum for accessing and storing the entities in hass."""
MEDIA_PLAYER = "media_player"
WEBSOCKET = "websocket"
# Dispatcher events
class WEBSOCKET_NOTIFICATION(StrEnum):
"""Enum for WebSocket notification types."""
@@ -148,7 +107,6 @@ VOLUME_STEP_RANGE: Final[range] = range(1, 20, 1)
# Configuration.
CONF_DEFAULT_VOLUME: Final = "default_volume"
CONF_MAX_VOLUME: Final = "max_volume"
CONF_VOLUME_STEP: Final = "volume_step"
CONF_SERIAL_NUMBER: Final = "serial_number"
CONF_BEOLINK_JID: Final = "jid"
@@ -199,57 +157,57 @@ FALLBACK_SOURCES: Final[SourceArray] = SourceArray(
items=[
Source(
id="uriStreamer",
is_enabled=True,
is_playable=False,
isEnabled=True,
isPlayable=False,
name="Audio Streamer",
type=SourceTypeEnum(value="uriStreamer"),
),
Source(
id="bluetooth",
is_enabled=True,
is_playable=False,
isEnabled=True,
isPlayable=False,
name="Bluetooth",
type=SourceTypeEnum(value="bluetooth"),
),
Source(
id="spotify",
is_enabled=True,
is_playable=False,
isEnabled=True,
isPlayable=False,
name="Spotify Connect",
type=SourceTypeEnum(value="spotify"),
),
Source(
id="lineIn",
is_enabled=True,
is_playable=True,
isEnabled=True,
isPlayable=True,
name="Line-In",
type=SourceTypeEnum(value="lineIn"),
),
Source(
id="spdif",
is_enabled=True,
is_playable=True,
isEnabled=True,
isPlayable=True,
name="Optical",
type=SourceTypeEnum(value="spdif"),
),
Source(
id="netRadio",
is_enabled=True,
is_playable=True,
isEnabled=True,
isPlayable=True,
name="B&O Radio",
type=SourceTypeEnum(value="netRadio"),
),
Source(
id="deezer",
is_enabled=True,
is_playable=True,
isEnabled=True,
isPlayable=True,
name="Deezer",
type=SourceTypeEnum(value="deezer"),
),
Source(
id="tidalConnect",
is_enabled=True,
is_playable=True,
isEnabled=True,
isPlayable=True,
name="Tidal Connect",
type=SourceTypeEnum(value="tidalConnect"),
),
@@ -265,44 +223,3 @@ CONNECTION_STATUS: Final[str] = "CONNECTION_STATUS"
# Misc.
WEBSOCKET_CONNECTION_DELAY: Final[float] = 3.0
def get_device(hass: HomeAssistant | None, unique_id: str) -> DeviceEntry | None:
"""Get the device."""
if not isinstance(hass, HomeAssistant):
return None
device_registry = dr.async_get(hass)
device = cast(DeviceEntry, device_registry.async_get_device({(DOMAIN, unique_id)}))
return device
class BangOlufsenVariables:
"""Shared variables for various classes."""
def __init__(self, entry: ConfigEntry) -> None:
"""Initialize the object."""
# get the input from the config entry.
self.entry: ConfigEntry = entry
# Set the configuration variables.
self._host: str = self.entry.data[CONF_HOST]
self._name: str = self.entry.title
self._unique_id: str = cast(str, self.entry.unique_id)
self._client: MozartClient = MozartClient(
host=self._host,
websocket_reconnect=True,
urllib3_logging_level=logging.ERROR,
)
# Objects that get directly updated by notifications.
self._playback_metadata: PlaybackContentMetadata = PlaybackContentMetadata()
self._playback_progress: PlaybackProgress = PlaybackProgress(total_duration=0)
self._playback_source: Source = Source()
self._playback_state: RenderingState = RenderingState()
self._source_change: Source = Source()
self._volume: VolumeState = VolumeState(
level=VolumeLevel(level=0), muted=VolumeMute(muted=False)
)

67
homeassistant/components/bangolufsen/entity.py
View File

@@ -2,13 +2,59 @@
from __future__ import annotations
from collections.abc import Callable
import logging
from typing import cast
from mozart_api.models import (
PlaybackContentMetadata,
PlaybackProgress,
RenderingState,
Source,
VolumeLevel,
VolumeMute,
VolumeState,
)
from mozart_api.mozart_client import MozartClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.const import CONF_HOST
from homeassistant.helpers.device_registry import DeviceEntry, DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import CONNECTION_STATUS, DOMAIN, BangOlufsenVariables
from .const import DOMAIN
class BangOlufsenVariables:
"""Shared variables for various classes."""
def __init__(self, entry: ConfigEntry) -> None:
"""Initialize the object."""
# get the input from the config entry.
self.entry: ConfigEntry = entry
self._device: DeviceEntry | None = None
# Set the configuration variables.
self._host: str = self.entry.data[CONF_HOST]
self._name: str = self.entry.title
self._unique_id: str = cast(str, self.entry.unique_id)
self._client: MozartClient = MozartClient(
host=self._host,
websocket_reconnect=True,
urllib3_logging_level=logging.ERROR,
)
# Objects that get directly updated by notifications.
self._playback_metadata: PlaybackContentMetadata = PlaybackContentMetadata()
self._playback_progress: PlaybackProgress = PlaybackProgress(totalDuration=0)
self._playback_source: Source = Source()
self._playback_state: RenderingState = RenderingState()
self._source_change: Source = Source()
self._volume: VolumeState = VolumeState(
level=VolumeLevel(level=0), muted=VolumeMute(muted=False)
)
class BangOlufsenEntity(Entity, BangOlufsenVariables):
@@ -26,21 +72,6 @@ class BangOlufsenEntity(Entity, BangOlufsenVariables):
self._attr_entity_category = None
self._attr_should_poll = False
async def async_added_to_hass(self) -> None:
"""Turn on the dispatchers."""
self._dispatchers = [
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{CONNECTION_STATUS}",
self._update_connection_state,
)
]
async def async_will_remove_from_hass(self) -> None:
"""Turn off the dispatchers."""
for dispatcher in self._dispatchers:
dispatcher()
async def _update_connection_state(self, connection_state: bool) -> None:
"""Update entity connection state."""
self._attr_available = connection_state

2
homeassistant/components/bangolufsen/manifest.json
View File

@@ -6,6 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/bangolufsen",
"integration_type": "device",
"iot_class": "local_push",
"requirements": ["mozart-api==3.2.1.150.1"],
"requirements": ["mozart-api==3.2.1.150.4"],
"zeroconf": ["_bangolufsen._tcp.local."]
}

360
homeassistant/components/bangolufsen/media_player.py
View File

@@ -1,10 +1,9 @@
"""Media player entity for the Bang & Olufsen integration."""
from __future__ import annotations
from datetime import datetime, timedelta
from datetime import timedelta
import json
import logging
from multiprocessing.pool import ApplyResult
from typing import Any, cast
from mozart_api import __version__ as MOZART_API_VERSION
@@ -12,23 +11,17 @@ from mozart_api.exceptions import ApiException
from mozart_api.models import (
Action,
Art,
BeolinkLeader,
BeolinkListener,
OverlayPlayRequest,
PlaybackContentMetadata,
PlaybackError,
PlaybackProgress,
PlayQueueItem,
PlayQueueItemType,
PlayQueueSettings,
ProductState,
RemoteMenuItem,
RenderingState,
SceneProperties,
SoftwareUpdateState,
SoftwareUpdateStatus,
Source,
SourceArray,
Uri,
UserFlow,
VolumeLevel,
@@ -36,6 +29,7 @@ from mozart_api.models import (
VolumeSettings,
VolumeState,
)
from mozart_api.mozart_client import get_highest_resolution_artwork
from homeassistant.components import media_source
from homeassistant.components.media_player import (
@@ -46,35 +40,32 @@ from homeassistant.components.media_player import (
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
RepeatMode,
async_process_play_media_url,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_MODEL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceEntry, DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.dt import utcnow
from .const import (
ART_SIZE_ENUM,
BANGOLUFSEN_MEDIA_TYPE,
BANGOLUFSEN_STATES,
CONF_BEOLINK_JID,
CONF_DEFAULT_VOLUME,
CONF_MAX_VOLUME,
CONF_VOLUME_STEP,
DOMAIN,
ENTITY_ENUM,
FALLBACK_SOURCES,
HIDDEN_SOURCE_IDS,
REPEAT_ENUM,
SOURCE_ENUM,
VALID_MEDIA_TYPES,
WEBSOCKET_NOTIFICATION,
)
from .entity import BangOlufsenEntity
from .util import get_device
_LOGGER = logging.getLogger(__name__)
@@ -86,19 +77,15 @@ BANGOLUFSEN_FEATURES = (
| MediaPlayerEntityFeature.PREVIOUS_TRACK
| MediaPlayerEntityFeature.NEXT_TRACK
| MediaPlayerEntityFeature.PLAY_MEDIA
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.CLEAR_PLAYLIST
| MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.SHUFFLE_SET
| MediaPlayerEntityFeature.BROWSE_MEDIA
| MediaPlayerEntityFeature.REPEAT_SET
| MediaPlayerEntityFeature.TURN_OFF
)
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(minutes=2)
@@ -108,9 +95,10 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up a Media Player entity from config entry."""
entity = hass.data[DOMAIN][config_entry.unique_id][ENTITY_ENUM.MEDIA_PLAYER]
# Add MediaPlayer entity
async_add_entities(new_entities=[entity], update_before_add=True)
async_add_entities(
new_entities=[BangOlufsenMediaPlayer(config_entry)], update_before_add=True
)
class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
@@ -128,9 +116,7 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
self._default_volume: int = self.entry.data[CONF_DEFAULT_VOLUME]
self._max_volume: int = self.entry.data[CONF_MAX_VOLUME]
self._model: str = self.entry.data[CONF_MODEL]
self._volume_step: int = self.entry.data[CONF_VOLUME_STEP]
self._attr_device_class = MediaPlayerDeviceClass.SPEAKER
self._attr_device_info = DeviceInfo(
configuration_url=f"http://{self._host}/#/",
identifiers={(DOMAIN, self._unique_id)},
@@ -139,19 +125,16 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
name=cast(str, self.name),
)
self._attr_name = self._name
self._attr_should_poll = True
self._attr_unique_id = self._unique_id
self._attr_device_class = MediaPlayerDeviceClass.SPEAKER
# Misc. variables.
self._audio_sources: dict[str, str] = {}
self._beolink_listeners: list[BeolinkListener] = []
self._last_update: datetime = datetime(1970, 1, 1, 0, 0, 0, 0)
self._media_image: Art = Art()
self._queue_settings: PlayQueueSettings = PlayQueueSettings()
self._remote_leader: BeolinkLeader | None = None
# self._queue_settings: PlayQueueSettings = PlayQueueSettings()
self._software_status: SoftwareUpdateStatus = SoftwareUpdateStatus(
software_version="",
state=SoftwareUpdateState(seconds_remaining=0, value="idle"),
softwareVersion="",
state=SoftwareUpdateState(secondsRemaining=0, value="idle"),
)
self._sources: dict[str, str] = {}
self._state: str = MediaPlayerState.IDLE
@@ -159,66 +142,72 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
async def async_added_to_hass(self) -> None:
"""Turn on the dispatchers."""
await self._initialize()
await super().async_added_to_hass()
self._dispatchers.extend(
[
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_METADATA}",
self._update_playback_metadata,
),
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_ERROR}",
self._update_playback_error,
),
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_PROGRESS}",
self._update_playback_progress,
),
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_STATE}",
self._update_playback_state,
),
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.SOURCE_CHANGE}",
self._update_source_change,
),
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.VOLUME}",
self._update_volume,
),
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.REMOTE_MENU_CHANGED}",
self._update_sources,
),
]
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_ERROR}",
self._update_playback_error,
)
)
async def async_update(self) -> None:
"""Update polling information."""
if self._attr_available:
self._queue_settings = cast(
ApplyResult[PlayQueueSettings],
self._client.get_settings_queue(async_req=True, _request_timeout=5),
).get()
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_METADATA}",
self._update_playback_metadata,
)
)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_PROGRESS}",
self._update_playback_progress,
)
)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.PLAYBACK_STATE}",
self._update_playback_state,
)
)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.REMOTE_MENU_CHANGED}",
self._update_sources,
)
)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.SOURCE_CHANGE}",
self._update_source_change,
)
)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.VOLUME}",
self._update_volume,
)
)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.SOFTWARE_UPDATE_STATE}",
self._update_device,
)
)
async def _initialize(self) -> None:
"""Initialize connection dependent variables."""
# Get software version.
self._software_status = cast(
ApplyResult[SoftwareUpdateStatus],
self._client.get_softwareupdate_status(async_req=True),
).get()
self._software_status = await self._client.get_softwareupdate_status()
_LOGGER.debug(
"Connected to: %s %s running SW %s",
@@ -228,18 +217,15 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
)
# Set the default and maximum volume of the product.
self._client.set_volume_settings(
await self._client.set_volume_settings(
volume_settings=VolumeSettings(
default=VolumeLevel(level=self._default_volume),
maximum=VolumeLevel(level=self._max_volume),
),
async_req=True,
)
)
# Get overall device state once. This is handled by WebSocket events the rest of the time.
product_state = cast(
ApplyResult[ProductState], self._client.get_product_state(async_req=True)
).get()
product_state = await self._client.get_product_state()
# Get volume information.
if product_state.volume:
@@ -260,10 +246,10 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
if self._playback_state.value:
self._state = self._playback_state.value
self._last_update = utcnow()
self._attr_media_position_updated_at = utcnow()
# Get the highest resolution available of the given images.
self._update_artwork()
self._media_image = get_highest_resolution_artwork(self._playback_metadata)
# If the device has been updated with new sources, then the API will fail here.
await self._update_sources()
@@ -277,10 +263,7 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
# Audio sources
try:
# Get all available sources.
sources = cast(
ApplyResult[SourceArray],
self._client.get_available_sources(target_remote=False, async_req=True),
).get()
sources = await self._client.get_available_sources(target_remote=False)
# Use a fallback list of sources
except ValueError:
@@ -308,10 +291,7 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
}
# Video sources from remote menu
menu_items = cast(
ApplyResult[dict[str, RemoteMenuItem]],
self._client.get_remote_menu(async_req=True),
).get()
menu_items = await self._client.get_remote_menu()
for key in menu_items:
menu_item = menu_items[key]
@@ -337,40 +317,12 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
if self.hass.is_running:
self.async_write_ha_state()
def _update_artwork(self) -> None:
"""Find the highest resolution image."""
# Ensure that the metadata doesn't change mid processing.
metadata = self._playback_metadata
# Check if the metadata is not null and that there is art.
if (
isinstance(metadata, PlaybackContentMetadata)
and isinstance(metadata.art, list)
and len(metadata.art) > 0
):
images = []
# Images either have a key for specifying resolution or a "size" for the image.
for image in metadata.art:
# Netradio.
if metadata.art[0].key is not None:
images.append(int(image.key.split("x")[0]))
# Everything else.
elif metadata.art[0].size is not None:
images.append(ART_SIZE_ENUM[image.size].value)
# Choose the largest image.
self._media_image = metadata.art[images.index(max(images))]
# Don't leave stale image metadata if there is no available artwork.
else:
self._media_image = Art()
async def _update_playback_metadata(self, data: PlaybackContentMetadata) -> None:
"""Update _playback_metadata and related."""
self._playback_metadata = data
# Update current artwork.
self._update_artwork()
self._media_image = get_highest_resolution_artwork(self._playback_metadata)
self.async_write_ha_state()
@@ -381,7 +333,7 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
async def _update_playback_progress(self, data: PlaybackProgress) -> None:
"""Update _playback_progress and last update."""
self._playback_progress = data
self._last_update = utcnow()
self._attr_media_position_updated_at = utcnow()
self.async_write_ha_state()
@@ -399,12 +351,35 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
"""Update _source_change and related."""
self._source_change = data
# Check if source is line-in or optical and progress should be updated
if self._source_change.id in (SOURCE_ENUM.lineIn, SOURCE_ENUM.spdif):
self._playback_progress = PlaybackProgress(progress=0)
async def _update_volume(self, data: VolumeState) -> None:
"""Update _volume."""
self._volume = data
self.async_write_ha_state()
async def _update_device(self, data: SoftwareUpdateState) -> None:
"""Update HA device SW version."""
# Get software version.
software_status = await self._client.get_softwareupdate_status()
# Update the HA device if the sw version does not match
if not isinstance(self._device, DeviceEntry):
self._device = get_device(self.hass, self._unique_id)
assert isinstance(self._device, DeviceEntry)
if software_status.software_version != self._device.sw_version:
device_registry = dr.async_get(self.hass)
device_registry.async_update_device(
device_id=self._device.id,
sw_version=software_status.software_version,
)
@property
def state(self) -> MediaPlayerState:
"""Return the current state of the media player."""
@@ -440,15 +415,7 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
@property
def media_position(self) -> int | None:
"""Return the current playback progress."""
# Don't show progress if the the device is a Beolink listener.
if self._remote_leader is None:
return self._playback_progress.progress
return None
@property
def media_position_updated_at(self) -> datetime:
"""Return the last time that the playback position was updated."""
return self._last_update
return self._playback_progress.progress
@property
def media_image_url(self) -> str | None:
@@ -524,59 +491,19 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
return self._source_change.name
@property
def shuffle(self) -> bool | None:
"""Return if queues should be shuffled."""
return self._queue_settings.shuffle
@property
def repeat(self) -> RepeatMode | None:
"""Return current repeat setting for queues."""
if self._queue_settings.repeat:
return cast(RepeatMode, REPEAT_ENUM(self._queue_settings.repeat).name)
return None
async def async_turn_off(self) -> None:
"""Set the device to "networkStandby"."""
self._client.post_standby(async_req=True)
async def async_volume_up(self) -> None:
"""Volume up the on media player."""
if not self._volume.level or not self._volume.level.level:
_LOGGER.warning("Error setting volume")
return
new_volume = min(self._volume.level.level + self._volume_step, self._max_volume)
self._client.set_current_volume_level(
volume_level=VolumeLevel(level=new_volume),
async_req=True,
)
async def async_volume_down(self) -> None:
"""Volume down the on media player."""
if not self._volume.level or not self._volume.level.level:
_LOGGER.warning("Error setting volume")
return
new_volume = max(self._volume.level.level - self._volume_step, 0)
self._client.set_current_volume_level(
volume_level=VolumeLevel(level=new_volume),
async_req=True,
)
await self._client.post_standby()
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
self._client.set_current_volume_level(
volume_level=VolumeLevel(level=int(volume * 100)),
async_req=True,
await self._client.set_current_volume_level(
volume_level=VolumeLevel(level=int(volume * 100))
)
async def async_mute_volume(self, mute: bool) -> None:
"""Mute or unmute media player."""
self._client.set_volume_mute(
volume_mute=VolumeMute(muted=mute),
async_req=True,
)
await self._client.set_volume_mute(volume_mute=VolumeMute(muted=mute))
async def async_media_play_pause(self) -> None:
"""Toggle play/pause media player."""
@@ -587,28 +514,26 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
async def async_media_pause(self) -> None:
"""Pause media player."""
self._client.post_playback_command(command="pause", async_req=True)
await self._client.post_playback_command(command="pause")
async def async_media_play(self) -> None:
"""Play media player."""
self._client.post_playback_command(command="play", async_req=True)
await self._client.post_playback_command(command="play")
async def async_media_stop(self) -> None:
"""Pause media player."""
self._client.post_playback_command(command="stop", async_req=True)
await self._client.post_playback_command(command="stop")
async def async_media_next_track(self) -> None:
"""Send the next track command."""
self._client.post_playback_command(command="skip", async_req=True)
await self._client.post_playback_command(command="skip")
async def async_media_seek(self, position: float) -> None:
"""Seek to position in ms."""
if self.source == SOURCE_ENUM.deezer:
self._client.seek_to_position(
position_ms=int(position * 1000), async_req=True
)
await self._client.seek_to_position(position_ms=int(position * 1000))
# Try to prevent the playback progress from bouncing in the UI.
self._last_update = utcnow()
self._attr_media_position_updated_at = utcnow()
self._playback_progress = PlaybackProgress(progress=int(position))
self.async_write_ha_state()
@@ -617,28 +542,11 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
async def async_media_previous_track(self) -> None:
"""Send the previous track command."""
self._client.post_playback_command(command="prev", async_req=True)
await self._client.post_playback_command(command="prev")
async def async_clear_playlist(self) -> None:
"""Clear the current playback queue."""
self._client.post_clear_queue(async_req=True)
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Set playback queues to shuffle."""
self._client.set_settings_queue(
play_queue_settings=PlayQueueSettings(shuffle=shuffle),
async_req=True,
)
self._queue_settings.shuffle = shuffle
async def async_set_repeat(self, repeat: RepeatMode) -> None:
"""Set playback queues to repeat."""
self._client.set_settings_queue(
play_queue_settings=PlayQueueSettings(repeat=REPEAT_ENUM[repeat]),
async_req=True,
)
self._queue_settings.repeat = REPEAT_ENUM[repeat]
await self._client.post_clear_queue()
async def async_select_source(self, source: str) -> None:
"""Select an input source."""
@@ -656,10 +564,10 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
# Check for source type
if source in self._audio_sources.values():
# Audio
self._client.set_active_source(source_id=key, async_req=True)
await self._client.set_active_source(source_id=key)
else:
# Video
self._client.post_remote_trigger(id=key, async_req=True)
await self._client.post_remote_trigger(id=key)
async def async_play_media(
self,
@@ -693,33 +601,31 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
media_id = media_id.replace(".m3u", "")
if media_type in (MediaType.URL, MediaType.MUSIC):
self._client.post_uri_source(uri=Uri(location=media_id), async_req=True)
await self._client.post_uri_source(uri=Uri(location=media_id))
# The "provider" media_type may not be suitable for overlay all the time.
# Use it for now.
elif media_type == BANGOLUFSEN_MEDIA_TYPE.TTS:
self._client.post_overlay_play(
await self._client.post_overlay_play(
overlay_play_request=OverlayPlayRequest(
uri=Uri(location=media_id),
),
async_req=True,
)
)
elif media_type == BANGOLUFSEN_MEDIA_TYPE.RADIO:
self._client.run_provided_scene(
await self._client.run_provided_scene(
scene_properties=SceneProperties(
action_list=[
actionList=[
Action(
type="radio",
radio_station_id=media_id,
radioStationId=media_id,
)
]
),
async_req=True,
)
)
elif media_type == BANGOLUFSEN_MEDIA_TYPE.FAVOURITE:
self._client.activate_preset(id=int(media_id), async_req=True)
await self._client.activate_preset(id=int(media_id))
elif media_type == BANGOLUFSEN_MEDIA_TYPE.DEEZER:
try:
@@ -730,8 +636,8 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
deezer_id = kwargs[ATTR_MEDIA_EXTRA]["id"]
# Play Deezer flow.
self._client.start_deezer_flow(
user_flow=UserFlow(user_id=deezer_id), async_req=True
await self._client.start_deezer_flow(
user_flow=UserFlow(userId=deezer_id)
)
# Play a Deezer playlist or album.
@@ -740,26 +646,24 @@ class BangOlufsenMediaPlayer(MediaPlayerEntity, BangOlufsenEntity):
if "start_from" in kwargs[ATTR_MEDIA_EXTRA]:
start_from = kwargs[ATTR_MEDIA_EXTRA]["start_from"]
self._client.add_to_queue(
await self._client.add_to_queue(
play_queue_item=PlayQueueItem(
provider=PlayQueueItemType(value="deezer"),
start_now_from_position=start_from,
startNowFromPosition=start_from,
type="playlist",
uri=media_id,
),
async_req=True,
)
)
# Play a Deezer track.
else:
self._client.add_to_queue(
await self._client.add_to_queue(
play_queue_item=PlayQueueItem(
provider=PlayQueueItemType(value="deezer"),
start_now_from_position=0,
startNowFromPosition=0,
type="track",
uri=media_id,
),
async_req=True,
)
)
except ApiException as error:

21
homeassistant/components/bangolufsen/util.py Normal file
View File

@@ -0,0 +1,21 @@
"""Various utilities for the Bang & Olufsen integration."""
from __future__ import annotations
from typing import cast
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceEntry
from .const import DOMAIN
def get_device(hass: HomeAssistant | None, unique_id: str) -> DeviceEntry | None:
"""Get the device."""
if not isinstance(hass, HomeAssistant):
return None
device_registry = dr.async_get(hass)
device = cast(DeviceEntry, device_registry.async_get_device({(DOMAIN, unique_id)}))
return device

36
homeassistant/components/bangolufsen/websocket.py
View File

@@ -5,8 +5,6 @@ from __future__ import annotations
from datetime import datetime
import logging
from multiprocessing.pool import ApplyResult
from typing import cast
from mozart_api.models import (
PlaybackContentMetadata,
@@ -14,7 +12,6 @@ from mozart_api.models import (
PlaybackProgress,
RenderingState,
SoftwareUpdateState,
SoftwareUpdateStatus,
Source,
VolumeState,
WebsocketNotificationTag,
@@ -22,7 +19,6 @@ from mozart_api.models import (
from homeassistant.config_entries import ConfigEntry
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
@@ -30,9 +26,9 @@ from .const import (
BANGOLUFSEN_WEBSOCKET_EVENT,
CONNECTION_STATUS,
WEBSOCKET_NOTIFICATION,
BangOlufsenVariables,
get_device,
)
from .entity import BangOlufsenVariables
from .util import get_device
_LOGGER = logging.getLogger(__name__)
@@ -46,7 +42,6 @@ class BangOlufsenWebsocket(BangOlufsenVariables):
BangOlufsenVariables.__init__(self, entry)
self.hass = hass
self._device: DeviceEntry | None = None
# WebSocket callbacks
self._client.get_on_connection(self.on_connection)
@@ -165,27 +160,12 @@ class BangOlufsenWebsocket(BangOlufsenVariables):
)
def on_software_update_state(self, notification: SoftwareUpdateState) -> None:
"""Check device sw version."""
# Get software version.
software_status = cast(
ApplyResult[SoftwareUpdateStatus],
self._client.get_softwareupdate_status(async_req=True),
).get()
# Update the HA device if the sw version does not match
if not isinstance(self._device, DeviceEntry):
self._device = get_device(self.hass, self._unique_id)
assert isinstance(self._device, DeviceEntry)
if software_status.software_version != self._device.sw_version:
device_registry = dr.async_get(self.hass)
device_registry.async_update_device(
device_id=self._device.id,
sw_version=software_status.software_version,
)
"""Send software_update_state dispatch."""
async_dispatcher_send(
self.hass,
f"{self._unique_id}_{WEBSOCKET_NOTIFICATION.SOFTWARE_UPDATE_STATE}",
notification,
)
def on_all_notifications_raw(self, notification: dict) -> None:
"""Receive all notifications."""

2
requirements_all.txt
View File

@@ -1270,7 +1270,7 @@ motionblinds==0.6.18
motioneye-client==0.3.14
# homeassistant.components.bangolufsen
mozart-api==3.2.1.150.1
mozart-api==3.2.1.150.4
# homeassistant.components.mullvad
mullvad-api==1.0.0

2
requirements_test_all.txt
View File

@@ -994,7 +994,7 @@ motionblinds==0.6.18
motioneye-client==0.3.14
# homeassistant.components.bangolufsen
mozart-api==3.2.1.150.1
mozart-api==3.2.1.150.4
# homeassistant.components.mullvad
mullvad-api==1.0.0

10
tests/components/bangolufsen/conftest.py
View File

@@ -1,6 +1,6 @@
"""Test fixtures for bangolufsen."""
from unittest.mock import Mock, patch
from unittest.mock import AsyncMock, patch
from mozart_api.models import BeolinkPeer, VolumeLevel, VolumeSettings
import pytest
@@ -34,11 +34,11 @@ class MockMozartClient:
)
# API endpoints
get_beolink_self = Mock()
get_beolink_self.return_value.get.return_value = get_beolink_self_result
get_beolink_self = AsyncMock()
get_beolink_self.return_value = get_beolink_self_result
get_volume_settings = Mock()
get_volume_settings.return_value.get.return_value = get_volume_settings_result
get_volume_settings = AsyncMock()
get_volume_settings.return_value = get_volume_settings_result
@pytest.fixture

30
tests/components/bangolufsen/const.py
View File

@@ -1,6 +1,8 @@
"""Constants used for testing the bangolufsen integration."""
from ipaddress import IPv4Address
from homeassistant.components.bangolufsen.const import (
ATTR_FRIENDLY_NAME,
ATTR_ITEM_NUMBER,
@@ -9,10 +11,9 @@ from homeassistant.components.bangolufsen.const import (
CONF_BEOLINK_JID,
CONF_DEFAULT_VOLUME,
CONF_MAX_VOLUME,
CONF_VOLUME_STEP,
)
from homeassistant.components.zeroconf import ZeroconfServiceInfo
from homeassistant.const import CONF_HOST, CONF_MODEL, CONF_NAME
from homeassistant.const import CONF_HOST, CONF_MODEL
TEST_HOST = "192.168.0.1"
TEST_HOST_INVALID = "192.168.0"
@@ -48,14 +49,12 @@ TEST_DATA_USER = {CONF_HOST: TEST_HOST, CONF_MODEL: TEST_MODEL_BALANCE}
TEST_DATA_USER_INVALID = {CONF_HOST: TEST_HOST_INVALID, CONF_MODEL: TEST_MODEL_BALANCE}
TEST_DATA_NO_HOST = {
CONF_VOLUME_STEP: TEST_VOLUME_STEP,
CONF_DEFAULT_VOLUME: TEST_DEFAULT_VOLUME,
CONF_MAX_VOLUME: TEST_MAX_VOLUME,
}
TEST_DATA_CONFIRM = {
CONF_HOST: TEST_HOST,
CONF_VOLUME_STEP: TEST_VOLUME_STEP,
CONF_DEFAULT_VOLUME: TEST_DEFAULT_VOLUME,
CONF_MAX_VOLUME: TEST_MAX_VOLUME,
CONF_MODEL: TEST_MODEL_BALANCE,
@@ -63,8 +62,8 @@ TEST_DATA_CONFIRM = {
}
TEST_DATA_ZEROCONF = ZeroconfServiceInfo(
addresses=[TEST_HOST],
host=TEST_HOST,
ip_address=IPv4Address(TEST_HOST),
ip_addresses=[IPv4Address(TEST_HOST)],
port=80,
hostname=TEST_HOSTNAME_ZEROCONF,
type=TEST_TYPE_ZEROCONF,
@@ -78,26 +77,11 @@ TEST_DATA_ZEROCONF = ZeroconfServiceInfo(
)
TEST_DATA_ZEROCONF_NOT_MOZART = ZeroconfServiceInfo(
addresses=[TEST_HOST],
host=TEST_HOST,
ip_address=IPv4Address(TEST_HOST),
ip_addresses=[IPv4Address(TEST_HOST)],
port=80,
hostname=TEST_HOSTNAME_ZEROCONF,
type=TEST_TYPE_ZEROCONF,
name=TEST_NAME_ZEROCONF,
properties={ATTR_SERIAL_NUMBER: TEST_SERIAL_NUMBER},
)
TEST_DATA_OPTIONS = {
CONF_NAME: TEST_NAME_OPTIONS,
CONF_VOLUME_STEP: TEST_VOLUME_STEP_OPTIONS,
CONF_DEFAULT_VOLUME: TEST_DEFAULT_VOLUME_OPTIONS,
CONF_MAX_VOLUME: TEST_MAX_VOLUME_OPTIONS,
}
TEST_DATA_OPTIONS_FULL = {
CONF_HOST: TEST_HOST,
CONF_VOLUME_STEP: TEST_VOLUME_STEP_OPTIONS,
CONF_DEFAULT_VOLUME: TEST_DEFAULT_VOLUME_OPTIONS,
CONF_MAX_VOLUME: TEST_MAX_VOLUME_OPTIONS,
CONF_MODEL: TEST_MODEL_BALANCE,
CONF_BEOLINK_JID: TEST_JID_1,
}

43
tests/components/bangolufsen/test_config_flow.py
View File

@@ -1,6 +1,8 @@
"""Test the bangolufsen config_flow."""
from unittest.mock import Mock
from mozart_api.exceptions import ApiException, NotFoundException
import pytest
from urllib3.exceptions import MaxRetryError, NewConnectionError
@@ -14,7 +16,6 @@ from homeassistant.data_entry_flow import FlowResultType
from .conftest import MockMozartClient
from .const import (
TEST_DATA_CONFIRM,
TEST_DATA_OPTIONS,
TEST_DATA_USER,
TEST_DATA_USER_INVALID,
TEST_DATA_ZEROCONF,
@@ -28,7 +29,7 @@ async def test_config_flow_max_retry_error(
hass: HomeAssistant, mock_client: MockMozartClient
) -> None:
"""Test we handle not_mozart_device."""
mock_client.get_beolink_self.side_effect = MaxRetryError(pool=None, url=None)
mock_client.get_beolink_self.side_effect = MaxRetryError(pool=Mock(), url="")
result_user = await hass.config_entries.flow.async_init(
handler=DOMAIN,
@@ -64,9 +65,7 @@ async def test_config_flow_new_connection_error(
hass: HomeAssistant, mock_client: MockMozartClient
) -> None:
"""Test we handle new_connection_error."""
mock_client.get_beolink_self.side_effect = NewConnectionError(
pool=None, message=None
)
mock_client.get_beolink_self.side_effect = NewConnectionError(Mock(), "")
result_user = await hass.config_entries.flow.async_init(
handler=DOMAIN,
@@ -184,26 +183,26 @@ async def test_config_flow_zeroconf_not_mozart_device(hass: HomeAssistant) -> No
assert result_user["reason"] == "not_mozart_device"
async def test_config_flow_options(hass: HomeAssistant, mock_config_entry) -> None:
"""Test config flow options."""
# async def test_config_flow_options(hass: HomeAssistant, mock_config_entry) -> None:
# """Test config flow options."""
mock_config_entry.add_to_hass(hass)
# mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
# assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
result_user = await hass.config_entries.options.async_init(
mock_config_entry.entry_id
)
# result_user = await hass.config_entries.options.async_init(
# mock_config_entry.entry_id
# )
assert result_user["type"] == FlowResultType.FORM
assert result_user["step_id"] == "init"
# assert result_user["type"] == FlowResultType.FORM
# assert result_user["step_id"] == "init"
result_confirm = await hass.config_entries.options.async_configure(
flow_id=result_user["flow_id"],
user_input=TEST_DATA_OPTIONS,
)
# result_confirm = await hass.config_entries.options.async_configure(
# flow_id=result_user["flow_id"],
# user_input=TEST_DATA_OPTIONS,
# )
assert result_confirm["type"] == FlowResultType.CREATE_ENTRY
new_data = TEST_DATA_CONFIRM
new_data.update(TEST_DATA_OPTIONS)
assert result_confirm["data"] == new_data
# assert result_confirm["type"] == FlowResultType.CREATE_ENTRY
# new_data = TEST_DATA_CONFIRM
# new_data.update(TEST_DATA_OPTIONS)
# assert result_confirm["data"] == new_data