YoursFunny/ArisuAutoSweeper
1
0
Fork 0
mirror of https://github.com/TheFunny/ArisuAutoSweeper synced 2026-10-10 20:02:14 +00:00
Code Releases Wiki Activity

feat: enhance device connection handling and droidcast resolution management

Browse Source
This commit is contained in:
YoursFunny
2026-07-05 20:06:53 +08:00
parent 3cef2ee01b
commit 50c4bc5bba
7 changed files with 197 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/device/connection.py
+115 -24
View File
@@ -3,6 +3,7 @@ import logging
import re
import socket
import subprocess
import sys
import time
from functools import wraps
@@ -12,7 +13,7 @@ from adbutils.errors import AdbError
import module.config.server as server_
import platform
from module.base.decorator import Config, cached_property, del_cached_property
from module.base.decorator import Config, cached_property, del_cached_property, run_once
from module.base.utils import SelectedGrids, ensure_time
from module.device.connection_attr import ConnectionAttr
from module.device.method.utils import (
@@ -84,10 +85,17 @@ class AdbDeviceWithStatus(AdbDevice):
def __bool__(self):
return True
@cached_property
def port(self) -> int:
try:
return int(self.serial.split(':')[1])
except (IndexError, ValueError):
return 0
@cached_property
def may_mumu12_family(self):
# 127.0.0.1:16XXX
return len(self.serial) == 15 and self.serial.startswith('127.0.0.1:16')
return 16384 <= self.port <= 17408
class Connection(ConnectionAttr):
@@ -274,6 +282,7 @@ class Connection(ConnectionAttr):
@cached_property
def nemud_app_keep_alive(self) -> str:
res = self.adb_getprop('nemud.app_keep_alive')
logger.attr('nemud.app_keep_alive', res)
return res
@retry
@@ -282,7 +291,6 @@ class Connection(ConnectionAttr):
return False
res = self.nemud_app_keep_alive
logger.attr('nemud.app_keep_alive', res)
if res == '':
# Empty property, probably MuMu6 or MuMu12 version < 3.5.6
return True
@@ -297,6 +305,15 @@ class Connection(ConnectionAttr):
logger.warning(f'Invalid nemud.app_keep_alive value: {res}')
return False
@cached_property
def is_mumu_over_version_356(self) -> bool:
"""
Returns:
bool: If MuMu12 version >= 3.5.6,
which has nemud.app_keep_alive and always be a vertical device
"""
return self.nemud_app_keep_alive != ''
@cached_property
def _nc_server_host_port(self):
"""
@@ -547,14 +564,14 @@ class Connection(ConnectionAttr):
# Disconnect offline device before connecting
for device in self.list_device():
if device.status == 'offline':
logger.warning(f'Device {serial} is offline, disconnect it before connecting')
self.adb_disconnect(serial)
logger.warning(f'Device {device.serial} is offline, disconnect it before connecting')
self.adb_disconnect(device.serial)
elif device.status == 'unauthorized':
logger.error(f'Device {serial} is unauthorized, please accept ADB debugging on your device')
logger.error(f'Device {device.serial} is unauthorized, please accept ADB debugging on your device')
elif device.status == 'device':
pass
else:
logger.warning(f'Device {serial} is is having a unknown status: {device.status}')
logger.warning(f'Device {device.serial} is is having a unknown status: {device.status}')
# Skip for emulator-5554
if 'emulator-' in serial:
@@ -762,23 +779,45 @@ class Connection(ConnectionAttr):
If serial=='auto' and only 1 device detected, use it
"""
logger.hr('Detect device')
logger.info('Here are the available devices, '
'copy to Alas.Emulator.Serial to use it or set Alas.Emulator.Serial="auto"')
devices = self.list_device()
available = SelectedGrids([])
devices = SelectedGrids([])
# Show available devices
available = devices.select(status='device')
for device in available:
logger.info(device.serial)
if not len(available):
logger.info('No available devices')
@run_once
def brute_force_connect():
logger.info('Brute force connect')
from deploy.Windows.emulator import EmulatorManager
manager = EmulatorManager()
manager.brute_force_connect()
# Show unavailable devices if having any
unavailable = devices.delete(available)
if len(unavailable):
logger.info('Here are the devices detected but unavailable')
for device in unavailable:
logger.info(f'{device.serial} ({device.status})')
for _ in range(2):
logger.info('Here are the available devices, '
'copy to Alas.Emulator.Serial to use it or set Alas.Emulator.Serial="auto"')
devices = self.list_device()
# Show available devices
available = devices.select(status='device')
for device in available:
logger.info(device.serial)
if not len(available):
logger.info('No available devices')
# Show unavailable devices if having any
unavailable = devices.delete(available)
if len(unavailable):
logger.info('Here are the devices detected but unavailable')
for device in unavailable:
logger.info(f'{device.serial} ({device.status})')
# brute_force_connect
if self.config.Emulator_Serial == 'auto' and available.count == 0:
logger.warning(f'No available device found')
if sys.platform == 'win32':
brute_force_connect()
continue
else:
break
else:
break
# Auto device detection
if self.config.Emulator_Serial == 'auto':
@@ -788,7 +827,7 @@ class Connection(ConnectionAttr):
raise RequestHumanTakeover
elif available.count == 1:
logger.info(f'Auto device detection found only one device, using it')
self.serial = available[0].serial
self.config.Emulator_Serial = self.serial = available[0].serial
del_cached_property(self, 'adb')
elif available.count == 2 \
and available.select(serial='127.0.0.1:7555') \
@@ -797,7 +836,7 @@ class Connection(ConnectionAttr):
# For MuMu12 serials like 127.0.0.1:7555 and 127.0.0.1:16384
# ignore 7555 use 16384
remain = available.select(may_mumu12_family=True).first_or_none()
self.serial = remain.serial
self.config.Emulator_Serial = self.serial = remain.serial
del_cached_property(self, 'adb')
else:
logger.critical('Multiple devices found, auto device detection cannot decide which to choose, '
@@ -806,6 +845,7 @@ class Connection(ConnectionAttr):
# Handle LDPlayer
# LDPlayer serial jumps between `127.0.0.1:5555+{X}` and `emulator-5554+{X}`
# No config write since it's dynamic
port_serial, emu_serial = get_serial_pair(self.serial)
if port_serial and emu_serial:
# Might be LDPlayer, check connected devices
@@ -832,6 +872,57 @@ class Connection(ConnectionAttr):
f'Using serial: {emu_serial}')
self.serial = emu_serial
# Redirect MuMu12 from 127.0.0.1:7555 to 127.0.0.1:16xxx
if self.serial == '127.0.0.1:7555':
for _ in range(2):
mumu12 = available.select(may_mumu12_family=True)
if mumu12.count == 1:
emu_serial = mumu12.first_or_none().serial
logger.warning(f'Redirect MuMu12 {self.serial} to {emu_serial}')
self.config.Emulator_Serial = self.serial = emu_serial
break
elif mumu12.count >= 2:
logger.warning(f'Multiple MuMu12 serial found, cannot redirect')
break
else:
# Only 127.0.0.1:7555
if self.is_mumu_over_version_356:
# is_mumu_over_version_356 and nemud_app_keep_alive was cached
# Acceptable since it's the same device
logger.warning(f'Device {self.serial} is MuMu12 but corresponding port not found')
brute_force_connect()
devices = self.list_device()
# Show available devices
available = devices.select(status='device')
for device in available:
logger.info(device.serial)
if not len(available):
logger.info('No available devices')
continue
else:
# MuMu6
break
# MuMu12 uses 127.0.0.1:16385 if port 16384 is occupied, auto redirect
# No config write since it's dynamic
if self.is_mumu12_family:
matched = False
for device in available.select(may_mumu12_family=True):
if device.port == self.port:
# Exact match
matched = True
break
if not matched:
for device in available.select(may_mumu12_family=True):
if -2 <= device.port - self.port <= 2:
# Port switched
logger.info(f'MuMu12 port switches from {self.serial} to {device.serial}')
del_cached_property(self, 'port')
del_cached_property(self, 'is_mumu12_family')
del_cached_property(self, 'is_mumu_family')
self.serial = device.serial
break
@retry
def list_package(self, show_log=True):
"""
module/device/connection_attr.py
+16 -4
View File
@@ -122,6 +122,18 @@ class ConnectionAttr:
def is_wsa(self):
return bool(re.match(r'^wsa', self.serial))
@cached_property
def port(self) -> int:
try:
return int(self.serial.split(':')[1])
except (IndexError, ValueError):
return 0
@cached_property
def is_mumu12_family(self):
# 127.0.0.1:16XXX
return 16384 <= self.port <= 17408
@cached_property
def is_mumu_family(self):
# 127.0.0.1:7555
@@ -129,9 +141,8 @@ class ConnectionAttr:
return self.serial == '127.0.0.1:7555' or self.is_mumu12_family
@cached_property
def is_mumu12_family(self):
# 127.0.0.1:16384 + 32*n
return len(self.serial) == 15 and self.serial.startswith('127.0.0.1:16')
def is_nox_family(self):
return 62001 <= self.port <= 63025
@cached_property
def is_emulator(self):
@@ -177,7 +188,8 @@ class ConnectionAttr:
rf"SOFTWARE\BlueStacks_bgp64_hyperv\Guests\{folder_name}\Config") as key:
port = QueryValueEx(key, "BstAdbPort")[0]
except FileNotFoundError:
logger.error(rf'Unable to find registry HKEY_LOCAL_MACHINE\SOFTWARE\BlueStacks_bgp64_hyperv\Guests\{folder_name}\Config')
logger.error(
rf'Unable to find registry HKEY_LOCAL_MACHINE\SOFTWARE\BlueStacks_bgp64_hyperv\Guests\{folder_name}\Config')
logger.error('Please confirm that your are using BlueStack 4 hyper-v and not regular BlueStacks 4')
logger.error(r'Please check if there is any other emulator instances under '
r'registry HKEY_LOCAL_MACHINE\SOFTWARE\BlueStacks_bgp64_hyperv\Guests')
module/device/device.py
+1 -2
View File
@@ -3,7 +3,6 @@ import collections
from module.base.timer import Timer
from module.device.app_control import AppControl
from module.device.control import Control
from module.device.platform import Platform
from module.device.screenshot import Screenshot
from module.exception import (
EmulatorNotRunningError,
@@ -55,7 +54,7 @@ def show_function_call():
logger.info('Function calls:' + ''.join(func_list))
class Device(Screenshot, Control, AppControl, Platform):
class Device(Screenshot, Control, AppControl):
_screen_size_checked = False
detect_record = set()
click_record = collections.deque(maxlen=15)
module/device/method/adb.py
+1 -1
View File
@@ -128,7 +128,7 @@ class Adb(Connection):
if image is None:
raise ImageTruncated('Empty image after cv2.imdecode')
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB, dst=image)
cv2.cvtColor(image, cv2.COLOR_BGR2RGB, dst=image)
if image is None:
raise ImageTruncated('Empty image after cv2.cvtColor')
module/device/method/droidcast.py
+55 -1
View File
@@ -95,6 +95,8 @@ class DroidCast(Uiautomator2):
"""
_droidcast_port: int = 0
droidcast_width: int = 0
droidcast_height: int = 0
@cached_property
def droidcast_session(self):
@@ -114,14 +116,35 @@ class DroidCast(Uiautomator2):
"""
def droidcast_url(self, url='/preview'):
if self.is_mumu_over_version_356:
w, h = self.droidcast_width, self.droidcast_height
if self.orientation == 0:
return f'http://127.0.0.1:{self._droidcast_port}{url}?width={w}&height={h}'
elif self.orientation == 1:
return f'http://127.0.0.1:{self._droidcast_port}{url}?width={h}&height={w}'
else:
# logger.warning('DroidCast receives invalid device orientation')
pass
return f'http://127.0.0.1:{self._droidcast_port}{url}'
def droidcast_raw_url(self, url='/screenshot'):
if self.is_mumu_over_version_356:
w, h = self.droidcast_width, self.droidcast_height
if self.orientation == 0:
return f'http://127.0.0.1:{self._droidcast_port}{url}?width={w}&height={h}'
elif self.orientation == 1:
return f'http://127.0.0.1:{self._droidcast_port}{url}?width={h}&height={w}'
else:
# logger.warning('DroidCast receives invalid device orientation')
pass
return f'http://127.0.0.1:{self._droidcast_port}{url}'
def droidcast_init(self):
logger.hr('DroidCast init')
self.droidcast_stop()
self._droidcast_update_resolution()
logger.info('Pushing DroidCast apk')
self.adb_push(self.config.DROIDCAST_FILEPATH_LOCAL, self.config.DROIDCAST_FILEPATH_REMOTE)
@@ -151,10 +174,25 @@ class DroidCast(Uiautomator2):
else:
logger.error(f'Unknown DROIDCAST_VERSION: {self.config.DROIDCAST_VERSION}')
def _droidcast_update_resolution(self):
if self.is_mumu_over_version_356:
logger.info('Update droidcast resolution')
w, h = self.resolution_uiautomator2(cal_rotation=False)
self.get_orientation()
# 720, 1280
# mumu12 > 3.5.6 is always a vertical device
self.droidcast_width, self.droidcast_height = w, h
logger.info(f'Droicast resolution: {(w, h)}')
@retry
def screenshot_droidcast(self):
self.config.DROIDCAST_VERSION = 'DroidCast'
if self.is_mumu_over_version_356:
if not self.droidcast_width or not self.droidcast_height:
self._droidcast_update_resolution()
resp = self.droidcast_session.get(self.droidcast_url(), timeout=3)
if resp.status_code == 404:
raise DroidCastVersionIncompatible('DroidCast server does not have /preview')
image = resp.content
@@ -174,16 +212,27 @@ class DroidCast(Uiautomator2):
if image is None:
raise ImageTruncated('Empty image after cv2.cvtColor')
if self.is_mumu_over_version_356:
if self.orientation == 1:
image = cv2.rotate(image, cv2.ROTATE_90_CLOCKWISE)
return image
@retry
def screenshot_droidcast_raw(self):
self.config.DROIDCAST_VERSION = 'DroidCast_raw'
shape = (720, 1280)
if self.is_mumu_over_version_356:
if not self.droidcast_width or not self.droidcast_height:
self._droidcast_update_resolution()
if self.droidcast_height and self.droidcast_width:
shape = (self.droidcast_height, self.droidcast_width)
image = self.droidcast_session.get(self.droidcast_raw_url(), timeout=3).content
# DroidCast_raw returns a RGB565 bitmap
try:
arr = np.frombuffer(image, dtype=np.uint16).reshape((720, 1280))
arr = np.frombuffer(image, dtype=np.uint16).reshape(shape)
except ValueError as e:
if len(image) < 500:
logger.warning(f'Unexpected screenshot: {image}')
@@ -231,6 +280,11 @@ class DroidCast(Uiautomator2):
cv2.add(b, m, dst=b)
image = cv2.merge([r, g, b])
if self.is_mumu_over_version_356:
if self.orientation == 1:
image = cv2.rotate(image, cv2.ROTATE_90_CLOCKWISE)
return image
def droidcast_wait_startup(self):
module/device/method/uiautomator_2.py
+5 -4
View File
@@ -242,7 +242,7 @@ class Uiautomator2(Connection):
return hierarchy
@retry
def resolution_uiautomator2(self) -> t.Tuple[int, int]:
def resolution_uiautomator2(self, cal_rotation=True) -> t.Tuple[int, int]:
"""
Faster u2.window_size(), cause that calls `dumpsys display` twice.
@@ -251,9 +251,10 @@ class Uiautomator2(Connection):
"""
info = self.u2.http.get('/info').json()
w, h = info['display']['width'], info['display']['height']
rotation = self.get_orientation()
if (w > h) != (rotation % 2 == 1):
w, h = h, w
if cal_rotation:
rotation = self.get_orientation()
if (w > h) != (rotation % 2 == 1):
w, h = h, w
return w, h
def resolution_check_uiautomator2(self):
module/device/screenshot.py
+4
View File
@@ -70,6 +70,10 @@ class Screenshot(Adb, WSA, DroidCast, AScreenCap, Scrcpy):
return self.image
@property
def has_cached_image(self):
return hasattr(self, 'image') and self.image is not None
def _handle_orientated_image(self, image):
"""
Args: