Allow set ScreenCap interval as option for AndroidTV (#124470)

Co-authored-by: Joostlek <joostlek@outlook.com>
pull/129457/head
ollo69 2024-10-30 03:24:20 +08:00 committed by GitHub
parent 8cdd5de75c
commit 041282190a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 132 additions and 33 deletions

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
import logging
import os
from typing import Any
@ -40,6 +41,7 @@ from .const import (
CONF_ADB_SERVER_IP,
CONF_ADB_SERVER_PORT,
CONF_ADBKEY,
CONF_SCREENCAP_INTERVAL,
CONF_STATE_DETECTION_RULES,
DEFAULT_ADB_SERVER_PORT,
DEVICE_ANDROIDTV,
@ -66,6 +68,8 @@ RELOAD_OPTIONS = [CONF_STATE_DETECTION_RULES]
_INVALID_MACS = {"ff:ff:ff:ff:ff:ff"}
_LOGGER = logging.getLogger(__name__)
@dataclass
class AndroidTVRuntimeData:
@ -157,6 +161,32 @@ async def async_connect_androidtv(
return aftv, None
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Migrate old entry."""
_LOGGER.debug(
"Migrating configuration from version %s.%s", entry.version, entry.minor_version
)
if entry.version == 1:
new_options = {**entry.options}
# Migrate MinorVersion 1 -> MinorVersion 2: New option
if entry.minor_version < 2:
new_options = {**new_options, CONF_SCREENCAP_INTERVAL: 0}
hass.config_entries.async_update_entry(
entry, options=new_options, minor_version=2, version=1
)
_LOGGER.debug(
"Migration to configuration version %s.%s successful",
entry.version,
entry.minor_version,
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: AndroidTVConfigEntry) -> bool:
"""Set up Android Debug Bridge platform."""

View File

@ -34,7 +34,7 @@ from .const import (
CONF_APPS,
CONF_EXCLUDE_UNNAMED_APPS,
CONF_GET_SOURCES,
CONF_SCREENCAP,
CONF_SCREENCAP_INTERVAL,
CONF_STATE_DETECTION_RULES,
CONF_TURN_OFF_COMMAND,
CONF_TURN_ON_COMMAND,
@ -43,7 +43,7 @@ from .const import (
DEFAULT_EXCLUDE_UNNAMED_APPS,
DEFAULT_GET_SOURCES,
DEFAULT_PORT,
DEFAULT_SCREENCAP,
DEFAULT_SCREENCAP_INTERVAL,
DEVICE_CLASSES,
DOMAIN,
PROP_ETHMAC,
@ -76,6 +76,7 @@ class AndroidTVFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 1
MINOR_VERSION = 2
@callback
def _show_setup_form(
@ -253,10 +254,12 @@ class OptionsFlowHandler(OptionsFlowWithConfigEntry):
CONF_EXCLUDE_UNNAMED_APPS, DEFAULT_EXCLUDE_UNNAMED_APPS
),
): bool,
vol.Optional(
CONF_SCREENCAP,
default=options.get(CONF_SCREENCAP, DEFAULT_SCREENCAP),
): bool,
vol.Required(
CONF_SCREENCAP_INTERVAL,
default=options.get(
CONF_SCREENCAP_INTERVAL, DEFAULT_SCREENCAP_INTERVAL
),
): vol.All(vol.Coerce(int), vol.Clamp(min=0, max=15)),
vol.Optional(
CONF_TURN_OFF_COMMAND,
description={

View File

@ -9,6 +9,7 @@ CONF_APPS = "apps"
CONF_EXCLUDE_UNNAMED_APPS = "exclude_unnamed_apps"
CONF_GET_SOURCES = "get_sources"
CONF_SCREENCAP = "screencap"
CONF_SCREENCAP_INTERVAL = "screencap_interval"
CONF_STATE_DETECTION_RULES = "state_detection_rules"
CONF_TURN_OFF_COMMAND = "turn_off_command"
CONF_TURN_ON_COMMAND = "turn_on_command"
@ -18,7 +19,7 @@ DEFAULT_DEVICE_CLASS = "auto"
DEFAULT_EXCLUDE_UNNAMED_APPS = False
DEFAULT_GET_SOURCES = True
DEFAULT_PORT = 5555
DEFAULT_SCREENCAP = True
DEFAULT_SCREENCAP_INTERVAL = 5
DEVICE_ANDROIDTV = "androidtv"
DEVICE_FIRETV = "firetv"

View File

@ -2,10 +2,9 @@
from __future__ import annotations
from datetime import timedelta
from datetime import datetime, timedelta
import hashlib
import logging
from typing import Any
from androidtv.constants import APPS, KEYS
from androidtv.setup_async import AndroidTVAsync, FireTVAsync
@ -23,19 +22,19 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import Throttle
from homeassistant.util.dt import utcnow
from . import AndroidTVConfigEntry
from .const import (
CONF_APPS,
CONF_EXCLUDE_UNNAMED_APPS,
CONF_GET_SOURCES,
CONF_SCREENCAP,
CONF_SCREENCAP_INTERVAL,
CONF_TURN_OFF_COMMAND,
CONF_TURN_ON_COMMAND,
DEFAULT_EXCLUDE_UNNAMED_APPS,
DEFAULT_GET_SOURCES,
DEFAULT_SCREENCAP,
DEFAULT_SCREENCAP_INTERVAL,
DEVICE_ANDROIDTV,
SIGNAL_CONFIG_ENTITY,
)
@ -48,8 +47,6 @@ ATTR_DEVICE_PATH = "device_path"
ATTR_HDMI_INPUT = "hdmi_input"
ATTR_LOCAL_PATH = "local_path"
MIN_TIME_BETWEEN_SCREENCAPS = timedelta(seconds=60)
SERVICE_ADB_COMMAND = "adb_command"
SERVICE_DOWNLOAD = "download"
SERVICE_LEARN_SENDEVENT = "learn_sendevent"
@ -125,7 +122,8 @@ class ADBDevice(AndroidTVEntity, MediaPlayerEntity):
self._app_name_to_id: dict[str, str] = {}
self._get_sources = DEFAULT_GET_SOURCES
self._exclude_unnamed_apps = DEFAULT_EXCLUDE_UNNAMED_APPS
self._screencap = DEFAULT_SCREENCAP
self._screencap_delta: timedelta | None = None
self._last_screencap: datetime | None = None
self.turn_on_command: str | None = None
self.turn_off_command: str | None = None
@ -159,7 +157,13 @@ class ADBDevice(AndroidTVEntity, MediaPlayerEntity):
self._exclude_unnamed_apps = options.get(
CONF_EXCLUDE_UNNAMED_APPS, DEFAULT_EXCLUDE_UNNAMED_APPS
)
self._screencap = options.get(CONF_SCREENCAP, DEFAULT_SCREENCAP)
screencap_interval: int = options.get(
CONF_SCREENCAP_INTERVAL, DEFAULT_SCREENCAP_INTERVAL
)
if screencap_interval > 0:
self._screencap_delta = timedelta(minutes=screencap_interval)
else:
self._screencap_delta = None
self.turn_off_command = options.get(CONF_TURN_OFF_COMMAND)
self.turn_on_command = options.get(CONF_TURN_ON_COMMAND)
@ -183,7 +187,7 @@ class ADBDevice(AndroidTVEntity, MediaPlayerEntity):
async def _async_get_screencap(self, prev_app_id: str | None = None) -> None:
"""Take a screen capture from the device when enabled."""
if (
not self._screencap
not self._screencap_delta
or self.state in {MediaPlayerState.OFF, None}
or not self.available
):
@ -193,11 +197,18 @@ class ADBDevice(AndroidTVEntity, MediaPlayerEntity):
force: bool = prev_app_id is not None
if force:
force = prev_app_id != self._attr_app_id
await self._adb_get_screencap(no_throttle=force)
await self._adb_get_screencap(force)
@Throttle(MIN_TIME_BETWEEN_SCREENCAPS)
async def _adb_get_screencap(self, **kwargs: Any) -> None:
"""Take a screen capture from the device every 60 seconds."""
async def _adb_get_screencap(self, force: bool = False) -> None:
"""Take a screen capture from the device every configured minutes."""
time_elapsed = self._screencap_delta is not None and (
self._last_screencap is None
or (utcnow() - self._last_screencap) >= self._screencap_delta
)
if not (force or time_elapsed):
return
self._last_screencap = utcnow()
if media_data := await self._adb_screencap():
self._media_image = media_data, "image/png"
self._attr_media_image_hash = hashlib.sha256(media_data).hexdigest()[:16]

View File

@ -31,7 +31,7 @@
"apps": "Configure applications list",
"get_sources": "Retrieve the running apps as the list of sources",
"exclude_unnamed_apps": "Exclude apps with unknown name from the sources list",
"screencap": "Use screen capture for album art",
"screencap_interval": "Interval in minutes between screen capture for album art (set 0 to disable)",
"state_detection_rules": "Configure state detection rules",
"turn_off_command": "ADB shell turn off command (leave empty for default)",
"turn_on_command": "ADB shell turn on command (leave empty for default)"

View File

@ -100,7 +100,12 @@ CONFIG_FIRETV_DEFAULT = CONFIG_FIRETV_PYTHON_ADB
def setup_mock_entry(
config: dict[str, Any], entity_domain: str
config: dict[str, Any],
entity_domain: str,
*,
options=None,
version=1,
minor_version=2,
) -> tuple[str, str, MockConfigEntry]:
"""Prepare mock entry for entities tests."""
patch_key = config[ADB_PATCH_KEY]
@ -109,6 +114,9 @@ def setup_mock_entry(
domain=DOMAIN,
data=config[DOMAIN],
unique_id="a1:b1:c1:d1:e1:f1",
options=options,
version=version,
minor_version=minor_version,
)
return patch_key, entity_id, config_entry

View File

@ -22,7 +22,7 @@ from homeassistant.components.androidtv.const import (
CONF_APPS,
CONF_EXCLUDE_UNNAMED_APPS,
CONF_GET_SOURCES,
CONF_SCREENCAP,
CONF_SCREENCAP_INTERVAL,
CONF_STATE_DETECTION_RULES,
CONF_TURN_OFF_COMMAND,
CONF_TURN_ON_COMMAND,
@ -501,7 +501,7 @@ async def test_options_flow(hass: HomeAssistant) -> None:
user_input={
CONF_GET_SOURCES: True,
CONF_EXCLUDE_UNNAMED_APPS: True,
CONF_SCREENCAP: True,
CONF_SCREENCAP_INTERVAL: 1,
CONF_TURN_OFF_COMMAND: "off",
CONF_TURN_ON_COMMAND: "on",
},
@ -515,6 +515,6 @@ async def test_options_flow(hass: HomeAssistant) -> None:
assert config_entry.options[CONF_GET_SOURCES] is True
assert config_entry.options[CONF_EXCLUDE_UNNAMED_APPS] is True
assert config_entry.options[CONF_SCREENCAP] is True
assert config_entry.options[CONF_SCREENCAP_INTERVAL] == 1
assert config_entry.options[CONF_TURN_OFF_COMMAND] == "off"
assert config_entry.options[CONF_TURN_ON_COMMAND] == "on"

View File

@ -0,0 +1,34 @@
"""Tests for AndroidTV integration initialization."""
from homeassistant.components.androidtv.const import (
CONF_SCREENCAP,
CONF_SCREENCAP_INTERVAL,
)
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
from homeassistant.core import HomeAssistant
from . import patchers
from .common import CONFIG_ANDROID_DEFAULT, SHELL_RESPONSE_OFF, setup_mock_entry
async def test_migrate_version(
hass: HomeAssistant,
) -> None:
"""Test migration to new version."""
patch_key, _, mock_config_entry = setup_mock_entry(
CONFIG_ANDROID_DEFAULT,
MP_DOMAIN,
options={CONF_SCREENCAP: False},
minor_version=1,
)
mock_config_entry.add_to_hass(hass)
with (
patchers.patch_connect(True)[patch_key],
patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key],
):
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.options[CONF_SCREENCAP_INTERVAL] == 0
assert mock_config_entry.minor_version == 2

View File

@ -13,7 +13,7 @@ import pytest
from homeassistant.components.androidtv.const import (
CONF_APPS,
CONF_EXCLUDE_UNNAMED_APPS,
CONF_SCREENCAP,
CONF_SCREENCAP_INTERVAL,
CONF_STATE_DETECTION_RULES,
CONF_TURN_OFF_COMMAND,
CONF_TURN_ON_COMMAND,
@ -801,6 +801,9 @@ async def test_get_image_http(
"""
patch_key, entity_id, config_entry = _setup(CONFIG_ANDROID_DEFAULT)
config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
config_entry, options={CONF_SCREENCAP_INTERVAL: 2}
)
with (
patchers.patch_connect(True)[patch_key],
@ -828,21 +831,27 @@ async def test_get_image_http(
content = await resp.read()
assert content == b"image"
next_update = utcnow() + timedelta(seconds=30)
next_update = utcnow() + timedelta(minutes=1)
with (
patchers.patch_shell("11")[patch_key],
patchers.PATCH_SCREENCAP as patch_screen_cap,
patch("homeassistant.util.utcnow", return_value=next_update),
patch(
"homeassistant.components.androidtv.media_player.utcnow",
return_value=next_update,
),
):
async_fire_time_changed(hass, next_update, True)
await hass.async_block_till_done()
patch_screen_cap.assert_not_called()
next_update = utcnow() + timedelta(seconds=60)
next_update = utcnow() + timedelta(minutes=2)
with (
patchers.patch_shell("11")[patch_key],
patchers.PATCH_SCREENCAP as patch_screen_cap,
patch("homeassistant.util.utcnow", return_value=next_update),
patch(
"homeassistant.components.androidtv.media_player.utcnow",
return_value=next_update,
),
):
async_fire_time_changed(hass, next_update, True)
await hass.async_block_till_done()
@ -854,6 +863,9 @@ async def test_get_image_http_fail(hass: HomeAssistant) -> None:
patch_key, entity_id, config_entry = _setup(CONFIG_ANDROID_DEFAULT)
config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
config_entry, options={CONF_SCREENCAP_INTERVAL: 2}
)
with (
patchers.patch_connect(True)[patch_key],
@ -885,7 +897,7 @@ async def test_get_image_disabled(hass: HomeAssistant) -> None:
patch_key, entity_id, config_entry = _setup(CONFIG_ANDROID_DEFAULT)
config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
config_entry, options={CONF_SCREENCAP: False}
config_entry, options={CONF_SCREENCAP_INTERVAL: 0}
)
with (
@ -1133,7 +1145,7 @@ async def test_options_reload(hass: HomeAssistant) -> None:
with patchers.PATCH_SETUP_ENTRY as setup_entry_call:
# change an option that not require integration reload
hass.config_entries.async_update_entry(
config_entry, options={CONF_SCREENCAP: False}
config_entry, options={CONF_EXCLUDE_UNNAMED_APPS: True}
)
await hass.async_block_till_done()