Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions module/device/app_control.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ def app_is_running(self) -> bool:
def app_start(self):
method = self.config.script.device.screenshot_method
logger.info(f'App start: {self.package}')
# MuMu 6.0 assigns the app a new virtual display on every launch,
# whichever start method this config uses
self._reset_mumu_display_ids()
# if self.config.Emulator_Serial == 'wsa-0':
# self.app_start_wsa(display=0)
if method in AppControl._app_u2_family:
Expand All @@ -40,6 +43,8 @@ def app_start(self):
def app_stop(self):
method = self.config.script.device.screenshot_method
logger.info(f'App stop: {self.package}')
# MuMu 6.0 destroys the app's virtual display on stop; force re-detection
self._reset_mumu_display_ids()
if method in AppControl._app_u2_family:
self.app_stop_uiautomator2()
else:
Expand Down
134 changes: 125 additions & 9 deletions module/device/method/adb.py
Original file line number Diff line number Diff line change
Expand Up @@ -157,10 +157,118 @@ def __process_screenshot(self, screenshot):
logger.warning(f'Unexpected screenshot: {screenshot}')
raise OSError(f'cannot load screenshot')

# MuMu 6.0 (Android 15 engine) runs every app on its own virtual display
# (mumuscreenNNN), while `screencap` and `input` default to display 0 (the
# desktop). Detect the display the game is running on and target it.
_mumu_display_ids_cache = None
_mumu_display_ids_retry_at = 0.0
_mumu_display_ids_generation = 0
_MUMU_DISPLAY_RETRY_SEC = 20
_MUMU_DISPLAY_TIMEOUT = 3

def _reset_mumu_display_ids(self):
self._mumu_display_ids_cache = None
self._mumu_display_ids_retry_at = 0.0
self._mumu_display_ids_generation += 1

def _mumu_display_ids(self):
"""
Returns:
tuple[int, str] | None:
(display id for `input -d`, long display id for `screencap -d`)
or None if not applicable (single-display devices, or the game
is not running yet).
"""
if self._mumu_display_ids_cache is not None:
return self._mumu_display_ids_cache
# A provisional miss (dumpsys broken, or the game display not up yet) is
# retried on the next call; only the definitive "no virtual display at
# all" result is remembered, and only for a short window, so a plain
# single-display device does not pay a `dumpsys` call per click.
now = time.monotonic()
if now < self._mumu_display_ids_retry_at:
return None
generation = self._mumu_display_ids_generation
try:
# 1) Which display is the game on? Skip display 0, the desktop. The
# game's display is the one holding a window of our package; a
# system dialog on top of it takes the focus away from the
# package, so a non-launcher focus is the fallback.
out = self.adb_shell(['dumpsys', 'window', 'displays'], timeout=self._MUMU_DISPLAY_TIMEOUT)
display_id = None
fallback = None
virtual_display = False
current = None
for line in out.splitlines():
m = re.search(r'Display: mDisplayId=(\d+)', line)
if m:
current = int(m.group(1))
virtual_display = virtual_display or current != 0
continue
if not current:
continue
if self.package and self.package in line:
display_id = current
break
if fallback is None and 'mFocusedApp=ActivityRecord' in line and 'launcher' not in line.lower():
fallback = current
if display_id is None:
display_id = fallback
if display_id is None:
if not virtual_display:
# Plain single-display device: do not probe on every call.
self._mumu_display_ids_retry_at = now + self._MUMU_DISPLAY_RETRY_SEC
# Otherwise the game display is not up yet - retry next call.
return None
# 2) Display name, e.g. "mumuscreen002"
name = None
out = self.adb_shell(['dumpsys', 'display'], timeout=self._MUMU_DISPLAY_TIMEOUT)
idx = out.find('mDisplayId=%d' % display_id)
if idx >= 0:
m = re.search(r'name=(mumuscreen\d+)', out[idx:idx + 6000])
name = m.group(1) if m else None
if name is None:
return None
# 3) SurfaceFlinger long display id for `screencap -d`
long_id = None
out = self.adb_shell(['dumpsys', 'SurfaceFlinger', '--display-id'], timeout=self._MUMU_DISPLAY_TIMEOUT)
for line in out.splitlines():
if 'displayName="%s"' % name in line:
m = re.search(r'Display (\d+)', line)
if m:
long_id = m.group(1)
break
if long_id is None:
return None
if generation != self._mumu_display_ids_generation:
# The app was restarted while the display was being probed.
return None
logger.info(f'MuMu multi-display: game on display {display_id} ({name}), screencap id {long_id}')
self._mumu_display_ids_cache = (display_id, long_id)
Comment thread
sourcery-ai[bot] marked this conversation as resolved.
return self._mumu_display_ids_cache
except Exception as e:
logger.warning(f'Failed to detect MuMu multi-display setup: {e}')
return None

def _input_prefix(self):
"""
Returns:
list: `['input']` or `['input', '-d', '<display>']` for the game display.
"""
cmd = ['input']
ids = self._mumu_display_ids()
if ids:
cmd += ['-d', str(ids[0])]
return cmd

@retry
@Config.when(DEVICE_OVER_HTTP=False)
def screenshot_adb(self):
data = self.adb_shell(['screencap', '-p'], stream=True)
cmd = ['screencap', '-p']
ids = self._mumu_display_ids()
if ids:
cmd += ['-d', ids[1]]
data = self.adb_shell(cmd, stream=True)
if len(data) < 500:
logger.warning(f'Unexpected screenshot: {data}')

Expand All @@ -169,15 +277,23 @@ def screenshot_adb(self):
@retry
@Config.when(DEVICE_OVER_HTTP=True)
def screenshot_adb(self):
data = self.adb_shell(['screencap'], stream=True)
cmd = ['screencap']
ids = self._mumu_display_ids()
if ids:
cmd += ['-d', ids[1]]
data = self.adb_shell(cmd, stream=True)
if len(data) < 500:
logger.warning(f'Unexpected screenshot: {data}')

return load_screencap(data)

@retry
def screenshot_adb_nc(self):
data = self.adb_shell_nc(['screencap'])
cmd = ['screencap']
ids = self._mumu_display_ids()
if ids:
cmd += ['-d', ids[1]]
data = self.adb_shell_nc(cmd)
if len(data) < 500:
logger.warning(f'Unexpected screenshot: {data}')

Expand All @@ -186,7 +302,7 @@ def screenshot_adb_nc(self):
@retry
def click_adb(self, x, y):
start = time.time()
self.adb_shell(['input', 'tap', x, y])
self.adb_shell(self._input_prefix() + ['tap', x, y])
if time.time() - start <= 0.05:
self.sleep(0.05)

Expand All @@ -196,7 +312,7 @@ def swipe_adb(self, p1, p2, duration=0.1):
if getattr(self, 'is_desktop', False):
return self.swipe_window_message(p1, p2)
duration = int(duration * 1000)
self.adb_shell(['input', 'swipe', *p1, *p2, duration])
self.adb_shell(self._input_prefix() + ['swipe', *p1, *p2, duration])

def long_click_adb(self, x: float, y: float, duration: float):
"""
Expand All @@ -208,7 +324,7 @@ def long_click_adb(self, x: float, y: float, duration: float):
"""
if not 0 < duration < 3:
duration = duration / 1000
self.adb_shell(['input', 'swipe', x, y, x, y, int(duration*1000)])
self.adb_shell(self._input_prefix() + ['swipe', x, y, x, y, int(duration*1000)])

@retry
def app_current_adb(self):
Expand Down Expand Up @@ -369,17 +485,17 @@ def draw_adb(self, points: list, interval=0.01):
start_point = smooth_points[0]

# 按下起始点
self.adb_shell(['input', 'motionevent', 'DOWN', str(start_point[0]), str(start_point[1])])
self.adb_shell(self._input_prefix() + ['motionevent', 'DOWN', str(start_point[0]), str(start_point[1])])

# 移动到后续点
for i, point in enumerate(smooth_points[1:], start=1):
# 给点间隔
time.sleep(interval)
self.adb_shell(['input', 'motionevent', 'MOVE', str(point[0]), str(point[1])])
self.adb_shell(self._input_prefix() + ['motionevent', 'MOVE', str(point[0]), str(point[1])])

# 抬起手指
final_point = smooth_points[-1]
self.adb_shell(['input', 'motionevent', 'UP', str(final_point[0]), str(final_point[1])])
self.adb_shell(self._input_prefix() + ['motionevent', 'UP', str(final_point[0]), str(final_point[1])])


if __name__ == "__main__":
Expand Down
130 changes: 130 additions & 0 deletions tests/module/device/test_mumu_display_ids.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
"""MuMu 6.0 multi-display detection in `Adb._mumu_display_ids`.

Every test drives the detection with a fake `adb_shell`, so no device is
needed. The dumpsys excerpts are trimmed copies of a real MuMu 6.0
(Android 15) dump.
"""

from module.device.method.adb import Adb

PACKAGE = 'com.netease.onmyoji.wyzymnqsd_cps'
LONG_ID = '4619826888814064386'

# display 5 idle, display 0 the launcher, display 3 the game
WINDOW_DISPLAYS_MUMU6 = f'''\
Display: mDisplayId=5 (organized)
mCurrentFocus=null
mFocusedApp=null
Display: mDisplayId=0 (organized)
mFocusedApp=ActivityRecord{{1 u0 app.lawnchair/.LawnchairLauncher t2}}
Display: mDisplayId=3 (organized)
mCurrentFocus=Window{{ac9120e u0 {PACKAGE}/{PACKAGE}.Client}}
mFocusedApp=ActivityRecord{{2 u0 {PACKAGE}/{PACKAGE}.Client t153}}
'''

# old engine / plain phone: everything on display 0
WINDOW_DISPLAYS_SINGLE = f'''\
Display: mDisplayId=0 (organized)
mFocusedApp=ActivityRecord{{2 u0 {PACKAGE}/{PACKAGE}.Client t153}}
'''

# another app focused on its own display, listed before the game's
WINDOW_DISPLAYS_OTHER_APP_FIRST = f'''\
Display: mDisplayId=2 (organized)
mFocusedApp=ActivityRecord{{3 u0 com.other.app/.Main t10}}
Display: mDisplayId=3 (organized)
mFocusedApp=ActivityRecord{{2 u0 {PACKAGE}/{PACKAGE}.Client t153}}
'''

# virtual display exists, but nothing is focused on it yet
WINDOW_DISPLAYS_NOT_READY = f'''\
Display: mDisplayId=0 (organized)
mFocusedApp=ActivityRecord{{1 u0 app.lawnchair/.LawnchairLauncher t2}}
Display: mDisplayId=3 (organized)
mCurrentFocus=null
'''

# a system dialog takes the focus away from the game package
WINDOW_DISPLAYS_DIALOG = '''\
Display: mDisplayId=3 (organized)
mFocusedApp=ActivityRecord{4 u0 com.android.permissioncontroller/.GrantPermissionsActivity t5}
'''

DUMPSYS_DISPLAY = f'''\
mDisplayId=3
mPrimaryDisplayDevice=mumuscreen002(local:{LONG_ID})
mBaseDisplayInfo=DisplayInfo{{"mumuscreen002", ..., deviceProductInfo DeviceProductInfo{{name=mumuscreen002, manufacturerPnpId=GGL}}}}
'''

SURFACEFLINGER = f'''\
Display 4619827820427265280 (HWC display 0): port=0 pnpId=GGL displayName="mumuscreen000"
Display {LONG_ID} (HWC display 2): port=2 pnpId=GGL displayName="mumuscreen002"
'''


def make_adb(window_displays, dumpsys_display=DUMPSYS_DISPLAY, surfaceflinger=SURFACEFLINGER):
"""An Adb instance with only what `_mumu_display_ids` touches."""
adb = object.__new__(Adb)
adb.package = PACKAGE
adb._mumu_display_ids_cache = None
adb._mumu_display_ids_retry_at = 0.0
adb._mumu_display_ids_generation = 0
calls = []

def adb_shell(cmd, **kwargs):
calls.append(list(cmd))
return {
'window': window_displays,
'display': dumpsys_display,
'SurfaceFlinger': surfaceflinger,
}[cmd[1]]

adb.adb_shell = adb_shell
return adb, calls


def test_game_on_a_virtual_display_is_detected():
adb, calls = make_adb(WINDOW_DISPLAYS_MUMU6)
assert adb._mumu_display_ids() == (3, LONG_ID)
# cached: the next call does not probe again
assert adb._mumu_display_ids() == (3, LONG_ID)
assert len(calls) == 3


def test_plain_single_display_device_stays_on_display_0():
adb, calls = make_adb(WINDOW_DISPLAYS_SINGLE)
assert adb._mumu_display_ids() is None
# the definitive miss is remembered, so clicks do not probe per call
assert adb._mumu_display_ids() is None
assert len(calls) == 1


def test_another_app_on_an_earlier_display_does_not_win():
adb, _ = make_adb(WINDOW_DISPLAYS_OTHER_APP_FIRST)
assert adb._mumu_display_ids() == (3, LONG_ID)


def test_dialog_on_top_falls_back_to_the_focused_display():
adb, _ = make_adb(WINDOW_DISPLAYS_DIALOG)
assert adb._mumu_display_ids() == (3, LONG_ID)


def test_display_not_up_yet_is_retried_instead_of_remembered():
adb, calls = make_adb(WINDOW_DISPLAYS_NOT_READY)
assert adb._mumu_display_ids() is None
assert adb._mumu_display_ids() is None
assert len(calls) == 2


def test_input_prefix_targets_the_game_display():
adb, _ = make_adb(WINDOW_DISPLAYS_MUMU6)
adb._mumu_display_ids()
assert adb._input_prefix() == ['input', '-d', '3']


def test_a_reset_drops_the_cached_display():
adb, calls = make_adb(WINDOW_DISPLAYS_MUMU6)
assert adb._mumu_display_ids() == (3, LONG_ID)
adb._reset_mumu_display_ids()
assert adb._mumu_display_ids() == (3, LONG_ID)
assert len(calls) == 6
Loading