support delay config
This commit is contained in:
@@ -19,6 +19,7 @@ from phone_agent.adb import (
|
||||
tap,
|
||||
type_text,
|
||||
)
|
||||
from phone_agent.config.timing import TIMING_CONFIG
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -162,18 +163,18 @@ class ActionHandler:
|
||||
|
||||
# Switch to ADB keyboard
|
||||
original_ime = detect_and_set_adb_keyboard(self.device_id)
|
||||
time.sleep(1.0)
|
||||
time.sleep(TIMING_CONFIG.action.keyboard_switch_delay)
|
||||
|
||||
# Clear existing text and type new text
|
||||
clear_text(self.device_id)
|
||||
time.sleep(1.0)
|
||||
time.sleep(TIMING_CONFIG.action.text_clear_delay)
|
||||
|
||||
type_text(text, self.device_id)
|
||||
time.sleep(1.0)
|
||||
time.sleep(TIMING_CONFIG.action.text_input_delay)
|
||||
|
||||
# Restore original keyboard
|
||||
restore_keyboard(original_ime, self.device_id)
|
||||
time.sleep(1.0)
|
||||
time.sleep(TIMING_CONFIG.action.keyboard_restore_delay)
|
||||
|
||||
return ActionResult(True, False)
|
||||
|
||||
|
||||
@@ -6,6 +6,8 @@ from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
|
||||
from phone_agent.config.timing import TIMING_CONFIG
|
||||
|
||||
|
||||
class ConnectionType(Enum):
|
||||
"""Type of ADB connection."""
|
||||
@@ -244,7 +246,7 @@ class ADBConnection:
|
||||
output = result.stdout + result.stderr
|
||||
|
||||
if "restarting" in output.lower() or result.returncode == 0:
|
||||
time.sleep(2) # Wait for ADB to restart
|
||||
time.sleep(TIMING_CONFIG.connection.adb_restart_delay)
|
||||
return True, f"TCP/IP mode enabled on port {port}"
|
||||
else:
|
||||
return False, output.strip()
|
||||
@@ -312,7 +314,7 @@ class ADBConnection:
|
||||
[self.adb_path, "kill-server"], capture_output=True, timeout=5
|
||||
)
|
||||
|
||||
time.sleep(1)
|
||||
time.sleep(TIMING_CONFIG.connection.server_restart_delay)
|
||||
|
||||
# Start server
|
||||
subprocess.run(
|
||||
|
||||
@@ -6,6 +6,7 @@ import time
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from phone_agent.config.apps import APP_PACKAGES
|
||||
from phone_agent.config.timing import TIMING_CONFIG
|
||||
|
||||
|
||||
def get_current_app(device_id: str | None = None) -> str:
|
||||
@@ -35,7 +36,9 @@ def get_current_app(device_id: str | None = None) -> str:
|
||||
return "System Home"
|
||||
|
||||
|
||||
def tap(x: int, y: int, device_id: str | None = None, delay: float = 1.0) -> None:
|
||||
def tap(
|
||||
x: int, y: int, device_id: str | None = None, delay: float | None = None
|
||||
) -> None:
|
||||
"""
|
||||
Tap at the specified coordinates.
|
||||
|
||||
@@ -43,8 +46,11 @@ def tap(x: int, y: int, device_id: str | None = None, delay: float = 1.0) -> Non
|
||||
x: X coordinate.
|
||||
y: Y coordinate.
|
||||
device_id: Optional ADB device ID.
|
||||
delay: Delay in seconds after tap.
|
||||
delay: Delay in seconds after tap. If None, uses configured default.
|
||||
"""
|
||||
if delay is None:
|
||||
delay = TIMING_CONFIG.device.default_tap_delay
|
||||
|
||||
adb_prefix = _get_adb_prefix(device_id)
|
||||
|
||||
subprocess.run(
|
||||
@@ -54,7 +60,7 @@ def tap(x: int, y: int, device_id: str | None = None, delay: float = 1.0) -> Non
|
||||
|
||||
|
||||
def double_tap(
|
||||
x: int, y: int, device_id: str | None = None, delay: float = 1.0
|
||||
x: int, y: int, device_id: str | None = None, delay: float | None = None
|
||||
) -> None:
|
||||
"""
|
||||
Double tap at the specified coordinates.
|
||||
@@ -63,14 +69,17 @@ def double_tap(
|
||||
x: X coordinate.
|
||||
y: Y coordinate.
|
||||
device_id: Optional ADB device ID.
|
||||
delay: Delay in seconds after double tap.
|
||||
delay: Delay in seconds after double tap. If None, uses configured default.
|
||||
"""
|
||||
if delay is None:
|
||||
delay = TIMING_CONFIG.device.default_double_tap_delay
|
||||
|
||||
adb_prefix = _get_adb_prefix(device_id)
|
||||
|
||||
subprocess.run(
|
||||
adb_prefix + ["shell", "input", "tap", str(x), str(y)], capture_output=True
|
||||
)
|
||||
time.sleep(0.1)
|
||||
time.sleep(TIMING_CONFIG.device.double_tap_interval)
|
||||
subprocess.run(
|
||||
adb_prefix + ["shell", "input", "tap", str(x), str(y)], capture_output=True
|
||||
)
|
||||
@@ -82,7 +91,7 @@ def long_press(
|
||||
y: int,
|
||||
duration_ms: int = 3000,
|
||||
device_id: str | None = None,
|
||||
delay: float = 1.0,
|
||||
delay: float | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Long press at the specified coordinates.
|
||||
@@ -92,8 +101,11 @@ def long_press(
|
||||
y: Y coordinate.
|
||||
duration_ms: Duration of press in milliseconds.
|
||||
device_id: Optional ADB device ID.
|
||||
delay: Delay in seconds after long press.
|
||||
delay: Delay in seconds after long press. If None, uses configured default.
|
||||
"""
|
||||
if delay is None:
|
||||
delay = TIMING_CONFIG.device.default_long_press_delay
|
||||
|
||||
adb_prefix = _get_adb_prefix(device_id)
|
||||
|
||||
subprocess.run(
|
||||
@@ -111,7 +123,7 @@ def swipe(
|
||||
end_y: int,
|
||||
duration_ms: int | None = None,
|
||||
device_id: str | None = None,
|
||||
delay: float = 1.0,
|
||||
delay: float | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Swipe from start to end coordinates.
|
||||
@@ -123,8 +135,11 @@ def swipe(
|
||||
end_y: Ending Y coordinate.
|
||||
duration_ms: Duration of swipe in milliseconds (auto-calculated if None).
|
||||
device_id: Optional ADB device ID.
|
||||
delay: Delay in seconds after swipe.
|
||||
delay: Delay in seconds after swipe. If None, uses configured default.
|
||||
"""
|
||||
if delay is None:
|
||||
delay = TIMING_CONFIG.device.default_swipe_delay
|
||||
|
||||
adb_prefix = _get_adb_prefix(device_id)
|
||||
|
||||
if duration_ms is None:
|
||||
@@ -150,14 +165,17 @@ def swipe(
|
||||
time.sleep(delay)
|
||||
|
||||
|
||||
def back(device_id: str | None = None, delay: float = 1.0) -> None:
|
||||
def back(device_id: str | None = None, delay: float | None = None) -> None:
|
||||
"""
|
||||
Press the back button.
|
||||
|
||||
Args:
|
||||
device_id: Optional ADB device ID.
|
||||
delay: Delay in seconds after pressing back.
|
||||
delay: Delay in seconds after pressing back. If None, uses configured default.
|
||||
"""
|
||||
if delay is None:
|
||||
delay = TIMING_CONFIG.device.default_back_delay
|
||||
|
||||
adb_prefix = _get_adb_prefix(device_id)
|
||||
|
||||
subprocess.run(
|
||||
@@ -166,14 +184,17 @@ def back(device_id: str | None = None, delay: float = 1.0) -> None:
|
||||
time.sleep(delay)
|
||||
|
||||
|
||||
def home(device_id: str | None = None, delay: float = 1.0) -> None:
|
||||
def home(device_id: str | None = None, delay: float | None = None) -> None:
|
||||
"""
|
||||
Press the home button.
|
||||
|
||||
Args:
|
||||
device_id: Optional ADB device ID.
|
||||
delay: Delay in seconds after pressing home.
|
||||
delay: Delay in seconds after pressing home. If None, uses configured default.
|
||||
"""
|
||||
if delay is None:
|
||||
delay = TIMING_CONFIG.device.default_home_delay
|
||||
|
||||
adb_prefix = _get_adb_prefix(device_id)
|
||||
|
||||
subprocess.run(
|
||||
@@ -182,18 +203,23 @@ def home(device_id: str | None = None, delay: float = 1.0) -> None:
|
||||
time.sleep(delay)
|
||||
|
||||
|
||||
def launch_app(app_name: str, device_id: str | None = None, delay: float = 1.0) -> bool:
|
||||
def launch_app(
|
||||
app_name: str, device_id: str | None = None, delay: float | None = None
|
||||
) -> bool:
|
||||
"""
|
||||
Launch an app by name.
|
||||
|
||||
Args:
|
||||
app_name: The app name (must be in APP_PACKAGES).
|
||||
device_id: Optional ADB device ID.
|
||||
delay: Delay in seconds after launching.
|
||||
delay: Delay in seconds after launching. If None, uses configured default.
|
||||
|
||||
Returns:
|
||||
True if app was launched, False if app not found.
|
||||
"""
|
||||
if delay is None:
|
||||
delay = TIMING_CONFIG.device.default_launch_delay
|
||||
|
||||
if app_name not in APP_PACKAGES:
|
||||
return False
|
||||
|
||||
|
||||
@@ -4,6 +4,15 @@ from phone_agent.config.apps import APP_PACKAGES
|
||||
from phone_agent.config.i18n import get_message, get_messages
|
||||
from phone_agent.config.prompts_en import SYSTEM_PROMPT as SYSTEM_PROMPT_EN
|
||||
from phone_agent.config.prompts_zh import SYSTEM_PROMPT as SYSTEM_PROMPT_ZH
|
||||
from phone_agent.config.timing import (
|
||||
TIMING_CONFIG,
|
||||
ActionTimingConfig,
|
||||
ConnectionTimingConfig,
|
||||
DeviceTimingConfig,
|
||||
TimingConfig,
|
||||
get_timing_config,
|
||||
update_timing_config,
|
||||
)
|
||||
|
||||
|
||||
def get_system_prompt(lang: str = "cn") -> str:
|
||||
@@ -32,4 +41,11 @@ __all__ = [
|
||||
"get_system_prompt",
|
||||
"get_messages",
|
||||
"get_message",
|
||||
"TIMING_CONFIG",
|
||||
"TimingConfig",
|
||||
"ActionTimingConfig",
|
||||
"DeviceTimingConfig",
|
||||
"ConnectionTimingConfig",
|
||||
"get_timing_config",
|
||||
"update_timing_config",
|
||||
]
|
||||
|
||||
167
phone_agent/config/timing.py
Normal file
167
phone_agent/config/timing.py
Normal file
@@ -0,0 +1,167 @@
|
||||
"""Timing configuration for Phone Agent.
|
||||
|
||||
This module defines all configurable waiting times used throughout the application.
|
||||
Users can customize these values by modifying this file or by setting environment variables.
|
||||
"""
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class ActionTimingConfig:
|
||||
"""Configuration for action handler timing delays."""
|
||||
|
||||
# Text input related delays (in seconds)
|
||||
keyboard_switch_delay: float = 1.0 # Delay after switching to ADB keyboard
|
||||
text_clear_delay: float = 1.0 # Delay after clearing text
|
||||
text_input_delay: float = 1.0 # Delay after typing text
|
||||
keyboard_restore_delay: float = 1.0 # Delay after restoring original keyboard
|
||||
|
||||
def __post_init__(self):
|
||||
"""Load values from environment variables if present."""
|
||||
self.keyboard_switch_delay = float(
|
||||
os.getenv("PHONE_AGENT_KEYBOARD_SWITCH_DELAY", self.keyboard_switch_delay)
|
||||
)
|
||||
self.text_clear_delay = float(
|
||||
os.getenv("PHONE_AGENT_TEXT_CLEAR_DELAY", self.text_clear_delay)
|
||||
)
|
||||
self.text_input_delay = float(
|
||||
os.getenv("PHONE_AGENT_TEXT_INPUT_DELAY", self.text_input_delay)
|
||||
)
|
||||
self.keyboard_restore_delay = float(
|
||||
os.getenv("PHONE_AGENT_KEYBOARD_RESTORE_DELAY", self.keyboard_restore_delay)
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DeviceTimingConfig:
|
||||
"""Configuration for device operation timing delays."""
|
||||
|
||||
# Default delays for various device operations (in seconds)
|
||||
default_tap_delay: float = 1.0 # Default delay after tap
|
||||
default_double_tap_delay: float = 1.0 # Default delay after double tap
|
||||
double_tap_interval: float = 0.1 # Interval between two taps in double tap
|
||||
default_long_press_delay: float = 1.0 # Default delay after long press
|
||||
default_swipe_delay: float = 1.0 # Default delay after swipe
|
||||
default_back_delay: float = 1.0 # Default delay after back button
|
||||
default_home_delay: float = 1.0 # Default delay after home button
|
||||
default_launch_delay: float = 1.0 # Default delay after launching app
|
||||
|
||||
def __post_init__(self):
|
||||
"""Load values from environment variables if present."""
|
||||
self.default_tap_delay = float(
|
||||
os.getenv("PHONE_AGENT_TAP_DELAY", self.default_tap_delay)
|
||||
)
|
||||
self.default_double_tap_delay = float(
|
||||
os.getenv("PHONE_AGENT_DOUBLE_TAP_DELAY", self.default_double_tap_delay)
|
||||
)
|
||||
self.double_tap_interval = float(
|
||||
os.getenv("PHONE_AGENT_DOUBLE_TAP_INTERVAL", self.double_tap_interval)
|
||||
)
|
||||
self.default_long_press_delay = float(
|
||||
os.getenv("PHONE_AGENT_LONG_PRESS_DELAY", self.default_long_press_delay)
|
||||
)
|
||||
self.default_swipe_delay = float(
|
||||
os.getenv("PHONE_AGENT_SWIPE_DELAY", self.default_swipe_delay)
|
||||
)
|
||||
self.default_back_delay = float(
|
||||
os.getenv("PHONE_AGENT_BACK_DELAY", self.default_back_delay)
|
||||
)
|
||||
self.default_home_delay = float(
|
||||
os.getenv("PHONE_AGENT_HOME_DELAY", self.default_home_delay)
|
||||
)
|
||||
self.default_launch_delay = float(
|
||||
os.getenv("PHONE_AGENT_LAUNCH_DELAY", self.default_launch_delay)
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConnectionTimingConfig:
|
||||
"""Configuration for ADB connection timing delays."""
|
||||
|
||||
# ADB server and connection delays (in seconds)
|
||||
adb_restart_delay: float = 2.0 # Wait time after enabling TCP/IP mode
|
||||
server_restart_delay: float = (
|
||||
1.0 # Wait time between killing and starting ADB server
|
||||
)
|
||||
|
||||
def __post_init__(self):
|
||||
"""Load values from environment variables if present."""
|
||||
self.adb_restart_delay = float(
|
||||
os.getenv("PHONE_AGENT_ADB_RESTART_DELAY", self.adb_restart_delay)
|
||||
)
|
||||
self.server_restart_delay = float(
|
||||
os.getenv("PHONE_AGENT_SERVER_RESTART_DELAY", self.server_restart_delay)
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TimingConfig:
|
||||
"""Master timing configuration combining all timing settings."""
|
||||
|
||||
action: ActionTimingConfig
|
||||
device: DeviceTimingConfig
|
||||
connection: ConnectionTimingConfig
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize all timing configurations."""
|
||||
self.action = ActionTimingConfig()
|
||||
self.device = DeviceTimingConfig()
|
||||
self.connection = ConnectionTimingConfig()
|
||||
|
||||
|
||||
# Global timing configuration instance
|
||||
# Users can modify these values at runtime or through environment variables
|
||||
TIMING_CONFIG = TimingConfig()
|
||||
|
||||
|
||||
def get_timing_config() -> TimingConfig:
|
||||
"""
|
||||
Get the global timing configuration.
|
||||
|
||||
Returns:
|
||||
The global TimingConfig instance.
|
||||
"""
|
||||
return TIMING_CONFIG
|
||||
|
||||
|
||||
def update_timing_config(
|
||||
action: ActionTimingConfig | None = None,
|
||||
device: DeviceTimingConfig | None = None,
|
||||
connection: ConnectionTimingConfig | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Update the global timing configuration.
|
||||
|
||||
Args:
|
||||
action: New action timing configuration.
|
||||
device: New device timing configuration.
|
||||
connection: New connection timing configuration.
|
||||
|
||||
Example:
|
||||
>>> from phone_agent.config.timing import update_timing_config, ActionTimingConfig
|
||||
>>> custom_action = ActionTimingConfig(
|
||||
... keyboard_switch_delay=0.5,
|
||||
... text_input_delay=0.5
|
||||
... )
|
||||
>>> update_timing_config(action=custom_action)
|
||||
"""
|
||||
global TIMING_CONFIG
|
||||
if action is not None:
|
||||
TIMING_CONFIG.action = action
|
||||
if device is not None:
|
||||
TIMING_CONFIG.device = device
|
||||
if connection is not None:
|
||||
TIMING_CONFIG.connection = connection
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ActionTimingConfig",
|
||||
"DeviceTimingConfig",
|
||||
"ConnectionTimingConfig",
|
||||
"TimingConfig",
|
||||
"TIMING_CONFIG",
|
||||
"get_timing_config",
|
||||
"update_timing_config",
|
||||
]
|
||||
Reference in New Issue
Block a user