<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-

from __future__ import print_function
from __future__ import unicode_literals


import json
import base64
import copy
import time
import re
from collections import namedtuple

import six
import requests
from . import xcui_element_types

if six.PY3:
    from urllib.parse import urljoin as _urljoin
    from functools import reduce
else:
    from urlparse import urljoin as _urljoin

DEBUG = False


def convert(dictionary):
    """
    Convert dict to namedtuple
    """
    return namedtuple('GenericDict', list(dictionary.keys()))(**dictionary)


def urljoin(*urls):
    """
    The default urlparse.urljoin behavior look strange
    Standard urlparse.urljoin('http://a.com/foo', '/bar')
    Expect: http://a.com/foo/bar
    Actually: http://a.com/bar

    This function fix that.
    """
    return reduce(_urljoin, [u.strip('/')+'/' for u in urls if u.strip('/')], '').rstrip('/')

def roundint(i):
    return int(round(i, 0))


def httpdo(url, method='GET', data=None):
    """
    Do HTTP Request
    """
    if isinstance(data, dict):
        data = json.dumps(data)
    if DEBUG:
        print("Shell: curl -X {method} -d '{data}' '{url}'".format(method=method, data=data or '', url=url))

    fn = dict(GET=requests.get, POST=requests.post, DELETE=requests.delete)[method]
    try:
        response = fn(url, data=data, timeout=20)
    except requests.exceptions.ConnectionError:
        # retry again
        print('retry to connect, error: {}'.format(e))
        time.sleep(1.0)
        response = fn(url, data=data, timeout=10)

    retjson = response.json()
    if DEBUG:
        print('Return: {}'.format(json.dumps(retjson, indent=4)))
    r = convert(retjson)
    if r.status != 0:
        raise WDAError(r.status, r.value)
    return r


class WDAError(Exception):
    def __init__(self, status, value):
        self.status = status
        self.value = value

    def __str__(self):
        return 'WDAError(status=%d, value=%s)' % (self.status, self.value)


class Rect(object):
    def __init__(self, x, y, width, height):
        self.x = x
        self.y = y
        self.width = width
        self.height = height

    def __str__(self):
        return 'Rect(x={x}, y={y}, width={w}, height={h})'.format(
            x=self.x, y=self.y, w=self.width, h=self.height)

    def __repr__(self):
        return str(self)

    @property
    def center(self):
        return namedtuple('Point', ['x', 'y'])(self.x+self.width/2, self.y+self.height/2)

    @property
    def origin(self):
        return namedtuple('Point', ['x', 'y'])(self.x, self.y)

    @property
    def left(self):
        return self.x

    @property
    def top(self):
        return self.y

    @property
    def right(self):
        return self.x+self.width

    @property
    def bottom(self):
        return self.y+self.height
    

class Client(object):
    def __init__(self, target='http://127.0.0.1:8100'):
        """
        Args:
            - target(string): base URL of your iPhone, ex http://10.0.0.1:8100
        """
        self._target = target

    def _request(self, base_url, method='GET', data=None):
        return httpdo(urljoin(self._target, base_url), method, data)

    def status(self):
        res = self._request('status')
        sid = res.sessionId
        res.value['sessionId'] = sid
        return res.value

    def home(self):
        """ Press home button """
        return self._request('homescreen', 'POST')

    def session(self, bundle_id=None):
        """
        Args:
            - bundle_id(str): the app bundle id

        WDA Return json like

        {
            "value": {
                "sessionId": "69E6FDBA-8D59-4349-B7DE-A9CA41A97814",
                "capabilities": {
                    "device": "iphone",
                    "browserName": "部落冲突",
                    "sdkVersion": "9.3.2",
                    "CFBundleIdentifier": "com.supercell.magic"
                }
            },
            "sessionId": "69E6FDBA-8D59-4349-B7DE-A9CA41A97814",
            "status": 0
        }
        """
        if bundle_id is None:
            sid = self.status()['sessionId']
            if not sid:
                raise RuntimeError("no session created ever")
            return Session(self._target, sid)
        else:
            data = json.dumps({'desiredCapabilities': {'bundleId': bundle_id}})
            res = self._request('session', 'POST', data=data)
            return Session(self._target, res.sessionId)

    def screenshot(self, png_filename=None):
        """
        Screenshot with PNG format

        Args:
            - png_filename(string): optional, save file name

        Returns:
            png raw data
        """
        value = self._request('screenshot').value
        raw_value = base64.b64decode(value)
        if png_filename:
            with open(png_filename, 'w') as f:
                f.write(raw_value)
        return raw_value

    def source(self):
        # TODO: not tested
        return self._request('source', 'GET').value


class Session(object):
    def __init__(self, target, session_id):
        """
        Args:
            - target(string): for example, http://127.0.0.1:8100
            - session_id(string): wda session id
        """
        self._target = target.rstrip('/')
        self._sid = session_id

    def __str__(self):
        return 'wda.Session (id=%s)' % self._sid

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def _request(self, base_url, method='POST', data=None):
        url = urljoin(self._target, 'session', self._sid, base_url)
        return httpdo(url, method, data)

    def tap(self, x, y):
        return self._request('/tap/0', data=json.dumps(dict(x=x, y=y)))

    def swipe(self, x1, y1, x2, y2, duration=0.2):
        """
        duration(float), in the unit of second(NSTimeInterval)

        [[FBRoute POST:@"/uiaTarget/:uuid/dragfromtoforduration"] respondWithTarget:self action:@selector(handleDrag:)],
        """
        data = dict(fromX=x1, fromY=y1, toX=x2, toY=y2, duration=duration)
        return self._request('/uiaTarget/0/dragfromtoforduration', data=json.dumps(data))

    def dump(self):
        """ Bad """
        return self._request('source', 'GET')

    @property
    def orientation(self):
        """
        Return string
        One of <PORTRAIT | LANDSCAPE>
        """
        return self._request('orientation', 'GET').value

    def window_size(self):
        """
        Return namedtuple

        For example:
            Size(width=320, height=568)
        """
        value = self._request('/window/0/size', 'GET').value
        w = roundint(value['width'])
        h = roundint(value['height'])
        return namedtuple('Size', ['width', 'height'])(w, h)

    def send_keys(self, value):
        """
        send keys, yet I know not, todo function
        """
        if isinstance(value, six.string_types):
            value = list(value)
        return self._request('/keys', data=json.dumps({'value': value}))

    @property
    def alert(self):
        return Alert(self)

    def close(self):
        return self._request('/', 'DELETE')

    def __call__(self, **kwargs):
        if kwargs.get('className'):
            kwargs['class_name'] = kwargs.get('class_name') or kwargs.pop('className')
        return Selector(urljoin(self._target, '/session', self._sid), **kwargs)


class Alert(object):
    def __init__(self, session):
        self._s = session
        self._request = session._request

    @property
    def text(self):
        return self._request('/alert/text', 'GET').value

    def accept(self):
        return self._request('/alert/accept', 'POST')

    def dismiss(self):
        return self._request('/alert/dismiss', 'POST')
    

