Merge pull request #192 from zai-org/support-delay-config

support delay config
This commit is contained in:
yongbin-buaa
2025-12-15 13:17:10 +08:00
committed by GitHub
5 changed files with 233 additions and 21 deletions

View File

@@ -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)

View File

@@ -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(

View File

@@ -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

View File

@@ -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",
]

View 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",
]