Add stream_music_properties() method

androidtv/basetv/basetv.py

@@ -554,13 +554,11 @@ def _screen_on_awake_wake_lock_size(output):
         if not output:
             return False, False, None
 
-        if output == "1":
-            return True, False, None
+        screen_on = output[0] == "1"
+        awake = None if len(output) < 2 else output[1] == "1"
+        wake_lock_size = None if len(output) < 3 else BaseTV._wake_lock_size(output[2:])
 
-        if output == "11":
-            return True, True, None
-
-        return True, True, BaseTV._wake_lock_size(output[2:])
+        return screen_on, awake, wake_lock_size
 
     def _volume(self, stream_music, audio_output_device):
         """Get the absolute volume level from the ``STREAM_MUSIC`` block from ``adb shell dumpsys audio``.
@@ -585,8 +583,6 @@ def _volume(self, stream_music, audio_output_device):
             max_volume_matches = re.findall(constants.MAX_VOLUME_REGEX_PATTERN, stream_music, re.DOTALL | re.MULTILINE)
             if max_volume_matches:
                 self.max_volume = float(max_volume_matches[0])
-            else:
-                self.max_volume = 15.0
 
         if not audio_output_device:
             return None

androidtv/basetv/basetv_async.py

@@ -282,6 +282,7 @@ async def current_app_media_session_state(self):
             The state from the output of the ADB shell command ``dumpsys media_session``, or ``None`` if it could not be determined
 
         """
+        # This needs to use different commands depending on the device
         media_session_state_response = await self._adb.shell(constants.CMD_MEDIA_SESSION_STATE_FULL)
 
         return self._current_app_media_session_state(media_session_state_response)
@@ -364,6 +365,29 @@ async def screen_on_awake_wake_lock_size(self):
 
         return self._screen_on_awake_wake_lock_size(output)
 
+    async def stream_music_properties(self):
+        """Get various properties from the "STREAM_MUSIC" block from ``dumpsys audio``..
+
+        Returns
+        -------
+        audio_output_device : str, None
+            The current audio playback device, or ``None`` if it could not be determined
+        is_volume_muted : bool, None
+            Whether or not the volume is muted, or ``None`` if it could not be determined
+        volume : int, None
+            The absolute volume level, or ``None`` if it could not be determined
+        volume_level : float, None
+            The volume level (between 0 and 1), or ``None`` if it could not be determined
+
+        """
+        stream_music = await self._get_stream_music()
+        audio_output_device = self._audio_output_device(stream_music)
+        volume = self._volume(stream_music, audio_output_device)
+        volume_level = self._volume_level(volume)
+        is_volume_muted = self._is_volume_muted(stream_music)
+
+        return audio_output_device, is_volume_muted, volume, volume_level
+
     async def volume(self):
         """Get the absolute volume level.
 

androidtv/basetv/basetv_sync.py

@@ -282,6 +282,7 @@ def current_app_media_session_state(self):
             The state from the output of the ADB shell command ``dumpsys media_session``, or ``None`` if it could not be determined
 
         """
+        # This needs to use different commands depending on the device
         media_session_state_response = self._adb.shell(constants.CMD_MEDIA_SESSION_STATE_FULL)
 
         return self._current_app_media_session_state(media_session_state_response)
@@ -364,6 +365,29 @@ def screen_on_awake_wake_lock_size(self):
 
         return self._screen_on_awake_wake_lock_size(output)
 
