You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
206 lines
8.7 KiB
206 lines
8.7 KiB
"""
|
|
滑动手势服务模块
|
|
集成adb_swipe_norm.py的功能
|
|
"""
|
|
import re
|
|
from typing import Optional, Tuple, Dict, Any
|
|
from app.core.device.manager import device_manager, DeviceSource
|
|
from app.core.device.dispatcher import device_dispatcher
|
|
from app.services.auto_discovery_adb_service import AutoDiscoveryAdbService
|
|
from app.utils.structured_log import get_structured_logger
|
|
|
|
logger = get_structured_logger(__name__)
|
|
|
|
class SwipeService:
|
|
"""滑动手势服务类 - 遵循设备服务统一模式"""
|
|
|
|
def __init__(self):
|
|
self.auto_discovery_adb_service = AutoDiscoveryAdbService()
|
|
|
|
def clamp01(self, v: float) -> float:
|
|
"""将值限制在0-1范围内"""
|
|
return max(0.0, min(1.0, v))
|
|
|
|
async def get_device_source(self, device_id: str) -> DeviceSource:
|
|
"""获取设备来源"""
|
|
return await device_manager.get_device_source(device_id)
|
|
|
|
async def get_screen_size(self, device_id: str) -> Tuple[int, int]:
|
|
"""获取屏幕尺寸"""
|
|
try:
|
|
# 使用auto_discovery_adb_service获取屏幕尺寸
|
|
result = await self.auto_discovery_adb_service.execute_shell_command(device_id, "wm size")
|
|
if result.success:
|
|
output = result.output
|
|
m = re.search(r"(\d+)\s*x\s*(\d+)", output)
|
|
if m:
|
|
w, h = int(m.group(1)), int(m.group(2))
|
|
return w, h
|
|
|
|
# 备用方案：dumpsys display
|
|
result = await self.auto_discovery_adb_service.execute_shell_command(device_id, "dumpsys display")
|
|
if result.success:
|
|
output = result.output
|
|
m = re.search(r"(\d+)\s*x\s*(\d+)", output)
|
|
if m:
|
|
w, h = int(m.group(1)), int(m.group(2))
|
|
return w, h
|
|
|
|
raise RuntimeError("failed to detect screen size via 'wm size' or 'dumpsys display'.")
|
|
|
|
except Exception as e:
|
|
logger.error(f"获取屏幕尺寸失败: {device_id}", error=str(e))
|
|
raise
|
|
|
|
async def perform_swipe(self,
|
|
device_id: str,
|
|
x: float = 0.5,
|
|
y: float = 0.2,
|
|
direction: str = "down",
|
|
distance: float = 0.6,
|
|
duration: int = 300) -> Dict[str, Any]:
|
|
"""
|
|
执行标准化坐标滑动手势（统一入口）
|
|
|
|
Args:
|
|
device_id: 设备ID
|
|
x: 标准化起始x坐标 (0-1)
|
|
y: 标准化起始y坐标 (0-1)
|
|
direction: 滑动方向 ("up", "down", "left", "right")
|
|
distance: 滑动距离占屏幕比例 (0-1)
|
|
duration: 滑动持续时间(毫秒)
|
|
|
|
Returns:
|
|
包含执行结果的字典
|
|
"""
|
|
try:
|
|
device_source = await self.get_device_source(device_id)
|
|
logger.info(f"执行滑动手势: {device_id} -> {direction}",
|
|
device_source=device_source.value)
|
|
|
|
if device_source == DeviceSource.REGISTERED:
|
|
return await self._execute_registered_device_swipe(device_id, x, y, direction, distance, duration)
|
|
else:
|
|
return await self._execute_auto_discovered_device_swipe(device_id, x, y, direction, distance, duration)
|
|
|
|
except Exception as e:
|
|
logger.error(f"执行滑动手势失败: {device_id} -> {direction}", error=str(e))
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"device_id": device_id
|
|
}
|
|
|
|
async def _execute_registered_device_swipe(self, device_id: str, x: float, y: float,
|
|
direction: str, distance: float, duration: int) -> Dict[str, Any]:
|
|
"""执行注册设备的滑动手势"""
|
|
# 通过设备分发器执行操作
|
|
return await device_dispatcher.execute_operation(
|
|
device_id=device_id,
|
|
protocol_type="adb", # 滑动手势只支持ADB设备
|
|
operation="perform_swipe",
|
|
x=x,
|
|
y=y,
|
|
direction=direction,
|
|
distance=distance,
|
|
duration=duration
|
|
)
|
|
|
|
async def _execute_auto_discovered_device_swipe(self, device_id: str, x: float, y: float,
|
|
direction: str, distance: float, duration: int) -> Dict[str, Any]:
|
|
"""执行自动发现设备的滑动手势"""
|
|
try:
|
|
# 验证和限制参数
|
|
sx_n = self.clamp01(x)
|
|
sy_n = self.clamp01(y)
|
|
dist = max(0.0, min(1.0, distance))
|
|
|
|
# 获取屏幕尺寸
|
|
width, height = await self.get_screen_size(device_id)
|
|
|
|
# 转换标准化起始点到像素
|
|
sx = int(round(sx_n * width))
|
|
sy = int(round(sy_n * height))
|
|
|
|
# 根据方向和距离计算结束点
|
|
ex, ey = sx, sy
|
|
if direction == "down":
|
|
ey = sy + int(round(height * dist))
|
|
elif direction == "up":
|
|
ey = sy - int(round(height * dist))
|
|
elif direction == "right":
|
|
ex = sx + int(round(width * dist))
|
|
elif direction == "left":
|
|
ex = sx - int(round(width * dist))
|
|
else:
|
|
raise ValueError(f"无效的滑动方向: {direction}")
|
|
|
|
# 限制结束点在屏幕边界内
|
|
ex = max(0, min(width - 1, ex))
|
|
ey = max(0, min(height - 1, ey))
|
|
|
|
# 构建滑动命令
|
|
swipe_command = f"input swipe {sx} {sy} {ex} {ey} {duration}"
|
|
|
|
# 使用auto_discovery_adb_service执行命令
|
|
result = await self.auto_discovery_adb_service.execute_shell_command(device_id, swipe_command)
|
|
|
|
if not result.success:
|
|
error_msg = result.error or "Unknown error"
|
|
logger.error(f"ADB滑动命令执行失败: {error_msg}")
|
|
return {
|
|
"success": False,
|
|
"error": f"ADB error: {error_msg}",
|
|
"device_id": device_id,
|
|
"command": swipe_command
|
|
}
|
|
|
|
result_msg = f"swipe {direction} from ({sx},{sy}) to ({ex},{ey}) in {duration} ms on {width}x{height}"
|
|
logger.info(f"滑动成功: {result_msg}")
|
|
|
|
return {
|
|
"success": True,
|
|
"message": result_msg,
|
|
"device_id": device_id,
|
|
"start_point": {"x": sx, "y": sy},
|
|
"end_point": {"x": ex, "y": ey},
|
|
"screen_size": {"width": width, "height": height},
|
|
"normalized_start": {"x": sx_n, "y": sy_n},
|
|
"direction": direction,
|
|
"distance": dist,
|
|
"duration": duration
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"执行自动发现设备滑动手势失败: {e}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"device_id": device_id
|
|
}
|
|
|
|
async def swipe_up(self, device_id: str, distance: float = 0.6, duration: int = 300) -> Dict[str, Any]:
|
|
"""向上滑动"""
|
|
return await self.perform_swipe(device_id, direction="up", distance=distance, duration=duration)
|
|
|
|
async def swipe_down(self, device_id: str, distance: float = 0.6, duration: int = 300) -> Dict[str, Any]:
|
|
"""向下滑动"""
|
|
return await self.perform_swipe(device_id, direction="down", distance=distance, duration=duration)
|
|
|
|
async def swipe_left(self, device_id: str, distance: float = 0.6, duration: int = 300) -> Dict[str, Any]:
|
|
"""向左滑动"""
|
|
return await self.perform_swipe(device_id, direction="left", distance=distance, duration=duration)
|
|
|
|
async def swipe_right(self, device_id: str, distance: float = 0.6, duration: int = 300) -> Dict[str, Any]:
|
|
"""向右滑动"""
|
|
return await self.perform_swipe(device_id, direction="right", distance=distance, duration=duration)
|
|
|
|
async def swipe_from_point(self,
|
|
device_id: str,
|
|
x: float,
|
|
y: float,
|
|
direction: str,
|
|
distance: float = 0.6,
|
|
duration: int = 300) -> Dict[str, Any]:
|
|
"""从指定点开始滑动"""
|
|
return await self.perform_swipe(device_id, x=x, y=y, direction=direction, distance=distance, duration=duration)
|
|
|