class Selector(object):
    def __init__(self, base_url, name=None, text=None, class_name=None, value=None, label=None, xpath=None, index=0):
        '''
        Args:
            - name(str): attr for name
            - text(str): alias of name
            - class_name(str): attr of className
            - value(str): attr
            - label(str): attr for label
            - xpath(str): xpath string, a little slow, but works fine
            - index(int): useful when found multi elements
        '''
        self._base_url = base_url
        if text:
            name = text
        self._name = six.u(name) if name else None
        self._value = six.u(value) if value else None
        self._label = six.u(label) if label else None
        self._class_name = six.u(class_name) if class_name else None
        self._xpath = six.u(xpath) if xpath else None
        self._index = index
        if class_name and not class_name.startswith('XCUIElementType'):
            self._class_name = 'XCUIElementType' + class_name
        if xpath and not xpath.startswith('//XCUIElementType'):
            element = '|'.join(xcui_element_types.xcui_element)
            self._xpath = re.sub(r'/('+element+')', '/XCUIElementType\g<1>', xpath)

    def _request(self, data, suburl='elements', method='POST'):
        return httpdo(urljoin(self._base_url, suburl), method, data=data)

    @property
    def elements(self):
        """
        xpath: //XCUIElementTypeButton[@name='Share']
        Return like<commit_msg>fix unicode convert to unicode error<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-

from __future__ import print_function
from __future__ import unicode_literals


import json
import base64
import copy
import time
import re
from collections import namedtuple

import six
import requests
from . import xcui_element_types

if six.PY3:
    from urllib.parse import urljoin as _urljoin
    from functools import reduce
else:
    from urlparse import urljoin as _urljoin

DEBUG = False


def convert(dictionary):
    """
    Convert dict to namedtuple
    """
    return namedtuple('GenericDict', list(dictionary.keys()))(**dictionary)


def urljoin(*urls):
    """
    The default urlparse.urljoin behavior look strange
    Standard urlparse.urljoin('http://a.com/foo', '/bar')
    Expect: http://a.com/foo/bar
    Actually: http://a.com/bar

    This function fix that.
    """
    return reduce(_urljoin, [u.strip('/')+'/' for u in urls if u.strip('/')], '').rstrip('/')

def roundint(i):
    return int(round(i, 0))


def httpdo(url, method='GET', data=None):
    """
    Do HTTP Request
    """
    if isinstance(data, dict):
        data = json.dumps(data)
    if DEBUG:
        print("Shell: curl -X {method} -d '{data}' '{url}'".format(method=method, data=data or '', url=url))

    fn = dict(GET=requests.get, POST=requests.post, DELETE=requests.delete)[method]
    try:
        response = fn(url, data=data, timeout=20)
    except requests.exceptions.ConnectionError:
        # retry again
        print('retry to connect, error: {}'.format(e))
        time.sleep(1.0)
        response = fn(url, data=data, timeout=10)

    retjson = response.json()
    if DEBUG:
        print('Return: {}'.format(json.dumps(retjson, indent=4)))
    r = convert(retjson)
    if r.status != 0:
        raise WDAError(r.status, r.value)
    return r


class WDAError(Exception):
    def __init__(self, status, value):
        self.status = status
        self.value = value

    def __str__(self):
        return 'WDAError(status=%d, value=%s)' % (self.status, self.value)


class Rect(object):
    def __init__(self, x, y, width, height):
        self.x = x
        self.y = y
        self.width = width
        self.height = height

    def __str__(self):
        return 'Rect(x={x}, y={y}, width={w}, height={h})'.format(
            x=self.x, y=self.y, w=self.width, h=self.height)

    def __repr__(self):
        return str(self)

    @property
    def center(self):
        return namedtuple('Point', ['x', 'y'])(self.x+self.width/2, self.y+self.height/2)

    @property
    def origin(self):
        return namedtuple('Point', ['x', 'y'])(self.x, self.y)

    @property
    def left(self):
        return self.x

    @property
    def top(self):
        return self.y

    @property
    def right(self):
        return self.x+self.width

    @property
    def bottom(self):
        return self.y+self.height
    

class Client(object):
    def __init__(self, target='http://127.0.0.1:8100'):
        """
        Args:
            - target(string): base URL of your iPhone, ex http://10.0.0.1:8100
        """
        self._target = target

    def _request(self, base_url, method='GET', data=None):
        return httpdo(urljoin(self._target, base_url), method, data)

    def status(self):
        res = self._request('status')
        sid = res.sessionId
        res.value['sessionId'] = sid
        return res.value

    def home(self):
        """ Press home button """
        return self._request('homescreen', 'POST')

    def session(self, bundle_id=None):
        """
        Args:
            - bundle_id(str): the app bundle id

        WDA Return json like

        {
            "value": {
                "sessionId": "69E6FDBA-8D59-4349-B7DE-A9CA41A97814",
                "capabilities": {
                    "device": "iphone",
                    "browserName": "部落冲突",
                    "sdkVersion": "9.3.2",
                    "CFBundleIdentifier": "com.supercell.magic"
                }
            },
            "sessionId": "69E6FDBA-8D59-4349-B7DE-A9CA41A97814",
            "status": 0
        }
        """
        if bundle_id is None:
            sid = self.status()['sessionId']
            if not sid:
                raise RuntimeError("no session created ever")
            return Session(self._target, sid)
        else:
            data = json.dumps({'desiredCapabilities': {'bundleId': bundle_id}})
            res = self._request('session', 'POST', data=data)
            return Session(self._target, res.sessionId)

    def screenshot(self, png_filename=None):
        """
        Screenshot with PNG format

        Args:
            - png_filename(string): optional, save file name

        Returns:
            png raw data
        """
        value = self._request('screenshot').value
        raw_value = base64.b64decode(value)
        if png_filename:
            with open(png_filename, 'w') as f:
                f.write(raw_value)
        return raw_value

    def source(self):
        # TODO: not tested
        return self._request('source', 'GET').value


class Session(object):
    def __init__(self, target, session_id):
        """
        Args:
            - target(string): for example, http://127.0.0.1:8100
            - session_id(string): wda session id
        """
        self._target = target.rstrip('/')
        self._sid = session_id

    def __str__(self):
        return 'wda.Session (id=%s)' % self._sid

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def _request(self, base_url, method='POST', data=None):
        url = urljoin(self._target, 'session', self._sid, base_url)
        return httpdo(url, method, data)

    def tap(self, x, y):
        return self._request('/tap/0', data=json.dumps(dict(x=x, y=y)))

    def swipe(self, x1, y1, x2, y2, duration=0.2):
        """
        duration(float), in the unit of second(NSTimeInterval)

        [[FBRoute POST:@"/uiaTarget/:uuid/dragfromtoforduration"] respondWithTarget:self action:@selector(handleDrag:)],
        """
        data = dict(fromX=x1, fromY=y1, toX=x2, toY=y2, duration=duration)
        return self._request('/uiaTarget/0/dragfromtoforduration', data=json.dumps(data))

    def dump(self):
        """ Bad """
        return self._request('source', 'GET')

    @property
    def orientation(self):
        """
        Return string
        One of <PORTRAIT | LANDSCAPE>
        """
        return self._request('orientation', 'GET').value

    def window_size(self):
        """
        Return namedtuple

        For example:
            Size(width=320, height=568)
        """
        value = self._request('/window/0/size', 'GET').value
        w = roundint(value['width'])
        h = roundint(value['height'])
        return namedtuple('Size', ['width', 'height'])(w, h)

    def send_keys(self, value):
        """
        send keys, yet I know not, todo function
        """
        if isinstance(value, six.string_types):
            value = list(value)
        return self._request('/keys', data=json.dumps({'value': value}))

    @property
    def alert(self):
        return Alert(self)

    def close(self):
        return self._request('/', 'DELETE')

    def __call__(self, **kwargs):
        if kwargs.get('className'):
            kwargs['class_name'] = kwargs.get('class_name') or kwargs.pop('className')
        return Selector(urljoin(self._target, '/session', self._sid), **kwargs)


class Alert(object):
    def __init__(self, session):
        self._s = session
        self._request = session._request

    @property
    def text(self):
        return self._request('/alert/text', 'GET').value

    def accept(self):
        return self._request('/alert/accept', 'POST')

    def dismiss(self):
        return self._request('/alert/dismiss', 'POST')
    

class Selector(object):
    def __init__(self, base_url, name=None, text=None, class_name=None, value=None, label=None, xpath=None, index=0):
        '''
        Args:
            - name(str): attr for name
            - text(str): alias of name
            - class_name(str): attr of className
            - value(str): attr
            - label(str): attr for label
            - xpath(str): xpath string, a little slow, but works fine
            - index(int): useful when found multi elements
        '''
        self._base_url = base_url
        if text:
            name = text
        self._index = index
        self._name = six.text_type(name) if name else None
        self._value = six.text_type(value) if value else None
        self._label = six.text_type(label) if label else None
        self._xpath = six.text_type(xpath) if xpath else None
        self._class_name = six.text_type(class_name) if class_name else None
        if class_name and not class_name.startswith('XCUIElementType'):
            self._class_name = 'XCUIElementType' + class_name
        if xpath and not xpath.startswith('//XCUIElementType'):
            element = '|'.join(xcui_element_types.xcui_element)
            self._xpath = re.sub(r'/('+element+')', '/XCUIElementType\g<1>', xpath)

    def _request(self, data, suburl='elements', method='POST'):
        return httpdo(urljoin(self._base_url, suburl), method, data=data)

    @property
    def elements(self):
        """
        xpath: //XCUIElementTypeButton[@name='Share']
        Return like
{'fraction_non_alphanumeric': 0.13592233009708737, 'fraction_numerical': 0.011777813146586026, 'mean_word_length': 2.262374524056767, 'pattern_counts': {'":': 18, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 6, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '180251', 'n_tokens_mistral': 5938, 'n_tokens_neox': 5711, 'n_words': 1396}