feat: Add sound content control for v2 devices (#21)

.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+    "cSpell.words": [
+        "mqtt"
+    ]
+}

requirements.txt

@@ -2,6 +2,6 @@ setuptools>=42
 wheel
 colorlog==6.7.0
 pip>=8.0.3,<23.1
-ruff==0.0.257
+ruff>=0.0.291
 aiohttp>=3.8.1
 awsiotsdk>=1.16.0

src/hatch_rest_api/riot.py

@@ -1,5 +1,7 @@
 import logging
 
+from .types import SoundContent
+
 from .util import (
     convert_to_percentage,
     safely_get_json_value,
@@ -161,7 +163,12 @@ def set_toddler_lock(self, on: bool):
     def set_clock(self, brightness: int = 0):
         _LOGGER.debug(f"Setting clock on: {brightness}")
         self._update(
-            {"clock": {"flags": self.flags | RIOT_FLAGS_CLOCK_ON, "i": convert_from_percentage(brightness)}}
+            {
+                "clock": {
+                    "flags": self.flags | RIOT_FLAGS_CLOCK_ON,
+                    "i": convert_from_percentage(brightness),
+                }
+            }
         )
 
     def turn_clock_off(self):
@@ -174,6 +181,43 @@ def set_favorite(self, favorite_name_id: str):
         fav_id = int(favorite_name_id.split("-")[1])
         self._update({"current": {"srId": fav_id, "step": 1, "playing": "routine"}})
 
+    def set_sound(self, sound: SoundContent, duration: int = 0, until="indefinite"):
+        """
+        Pass a SoundContent item from self.sounds
+        """
+        _LOGGER.debug(f"Setting sound: {sound['title']}")
+        self._update(
+            {
+                "current": {
+                    "playing": "remote",
+                    "sound": {
+                        # not clear if this is the right ID, but it also doesn't appear to matter?
+                        "id": sound["contentId"],
+                        "mute": False,
+                        "url": sound.get("wavUrl") or sound.get("mp3Url"),
+                        "duration": duration,
+                        "until": until,
+                    },
+                }
+            }
+        )
+
+    def set_sound_url(self, sound_url: str):
+        """
+        appears to work with some but not all public wav and mp3 urls
+
+        i.e. http://codeskulptor-demos.commondatastorage.googleapis.com/GalaxyInvaders/theme_01.mp3
+        """
+        _LOGGER.debug(f"Setting sound URL: {sound_url}")
+        self._update(
+            {
+                "current": {
+                    "playing": "remote",
+                    "sound": {"mute": False, "url": sound_url},
+                }
+            }
+        )
+
     def turn_off(self):
         _LOGGER.debug("Turning off sound")
         self._update({"current": {"srId": 0, "step": 0, "playing": "none"}})

src/hatch_rest_api/shadow_client_subscriber.py

@@ -9,6 +9,8 @@
     ShadowState,
 )
 
+from .types import SoundContent
+
 from .callbacks import CallbacksMixin
 
 _LOGGER = logging.getLogger(__name__)
@@ -24,12 +26,14 @@ def __init__(
         mac: str,
         shadow_client: IotShadowClient,
         favorites: list = [],
+        sounds: "list[SoundContent|dict]" = [],
     ):
         self.device_name = device_name
         self.thing_name = thing_name
         self.mac = mac
         self.shadow_client = shadow_client
         self.favorites = favorites
+        self.sounds = sounds
         _LOGGER.debug(f"creating {self.__class__.__name__}: {device_name}")
 
         def update_shadow_accepted(response: UpdateShadowResponse):

src/hatch_rest_api/stub.py

@@ -35,8 +35,18 @@ def output():
 
             iot_device.register_callback(output)
 
+            if iot_device.set_sound:
+                # set volume safely low
+                print("******-Adjusting volume to safe level")
+                iot_device.set_volume(25)
+                # play each available sound
+                for sound in iot_device.sounds:
+                    print(f"******-PLAYING SOUND {sound['title']}")
+                    iot_device.set_sound(sound)
+                    await asyncio.sleep(5)
+                iot_device.turn_off()
+
         await asyncio.sleep(60)
