scrcpy代码改进
This commit is contained in:
parent
a478ba14f2
commit
0bfc88a3b0
2 changed files with 10 additions and 184 deletions
|
|
@ -1,5 +1,4 @@
|
|||
import functools
|
||||
import socket
|
||||
import struct
|
||||
from time import sleep
|
||||
from typing import TYPE_CHECKING
|
||||
|
|
@ -48,17 +47,6 @@ class ControlSender:
|
|||
"""
|
||||
return struct.pack("!BIII", action, keycode, repeat, 0)
|
||||
|
||||
@inject(const.TYPE_INJECT_TEXT)
|
||||
def text(self, text: str) -> bytes:
|
||||
"""
|
||||
Send text to device
|
||||
Args:
|
||||
text: text to send
|
||||
"""
|
||||
|
||||
buffer = text.encode("utf-8")
|
||||
return struct.pack(">i", len(buffer)) + buffer
|
||||
|
||||
@inject(const.TYPE_INJECT_TOUCH_EVENT)
|
||||
def touch(
|
||||
self, x: int, y: int, action: int = const.ACTION_DOWN, touch_id: int = -1
|
||||
|
|
@ -95,170 +83,6 @@ class ControlSender:
|
|||
buttons,
|
||||
)
|
||||
|
||||
@inject(const.TYPE_INJECT_SCROLL_EVENT)
|
||||
def scroll(self, x: int, y: int, h: int, v: int) -> bytes:
|
||||
"""
|
||||
Scroll screen
|
||||
Args:
|
||||
x: horizontal position
|
||||
y: vertical position
|
||||
h: horizontal movement
|
||||
v: vertical movement
|
||||
"""
|
||||
|
||||
x, y = max(x, 0), max(y, 0)
|
||||
return struct.pack(
|
||||
">iiHHii",
|
||||
int(x),
|
||||
int(y),
|
||||
int(self.parent.resolution[0]),
|
||||
int(self.parent.resolution[1]),
|
||||
int(h),
|
||||
int(v),
|
||||
)
|
||||
|
||||
@inject(const.TYPE_BACK_OR_SCREEN_ON)
|
||||
def back_or_turn_screen_on(self, action: int = const.ACTION_DOWN) -> bytes:
|
||||
"""
|
||||
If the screen is off, it is turned on only on ACTION_DOWN
|
||||
Args:
|
||||
action: ACTION_DOWN | ACTION_UP
|
||||
"""
|
||||
return struct.pack(">B", action)
|
||||
|
||||
@inject(const.TYPE_EXPAND_NOTIFICATION_PANEL)
|
||||
def expand_notification_panel(self) -> bytes:
|
||||
"""
|
||||
Expand notification panel
|
||||
"""
|
||||
return b""
|
||||
|
||||
@inject(const.TYPE_EXPAND_SETTINGS_PANEL)
|
||||
def expand_settings_panel(self) -> bytes:
|
||||
"""
|
||||
Expand settings panel
|
||||
"""
|
||||
return b""
|
||||
|
||||
@inject(const.TYPE_COLLAPSE_PANELS)
|
||||
def collapse_panels(self) -> bytes:
|
||||
"""
|
||||
Collapse all panels
|
||||
"""
|
||||
return b""
|
||||
|
||||
def get_clipboard(self, copy_key=const.COPY_KEY_NONE) -> str:
|
||||
"""
|
||||
Get clipboard
|
||||
"""
|
||||
# Since this function need socket response, we can't auto inject it any more
|
||||
s: socket.socket = self.parent.control_socket
|
||||
|
||||
with self.parent.control_socket_lock:
|
||||
# Flush socket
|
||||
s.setblocking(False)
|
||||
while True:
|
||||
try:
|
||||
s.recv(1024)
|
||||
except BlockingIOError:
|
||||
break
|
||||
s.setblocking(True)
|
||||
|
||||
# Read package
|
||||
package = struct.pack(">BB", const.TYPE_GET_CLIPBOARD, copy_key)
|
||||
s.send(package)
|
||||
(code,) = struct.unpack(">B", s.recv(1))
|
||||
assert code == 0
|
||||
(length,) = struct.unpack(">i", s.recv(4))
|
||||
|
||||
return s.recv(length).decode("utf-8")
|
||||
|
||||
@inject(const.TYPE_SET_CLIPBOARD)
|
||||
def set_clipboard(self, text: str, paste: bool = False) -> bytes:
|
||||
"""
|
||||
Set clipboard
|
||||
Args:
|
||||
text: the string you want to set
|
||||
paste: paste now
|
||||
"""
|
||||
buffer = text.encode("utf-8")
|
||||
return struct.pack(">?i", paste, len(buffer)) + buffer
|
||||
|
||||
@inject(const.TYPE_SET_SCREEN_POWER_MODE)
|
||||
def set_screen_power_mode(self, mode: int = const.POWER_MODE_NORMAL) -> bytes:
|
||||
"""
|
||||
Set screen power mode
|
||||
Args:
|
||||
mode: POWER_MODE_OFF | POWER_MODE_NORMAL
|
||||
"""
|
||||
return struct.pack(">b", mode)
|
||||
|
||||
@inject(const.TYPE_ROTATE_DEVICE)
|
||||
def rotate_device(self) -> bytes:
|
||||
"""
|
||||
Rotate device
|
||||
"""
|
||||
return b""
|
||||
|
||||
def swipe(
|
||||
self,
|
||||
start_x: int,
|
||||
start_y: int,
|
||||
end_x: int,
|
||||
end_y: int,
|
||||
move_step_length: int = 5,
|
||||
move_steps_delay: float = 0.005,
|
||||
) -> None:
|
||||
"""
|
||||
Swipe on screen
|
||||
Args:
|
||||
start_x: start horizontal position
|
||||
start_y: start vertical position
|
||||
end_x: start horizontal position
|
||||
end_y: end vertical position
|
||||
move_step_length: length per step
|
||||
move_steps_delay: sleep seconds after each step
|
||||
:return:
|
||||
"""
|
||||
|
||||
self.touch(start_x, start_y, const.ACTION_DOWN)
|
||||
next_x = start_x
|
||||
next_y = start_y
|
||||
|
||||
if end_x > self.parent.resolution[0]:
|
||||
end_x = self.parent.resolution[0]
|
||||
|
||||
if end_y > self.parent.resolution[1]:
|
||||
end_y = self.parent.resolution[1]
|
||||
|
||||
decrease_x = True if start_x > end_x else False
|
||||
decrease_y = True if start_y > end_y else False
|
||||
while True:
|
||||
if decrease_x:
|
||||
next_x -= move_step_length
|
||||
if next_x < end_x:
|
||||
next_x = end_x
|
||||
else:
|
||||
next_x += move_step_length
|
||||
if next_x > end_x:
|
||||
next_x = end_x
|
||||
|
||||
if decrease_y:
|
||||
next_y -= move_step_length
|
||||
if next_y < end_y:
|
||||
next_y = end_y
|
||||
else:
|
||||
next_y += move_step_length
|
||||
if next_y > end_y:
|
||||
next_y = end_y
|
||||
|
||||
self.touch(next_x, next_y, const.ACTION_MOVE)
|
||||
|
||||
if next_x == end_x and next_y == end_y:
|
||||
self.touch(next_x, next_y, const.ACTION_UP)
|
||||
break
|
||||
sleep(move_steps_delay)
|
||||
|
||||
def tap(self, x, y, hold_time: float = 0.07) -> None:
|
||||
"""
|
||||
Tap on screen
|
||||
|
|
|
|||
|
|
@ -8,19 +8,20 @@ from typing import Any, Callable, Optional
|
|||
|
||||
import numpy as np
|
||||
|
||||
from mower import __rootdir__
|
||||
from mower.utils import config
|
||||
from mower.utils import typealias as tp
|
||||
from mower.utils.device.adb_client.const import KeyCode
|
||||
from mower.utils.device.adb_client.core import Client as ADBClient
|
||||
from mower.utils.device.adb_client.socket import Socket
|
||||
from mower.utils.log import logger
|
||||
from mower.utils.path import get_path
|
||||
from mower.utils.vector import sm, va
|
||||
|
||||
from . import const
|
||||
from .control import ControlSender
|
||||
|
||||
SCR_PATH = "/data/local/tmp/scrcpy-server.jar"
|
||||
SCR_VER = "2.7"
|
||||
|
||||
|
||||
swipe_update_result = None
|
||||
|
|
@ -93,7 +94,7 @@ class Client:
|
|||
"""
|
||||
Start server and get the connection
|
||||
"""
|
||||
cmdline = f"CLASSPATH={SCR_PATH} app_process / com.genymobile.scrcpy.Server 2.7 video=false audio=false control=true tunnel_forward=true"
|
||||
cmdline = f"CLASSPATH={SCR_PATH} app_process / com.genymobile.scrcpy.Server {SCR_VER} video=false audio=false control=true tunnel_forward=true"
|
||||
if self.displayid is not None:
|
||||
cmdline += f" display_id={self.displayid}"
|
||||
self.__server_stream: Socket = self.client.stream_shell(cmdline)
|
||||
|
|
@ -109,16 +110,12 @@ class Client:
|
|||
"""
|
||||
Deploy server to android device
|
||||
"""
|
||||
server_file_path = __rootdir__ / "vendor" / "scrcpy-server-v2.7"
|
||||
server_file_path = get_path(f"@install/mower/vendor/scrcpy-server-v{SCR_VER}")
|
||||
server_buf = server_file_path.read_bytes()
|
||||
self.client.push(SCR_PATH, server_buf)
|
||||
self.__start_server()
|
||||
|
||||
def __init_server_connection(self) -> None:
|
||||
"""
|
||||
Connect to android server, there will be two sockets, video and control socket.
|
||||
This method will set: video_socket, control_socket, resolution variables
|
||||
"""
|
||||
try:
|
||||
self.control_socket = self.client.stream("localabstract:scrcpy")
|
||||
except socket.timeout:
|
||||
|
|
@ -176,7 +173,12 @@ class Client:
|
|||
while try_count < 3:
|
||||
try:
|
||||
return f(self, *args, **kwargs)
|
||||
except (ConnectionResetError, BrokenPipeError, ConnectionAbortedError):
|
||||
except (
|
||||
ConnectionResetError,
|
||||
BrokenPipeError,
|
||||
ConnectionAbortedError,
|
||||
) as e:
|
||||
logger.debug_exception(e)
|
||||
self.stop()
|
||||
time.sleep(1)
|
||||
self.check_adb_alive()
|
||||
|
|
|
|||
Loading…
Reference in a new issue