"""Asynchronous Python client for StreamMagic API.""" import asyncio import json from asyncio import AbstractEventLoop, Future, Task, Queue from datetime import datetime, UTC from typing import Any, Optional, Callable, Awaitable from aiohttp import ClientWebSocketResponse, ClientSession from aiostreammagic.exceptions import StreamMagicError from aiostreammagic.models import ( Info, Source, State, PlayState, NowPlaying, ShuffleMode, RepeatMode, CallbackType, AudioOutput, Display, DisplayBrightness, Update, PresetList, ControlBusMode, StandbyMode, Audio, EQBand, EQFilterType, EQ_PRESETS, ) from aiostreammagic.util import eq_bands_to_param_string from . import endpoints as ep from .const import _LOGGER, WS_HEARTBEAT_TIME class StreamMagicClient: """Client for handling connections with StreamMagic enabled devices.""" def __init__( self, host: str, session: ClientSession | None = None, *, should_close_session: bool = True, ) -> None: self.host = host self.session: Optional[ClientSession] = session self._should_close_session: bool = should_close_session self.connection: ClientWebSocketResponse | None = None self.futures: dict[str, list[Future[Any]]] = {} self._subscriptions: dict[str, Any] = {} self._loop: AbstractEventLoop = asyncio.get_running_loop() self.connect_result: Future[bool] | None = None self.connect_task: Task[Any] | None = None self.state_update_callbacks: list[Any] = [] self._allow_state_update = False self._info: Optional[Info] = None self.sources: list[Source] = [] self._state: Optional[State] = None self._play_state: Optional[PlayState] = None self._now_playing: Optional[NowPlaying] = None self._audio: Audio | None = None self._audio_output: Optional[AudioOutput] = None self._display: Optional[Display] = None self._update: Optional[Update] = None self._preset_list: Optional[PresetList] = None self._attempt_reconnection = False self._reconnect_task: Optional[Task[Any]] = None self.position_last_updated: datetime = datetime.now() self._subscription_tasks: dict[str, asyncio.Task[Any]] = {} async def register_state_update_callbacks(self, callback: Any) -> None: """Register state update callback.""" self.state_update_callbacks.append(callback) if self._allow_state_update: await callback(self, CallbackType.STATE) def unregister_state_update_callbacks(self, callback: Any) -> None: """Unregister state update callback.""" if callback in self.state_update_callbacks: self.state_update_callbacks.remove(callback) def clear_state_update_callbacks(self) -> None: """Clear state update callbacks.""" self.state_update_callbacks.clear() async def do_state_update_callbacks( self, callback_type: CallbackType = CallbackType.STATE ) -> None: """Call state update callbacks.""" if not self.state_update_callbacks: return callbacks = set() for callback in self.state_update_callbacks: callbacks.add(callback(self, callback_type)) if callbacks: await asyncio.gather(*callbacks) async def connect(self) -> Any: """Connect to StreamMagic enabled devices.""" if not self.is_connected(): self.connect_result = self._loop.create_future() self._reconnect_task = asyncio.create_task( self._reconnect_handler(self.connect_result) ) return await self.connect_result # Already connected, just return True return True async def disconnect(self) -> None: """Disconnect from StreamMagic enabled devices.""" self._attempt_reconnection = False if self.connection is not None and not self.connection.closed: await self.connection.close() self.connection = None if self._reconnect_task: self._reconnect_task.cancel() await asyncio.gather(self._reconnect_task, return_exceptions=True) self._reconnect_task = None if self.connect_task: self.connect_task.cancel() await asyncio.gather(self.connect_task, return_exceptions=True) self.connect_task = None # Cancel all subscription handler tasks for task in self._subscription_tasks.values(): task.cancel() await asyncio.gather(*self._subscription_tasks.values(), return_exceptions=True) self._subscription_tasks.clear() await self.do_state_update_callbacks(CallbackType.CONNECTION) # Properly close the aiohttp session if it was created by this client if self._should_close_session and self.session is not None: if not self.session.closed: await self.session.close() self.session = None def is_connected(self) -> bool: """Return True if device is connected.""" return ( self.connection is not None and not self.connection.closed and self.connect_task is not None and not self.connect_task.done() ) async def _ws_connect(self, uri: str) -> ClientWebSocketResponse: """Establish a connection with a WebSocket.""" if self.session is None: self.session = ClientSession() return await self.session.ws_connect( uri, headers={ "Origin": f"ws://{self.host}", "Host": f"{self.host}:80", }, heartbeat=WS_HEARTBEAT_TIME, ) async def _reconnect_handler(self, res: Future[bool]) -> None: reconnect_delay = 0.5 while True: try: self.connect_task = asyncio.create_task(self._connect_handler(res)) await self.connect_task except asyncio.CancelledError: raise except Exception: _LOGGER.exception("StreamMagic connection handler failed") await self.do_state_update_callbacks(CallbackType.CONNECTION) if not self._attempt_reconnection: _LOGGER.debug( "Failed to connect to device on initial pass, skipping reconnect." ) break reconnect_delay = min(reconnect_delay * 2, 30) _LOGGER.debug( f"Attempting reconnection to Cambridge Audio device in {reconnect_delay} seconds..." ) await asyncio.sleep(reconnect_delay) async def _connect_handler(self, res: Future[bool]) -> None: """Handle connection for StreamMagic.""" try: self.futures = {} self._allow_state_update = False uri = f"ws://{self.host}/smoip" ws = await self._ws_connect(uri) self.connection = ws x = asyncio.create_task( self.consumer_handler(ws, self._subscriptions, self.futures) ) # mypy/typeshed bug: https://github.com/python/mypy/issues/17030 # The following ignore is safe because we know the return types. ( self._info, self.sources, self._state, self._play_state, self._now_playing, self._audio, self._audio_output, self._display, self._update, self._preset_list, ) = await asyncio.gather( # type: ignore[assignment] self.get_info(), self.get_sources(), self.get_state(), self.get_play_state(), self.get_now_playing(), self.get_audio(), self.get_audio_output(), self.get_display(), self.get_update(), self.get_preset_list(), ) subscribe_state_updates = { self.subscribe(self._async_handle_info, ep.INFO), self.subscribe(self._async_handle_sources, ep.SOURCES), self.subscribe(self._async_handle_zone_state, ep.ZONE_STATE), self.subscribe(self._async_handle_play_state, ep.PLAY_STATE), self.subscribe(self._async_handle_position, ep.POSITION), self.subscribe(self._async_handle_now_playing, ep.NOW_PLAYING), self.subscribe(self._async_handle_audio_output, ep.ZONE_AUDIO_OUTPUT), self.subscribe(self._async_handle_display, ep.DISPLAY), self.subscribe(self._async_handle_update, ep.UPDATE), self.subscribe(self._async_handle_preset_list, ep.PRESET_LIST), self.subscribe(self._async_handle_audio, ep.AUDIO), } subscribe_tasks = set() for state_update in subscribe_state_updates: subscribe_tasks.add(asyncio.create_task(state_update)) await asyncio.gather(*subscribe_tasks) self._allow_state_update = True await self.do_state_update_callbacks(CallbackType.CONNECTION) self._attempt_reconnection = True if not res.done(): res.set_result(True) await x except asyncio.CancelledError: raise except Exception as ex: if not res.done(): res.set_exception(ex) raise @staticmethod async def subscription_handler( queue: Queue[dict[str, Any]], callback: Callable[[dict[str, Any]], Awaitable[None]], ) -> None: """Handle subscriptions.""" try: while True: msg = await queue.get() await callback(msg) except asyncio.CancelledError: pass async def consumer_handler( self, ws: ClientWebSocketResponse, subscriptions: dict[str, list[Any]], futures: dict[str, list[asyncio.Future[Any]]], ) -> None: """Callback consumer handler.""" subscription_queues: dict[str, Queue[dict[str, Any]]] = {} try: async for raw_msg in ws: try: if futures or subscriptions: _LOGGER.debug("recv(%s): %s", self.host, raw_msg) msg = json.loads(raw_msg.data) path = msg["path"] path_futures = self.futures.get(path) subscription = self._subscriptions.get(path) if path_futures and msg.get("type") == "response": for future in path_futures: if not future.done(): future.set_result(msg) if subscription: queue = subscription_queues.get(path) if queue is None: queue = asyncio.Queue() subscription_queues[path] = queue old_task = self._subscription_tasks.pop(path, None) if old_task: old_task.cancel() await asyncio.gather( old_task, return_exceptions=True ) self._subscription_tasks[path] = asyncio.create_task( self.subscription_handler(queue, subscription) ) queue.put_nowait(msg) except Exception: _LOGGER.exception( "Failed handling StreamMagic websocket message: %s", raw_msg ) raise except asyncio.CancelledError: raise except Exception: _LOGGER.exception("StreamMagic websocket consumer failed") raise finally: if self.connection is ws: self.connection = None for task in self._subscription_tasks.values(): task.cancel() await asyncio.gather( *self._subscription_tasks.values(), return_exceptions=True ) self._subscription_tasks.clear() for path_futures in self.futures.values(): for future in path_futures: if not future.done(): future.set_exception( StreamMagicError( "StreamMagic consumer stopped before response was received" ) ) self.futures.clear() async def _send( self, path: str, params: Optional[dict[str, str | int | float | bool]] = None ) -> None: """Send a command to the device.""" message = { "path": path, "params": params or {}, } if not self.connection: raise StreamMagicError("Not connected to device.") _LOGGER.debug("Sending command: %s", message) await self.connection.send_str(json.dumps(message)) async def request( self, path: str, params: Optional[dict[str, str | int | float | bool]] = None ) -> Any: res = self._loop.create_future() path_futures = self.futures.get(path, []) path_futures.append(res) self.futures[path] = path_futures try: await self._send(path, params) except (asyncio.CancelledError, StreamMagicError): path_futures.remove(res) raise try: response = await res except asyncio.CancelledError: if res in path_futures: path_futures.remove(res) raise path_futures.remove(res) message = response["message"] result = response["result"] if result != 200: raise StreamMagicError(message) return response async def subscribe(self, callback: Any, path: str) -> Any: self._subscriptions[path] = callback try: await self._send(path, {"update": 100, "zone": "ZONE1"}) except (asyncio.CancelledError, StreamMagicError): del self._subscriptions[path] raise @property def info(self) -> Info: """Return a type-guaranteed instance of Info""" if not self._info: raise StreamMagicError("Info not available.") return self._info @property def state(self) -> State: """Return a type-guaranteed instance of State""" if not self._state: raise StreamMagicError("State not available.") return self._state @property def play_state(self) -> PlayState: """Return a type-guaranteed instance of PlayState""" if not self._play_state: raise StreamMagicError("Play state not available.") return self._play_state @property def now_playing(self) -> NowPlaying: """Return a type-guaranteed instance of NowPlaying""" if not self._now_playing: raise StreamMagicError("NowPlaying not available.") return self._now_playing @property def audio(self) -> Audio: """Return a type-guaranteed instance of Audio""" if not self._audio: raise StreamMagicError("Audio not available.") return self._audio @property def audio_output(self) -> AudioOutput: """Return a type-guaranteed instance of AudioOutput""" if not self._audio_output: raise StreamMagicError("AudioOutput not available.") return self._audio_output @property def display(self) -> Display: """Return a type-guaranteed instance of Display""" if not self._display: raise StreamMagicError("Display not available.") return self._display @property def update(self) -> Update: """Return a type-guaranteed instance of Update""" if not self._update: raise StreamMagicError("Update not available.") return self._update @property def preset_list(self) -> PresetList: """Return a type-guaranteed instance of PresetList""" if not self._preset_list: raise StreamMagicError("PresetList not available.") return self._preset_list async def get_info(self) -> Info: """Get device information from device.""" data = await self.request(ep.INFO) return Info.from_dict(data["params"]["data"]) async def get_sources(self) -> list[Source]: """Get source information from device.""" data = await self.request(ep.SOURCES) sources = [Source.from_dict(x) for x in data["params"]["data"]["sources"]] return sources async def get_state(self) -> State: """Get state information from device.""" data = await self.request(ep.ZONE_STATE) return State.from_dict(data["params"]["data"]) async def get_play_state(self) -> PlayState: """Get play state information from device.""" data = await self.request(ep.PLAY_STATE) return PlayState.from_dict(data["params"]["data"]) async def get_now_playing(self) -> NowPlaying: """Get now playing information from device.""" data = await self.request(ep.NOW_PLAYING) return NowPlaying.from_dict(data["params"]["data"]) async def get_audio(self) -> Audio | None: """Get audio information from device.""" data = await self.request(ep.AUDIO) return Audio.from_dict(data["params"]["data"]) async def get_audio_output(self) -> AudioOutput: """Get audio output information from device.""" data = await self.request(ep.ZONE_AUDIO_OUTPUT) return AudioOutput.from_dict(data["params"]["data"]) async def get_display(self) -> Display: """Get display information from device.""" data = await self.request(ep.DISPLAY) return Display.from_dict(data["params"]["data"]) async def get_update(self) -> Update: """Get display information from device.""" data = await self.request(ep.UPDATE) return Update.from_dict(data["params"]["data"]) async def get_preset_list(self) -> PresetList: """Get preset list information from device.""" data = await self.request(ep.PRESET_LIST) return PresetList.from_dict(data["params"]["data"]) async def _async_handle_info(self, payload: dict[str, Any]) -> None: """Handle async info update.""" params = payload["params"] if "data" in params: self._info = Info.from_dict(params["data"]) await self.do_state_update_callbacks() async def _async_handle_sources(self, payload: dict[str, Any]) -> None: """Handle async sources update.""" params = payload["params"] if "data" in params: self.sources = [Source.from_dict(x) for x in params["data"]["sources"]] await self.do_state_update_callbacks() async def _async_handle_zone_state(self, payload: dict[str, Any]) -> None: """Handle async zone state update.""" params = payload["params"] if "data" in params: self._state = State.from_dict(params["data"]) await self.do_state_update_callbacks() async def _async_handle_play_state(self, payload: dict[str, Any]) -> None: """Handle async zone state update.""" params = payload["params"] if "data" in params: self._play_state = PlayState.from_dict(params["data"]) self.position_last_updated = datetime.now() await self.do_state_update_callbacks() async def _async_handle_position(self, payload: dict[str, Any]) -> None: """Handle async position update.""" params = payload["params"] if "data" in params and params["data"]["position"] and self.play_state: self.play_state.position = params["data"]["position"] self.position_last_updated = datetime.now(UTC) await self.do_state_update_callbacks() async def _async_handle_now_playing(self, payload: dict[str, Any]) -> None: """Handle async now playing update.""" params = payload["params"] if "data" in params: self._now_playing = NowPlaying.from_dict(params["data"]) await self.do_state_update_callbacks() async def _async_handle_audio(self, payload: dict[str, Any]) -> None: """Handle async audio update.""" params = payload["params"] if "data" in params: self._audio = Audio.from_dict(params["data"]) await self.do_state_update_callbacks() async def _async_handle_audio_output(self, payload: dict[str, Any]) -> None: """Handle async audio output update.""" params = payload["params"] if "data" in params: self._audio_output = AudioOutput.from_dict(params["data"]) await self.do_state_update_callbacks() async def _async_handle_display(self, payload: dict[str, Any]) -> None: """Handle async display update.""" params = payload["params"] if "data" in params: self._display = Display.from_dict(params["data"]) await self.do_state_update_callbacks() async def _async_handle_update(self, payload: dict[str, Any]) -> None: """Handle async display update.""" params = payload["params"] if "data" in params: self._update = Update.from_dict(params["data"]) await self.do_state_update_callbacks() async def _async_handle_preset_list(self, payload: dict[str, Any]) -> None: """Handle async preset list update.""" params = payload["params"] if "data" in params: self._preset_list = PresetList.from_dict(params["data"]) await self.do_state_update_callbacks() async def power_on(self) -> None: """Set the power of the device to on.""" await self.request(ep.POWER, params={"power": "ON"}) async def power_off(self) -> None: """Set the power of the device to network.""" await self.request(ep.POWER, params={"power": "NETWORK"}) async def volume_up(self) -> None: """Increase the volume of the device by 1.""" await self.request( ep.ZONE_STATE, params={"zone": "ZONE1", "volume_step_change": 1} ) async def volume_down(self) -> None: """Increase the volume of the device by -1.""" await self.request( ep.ZONE_STATE, params={"zone": "ZONE1", "volume_step_change": -1} ) async def set_volume(self, volume: int) -> None: """Set the volume of the device.""" if not 0 <= volume <= 100: raise StreamMagicError("Volume must be between 0 and 100") await self.request( ep.ZONE_STATE, params={"zone": "ZONE1", "volume_percent": volume} ) async def set_mute(self, mute: bool) -> None: """Set the mute of the device.""" await self.request(ep.ZONE_STATE, params={"zone": "ZONE1", "mute": mute}) async def set_source(self, source: Source) -> None: """Set the source of the device.""" await self.set_source_by_id(source.id) async def set_source_by_id(self, source_id: str) -> None: """Set the source of the device.""" await self.request(ep.ZONE_STATE, params={"zone": "ZONE1", "source": source_id}) async def media_seek(self, position: int) -> None: """Set the media position of the device.""" await self.request( ep.PLAY_CONTROL, params={"zone": "ZONE1", "position": position} ) async def next_track(self) -> None: """Skip the next track.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "skip_track": 1} ) async def previous_track(self) -> None: """Skip the next track.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "skip_track": -1} ) async def play_pause(self) -> None: """Toggle play/pause.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "action": "toggle"}, ) async def play(self) -> None: """Play the device.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "action": "play"}, ) async def pause(self) -> None: """Pause the device.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "action": "pause"}, ) async def stop(self) -> None: """Pause the device.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "action": "stop"} ) async def set_shuffle(self, shuffle: ShuffleMode) -> None: """Set the shuffle of the device.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "mode_shuffle": shuffle}, ) async def set_repeat(self, repeat: RepeatMode) -> None: """Set the repeat of the device.""" await self.request( ep.PLAY_CONTROL, params={"match": "none", "zone": "ZONE1", "mode_repeat": repeat}, ) async def play_radio_airable(self, name: str, airable_radio_id: int) -> None: """Play an airable radio station.""" await self.request( ep.STREAM_RADIO, params={ "zone": "ZONE1", "airable_radio_id": airable_radio_id, "name": name, }, ) async def play_radio_url(self, name: str, url: str) -> None: """Play a radio station from a provided url.""" await self.request( ep.STREAM_RADIO, params={"zone": "ZONE1", "url": url, "name": name} ) async def set_audio_output(self, output_id: str) -> None: """Set the audio output of the device.""" await self.request( ep.ZONE_AUDIO_OUTPUT, params={"zone": "ZONE1", "id": output_id} ) async def set_pre_amp_mode(self, enabled: bool) -> None: """Sets whether the internal pre-amp is enabled.""" await self.request(ep.ZONE_STATE, params={"pre_amp_mode": enabled}) async def set_equalizer_mode(self, enabled: bool) -> None: """Sets whether the internal equalizer is enabled.""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") await self.request(ep.AUDIO, params={"zone": "ZONE1", "user_eq": enabled}) async def set_equalizer_band_filter( self, band_index: int, filter_type: EQFilterType ) -> None: """Sets the filter type for a specific equalizer band.""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") band = EQBand(index=band_index, filter=filter_type) await self.set_equalizer_params([band]) async def set_equalizer_band_frequency( self, band_index: int, frequency: int ) -> None: """Sets the frequency for a specific equalizer band.""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") if not 20 <= frequency <= 20000: raise StreamMagicError("Frequency must be between 20 Hz and 20 kHz") band = EQBand(index=band_index, freq=frequency) await self.set_equalizer_params([band]) async def set_equalizer_band_gain(self, band_index: int, gain: float) -> None: """Sets the gain for a specific equalizer band.""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") if not -6 <= gain <= 3: raise StreamMagicError("Gain must be between -6 dB and 3 dB") band = EQBand(index=band_index, gain=gain) await self.set_equalizer_params([band]) async def set_equalizer_band_q_factor(self, band_index: int, q: float) -> None: """Sets the Q factor for a specific equalizer band.""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") if not 0.1 <= q <= 10: raise StreamMagicError("Q factor must be between 0.1 and 10") band = EQBand(index=band_index, q=q) await self.set_equalizer_params([band]) async def set_equalizer_defaults(self) -> None: """Sets the equalizer to the default settings.""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") bands = [ EQBand(index=0, filter=EQFilterType.LOWSHELF, freq=80, gain=0.0, q=0.8), EQBand(index=1, filter=EQFilterType.PEAKING, freq=120, gain=0.0, q=1.24), EQBand(index=2, filter=EQFilterType.PEAKING, freq=315, gain=0.0, q=1.24), EQBand(index=3, filter=EQFilterType.PEAKING, freq=800, gain=0.0, q=1.24), EQBand(index=4, filter=EQFilterType.PEAKING, freq=2000, gain=0.0, q=1.24), EQBand(index=5, filter=EQFilterType.PEAKING, freq=5000, gain=0.0, q=1.24), EQBand(index=6, filter=EQFilterType.HIGHSHELF, freq=8000, gain=0.0, q=0.8), ] await self.set_equalizer_params(bands) async def set_equalizer_preset(self, eq_preset_name: str) -> None: """Sets the equalizer to a preset configuration.""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") if eq_preset_name not in EQ_PRESETS: available = ", ".join(sorted(EQ_PRESETS.keys())) raise StreamMagicError( f"Unknown preset '{eq_preset_name}'. Available presets: {available}" ) gains = EQ_PRESETS[eq_preset_name] bands = [EQBand(index=i, gain=gain) for i, gain in enumerate(gains)] await self.set_equalizer_params(bands) async def set_equalizer_params(self, bands: list[EQBand]) -> None: """Sets the internal equalizer to the provided band settings""" if self.audio.user_eq is None: raise StreamMagicError("Equalizer is not supported on this device") await self.request( ep.AUDIO, params={"zone": "ZONE1", "user_eq_bands": eq_bands_to_param_string(bands)}, ) async def set_room_correction_mode(self, enabled: bool) -> None: """Sets whether the internal room correction is enabled.""" if self.audio.tilt_eq is None: raise StreamMagicError("Room correction is not supported on this device") await self.request(ep.AUDIO, params={"zone": "ZONE1", "tilt_eq": enabled}) async def set_room_correction_intensity(self, intensity: int) -> None: """Sets the intensity of the room correction.""" if self.audio.tilt_eq is None: raise StreamMagicError("Room correction is not supported on this device") if not -15 <= intensity <= 15: raise StreamMagicError("Intensity must be between -15 and 15") await self.request( ep.AUDIO, params={"zone": "ZONE1", "tilt_intensity": intensity} ) async def set_balance(self, balance: int) -> None: """Sets the balance for the internal pre-amp of the device.""" if self.audio.balance is None: raise StreamMagicError("Balance is not supported on this device") if not -15 <= balance <= 15: raise StreamMagicError("Balance must be between -15 and 15") await self.request(ep.AUDIO, params={"zone": "ZONE1", "balance": balance}) async def set_volume_limit(self, volume_limit_percent: int) -> None: """Sets the volume limit for the internal pre-amp. Value must be between 1 and 100.""" if not 1 <= volume_limit_percent <= 100: raise StreamMagicError("Volume limit must be between 1 and 100") await self.request( ep.AUDIO, params={"volume_limit_percent": volume_limit_percent} ) async def set_device_name(self, device_name: str) -> None: """Set the device name.""" await self.request(ep.INFO, params={"name": device_name}) async def set_display_brightness( self, display_brightness: DisplayBrightness ) -> None: """Set the display brightness of the device.""" await self.request(ep.DISPLAY, params={"brightness": display_brightness}) async def set_early_update(self, early_update: bool) -> None: """Set whether the device should be on the early update channel.""" await self.request( ep.UPDATE, params={"early_update": early_update, "action": "CHECK"} ) async def recall_preset(self, preset: int) -> None: """Recall a preset for the device.""" await self.request(ep.RECALL_PRESET, params={"preset": preset, "zone": "ZONE1"}) async def set_control_bus_mode(self, control_bus: ControlBusMode) -> None: """Set the control bus mode.""" await self.request(ep.ZONE_STATE, params={"cbus": control_bus}) async def set_standby_mode(self, standby_mode: StandbyMode) -> None: """Set the standby mode.""" await self.request(ep.POWER, params={"standby_mode": standby_mode}) async def set_auto_power_down(self, auto_power_down_time_seconds: int) -> None: """Set the automatic power down time.""" await self.request( ep.POWER, params={"auto_power_down": auto_power_down_time_seconds} ) async def __aenter__(self) -> "StreamMagicClient": await self.connect() return self async def __aexit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: object | None, ) -> None: await self.disconnect()