前言
来啦老铁!
前段时间写了一篇文章:快看,这有一款远程访问工具!
当时手搓了一个看起来很 cheap 的前端,实在上不了台面!
于是抽空对其做了一次升级,同时也在后端做了一点提升,先睹为快:
简单做一下说明:
- 前端改用 vue3+element-plus+axios;
- 前端页面做了重构,分成主页(图1)跟设备实时浏览页面(图4),这样就能一边执行代码,一边实时查看被测对象的表现了;
- 首页增加 Ping、Boot Up 入口,Ping 能对所有远程服务进行 ping 操作,获得远程服务的状态,Boot Up 入口能对掉线的远程服务进行重启操作(也可开放重启远程服务等操作);
- 首页右侧的电脑画面更新频率改为0.8秒一次(原 1秒一次),让画面更为实时,同时具备隐藏画面等功能入口;
- 左侧侧边栏,每个服务 item 中有一个小小的手机 icon,点击打开另一个“被测对象实时查看器”页面 tab,默认选择所有设备进行展示,让用户能一次性知道所有发生的情况,能对每个设备进行一些基本的操作;
- 考虑到存在多个用户打开同一服务的“被测对象实时查看器”,对后端服务带来过分的压力,从而容易导致后端服务挂掉,因此在后端采用:减小图片尺寸+降低图片质量+多线程+缓存策略(3 秒以内,同一台设备的截图都一样,3 秒后会刷新截图),目前就简单缓存在后端服务的内存中(多线程+缓存能解决 ADB I/O 阻塞问题,代价是没那么“实时”)。后端代码如:
cache_duration = 3
@app.get("/bridge/adb_screenshot")
async def do_adb_screenshot(device_id: str):
"""
do adb screenshot
:param device_id: android device's serial number
:return: screenshot of current android device
"""
try:
if device_id in device_screenshots:
cached_img_stream, screenshot_time = device_screenshots[device_id]
if (time.time() - screenshot_time) > cache_duration:
thread = threading.Thread(target=ScreenshotUtil.get_adb_screenshot, args=(device_id,))
thread.start()
return StreamingResponse(BytesIO(cached_img_stream), media_type='image/jpeg')
thread = threading.Thread(target=ScreenshotUtil.get_adb_screenshot, args=(device_id,))
thread.start()
return JSONResponse(status_code=200, content={"message": "Please wait..."})
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})
其中,使用 screenshots.py 文件进行图片数据缓存,screenshots.py 内的代码就一行:
device_screenshots = {}
ScreenshotUtil 内的 get_adb_screenshot 方法如下:
@staticmethod
def get_adb_screenshot(device_id):
try:
print("Take screenshot...")
result = subprocess.run(
["adb", "-s", device_id, "exec-out", "screencap", "-p"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=5
)
if result.returncode != 0:
return {"error": "Failed to capture screenshot by adb", "details": result.stderr.decode()}
img = Image.open(BytesIO(result.stdout))
new_width = int(img.width * 0.4)
new_height = int(img.height * 0.4)
img = img.resize((new_width, new_height))
# set image's quality to 40% of original one
output_stream = BytesIO()
img.convert("RGB").save(output_stream, format='JPEG', quality=40)
output_stream.seek(0)
device_screenshots[device_id] = (output_stream.getvalue(), time.time())
return {"message": "success", "device_id": device_id}
except subprocess.TimeoutExpired:
return {"error": "Timeout expired while trying to capture screenshot."}
except Exception as e:
return {"error": "An error occurred.", "details": str(e)}
- 前端代码仓库:由于内含 ip,较为敏感,暂时不公开,有需要可以私信我;
- 后端代码仓库:https://github.com/dylanz666/debug-bridge-server.git
展望:
- 解决后端容易挂掉的问题;
- 考虑加入截图生成视频的功能;
- 研究更为实时的“被测对象实时查看器”;
(原本想考虑使用 websocket 或者 SSE(Server-Sent Events)来更为实时同步设备画面,但是会卡在画面无法渲染方面,后续可以考虑再调研调研。)
好了,今天就记录到这里啦,下期再见喽~
如果本文对您有帮助,麻烦动动手指点点赞?
谢谢!