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

import collections
import json
import os
import platform
import re
import sys
import subprocess
import time
import threading
import warnings
import logging

import cv2
import numpy as np
import aircv as ac
from uiautomator import device as d
from uiautomator import Device as UiaDevice
from uiautomator import AutomatorDeviceObject
from PIL import Image

from atx import consts
from atx import errors
from atx import patch
from atx import logutils


log = logutils.getLogger(__name__)
log.setLevel(logging.DEBUG)
FindPoint = collections.namedtuple('FindPoint', ['pos', 'confidence', 'method'])

__dir__ = os.path.dirname(os.path.abspath(__file__))
__tmp__ = os.path.join(__dir__, '__cache__')

DISPLAY_RE = re.compile(
    '.*DisplayViewport{valid=true, .*orientation=(?P<orientation>\d+), .*deviceWidth=(?P<width>\d+), deviceHeight=(?P<height>\d+).*')


class ImageSelector(object):
    def __init__(self, image):
        self._name = None
        self._image = image
        if isinstance(image, basestring):
            self._name = image

    @property
    def image(self):
        return self._image
    

class Watcher(object):
    ACTION_TOUCH = 1 <<0
    ACTION_QUIT = 1 <<1
    Event = collections.namedtuple('WatchEvent', ['selector', 'actions'])

    def __init__(self, device, name=None, timeout=None):
        self._events = []
        self._dev = device
        self._run = False
        self._stored_selector = None

        self.name = name
        self.touched = {}
        self.timeout = timeout

    def on(self, image=None, actions=None, text=None):
        """Trigger when some object exists
        Args:
            image: string location of an image
            flags: ACTION_TOUCH | ACTION_QUIT

        Returns:
            None
        """
        if isinstance(image, basestring):
            self._stored_selector = ImageSelector(image)
        elif text:
            print 'stored selector'
            self._stored_selector = self._dev(text=text)

        if actions:
            self._events.append(Watcher.Event(image, actions))
            self._stored_selector = None
        return self

    def touch(self):
        self._events.append(Watcher.Event(self._stored_selector, Watcher.ACTION_TOUCH))
        return self

    def quit(self):
        self._events.append(Watcher.Event(self._stored_selector, Watcher.ACTION_QUIT))

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        self._run_watch()

    def _match(self, selector, screen):
        ''' returns position(x, y) or None'''
        if isinstance(selector, ImageSelector):
            ret = self._dev.exists(selector.image, screen=screen)
            if ret is None:
                return None

            # FIXME(ssx): Image match confidence should can set
            exists = False
            if ret.method == consts.IMAGE_MATCH_METHOD_TMPL:
                if ret.confidence > 0.8:
                    exists = True
                # else:
                    # print("Skip confidence:", ret.confidence)
            elif ret.method == consts.IMAGE_MATCH_METHOD_SIFT:
                matches, total = ret.confidence
                if 1.0*matches/total > 0.5:
                    exists = True

            if exists:
                return ret.pos
        elif isinstance(selector, AutomatorDeviceObject):
            if not selector.exists:
                return None
            info = selector.info['visibleBounds']
            x = (info['left'] + info['right']) / 2
            y = (info['bottom'] + info['top']) / 2
            return (x, y)

    def _hook(self, screen):
        for evt in self._events:
            pos = self._match(evt.selector, screen)
            if pos is None:
                continue

            if evt.actions & Watcher.ACTION_TOUCH:
                log.debug('trigger watch touch: %s', pos)
                self._dev.touch(*pos)

            if evt.actions & Watcher.ACTION_QUIT:
                self._run = False

    def _run_watch(self):
        self._run = True
        start_time = time.time()
        
        while self._run:
            screen = self._dev.screenshot()
            self._hook(screen)
            if self.timeout is not None:
                if time.time() - start_time > self.timeout:
                    raise errors.WatchTimeoutError("Watcher(%s) timeout %s" % (self.name, self.timeout,))
                # sys.stdout.write("\rWatching %4.1fs left: %4.1fs\r" %(self.timeout, self.timeout-time.time()+start_time))
                log.debug("Watching %4.1fs left: %4.1fs\r" %(self.timeout, self.timeout-time.time()+start_time))
                # sys.stdout.flush()
        # sys.stdout.write('\n')

def read_image(self, img):
    if isinstance(img, basestring):
        return ac.imread(img)
    # FIXME(ssx): need support other types
    return img


def pil_to_opencv(pil_image):
    # convert PIL to OpenCV
    pil_image = pil_image.convert('RGB')
    cv2_image = np.array(pil_image)
    # Convert RGB to BGR 
    cv2_image = cv2_image[:, :, ::-1].copy()
    return cv2_image


class CommonWrap(object):
    def __init__(self):
        self.image_match_method = consts.IMAGE_MATCH_METHOD_TMPL
        self.resolution = None
<commit_msg>fix bounds bug, and read_image bug<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# License under MIT

import collections
import json
import os
import platform
import re
import sys
import subprocess
import time
import threading
import warnings
import logging

