Add new audio endpoint (noahhusby#165)

Co-authored-by: Noah Husby <32528627+noahhusby@users.noreply.github.com>

Author: Solmath

aiostreammagic/endpoints.py

@@ -10,6 +10,7 @@
 PLAY_CONTROL = "/zone/play_control"
 STREAM_RADIO = "/stream/radio"
 POWER = "/system/power"
+AUDIO = "/zone/audio"
 ZONE_AUDIO_OUTPUT = "/zone/audio/output"
 DISPLAY = "/system/display"
 PRESET_LIST = "/presets/list"

aiostreammagic/models.py

@@ -63,13 +63,32 @@ class ControlBusMode(StrEnum):
     RECEIVER = "receiver"
     OFF = "off"
 
+
 class StandbyMode(StrEnum):
     """Standby mode"""
 
     ECO = "ECO_MODE"
     NETWORK = "NETWORK"
 
 
+class EQFilterType(StrEnum):
+    """EQ filter type."""
+
+    LOWSHELF = "LOWSHELF"
+    PEAKING = "PEAKING"
+    HIGHSHELF = "HIGHSHELF"
+    LOWPASS = "LOWPASS"
+    HIGHPASS = "HIGHPASS"
+    NOTCH = "NOTCH"
+
+
+class Pipeline(StrEnum):
+    """Pipeline type."""
+
+    DSP = "DSP"
+    DIRECT = "DIRECT"
+
+
 @dataclass
 class Info(DataClassORJSONMixin):
     """Cambridge Audio device metadata."""
@@ -268,3 +287,59 @@ class Update(DataClassORJSONMixin):
         metadata=field_options(alias="update_available"), default=False
     )
     updating: bool = field(metadata=field_options(alias="updating"), default=False)
+
+
+@dataclass
+class EQBand(DataClassORJSONMixin):
+    """Represents a single EQ band."""
+
+    index: int = field(metadata=field_options(alias="index"))
+    filter: EQFilterType = field(metadata=field_options(alias="filter"))
+    freq: int = field(metadata=field_options(alias="freq"))
+    gain: float = field(metadata=field_options(alias="gain"))
+    q: float = field(metadata=field_options(alias="q"))
+
+
+@dataclass
+class UserEQ(DataClassORJSONMixin):
+    """Represents user EQ settings."""
+
+    enabled: bool = field(metadata=field_options(alias="enabled"))
+    bands: list[EQBand] = field(
+        metadata=field_options(alias="bands"), default_factory=list
+    )
+
+
+@dataclass
+class TiltEQ(DataClassORJSONMixin):
+    """Represents tilt EQ settings."""
+
+    enabled: bool = field(metadata=field_options(alias="enabled"))
+    intensity: int = field(metadata=field_options(alias="intensity"))
+
+
+@dataclass
+class Audio(DataClassORJSONMixin):
+    """Represents audio settings including EQ and balance."""
+
+    digital_filter: Optional[str] = field(
+        metadata=field_options(alias="digital_filter"), default=None
+    )
+    phase_invert: Optional[bool] = field(
+        metadata=field_options(alias="phase_invert"), default=None
+    )
+    volume_limit_percent: Optional[int] = field(
+        metadata=field_options(alias="volume_limit_percent"), default=None
+    )
+    tilt_eq: Optional[TiltEQ] = field(
+        metadata=field_options(alias="tilt_eq"), default=None
+    )
+    user_eq: Optional[UserEQ] = field(
+        metadata=field_options(alias="user_eq"), default=None
+    )
+    balance: Optional[int] = field(
+        metadata=field_options(alias="balance"), default=None
+    )
+    pipeline: Optional[Pipeline] = field(
+        metadata=field_options(alias="pipeline"), default=None
+    )

aiostreammagic/stream_magic.py

@@ -25,12 +25,11 @@
     PresetList,
     ControlBusMode,
     StandbyMode,
+    Audio,
 )
 from . import endpoints as ep
 from .const import _LOGGER
 
-VERSION = "1.0.0"
-
 
 class StreamMagicClient:
     """Client for handling connections with StreamMagic enabled devices."""
@@ -51,6 +50,7 @@ def __init__(self, host: str, session: ClientSession | None = None) -> None:
         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
@@ -167,6 +167,7 @@ async def _connect_handler(self, res: Future[bool]) -> None:
             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.
             (
@@ -175,6 +176,7 @@ async def _connect_handler(self, res: Future[bool]) -> None:
                 self._state,
                 self._play_state,
                 self._now_playing,
+                self._audio,
                 self._audio_output,
                 self._display,
                 self._update,
@@ -185,11 +187,13 @@ async def _connect_handler(self, res: Future[bool]) -> None:
                 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),
@@ -201,7 +205,9 @@ async def _connect_handler(self, res: Future[bool]) -> None:
                 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))
@@ -339,6 +345,13 @@ def now_playing(self) -> NowPlaying:
             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"""
@@ -393,6 +406,11 @@ async def get_now_playing(self) -> NowPlaying:
         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)
@@ -457,6 +475,13 @@ async def _async_handle_now_playing(self, payload: dict[str, Any]) -> None:
             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"]
@@ -653,9 +678,14 @@ async def set_auto_power_down(self, auto_power_down_time_seconds: int) -> None:
             ep.POWER, params={"auto_power_down": auto_power_down_time_seconds}
         )
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> "StreamMagicClient":
         await self.connect()
         return self
 
-    async def __aexit__(self, exc_type, exc, tb):
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: object | None,
+    ) -> None:
         await self.disconnect()