diff --git a/androidtv/setup_async.py b/androidtv/setup_async.py new file mode 100644 index 00000000..87527037 --- /dev/null +++ b/androidtv/setup_async.py @@ -0,0 +1,74 @@ +"""Connect to a device and determine whether it's an Android TV or an Amazon Fire TV. + +ADB Debugging must be enabled. +""" + +from .androidtv.androidtv_async import AndroidTVAsync +from .basetv.basetv_async import BaseTVAsync +from .constants import DEFAULT_AUTH_TIMEOUT_S +from .firetv.firetv_async import FireTVAsync + + +__version__ = '0.0.4' + + +async def setup(host, port=5555, adbkey='', adb_server_ip='', adb_server_port=5037, state_detection_rules=None, device_class='auto', auth_timeout_s=DEFAULT_AUTH_TIMEOUT_S): + """Connect to a device and determine whether it's an Android TV or an Amazon Fire TV. + + Parameters + ---------- + host : str + The address of the device; may be an IP address or a host name + port : int + The device port to which we are connecting (default is 5555) + adbkey : str + The path to the ``adbkey`` file for ADB authentication + adb_server_ip : str + The IP address of the ADB server + adb_server_port : int + The port for the ADB server + state_detection_rules : dict, None + A dictionary of rules for determining the state (see :class:`~aio_androidtv.basetv.BaseTV`) + device_class : str + The type of device: ``'auto'`` (detect whether it is an Android TV or Fire TV device), ``'androidtv'``, or ``'firetv'``` + auth_timeout_s : float + Authentication timeout (in seconds) + + Returns + ------- + aftv : AndroidTVAsync, FireTVAsync + The representation of the device + + """ + if device_class == 'androidtv': + atv = AndroidTVAsync(host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules) + await atv.adb_connect(auth_timeout_s=auth_timeout_s) + atv.device_properties = await atv.get_device_properties() + return atv + + if device_class == 'firetv': + ftv = FireTVAsync(host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules) + await ftv.adb_connect(auth_timeout_s=auth_timeout_s) + ftv.device_properties = await ftv.get_device_properties() + return ftv + + if device_class != 'auto': + raise ValueError("`device_class` must be 'androidtv', 'firetv', or 'auto'.") + + aftv = BaseTVAsync(host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules) + + # establish the ADB connection + await aftv.adb_connect(auth_timeout_s=auth_timeout_s) + + # get device properties + aftv.device_properties = await aftv.get_device_properties() + + # Fire TV + if aftv.device_properties.get('manufacturer') == 'Amazon': + aftv.__class__ = FireTVAsync + + # Android TV + else: + aftv.__class__ = AndroidTVAsync + + return aftv diff --git a/tests/test_setup_async.py b/tests/test_setup_async.py new file mode 100644 index 00000000..89c4c8e6 --- /dev/null +++ b/tests/test_setup_async.py @@ -0,0 +1,67 @@ +import sys +import unittest +from unittest.mock import patch + + +sys.path.insert(0, '..') + +from androidtv.setup_async import setup +from androidtv.androidtv.androidtv_async import AndroidTVAsync +from androidtv.firetv.firetv_async import FireTVAsync + +from . import async_patchers +from .async_wrapper import awaiter + + +DEVICE_PROPERTIES_OUTPUT1 = "Amazon\n\n\n\n\n." + +DEVICE_PROPERTIES_DICT1 = {'manufacturer': 'Amazon', + 'model': '', + 'serialno': None, + 'sw_version': '', + 'wifimac': None, + 'ethmac': None} + +DEVICE_PROPERTIES_OUTPUT2 = "Not Amazon\n\n\n\n\n." + +DEVICE_PROPERTIES_DICT2 = {'manufacturer': 'Not Amazon', + 'model': '', + 'serialno': None, + 'sw_version': '', + 'wifimac': None, + 'ethmac': None} + + +class TestSetup(unittest.TestCase): + PATCH_KEY = 'python' + + @awaiter + async def test_setup(self): + """Test that the ``setup`` function works correctly. + """ + with self.assertRaises(ValueError): + await setup('HOST', 5555, device_class='INVALID') + + with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT1)[self.PATCH_KEY]: + ftv = await setup('HOST', 5555) + self.assertIsInstance(ftv, FireTVAsync) + self.assertDictEqual(ftv.device_properties, DEVICE_PROPERTIES_DICT1) + + with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT2)[self.PATCH_KEY]: + atv = await setup('HOST', 5555) + self.assertIsInstance(atv, AndroidTVAsync) + self.assertDictEqual(atv.device_properties, DEVICE_PROPERTIES_DICT2) + + with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT1)[self.PATCH_KEY]: + ftv = await setup('HOST', 5555, device_class='androidtv') + self.assertIsInstance(ftv, AndroidTVAsync) + self.assertDictEqual(ftv.device_properties, DEVICE_PROPERTIES_DICT1) + + with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT2)[self.PATCH_KEY]: + atv = await setup('HOST', 5555, device_class='firetv') + self.assertIsInstance(atv, FireTVAsync) + self.assertDictEqual(atv.device_properties, DEVICE_PROPERTIES_DICT2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_setup.py b/tests/test_setup_sync.py similarity index 100% rename from tests/test_setup.py rename to tests/test_setup_sync.py