Supports initializing poco using ip:port, and connecting to android:/// by default when the device is not connected.

Author: z

poco/drivers/android/uiautomation.py

@@ -9,7 +9,6 @@
 import threading
 import atexit
 
-from airtest.core.api import connect_device, device as current_device
 from airtest.core.android.ime import YosemiteIme
 from airtest.core.error import AdbShellError, AirtestError
 
@@ -22,6 +21,7 @@
 from poco.utils.hrpc.hierarchy import RemotePocoHierarchy
 from poco.utils.airtest.input import AirtestInput
 from poco.utils import six
+from poco.utils.device import default_device
 from poco.drivers.android.utils.installation import install, uninstall
 
 __all__ = ['AndroidUiautomationPoco', 'AndroidUiautomationHelper']
@@ -150,9 +150,7 @@ def __init__(self, device=None, using_proxy=True, force_restart=False, use_airte
         if options.get('screenshot_each_action') is False:
             self.screenshot_each_action = False
 
-        self.device = device or current_device()
-        if not self.device:
-            self.device = connect_device("Android:///")
+        self.device = device or default_device()
 
         self.adb_client = self.device.adb
         if using_proxy:

poco/drivers/cocosjs/__init__.py

@@ -17,6 +17,7 @@
     from urlparse import urlparse
 
 from airtest.core.api import connect_device, device as current_device
+from poco.utils.device import default_device
 from airtest.core.helper import device_platform
 
 
@@ -26,25 +27,23 @@
 
 
 class CocosJsPocoAgent(PocoAgent):
-    def __init__(self, port, device=None):
-        self.device = device or current_device()
-        if not self.device:
-            self.device = connect_device("Android:///")
-
-        platform_name = device_platform(self.device)
-        if platform_name == 'Android':
-            local_port, _ = self.device.adb.setup_forward('tcp:{}'.format(port))
-            ip = self.device.adb.host or 'localhost'
-            port = local_port
-        elif platform_name == 'IOS':
-            # Note: ios is now support for now.
-            # ip = device.get_ip_address()
-            # use iproxy first
-            ip = 'localhost'
-            local_port, _ = self.device.instruct_helper.setup_proxy(port)
-            port = local_port
-        else:
-            ip = self.device.get_ip_address()
+    def __init__(self, port, device=None, ip=None):
+        if ip is None or ip == "localhost":
+            self.device = device or default_device()
+
+            platform_name = device_platform(self.device)
+            if platform_name == 'Android':
+                local_port, _ = self.device.adb.setup_forward('tcp:{}'.format(port))
+                ip = self.device.adb.host or 'localhost'
+                port = local_port
+            elif platform_name == 'IOS':
+                port, _ = self.device.setup_forward(port)
+                if self.device.is_local_device:
+                    ip = 'localhost'
+                else:
+                    ip = self.device.ip
+            else:
+                ip = self.device.get_ip_address()
 
         # transport
         self.conn = WebSocketClient('ws://{}:{}'.format(ip, port))
@@ -90,11 +89,12 @@ def __init__(self, addr=DEFAULT_ADDR, device=None, **options):
                 port = urlparse(addr).port
                 if not port:
                     raise ValueError
+                ip = urlparse(addr).hostname
         except ValueError:
             raise ValueError('Argument "addr" should be a tuple[2] or string format. e.g. '
                              '["localhost", 5003] or "ws://localhost:5003". Got {}'.format(repr(addr)))
 
-        agent = CocosJsPocoAgent(port, device)
+        agent = CocosJsPocoAgent(port, device, ip=ip)
         if 'action_interval' not in options:
             options['action_interval'] = 0.5
         super(CocosJsPoco, self).__init__(agent, **options)

poco/drivers/std/__init__.py

@@ -11,6 +11,7 @@
 from poco.utils.simplerpc.rpcclient import RpcClient
 from poco.utils.simplerpc.transport.tcp.main import TcpClient
 from poco.utils.simplerpc.utils import sync_wrapper
+from poco.utils.device import default_device
 
 from airtest.core.api import connect_device, device as current_device
 from airtest.core.helper import device_platform
@@ -70,38 +71,41 @@ class StdPoco(Poco):
             device = connect_device('Android:///')
             poco = StdPoco(10054, device)
 
+            # or use ip:port to initialize poco object
+            poco = StdPoco(port=10054, ip='xx.xx.xx.xx')
+
             # now you can play with poco
             ui = poco('...')
             ui.click()
             ...
 
     """
 
-    def __init__(self, port=DEFAULT_PORT, device=None, use_airtest_input=True, **kwargs):
-        self.device = device or current_device()
-        if not self.device:
-            self.device = connect_device("Android:///")
-
-        platform_name = device_platform(self.device)
-        if platform_name == 'Android':
-            # always forward for android device to avoid network unreachable
-            local_port, _ = self.device.adb.setup_forward('tcp:{}'.format(port))
-            ip = self.device.adb.host or 'localhost'
-            port = local_port
-        elif platform_name == 'IOS':
-            # ip = device.get_ip_address()
-            # use iproxy first
-            ip = 'localhost'
-            port, _ = self.device.instruct_helper.setup_proxy(port)
-        else:
-            try:
-                ip = self.device.get_ip_address()
-            except AttributeError:
-                try:
-                    ip = socket.gethostbyname(socket.gethostname())
-                except socket.gaierror:
-                    # 某些特殊情况下会出现这个error，无法正确获取本机ip地址
+    def __init__(self, port=DEFAULT_PORT, device=None, use_airtest_input=True, ip=None, **kwargs):
+        if ip is None or ip == "localhost":
+            self.device = device or default_device()
+
+            platform_name = device_platform(self.device)
+            if platform_name == 'Android':
+                # always forward for android device to avoid network unreachable
+                local_port, _ = self.device.adb.setup_forward('tcp:{}'.format(port))
+                ip = self.device.adb.host or 'localhost'
+                port = local_port
+            elif platform_name == 'IOS':
+                port, _ = self.device.setup_forward(port)
+                if self.device.is_local_device:
                     ip = 'localhost'
+                else:
+                    ip = self.device.ip
+            else:
+                try:
+                    ip = self.device.get_ip_address()
+                except AttributeError:
+                    try:
+                        ip = socket.gethostbyname(socket.gethostname())
+                    except socket.gaierror:
+                        # 某些特殊情况下会出现这个error，无法正确获取本机ip地址
+                        ip = 'localhost'
 
         agent = StdPocoAgent((ip, port), use_airtest_input)
         kwargs['reevaluate_volatile_attributes'] = True

poco/drivers/ue4/ue4_poco.py

@@ -23,4 +23,4 @@ def __init__(self, addr=DEFAULT_ADDR, ue4_editor=False, connect_default_device=T
         if dev is None and connect_default_device and not current_device():
             dev = connect_device("Android:///")
 
-        super(UE4Poco, self).__init__(addr[1], dev, **options)
+        super(UE4Poco, self).__init__(addr[1], dev, ip=addr[0], **options)

poco/drivers/unity3d/unity3d_poco.py

@@ -77,6 +77,6 @@ def __init__(self, addr=DEFAULT_ADDR, unity_editor=False, connect_default_device
             # can apply auto detection in the future
             dev = connect_device("Android:///")
 
-        super(UnityPoco, self).__init__(addr[1], dev, **options)
+        super(UnityPoco, self).__init__(addr[1], dev, ip=addr[0], **options)
         # If some devices fail to initialize, the UI tree cannot be obtained
         # self.vr = UnityVRSupport(self.agent.rpc)

poco/utils/device.py

@@ -2,6 +2,8 @@
 from __future__ import absolute_import
 
 from airtest.core.device import Device
+from airtest.core.api import connect_device, device as current_device
+from airtest.core.error import NoDeviceError
 
 
 class VirtualDevice(Device):
@@ -18,3 +20,15 @@ def get_current_resolution(self):
 
     def get_ip_address(self):
         return self.ip
+
+
+def default_device():
+    """
+    Get default device, if no device connected, connect to first android device.
+
+    :return:
+    """
+    try:
+        return current_device()
+    except NoDeviceError:
+        return connect_device('Android:///')