-
Notifications
You must be signed in to change notification settings - Fork 0
/
AndroidViaWeb.py
197 lines (171 loc) · 6.7 KB
/
AndroidViaWeb.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
# pylint: disable=invalid-name,missing-docstring,bad-continuation,too-many-return-statements
import sys
import argparse
#import StringIO
import io
import re
import os
import time
import urllib
#import urlparse
import traceback
#from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from http.server import BaseHTTPRequestHandler, HTTPServer
from PIL import Image
homeDir = os.path.dirname(os.path.realpath(__file__))
sys.path.append(homeDir+'/AndroidViewClient/src')
# pylint: disable=wrong-import-position
from com.dtmilano.android.viewclient import ViewClient
# pylint: enable=wrong-import-position
class MyHandler(BaseHTTPRequestHandler):
def __init__(self, *args, **kwargs):
BaseHTTPRequestHandler.__init__(self, *args, **kwargs)
self.main = None
#Handler for the GET requests
def defaultHeader(self, contentType):
self.send_response(200)
self.send_header('Content-type', contentType)
self.end_headers()
def do_GET(self):
attempt = 0
while attempt < 3:
attempt += 5
try:
self.do_GET2()
except Exception as e:
print('error:'+str(e))
print('error0:'+str(e.args[0]))
traceback.print_exc()
print("reconnect, wait...")
time.sleep(5)
self.main.connect()
print("reconnected...")
def do_GET2(self):
self.main = self.server.androidViaWeb
if self.main.indexRe.match(self.path):
self.defaultHeader('text/html')
# Send the html message
f = open(homeDir+"/index.html")
self.wfile.write(bytes(f.read(), "utf-8"))
f.close()
return
screenshotM = self.main.screenshotRe.match(self.path)
if screenshotM is not None:
q = urllib.parse.parse_qs(screenshotM.group(1))
imageFormat = q['imageFormat'][0]
framebuffer = None
if 'framebuffer' in q:
framebuffer = bool(int(q['framebuffer'][0]))
img = self.main.screenshotPng(q['r'][0], imageFormat, framebuffer)
self.defaultHeader('image/'+imageFormat)
self.wfile.write(img)
return
typeM = self.main.typeRe.match(self.path)
clickM = self.main.clickRe.match(self.path)
if clickM is None and typeM is None:
self.send_response(404)
self.wfile.write(bytes("404 You are lost", "utf-8"))
return
if clickM is not None:
if clickM.group(1) == 'touch':
self.defaultHeader('application/json')
self.main.touch(clickM.group(2).split(','))
self.wfile.write(bytes('{"ok":true}', 'utf-8'))
return
elif clickM.group(1) == 'swipe':
self.defaultHeader('application/json')
self.main.swipe(clickM.group(2).split(','))
self.wfile.write(bytes('{"ok":true}', 'utf-8'))
return
elif typeM is not None:
typeStr = urllib.parse.unquote(typeM.group(2))
if typeM.group(1) == 'type':
self.defaultHeader('application/json')
self.main.device.type(typeStr)
self.wfile.write(bytes('{"ok":true}', 'utf-8'))
return
elif typeM.group(1) == 'press':
self.defaultHeader('application/json')
self.main.device.press(typeStr)
self.wfile.write(bytes('{"ok":true}', 'utf-8'))
return
class AndroidViaWeb():
def __init__(self):
self.port = 8080
self.serial = None
self.serialno = None
self.device = None
self.typeRe = re.compile(r'^/(press|type)\?k=(\S+)$')
self.clickRe = re.compile(r'^/(touch|swipe)\?l=([0-9,]+)$')
self.screenshotRe = re.compile(r'^/screenshot.png\?(.+)$')
self.indexRe = re.compile(r'^/(\?.*$|$)')
def connect(self):
(self.device, self.serialno) = ViewClient.connectToDeviceOrExit(
serialno=self.serial, ignoreversioncheck=True
)
def start(self):
self.parseArgs()
self.connect()
self.serveWeb()
def touch(self, arr):
self.device.longTouch(int(arr[0]), int(arr[1]), int(arr[2]))
def swipe(self, arr):
self.device.drag(
(int(arr[0]), int(arr[1])),
(int(arr[2]), int(arr[3])),
int(arr[4])
)
def screenshotPng(self, rotate, imageFormat, framebuffer):
# *** Bad: on some devices,
# this is compressing in png then uncompressing to PIL
# then re-compressing again.
# Need to reconnect after takeSnapshot...
# https://github.com/dtmilano/AndroidViewClient/issues/46
# image = self.device.takeSnapshot(reconnect=True, force_adb_framebuffer=framebuffer)
image = self.device.takeSnapshot(reconnect=True)
rotate = int(rotate)
transpose = None
if rotate == 90:
transpose = Image.ROTATE_90
elif rotate == 180:
transpose = Image.ROTATE_180
elif rotate == 270:
transpose = Image.ROTATE_270
if transpose is not None:
image = image.transpose(transpose)
output = io.BytesIO()
if imageFormat == 'JPEG':
imageRGB = Image.new("RGB", image.size, (0, 0, 0))
imageRGB.paste(image, mask=image.split()[3])
imageRGB.save(output, imageFormat, quality=10)
else:
image.save(output, imageFormat)
contents = output.getvalue()
output.close()
return contents
def parseArgs(self):
parser = argparse.ArgumentParser(description='Android')
parser.add_argument('--port', dest='port',
action='store', help='Port number')
parser.add_argument('--serial', dest='serial',
action='store', help='Serial number')
args = parser.parse_args()
if args.port:
self.port = int(args.port)
if args.serial:
self.serial = args.serial
def serveWeb(self):
try:
#Create a web server and define the handler to manage the
#incoming request
server = HTTPServer(('', self.port), MyHandler)
server.androidViaWeb = self
print('Started httpserver on port ', self.port)
#Wait forever for incoming htto requests
server.serve_forever()
except KeyboardInterrupt as e:
print('^C received, shutting down the web server', str(e))
server.socket.close()
androidViaWeb = AndroidViaWeb()
androidViaWeb.start()
# vim: set smartindent expandtab ts=4 sw=4: