How to use open_minitouch_stream method in ATX

Best Python code snippet using ATX

mixins.py

Source:mixins.py

...194    __minitouch_process = None195    def __install_minitouch(self):196        # install minicap & minitouch197        os.system('python -m atx minicap')198    def open_minitouch_stream(self, port=1111):199        if self.__touch_queue is None:200            self.__touch_queue = Queue.Queue()201        # ensure minicap installed202        out = self.raw_cmd('shell', 'ls', '"/data/local/tmp/minitouch"', stdout=subprocess.PIPE).communicate()[0]203        if 'No such file or directory' in out:204            self.__install_minitouch()205        if self.__minitouch_process is not None:206            self.__minitouch_process.kill()207        out = self.raw_cmd('shell', 'ps', '-C', '/data/local/tmp/minitouch', stdout=subprocess.PIPE).communicate()[0]208        out = out.strip().split('\n')209        if len(out) > 1:210            p = None211        else:212            p = self.raw_cmd('shell', '/data/local/tmp/minitouch')213            time.sleep(1)214            if p.poll() is not None:215                # print 'start minitouch failed.'216                return217        self.__minitouch_process = p218        self.raw_cmd('forward', 'tcp:%s' % port, 'localabstract:minitouch').wait()219        def send():220            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)221            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)222            try:223                s.connect(('127.0.0.1', port))224                while True:225                    cmd = self.__touch_queue.get()  # wait here226                    if not cmd:227                        continue228                    elif cmd[-1] != '\n':229                        cmd += '\n'230                    s.send(cmd)231            except:232                traceback.print_exc()233            finally:234                s.close()235                self.raw_cmd('forward', '--remove', 'tcp:%s' % port).wait()236        t = threading.Thread(target=send)237        t.setDaemon(True)238        t.start()239    def click(self, x, y):240        cmd = 'd 0 %d %d 30\nc\nu 0\nc\n' % (int(x), int(y))241        self.__touch_queue.put(cmd)242    def swipe(self, sx, sy, ex, ey, steps=20):243        x1, y1, x2, y2 = map(int, (sx, sy, ex, ey))244        dx = (x2 - x1) / steps245        dy = (y2 - y1) / steps246        send = self.touchqueue.put247        send('d 0 %d %d 30\nc\n' % (x1, y1))248        for i in range(steps - 1):249            x, y = x1 + (i + 1) * dx, y1 + (i + 1) * dy250            send('m 0 %d %d 30\nc\n' % (x, y))251        send('u 0 %d %d 30\nc\nu 0\nc\n' % (x2, y2))252    def pinchin(self, x1, y1, x2, y2, steps=10):253        pass254    def pinchout(self, x1, y1, x2, y2, steps=10):255        pass256class OpenSTFServiceMixin(object):257    pass258# -------------- examples ----------------#259class DummyDevice(object):260    def raw_cmd(self, *args, **kwargs):261        cmds = ['adb'] + list(args)262        # print cmds263        return subprocess.Popen(cmds, **kwargs)264# Mixins should come in front to override functions in Base265class TestDevice(MinitouchStreamMixin, MinicapStreamMixin, RotationWatcherMixin, DummyDevice):266    def __init__(self, *args, **kwargs):267        super(self.__class__, self).__init__(*args, **kwargs)268        self.open_rotation_watcher(on_rotation_change=lambda v: self.open_minicap_stream())269        self.open_minitouch_stream()270if __name__ == '__main__':271    import cv2272    dev = TestDevice()273    while True:274        img = dev.screenshot_cv2()275        if img is not None:276            cv2.imshow('screen', img)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.