import cv2
import numpy as np
import aircv as ac
from uiautomator import device as d
from uiautomator import Device as UiaDevice
from uiautomator import AutomatorDeviceObject
from PIL import Image

from atx import consts
from atx import errors
from atx import patch
from atx import logutils


log = logutils.getLogger(__name__)
log.setLevel(logging.DEBUG)
FindPoint = collections.namedtuple('FindPoint', ['pos', 'confidence', 'method'])

__dir__ = os.path.dirname(os.path.abspath(__file__))
__tmp__ = os.path.join(__dir__, '__cache__')

DISPLAY_RE = re.compile(
    '.*DisplayViewport{valid=true, .*orientation=(?P<orientation>\d+), .*deviceWidth=(?P<width>\d+), deviceHeight=(?P<height>\d+).*')


class ImageSelector(object):
    def __init__(self, image):
        self._name = None
        self._image = image
        if isinstance(image, basestring):
            self._name = image

    @property
    def image(self):
        return self._image
    

class Watcher(object):
    ACTION_TOUCH = 1 <<0
    ACTION_QUIT = 1 <<1
    Event = collections.namedtuple('WatchEvent', ['selector', 'actions'])

    def __init__(self, device, name=None, timeout=None):
        self._events = []
        self._dev = device
        self._run = False
        self._stored_selector = None

        self.name = name
        self.touched = {}
        self.timeout = timeout

    def on(self, image=None, actions=None, text=None):
        """Trigger when some object exists
        Args:
            image: string location of an image
            flags: ACTION_TOUCH | ACTION_QUIT

        Returns:
            None
        """
        if isinstance(image, basestring):
            self._stored_selector = ImageSelector(image)
        elif text:
            print 'stored selector'
            self._stored_selector = self._dev(text=text)

        if actions:
            self._events.append(Watcher.Event(image, actions))
            self._stored_selector = None
        return self

    def touch(self):
        self._events.append(Watcher.Event(self._stored_selector, Watcher.ACTION_TOUCH))
        return self

    def quit(self):
        self._events.append(Watcher.Event(self._stored_selector, Watcher.ACTION_QUIT))

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        self._run_watch()

    def _match(self, selector, screen):
        ''' returns position(x, y) or None'''
        if isinstance(selector, ImageSelector):
            ret = self._dev.exists(selector.image, screen=screen)
            if ret is None:
                return None

            # FIXME(ssx): Image match confidence should can set
            exists = False
            if ret.method == consts.IMAGE_MATCH_METHOD_TMPL:
                if ret.confidence > 0.8:
                    exists = True
                # else:
                    # print("Skip confidence:", ret.confidence)
            elif ret.method == consts.IMAGE_MATCH_METHOD_SIFT:
                matches, total = ret.confidence
                if 1.0*matches/total > 0.5:
                    exists = True

            if exists:
                return ret.pos
        elif isinstance(selector, AutomatorDeviceObject):
            if not selector.exists:
                return None
            info = selector.info['bounds']
            x = (info['left'] + info['right']) / 2
            y = (info['bottom'] + info['top']) / 2
            return (x, y)

    def _hook(self, screen):
        for evt in self._events:
            pos = self._match(evt.selector, screen)
            if pos is None:
                continue

            if evt.actions & Watcher.ACTION_TOUCH:
                log.debug('trigger watch touch: %s', pos)
                self._dev.touch(*pos)

            if evt.actions & Watcher.ACTION_QUIT:
                self._run = False

    def _run_watch(self):
        self._run = True
        start_time = time.time()
        
        while self._run:
            screen = self._dev.screenshot()
            self._hook(screen)
            if self.timeout is not None:
                if time.time() - start_time > self.timeout:
                    raise errors.WatchTimeoutError("Watcher(%s) timeout %s" % (self.name, self.timeout,))
                # sys.stdout.write("\rWatching %4.1fs left: %4.1fs\r" %(self.timeout, self.timeout-time.time()+start_time))
                log.debug("Watching %4.1fs left: %4.1fs\r" %(self.timeout, self.timeout-time.time()+start_time))
                # sys.stdout.flush()
        # sys.stdout.write('\n')

def read_image(img):
    if isinstance(img, basestring):
        return ac.imread(img)
    # FIXME(ssx): need support other types
    return img


def pil_to_opencv(pil_image):
    # convert PIL to OpenCV
    pil_image = pil_image.convert('RGB')
    cv2_image = np.array(pil_image)
    # Convert RGB to BGR 
    cv2_image = cv2_image[:, :, ::-1].copy()
    return cv2_image


class CommonWrap(object):
    def __init__(self):
        self.image_match_method = consts.IMAGE_MATCH_METHOD_TMPL
        self.resolution = None

{'fraction_non_alphanumeric': 0.10586471800637062, 'fraction_numerical': 0.00505902192242833, 'mean_word_length': 2.168596022558623, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5234524', 'n_tokens_mistral': 3182, 'n_tokens_neox': 3079, 'n_words': 790}