Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add support for 2020 Vizio TV's #116

Merged
merged 9 commits into from
Nov 19, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 46 additions & 20 deletions pyvizio/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
from asyncio import sleep
from datetime import datetime, timedelta
import logging
from typing import Any, Dict, List, Optional, Union
from typing import Any, Dict, KeysView, List, Optional, Union
from urllib.parse import urlsplit

from aiohttp import ClientSession
Expand All @@ -25,6 +25,9 @@
InputItem,
)
from pyvizio.api.item import (
GetAltESNCommand,
GetAltSerialNumberCommand,
GetAltVersionCommand,
GetCurrentPowerStateCommand,
GetDeviceInfoCommand,
GetESNCommand,
Expand Down Expand Up @@ -76,7 +79,7 @@ def __init__(
device_id: str,
ip: str,
name: str,
auth_token: str = "",
auth_token: Optional[str] = "",
device_type: str = DEFAULT_DEVICE_CLASS,
session: Optional[ClientSession] = None,
timeout: int = DEFAULT_TIMEOUT,
Expand Down Expand Up @@ -193,8 +196,9 @@ async def __remote_multiple(
return await self.__remote(key_codes, log_api_exception=log_api_exception)

async def __get_cached_apps_list(self) -> List[str]:
if self._latest_apps and datetime.now() - self._latest_apps_last_updated < timedelta(
days=1
if (
self._latest_apps
and datetime.now() - self._latest_apps_last_updated < timedelta(days=1)
):
await sleep(0)
return self._latest_apps
Expand Down Expand Up @@ -287,9 +291,11 @@ async def get_esn(self, log_api_exception: bool = True) -> Optional[str]:
"""Asynchronously get device's ESN (electronic serial number?)."""
item = await self.__invoke_api_may_need_auth(
GetESNCommand(self.device_type), log_api_exception=log_api_exception
) or await self.__invoke_api_may_need_auth(
GetAltESNCommand(self.device_type), log_api_exception=log_api_exception
)

if item:
if item and item.value:
return item.value

return None
Expand All @@ -299,9 +305,12 @@ async def get_serial_number(self, log_api_exception: bool = True) -> Optional[st
item = await self.__invoke_api(
GetSerialNumberCommand(self.device_type),
log_api_exception=log_api_exception,
) or await self.__invoke_api(
GetAltSerialNumberCommand(self.device_type),
log_api_exception=log_api_exception,
)

if item:
if item and item.value:
return item.value

return None
Expand All @@ -310,9 +319,11 @@ async def get_version(self, log_api_exception: bool = True) -> Optional[str]:
"""Asynchronously get SmartCast software version on device."""
item = await self.__invoke_api(
GetVersionCommand(self.device_type), log_api_exception=log_api_exception
) or await self.__invoke_api(
GetAltVersionCommand(self.device_type), log_api_exception=log_api_exception
)

if item:
if item and item.value:
return item.value

return None
Expand Down Expand Up @@ -340,7 +351,11 @@ async def stop_pair(self, log_api_exception: bool = True) -> Optional[bool]:
)

async def pair(
self, ch_type: int, token: int, pin: str = "", log_api_exception: bool = True
self,
ch_type: Union[int, str],
token: Union[int, str],
pin: str = "",
log_api_exception: bool = True,
) -> Optional[PairChallengeResponse]:
"""Asynchronously complete pairing process to obtain auth token."""
if self.device_type == DEVICE_CLASS_SPEAKER:
Expand Down Expand Up @@ -437,17 +452,20 @@ async def vol_down(

async def get_current_volume(self, log_api_exception: bool = True) -> Optional[int]:
"""Asynchronously get device's current volume level."""
return await VizioAsync.get_audio_setting(
volume = await VizioAsync.get_audio_setting(
self, "volume", log_api_exception=log_api_exception
)
return int(volume) if volume else None

async def is_muted(self, log_api_exception: bool = True) -> Optional[bool]:
"""Asynchronously determine whether or not device is muted."""
# If None is returned lower() will fail, if not we can do a simple boolean check
try:
return (
await VizioAsync.get_audio_setting(
self, "mute", log_api_exception=log_api_exception
str(
await VizioAsync.get_audio_setting(
self, "mute", log_api_exception=log_api_exception
)
).lower()
== "on"
)
Expand Down Expand Up @@ -502,7 +520,7 @@ async def remote(self, key: str, log_api_exception: bool = True) -> Optional[boo
"""Asynchronously emulate key press by key name."""
return await self.__remote(key, log_api_exception=log_api_exception)

def get_remote_keys_list(self) -> List[str]:
def get_remote_keys_list(self) -> KeysView[str]:
"""Get list of remote key names."""
return KEY_CODE[self.device_type].keys()

Expand Down Expand Up @@ -582,7 +600,7 @@ async def get_setting(

async def get_setting_options(
self, setting_type: str, setting_name: str, log_api_exception: bool = True
) -> Optional[Union[int, str]]:
) -> Optional[Union[List[str], Dict[str, Union[int, str]]]]:
"""Asynchronously get options of named setting."""
return await self.__invoke_api_may_need_auth(
GetSettingOptionsCommand(self.device_type, setting_type, setting_name),
Expand All @@ -591,7 +609,7 @@ async def get_setting_options(

async def get_setting_options_xlist(
self, setting_type: str, setting_name: str, log_api_exception: bool = True
) -> Optional[Union[int, str]]:
) -> Optional[List[str]]:
"""Asynchronously get options of named setting for settings based on a user defined list."""
return await self.__invoke_api_may_need_auth(
GetSettingOptionsXListCommand(self.device_type, setting_type, setting_name),
Expand Down Expand Up @@ -652,7 +670,7 @@ async def get_audio_setting(

async def get_audio_setting_options(
self, setting_name: str, log_api_exception: bool = True
) -> Optional[Union[int, str]]:
) -> Optional[Union[List[str], Dict[str, Union[int, str]]]]:
"""Asynchronously get options of named audio setting."""
return await VizioAsync.get_setting_options(
self, "audio", setting_name, log_api_exception=log_api_exception
Expand Down Expand Up @@ -685,14 +703,14 @@ async def get_apps_list(
APP_HOME["name"],
*sorted(
[
app["name"]
str(app["name"])
for app in apps_list
if "*" in app["country"] or country.lower() in app["country"]
]
),
]

return [APP_HOME["name"], *sorted([app["name"] for app in apps_list])]
return [APP_HOME["name"], *sorted([str(app["name"]) for app in apps_list])]

async def launch_app(
self,
Expand Down Expand Up @@ -1021,7 +1039,7 @@ async def remote(self, key: str, log_api_exception: bool = True) -> Optional[boo
"""Emulate key press by key name."""
return await super(Vizio, self).remote(key, log_api_exception=log_api_exception)

def get_remote_keys_list(self) -> List[str]:
def get_remote_keys_list(self) -> KeysView[str]:
"""Get list of remote key names."""
return super(Vizio, self).get_remote_keys_list()

Expand Down Expand Up @@ -1064,12 +1082,20 @@ async def get_setting(
@async_to_sync
async def get_setting_options(
self, setting_type: str, setting_name: str, log_api_exception: bool = True
) -> Optional[Union[int, str]]:
) -> Optional[Union[List[str], Dict[str, Union[int, str]]]]:
"""Get options of named setting."""
return await super(Vizio, self).get_setting_options(
setting_type, setting_name, log_api_exception=log_api_exception
)

async def get_setting_options_xlist(
self, setting_type: str, setting_name: str, log_api_exception: bool = True
) -> Optional[List[str]]:
"""Get options of named setting for settings based on a user defined list."""
return await super(Vizio, self).get_setting_options_xlist(
setting_type, setting_name, log_api_exception=log_api_exception
)

@async_to_sync
async def set_setting(
self,
Expand Down Expand Up @@ -1113,7 +1139,7 @@ async def get_audio_setting(
@async_to_sync
async def get_audio_setting_options(
self, setting_name: str, log_api_exception: bool = True
) -> Optional[Union[int, str]]:
) -> Optional[Union[List[str], Dict[str, Union[int, str]]]]:
"""Get options of named audio setting."""
return await super(Vizio, self).get_audio_setting_options(
setting_name, log_api_exception=log_api_exception
Expand Down
10 changes: 8 additions & 2 deletions pyvizio/api/_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@
"ESN": "/menu_native/dynamic/tv_settings/system/system_information/uli_information/esn",
"SERIAL_NUMBER": "/menu_native/dynamic/tv_settings/system/system_information/tv_information/serial_number",
"VERSION": "/menu_native/dynamic/tv_settings/system/system_information/tv_information/version",
"_ALT_ESN": "/menu_native/dynamic/tv_settings/admin_and_privacy/system_information/uli_information/esn",
"_ALT_SERIAL_NUMBER": "/menu_native/dynamic/tv_settings/admin_and_privacy/system_information/tv_information/serial_number",
"_ALT_VERSION": "/menu_native/dynamic/tv_settings/admin_and_privacy/system_information/tv_information/version",
"DEVICE_INFO": "/state/device/deviceinfo",
"POWER_MODE": "/state/device/power_mode",
"KEY_PRESS": "/key_command/",
Expand All @@ -57,6 +60,9 @@
"ESN": "/menu_native/dynamic/audio_settings/system/system_information/uli_information/esn",
"SERIAL_NUMBER": "/menu_native/dynamic/audio_settings/system/system_information/speaker_information/serial_number",
"VERSION": "/menu_native/dynamic/audio_settings/system/system_information/speaker_information/version",
"_ALT_ESN": "/menu_native/dynamic/audio_settings/admin_and_privacy/system_information/uli_information/esn",
"_ALT_SERIAL_NUMBER": "/menu_native/dynamic/audio_settings/admin_and_privacy/system_information/speaker_information/serial_number",
"_ALT_VERSION": "/menu_native/dynamic/audio_settings/admin_and_privacy/system_information/speaker_information/version",
"DEVICE_INFO": "/state/device/deviceinfo",
"POWER_MODE": "/state/device/power_mode",
"KEY_PRESS": "/key_command/",
Expand Down Expand Up @@ -229,7 +235,7 @@ async def async_invoke_api(
"method": "put",
"url": url,
"headers": headers,
"data": json.loads(data),
"data": json.loads(str(data)),
},
)
response = await session.put(
Expand All @@ -256,7 +262,7 @@ async def async_invoke_api(
"method": "put",
"url": url,
"headers": headers,
"data": json.loads(data),
"data": json.loads(str(data)),
},
)
response = await local_session.put(
Expand Down
6 changes: 4 additions & 2 deletions pyvizio/api/apps.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Vizio SmartCast API commands for apps."""

from typing import Any, Dict, List, Union
from typing import Any, Dict, List, Optional, Union

from pyvizio.api._protocol import ENDPOINT, ResponseKey
from pyvizio.api.base import CommandBase
Expand Down Expand Up @@ -35,7 +35,9 @@ def __bool__(self) -> bool:
return self != AppConfig()


def find_app_name(config_to_check: AppConfig, app_list: List[Dict[str, Any]]) -> str:
def find_app_name(
config_to_check: Optional[AppConfig], app_list: List[Dict[str, Any]]
) -> str:
"""
Return the app name for a given AppConfig based on a list of apps.

Expand Down
4 changes: 2 additions & 2 deletions pyvizio/api/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ def get_method(self) -> str:
return self._method

@abstractmethod
def process_response(self, json_obj: Dict[str, Any]) -> None:
def process_response(self, json_obj: Dict[str, Any]) -> Any:
"""Always return True when there is no custom process_response method for subclass."""
return True

Expand All @@ -67,6 +67,6 @@ def url(self, new_url: str) -> None:
"""Set endpoint for command."""
CommandBase.url.fset(self, new_url)

def process_response(self, json_obj: Dict[str, Any]) -> None:
def process_response(self, json_obj: Dict[str, Any]) -> Any:
"""Always return None when there is no custom process_response method for subclass."""
return None
50 changes: 47 additions & 3 deletions pyvizio/api/item.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Vizio SmartCast API commands and class for individual item settings."""

from typing import Any, Dict, Union
from typing import Any, Dict, Optional, Union

from pyvizio.api._protocol import (
ACTION_MODIFY,
Expand All @@ -21,7 +21,7 @@ def __init__(self, device_type: str) -> None:
super(GetDeviceInfoCommand, self).__init__(ENDPOINT[device_type]["DEVICE_INFO"])
self.paths = PATH_MODEL[device_type]

def process_response(self, json_obj: Dict[str, Any]) -> bool:
def process_response(self, json_obj: Dict[str, Any]) -> Dict[str, Any]:
"""Return response to command to get device info."""
return dict_get_case_insensitive(json_obj, ResponseKey.ITEMS, [{}])[0]

Expand All @@ -33,7 +33,7 @@ def __init__(self, device_type: str) -> None:
"""Initialize command to get device model name."""
super(GetModelNameCommand, self).__init__(device_type)

def process_response(self, json_obj: Dict[str, Any]) -> bool:
def process_response(self, json_obj: Dict[str, Any]) -> Optional[str]:
"""Return response to command to get device model name."""
return get_value_from_path(
dict_get_case_insensitive(
Expand Down Expand Up @@ -184,3 +184,47 @@ class GetVersionCommand(ItemInfoCommandBase):
def __init__(self, device_type: str) -> None:
"""Initialize command to get SmartCast software version."""
super(GetVersionCommand, self).__init__(device_type, "VERSION")


class AltItemInfoCommandBase(ItemInfoCommandBase):
"""Command to get individual item setting."""

def __init__(
self,
device_type: str,
endpoint_name: str,
item_name: str,
default_return: Union[int, str] = None,
) -> None:
"""Initialize command to get individual item setting."""
super(ItemInfoCommandBase, self).__init__(ENDPOINT[device_type][endpoint_name])
self.item_name = item_name.upper()
self.default_return = default_return


class GetAltESNCommand(AltItemInfoCommandBase):
"""Command to get device ESN (electronic serial number?)."""

def __init__(self, device_type: str) -> None:
"""Initialize command to get device ESN (electronic serial number?)."""
super(GetAltESNCommand, self).__init__(device_type, "_ALT_ESN", "ESN")


class GetAltSerialNumberCommand(AltItemInfoCommandBase):
"""Command to get device serial number."""

def __init__(self, device_type: str) -> None:
"""Initialize command to get device serial number."""
super(GetAltSerialNumberCommand, self).__init__(
device_type, "_ALT_SERIAL_NUMBER", "SERIAL_NUMBER"
)


class GetAltVersionCommand(AltItemInfoCommandBase):
"""Command to get SmartCast software version."""

def __init__(self, device_type: str) -> None:
"""Initialize command to get SmartCast software version."""
super(GetAltVersionCommand, self).__init__(
device_type, "_ALT_VERSION", "VERSION"
)
6 changes: 3 additions & 3 deletions pyvizio/api/pair.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Vizio SmartCast API commands and class for pairing."""

from typing import Any, Dict
from typing import Any, Dict, Union

from pyvizio.api._protocol import ENDPOINT, PairingResponseKey, ResponseKey
from pyvizio.api.base import CommandBase
Expand Down Expand Up @@ -69,8 +69,8 @@ class PairChallengeCommand(PairCommandBase):
def __init__(
self,
device_id: str,
challenge_type: int,
pairing_token: int,
challenge_type: Union[int, str],
pairing_token: Union[int, str],
pin: str,
device_type: str,
) -> None:
Expand Down
2 changes: 1 addition & 1 deletion pyvizio/api/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ def __init__(self, device_type: str, setting_type: str, setting_name: str) -> No
device_type, self.setting_type
)

def process_response(self, json_obj: Dict[str, Any]) -> List[str]:
def process_response(self, json_obj: Dict[str, Any]) -> Optional[List[str]]:
"""Return response to command to get options of an audio setting by name (used for setting of type XList)."""
return (
super(GetSettingOptionsXListCommand, self)
Expand Down
Loading