+    def stream_music_properties(self):
+        """Get various properties from the "STREAM_MUSIC" block from ``dumpsys audio``..
+
+        Returns
+        -------
+        audio_output_device : str, None
+            The current audio playback device, or ``None`` if it could not be determined
+        is_volume_muted : bool, None
+            Whether or not the volume is muted, or ``None`` if it could not be determined
+        volume : int, None
+            The absolute volume level, or ``None`` if it could not be determined
+        volume_level : float, None
+            The volume level (between 0 and 1), or ``None`` if it could not be determined
+
+        """
+        stream_music = self._get_stream_music()
+        audio_output_device = self._audio_output_device(stream_music)
+        volume = self._volume(stream_music, audio_output_device)
+        volume_level = self._volume_level(volume)
+        is_volume_muted = self._is_volume_muted(stream_music)
+
+        return audio_output_device, is_volume_muted, volume, volume_level
+
     def volume(self):
         """Get the absolute volume level.
 

androidtv/constants.py

@@ -95,7 +95,7 @@
 
 #: Determine if the device is on, the screen is on, and get the wake lock size
 CMD_SCREEN_ON_AWAKE_WAKE_LOCK_SIZE = (
-    CMD_SCREEN_ON + CMD_SUCCESS1 + " && " + CMD_AWAKE + CMD_SUCCESS1 + " && " + CMD_WAKE_LOCK_SIZE
+    CMD_SCREEN_ON + CMD_SUCCESS1_FAILURE0 + " && " + CMD_AWAKE + CMD_SUCCESS1_FAILURE0 + " && " + CMD_WAKE_LOCK_SIZE
 )
 
 #: Get the properties for an Android TV device (``lazy=True, get_running_apps=True``); see :py:meth:`androidtv.androidtv.androidtv_sync.AndroidTVSync.get_properties` and :py:meth:`androidtv.androidtv.androidtv_async.AndroidTVAsync.get_properties`

tests/test_androidtv_async.py

@@ -106,36 +106,35 @@ async def test_running_apps(self):
                 assert patched.called
 
     @awaiter
-    async def test_audio_output_device(self):
-        """Check that the ``device`` property works correctly."""
+    async def test_stream_music_properties(self):
+        """Check that the ``stream_music_properties`` method works correctly."""
         with async_patchers.patch_shell(None)[self.PATCH_KEY]:
-            with patch_calls(self.atv, self.atv._audio_output_device) as patched:
+            with patch_calls(self.atv, self.atv._audio_output_device) as audio_output_device, patch_calls(
+                self.atv, self.atv._is_volume_muted
+            ) as is_volume_muted, patch_calls(self.atv, self.atv._volume) as volume, patch_calls(
+                self.atv, self.atv._volume_level
+            ) as volume_level:
+                await self.atv.stream_music_properties()
+                assert audio_output_device.called
+                assert is_volume_muted.called
+                assert volume.called
+                assert volume_level.called
+
+            with patch_calls(self.atv, self.atv._audio_output_device) as audio_output_device:
                 await self.atv.audio_output_device()
-                assert patched.called
+                assert audio_output_device.called
 
-    @awaiter
-    async def test_volume(self):
-        """Check that the ``volume`` property works correctly."""
-        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
-            with patch_calls(self.atv, self.atv._volume) as patched:
+            with patch_calls(self.atv, self.atv._is_volume_muted) as is_volume_muted:
+                await self.atv.is_volume_muted()
+                assert is_volume_muted.called
+
+            with patch_calls(self.atv, self.atv._volume) as volume:
                 await self.atv.volume()
-                assert patched.called
+                assert volume.called
 
-    @awaiter
-    async def test_volume_level(self):
-        """Check that the ``volume_level`` property works correctly."""
-        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
-            with patch_calls(self.atv, self.atv._volume_level) as patched:
+            with patch_calls(self.atv, self.atv._volume_level) as volume_level:
                 await self.atv.volume_level()
-                assert patched.called
-
-    @awaiter
-    async def test_is_volume_muted(self):
-        """Check that the ``is_volume_muted`` property works correctly."""
-        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
-            with patch_calls(self.atv, self.atv._is_volume_muted) as patched:
-                await self.atv.is_volume_muted()
-                assert patched.called
+                assert volume_level.called
 
     @awaiter
     async def test_set_volume_level(self):

tests/test_androidtv_sync.py

@@ -473,122 +473,57 @@ def test_running_apps(self):
             running_apps = self.atv.running_apps()
             self.assertListEqual(running_apps, RUNNING_APPS_LIST)
 
-    def test_audio_output_device(self):
-        """Check that the ``device`` property works correctly."""
+    def test_stream_music_properties(self):
+        """Check that the ``stream_music_properties`` method works correctly."""
         with patchers.patch_shell(None)[self.PATCH_KEY]:
-            audio_output_device = self.atv.audio_output_device()
-            self.assertIsNone(audio_output_device)
+            self.assertTupleEqual(self.atv.stream_music_properties(), (None, None, None, None))
+            self.assertIsNone(self.atv.audio_output_device())
+            self.assertIsNone(self.atv.is_volume_muted())
+            self.assertIsNone(self.atv.volume())
+            self.assertIsNone(self.atv.volume_level())
 
         with patchers.patch_shell("")[self.PATCH_KEY]:
-            audio_output_device = self.atv.audio_output_device()
-            self.assertIsNone(audio_output_device)
+            self.assertTupleEqual(self.atv.stream_music_properties(), (None, None, None, None))
+            self.assertIsNone(self.atv.audio_output_device())
+            self.assertIsNone(self.atv.is_volume_muted())
+            self.assertIsNone(self.atv.volume())
+            self.assertIsNone(self.atv.volume_level())
 
         with patchers.patch_shell(" ")[self.PATCH_KEY]:
-            audio_output_device = self.atv.audio_output_device()
-            self.assertIsNone(audio_output_device)
+            self.assertTupleEqual(self.atv.stream_music_properties(), (None, None, None, None))
+            self.assertIsNone(self.atv.audio_output_device())
+            self.assertIsNone(self.atv.is_volume_muted())
+            self.assertIsNone(self.atv.volume())
+            self.assertIsNone(self.atv.volume_level())
 
         with patchers.patch_shell(STREAM_MUSIC_EMPTY)[self.PATCH_KEY]:
-            audio_output_device = self.atv.audio_output_device()
-            self.assertIsNone(audio_output_device)
-
-        with patchers.patch_shell(STREAM_MUSIC_OFF)[self.PATCH_KEY]:
-            audio_output_device = self.atv.audio_output_device()
-            self.assertEqual("speaker", audio_output_device)
-
-        with patchers.patch_shell(STREAM_MUSIC_ON)[self.PATCH_KEY]:
-            audio_output_device = self.atv.audio_output_device()
-            self.assertEqual("hmdi_arc", audio_output_device)
-
-    def test_volume(self):
-        """Check that the ``volume`` property works correctly."""
-        with patchers.patch_shell(None)[self.PATCH_KEY]:
-            volume = self.atv.volume()
-            self.assertIsNone(volume)
-
-        with patchers.patch_shell("")[self.PATCH_KEY]:
-            volume = self.atv.volume()
-            self.assertIsNone(volume)
-
-        with patchers.patch_shell(" ")[self.PATCH_KEY]:
-            volume = self.atv.volume()
-            self.assertIsNone(volume)
-
-        with patchers.patch_shell(STREAM_MUSIC_EMPTY)[self.PATCH_KEY]:
-            volume = self.atv.volume()
-            self.assertIsNone(volume)
-
-        with patchers.patch_shell(STREAM_MUSIC_NO_VOLUME)[self.PATCH_KEY]:
-            volume = self.atv.volume()
-            self.assertIsNone(volume)
-
-        self.atv.max_volume = None
-        with patchers.patch_shell(STREAM_MUSIC_OFF)[self.PATCH_KEY]:
-            volume = self.atv.volume()
-            self.assertEqual(volume, 20)
-            self.assertEqual(self.atv.max_volume, 60.0)
-
-        self.atv.max_volume = None
-        with patchers.patch_shell(STREAM_MUSIC_ON)[self.PATCH_KEY]:
-            volume = self.atv.volume()
-            self.assertEqual(volume, 22)
-            self.assertEqual(self.atv.max_volume, 60.0)
-
-    def test_volume_level(self):
-        """Check that the ``volume_level`` property works correctly."""
-        with patchers.patch_shell(None)[self.PATCH_KEY]:
-            volume_level = self.atv.volume_level()
-            self.assertIsNone(volume_level)
-
-        with patchers.patch_shell("")[self.PATCH_KEY]:
-            volume_level = self.atv.volume_level()
-            self.assertIsNone(volume_level)
-
-        with patchers.patch_shell(" ")[self.PATCH_KEY]:
-            volume_level = self.atv.volume_level()
-            self.assertIsNone(volume_level)
-
-        with patchers.patch_shell(STREAM_MUSIC_EMPTY)[self.PATCH_KEY]:
-            volume_level = self.atv.volume_level()
-            self.assertIsNone(volume_level)
+            self.assertTupleEqual(self.atv.stream_music_properties(), (None, None, None, None))
+            self.assertIsNone(self.atv.audio_output_device())
+            self.assertIsNone(self.atv.is_volume_muted())
+            self.assertIsNone(self.atv.volume())
+            self.assertIsNone(self.atv.volume_level())
 
         with patchers.patch_shell(STREAM_MUSIC_NO_VOLUME)[self.PATCH_KEY]:
-            volume_level = self.atv.volume_level()
-            self.assertIsNone(volume_level)
+            self.assertTupleEqual(self.atv.stream_music_properties(), ("speaker", False, None, None))
+            self.assertEqual("speaker", self.atv.audio_output_device())
+            self.assertFalse(self.atv.is_volume_muted())
+            self.assertIsNone(self.atv.volume())
+            self.assertIsNone(self.atv.volume_level())
 
-        self.atv.max_volume = None
         with patchers.patch_shell(STREAM_MUSIC_OFF)[self.PATCH_KEY]:
-            volume_level = self.atv.volume_level()
-            self.assertEqual(volume_level, 20.0 / 60)
+            self.assertTupleEqual(self.atv.stream_music_properties(), ("speaker", False, 20, 20 / 60.0))
+            self.assertEqual("speaker", self.atv.audio_output_device())
+            self.assertFalse(self.atv.is_volume_muted())
+            self.assertEqual(self.atv.volume(), 20)
             self.assertEqual(self.atv.max_volume, 60.0)
 
-        self.atv.max_volume = None
         with patchers.patch_shell(STREAM_MUSIC_ON)[self.PATCH_KEY]:
-            volume_level = self.atv.volume_level()
-            self.assertEqual(volume_level, 22.0 / 60)
+            self.assertTupleEqual(self.atv.stream_music_properties(), ("hmdi_arc", False, 22, 22 / 60.0))
+            self.assertEqual("hmdi_arc", self.atv.audio_output_device())
+            self.assertFalse(self.atv.is_volume_muted())
+            self.assertEqual(self.atv.volume(), 22)
             self.assertEqual(self.atv.max_volume, 60.0)
 
-    def test_is_volume_muted(self):
-        """Check that the ``is_volume_muted`` property works correctly."""
-        with patchers.patch_shell(None)[self.PATCH_KEY]:
-            is_volume_muted = self.atv.is_volume_muted()
-            self.assertIsNone(is_volume_muted)
-
-        with patchers.patch_shell("")[self.PATCH_KEY]:
-            is_volume_muted = self.atv.is_volume_muted()
-            self.assertIsNone(is_volume_muted)
-
-        with patchers.patch_shell(" ")[self.PATCH_KEY]:
-            is_volume_muted = self.atv.is_volume_muted()
-            self.assertIsNone(is_volume_muted)
-
-        with patchers.patch_shell(STREAM_MUSIC_EMPTY)[self.PATCH_KEY]:
-            is_volume_muted = self.atv.is_volume_muted()
-            self.assertIsNone(is_volume_muted)
-
-        with patchers.patch_shell(STREAM_MUSIC_OFF)[self.PATCH_KEY]:
-            is_volume_muted = self.atv.is_volume_muted()
-            self.assertFalse(is_volume_muted)
-
     def test_set_volume_level(self):
         """Check that the ``set_volume_level`` method works correctly."""
         with patchers.patch_shell(None)[self.PATCH_KEY]:

tests/test_basetv_async.py

@@ -414,7 +414,7 @@ async def test_screen_on_awake_wake_lock_size(self):
             self.assertTupleEqual(await self.btv.screen_on_awake_wake_lock_size(), (False, False, None))
 
         with async_patchers.patch_shell("1")[self.PATCH_KEY]:
-            self.assertTupleEqual(await self.btv.screen_on_awake_wake_lock_size(), (True, False, None))
+            self.assertTupleEqual(await self.btv.screen_on_awake_wake_lock_size(), (True, None, None))
 
         with async_patchers.patch_shell("11")[self.PATCH_KEY]:
             self.assertTupleEqual(await self.btv.screen_on_awake_wake_lock_size(), (True, True, None))

tests/test_basetv_sync.py

@@ -571,7 +571,7 @@ def test_screen_on_awake_wake_lock_size(self):
             self.assertTupleEqual(self.btv.screen_on_awake_wake_lock_size(), (False, False, None))
 
         with patchers.patch_shell("1")[self.PATCH_KEY]:
-            self.assertTupleEqual(self.btv.screen_on_awake_wake_lock_size(), (True, False, None))
+            self.assertTupleEqual(self.btv.screen_on_awake_wake_lock_size(), (True, None, None))
 
         with patchers.patch_shell("11")[self.PATCH_KEY]:
             self.assertTupleEqual(self.btv.screen_on_awake_wake_lock_size(), (True, True, None))