Skip to content

Commit

Permalink
Split get_properties() for Android TV (#282)
Browse files Browse the repository at this point in the history 
* Split `get_properties()` for Android TV

* Uncomment

* Fix test

* Fix

* Cleanup

* Copy & paste async tests

* Remove _get_properties() method

* Fix tests

* 100% coverage

* Cleanup

* Cleanup

* Cleanup

* Cleanup

* Cleanup

* Cleanup
  • Loading branch information
@JeffLIrion
JeffLIrion committed Jan 16, 2022
1 parent 043610a commit a242320
Show file tree
Hide file tree
Showing 6 changed files with 192 additions and 664 deletions.
38 changes: 27 additions & 11 deletions androidtv/androidtv/androidtv_async.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -158,19 +158,35 @@ async def get_properties(self, get_running_apps=True, lazy=False):
The HDMI input, or ``None`` if it could not be determined
"""
if lazy:
if get_running_apps:
output = await self._adb.shell(self._cmd_get_properties_lazy_running_apps)
else:
output = await self._adb.shell(self._cmd_get_properties_lazy_no_running_apps)
screen_on, awake, wake_lock_size = await self.screen_on_awake_wake_lock_size()

if lazy and not (screen_on and awake):
return screen_on, awake, None, wake_lock_size, None, None, None, None, None, None, None

audio_state = await self.audio_state()
current_app, media_session_state = await self.current_app_media_session_state()
audio_output_device, is_volume_muted, volume, _ = await self.stream_music_properties()

if get_running_apps:
running_apps = await self.running_apps()
else:
if get_running_apps:
output = await self._adb.shell(self._cmd_get_properties_not_lazy_running_apps)
else:
output = await self._adb.shell(self._cmd_get_properties_not_lazy_no_running_apps)
_LOGGER.debug("Android TV %s:%d `get_properties` response: %s", self.host, self.port, output)
running_apps = [current_app] if current_app else None

hdmi_input = await self.get_hdmi_input()

return self._get_properties(output, get_running_apps)
return (
screen_on,
awake,
audio_state,
wake_lock_size,
current_app,
media_session_state,
audio_output_device,
is_volume_muted,
volume,
running_apps,
hdmi_input,
)

async def get_properties_dict(self, get_running_apps=True, lazy=True):
"""Get the properties needed for Home Assistant updates and return them as a dictionary.
Expand Down
38 changes: 27 additions & 11 deletions androidtv/androidtv/androidtv_sync.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -160,19 +160,35 @@ def get_properties(self, get_running_apps=True, lazy=False):
The HDMI input, or ``None`` if it could not be determined
"""
if lazy:
if get_running_apps:
output = self._adb.shell(self._cmd_get_properties_lazy_running_apps)
else:
output = self._adb.shell(self._cmd_get_properties_lazy_no_running_apps)
screen_on, awake, wake_lock_size = self.screen_on_awake_wake_lock_size()

if lazy and not (screen_on and awake):
return screen_on, awake, None, wake_lock_size, None, None, None, None, None, None, None

audio_state = self.audio_state()
current_app, media_session_state = self.current_app_media_session_state()
audio_output_device, is_volume_muted, volume, _ = self.stream_music_properties()

if get_running_apps:
running_apps = self.running_apps()
else:
if get_running_apps:
output = self._adb.shell(self._cmd_get_properties_not_lazy_running_apps)
else:
output = self._adb.shell(self._cmd_get_properties_not_lazy_no_running_apps)
_LOGGER.debug("Android TV %s:%d `get_properties` response: %s", self.host, self.port, output)
running_apps = [current_app] if current_app else None

hdmi_input = self.get_hdmi_input()

return self._get_properties(output, get_running_apps)
return (
screen_on,
awake,
audio_state,
wake_lock_size,
current_app,
media_session_state,
audio_output_device,
is_volume_muted,
volume,
running_apps,
hdmi_input,
)

def get_properties_dict(self, get_running_apps=True, lazy=True):
"""Get the properties needed for Home Assistant updates and return them as a dictionary.
Expand Down
156 changes: 0 additions & 156 deletions androidtv/androidtv/base_androidtv.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -241,159 +241,3 @@ def _update(
state = constants.STATE_IDLE

return state, current_app, running_apps, audio_output_device, is_volume_muted, volume_level, hdmi_input

# ======================================================================= #
# #
# Properties #
# #
# ======================================================================= #
def _get_properties(self, output, get_running_apps):
"""Get the properties needed for Home Assistant updates.
Parameters
----------
output : str, None
The output of the ADB command used to retrieve the properties
get_running_apps : bool
Whether or not to get the ``running_apps`` property
Returns
-------
screen_on : bool, None
Whether or not the device is on, or ``None`` if it was not determined
awake : bool, None
Whether or not the device is awake (screensaver is not running), or ``None`` if it was not determined
audio_state : str, None
The audio state, as determined from "dumpsys audio", or ``None`` if it was not determined
wake_lock_size : int, None
The size of the current wake lock, or ``None`` if it was not determined
current_app : str, None
The current app property, or ``None`` if it was not determined
media_session_state : int, None
The state from the output of ``dumpsys media_session``, or ``None`` if it was not determined
audio_output_device : str, None
The current audio playback device, or ``None`` if it was not determined
is_volume_muted : bool, None
Whether or not the volume is muted, or ``None`` if it was not determined
volume : int, None
The absolute volume level, or ``None`` if it was not determined
running_apps : list, None
A list of the running apps, or ``None`` if it was not determined
hdmi_input : str, None
The HDMI input, or ``None`` if it could not be determined
"""
# ADB command was unsuccessful
if output is None:
return None, None, None, None, None, None, None, None, None, None, None

# `screen_on` property
if not output:
return False, False, None, -1, None, None, None, None, None, None, None
screen_on = output[0] == "1"

# `awake` property
if len(output) < 2:
return screen_on, False, None, -1, None, None, None, None, None, None, None
awake = output[1] == "1"

# `audio_state` property
if len(output) < 3:
return screen_on, awake, None, -1, None, None, None, None, None, None, None
audio_state = self._audio_state(output[2])

lines = output.strip().splitlines()

# `wake_lock_size` property
if len(lines[0]) < 4:
return screen_on, awake, audio_state, -1, None, None, None, None, None, None, None
wake_lock_size = self._wake_lock_size(lines[0])

# `current_app` property
if len(lines) < 2:
return screen_on, awake, audio_state, wake_lock_size, None, None, None, None, None, None, None
current_app = self._current_app(lines[1])

# `media_session_state` property
if len(lines) < 3:
return screen_on, awake, audio_state, wake_lock_size, current_app, None, None, None, None, None, None
media_session_state = self._media_session_state(lines[2], current_app)

# HDMI input property
if len(lines) < 4:
return (
screen_on,
awake,
audio_state,
wake_lock_size,
current_app,
media_session_state,
None,
None,
None,
None,
None,
)
hdmi_input = self._get_hdmi_input(lines[3])

# "STREAM_MUSIC" block
if len(lines) < 5:
return (
screen_on,
awake,
audio_state,
wake_lock_size,
current_app,
media_session_state,
None,
None,
None,
None,
hdmi_input,
)

# reconstruct the output of `constants.CMD_STREAM_MUSIC`
stream_music_raw = "\n".join(lines[4:])

# the "STREAM_MUSIC" block from `adb shell dumpsys audio`
stream_music = self._parse_stream_music(stream_music_raw)

# `audio_output_device` property
audio_output_device = self._audio_output_device(stream_music)

# `volume` property
volume = self._volume(stream_music, audio_output_device)

# `is_volume_muted` property
is_volume_muted = self._is_volume_muted(stream_music)

# `running_apps` property
if not get_running_apps or len(lines) < 17:
return (
screen_on,
awake,
audio_state,
wake_lock_size,
current_app,
media_session_state,
audio_output_device,
is_volume_muted,
volume,
None,
hdmi_input,
)
running_apps = self._running_apps(lines[16:])

return (
screen_on,
awake,
audio_state,
wake_lock_size,
current_app,
media_session_state,
audio_output_device,
is_volume_muted,
volume,
running_apps,
hdmi_input,
)
36 changes: 4 additions & 32 deletions androidtv/basetv/basetv.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -391,11 +391,11 @@ def _current_app_media_session_state(self, media_session_state_response):
current_app = self._current_app(lines[0].strip())

if len(lines) > 1:
media_session_state = self._media_session_state(lines[1], current_app)
else:
media_session_state = None
matches = constants.REGEX_MEDIA_SESSION_STATE.search(media_session_state_response)
if matches:
return current_app, int(matches.group("state"))

return current_app, media_session_state
return current_app, None

@staticmethod
def _get_hdmi_input(hdmi_response):
Expand Down Expand Up @@ -460,32 +460,6 @@ def _is_volume_muted(stream_music):

return None

@staticmethod
def _media_session_state(media_session_state_response, current_app):
"""Get the state from the output of :py:const:`androidtv.constants.CMD_MEDIA_SESSION_STATE`.
Parameters
----------
media_session_state_response : str, None
The output of :py:const:`androidtv.constants.CMD_MEDIA_SESSION_STATE`
current_app : str, None
The current app, or ``None`` if it could not be determined
Returns
-------
int, None
The state from the output of the ADB shell command, or ``None`` if it could not be determined
"""
if not media_session_state_response or not current_app:
return None

matches = constants.REGEX_MEDIA_SESSION_STATE.search(media_session_state_response)
if matches:
return int(matches.group("state"))

return None

@staticmethod
def _parse_stream_music(stream_music_raw):
"""Parse the output of the command :py:const:`androidtv.constants.CMD_STREAM_MUSIC`.
Expand Down Expand Up @@ -526,8 +500,6 @@ def _running_apps(running_apps_response):
"""
if running_apps_response:
if isinstance(running_apps_response, list):
return [line.strip().rsplit(" ", 1)[-1] for line in running_apps_response if line.strip()]
return [line.strip().rsplit(" ", 1)[-1] for line in running_apps_response.splitlines() if line.strip()]

return None
Expand Down
65 changes: 59 additions & 6 deletions tests/test_androidtv_async.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -222,13 +222,66 @@ async def test_volume_down(self):

@awaiter
async def test_get_properties(self):
"""Check that the ``get_properties`` method works correctly."""
"""Check that ``get_properties()`` works correctly."""
with async_patchers.patch_shell(None)[self.PATCH_KEY]:
for get_running_apps in [True, False]:
for lazy in [True, False]:
with patch_calls(self.atv, self.atv._get_properties) as patched:
await self.atv.get_properties_dict(get_running_apps, lazy)
assert patched.called
with patch_calls(
self.atv, self.atv.screen_on_awake_wake_lock_size
) as screen_on_awake_wake_lock_size, patch_calls(
self.atv, self.atv.current_app_media_session_state
) as current_app_media_session_state, patch_calls(
self.atv, self.atv.stream_music_properties
) as stream_music_properties, patch_calls(
self.atv, self.atv.running_apps
) as running_apps, patch_calls(
self.atv, self.atv.get_hdmi_input
) as get_hdmi_input:
await self.atv.get_properties(lazy=True)
assert screen_on_awake_wake_lock_size.called
assert not current_app_media_session_state.called
assert not running_apps.called
assert not get_hdmi_input.called

with patch_calls(
self.atv, self.atv.screen_on_awake_wake_lock_size
) as screen_on_awake_wake_lock_size, patch_calls(
self.atv, self.atv.current_app_media_session_state
) as current_app_media_session_state, patch_calls(
self.atv, self.atv.stream_music_properties
) as stream_music_properties, patch_calls(
self.atv, self.atv.running_apps
) as running_apps, patch_calls(
self.atv, self.atv.get_hdmi_input
) as get_hdmi_input:
await self.atv.get_properties(lazy=False, get_running_apps=True)
assert screen_on_awake_wake_lock_size.called
assert current_app_media_session_state.called
assert running_apps.called
assert get_hdmi_input.called

with patch_calls(
self.atv, self.atv.screen_on_awake_wake_lock_size
) as screen_on_awake_wake_lock_size, patch_calls(
self.atv, self.atv.current_app_media_session_state
) as current_app_media_session_state, patch_calls(
self.atv, self.atv.stream_music_properties
) as stream_music_properties, patch_calls(
self.atv, self.atv.running_apps
) as running_apps, patch_calls(
self.atv, self.atv.get_hdmi_input
) as get_hdmi_input:
await self.atv.get_properties(lazy=False, get_running_apps=False)
assert screen_on_awake_wake_lock_size.called
assert current_app_media_session_state.called
assert not running_apps.called
assert get_hdmi_input.called

@awaiter
async def test_get_properties_dict(self):
"""Check that ``get_properties_dict()`` works correctly."""
with async_patchers.patch_shell(None)[self.PATCH_KEY]:
with patch_calls(self.atv, self.atv.get_properties) as get_properties:
await self.atv.get_properties_dict()
assert get_properties.called

@awaiter
async def test_update(self):
Expand Down
Loading

0 comments on commit a242320

Please sign in to comment.