-        mqtt_connection.disconnect().result()
     finally:
         if mqtt_connection:
             mqtt_connection.disconnect().result()

src/hatch_rest_api/types.py

@@ -0,0 +1,55 @@
+from typing import TypedDict
+from typing import Any
+
+
+class SoundContent(TypedDict):
+    """
+    No guarantees here, some display fields may be nullable.
+    These types were inferred based off the free/included content
+    from a Rest Gen2 (riot) device
+    """
+
+    id: int
+    createDate: str
+    updateDate: str
+    title: str
+    description: str
+    hidden: bool
+    newItem: bool
+    displayOrder: int
+    contentType: str
+    category: str
+    tier: str
+    extent: str
+    alarmOnly: bool
+    author: "Any | None"
+    narrator: "Any | None"
+    duration: "Any | None"
+    imageUrl: str
+    mp3Url: "str | None"
+    wavUrl: "str | None"
+    color: Any
+    series: "list[Any]"
+    products: "list[str]"
+    libraryVersion: int
+    libraryVersionString: str
+    url: str
+    contentSeries: bool
+    hatchId: int
+    contentSource: str
+    seriesSize: int
+    tagIds: "list"
+    altTagIds: "list"
+    mixIds: "list"
+    red: "Any | None"
+    green: "Any | None"
+    blue: "Any | None"
+    white: "Any | None"
+    sixCharacterColor: str
+    rotateSeries: bool
+    contentfulId: "Any | None"
+    rotationType: str
+    legacy: bool
+    contentId: int
+    contentful: bool
+    personalized: bool

src/hatch_rest_api/util.py

@@ -12,14 +12,14 @@ def clean_dictionary_for_logging(dictionary: dict[str, any]) -> dict[str, any]:
     for key in dictionary:
         if key.lower() in SENSITIVE_FIELD_NAMES:
             mutable_dictionary[key] = "***"
-        if type(mutable_dictionary[key]) is dict:
+        if isinstance(mutable_dictionary[key], dict):
             mutable_dictionary[key] = clean_dictionary_for_logging(
                 mutable_dictionary[key].copy()
             )
-        if type(mutable_dictionary[key]) is list:
+        if isinstance(mutable_dictionary[key], list):
             new_array = []
             for item in mutable_dictionary[key]:
-                if type(item) is dict:
+                if isinstance(item, dict):
                     new_array.append(clean_dictionary_for_logging(item.copy()))
                 else:
                     new_array.append(item)

src/hatch_rest_api/util_bootstrap.py

@@ -33,8 +33,7 @@ async def get_rest_devices(
     aws_token = await api.token(auth_token=token)
     favorites_map = await _get_favorites_for_all_v2_devices(api, token, iot_devices)
     routines_map = await _get_routines_for_all_v2_devices(api, token, iot_devices)
-    # This call will fetch sounds for a v2 device but the official app doesn't appear to use these sounds
-    # sounds = await _get_sound_content_for_v2_devices(api, token, iot_devices)
+    sounds = await _get_sound_content_for_v2_devices(api, token, iot_devices)
     aws_http: AwsHttp = AwsHttp(api.api_session)
     aws_credentials = await aws_http.aws_credentials(
         region=aws_token["region"],
@@ -85,6 +84,7 @@ def create_rest_devices(iot_device):
                 mac=iot_device["macAddress"],
                 shadow_client=shadow_client,
                 favorites=favorites_map[iot_device["macAddress"]],
+                sounds=sounds,
             )
         elif iot_device["product"] == "restoreIot":
             return RestoreIot(
@@ -121,10 +121,11 @@ async def _get_favorites_for_all_v2_devices(api, token, iot_devices):
             mac_to_fav[mac] = favorites
     return mac_to_fav
 
+
 async def _get_routines_for_all_v2_devices(api, token, iot_devices):
     mac_to_fav = {}
     for device in iot_devices:
-        if device["product"] in ["restoreIot"]:
+        if device["product"] in ["riot", "restoreIot"]:
             mac = device["macAddress"]
             routines = await api.routines(auth_token=token, mac=mac)
             _LOGGER.debug(f"Routines for {mac}: {routines}")