chenmingxuan
4ea7939d9c
Add optional ensure_state, gesture region and final value feedback shared by Flash and Pro. Preserve completed inputs when verification fails and stop repeated ineffective strategies. Regenerate pinned tool schemas for the new optional arguments.
339 lines
13 KiB
Python
339 lines
13 KiB
Python
"""Validated native actions shared by Flash, Pro and the in-process action server."""
|
|
|
|
import math
|
|
from typing import Annotated, Literal
|
|
|
|
from pydantic import ConfigDict, Field, StrictInt, StringConstraints, model_validator
|
|
|
|
from artemis.mcp.action_parameters import (
|
|
CoordinateSpace,
|
|
Duration,
|
|
GestureModel,
|
|
Point,
|
|
Region,
|
|
pixel_point,
|
|
pixel_region,
|
|
require_in_region,
|
|
)
|
|
|
|
ShortText = Annotated[str, Field(min_length=1, max_length=512)]
|
|
Bounds = Annotated[list[StrictInt], Field(min_length=4, max_length=4)]
|
|
|
|
|
|
class ControlSelector(GestureModel):
|
|
"""Every supplied attribute is an exact AND match against a fresh native node."""
|
|
|
|
model_config = ConfigDict(str_strip_whitespace=False)
|
|
|
|
resource_id: ShortText | None = None
|
|
text: Annotated[str, StringConstraints(strip_whitespace=False, max_length=4096)] | None = None
|
|
content_desc: ShortText | None = None
|
|
class_name: ShortText | None = None
|
|
package_name: ShortText | None = None
|
|
bounds: Bounds | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def meaningful_selector(self):
|
|
if not any((self.resource_id, self.text, self.content_desc, self.class_name)):
|
|
raise ValueError("Copy at least one observed resource_id/text/content_desc/class_name.")
|
|
if self.bounds is not None:
|
|
left, top, right, bottom = self.bounds
|
|
if right <= left or bottom <= top:
|
|
raise ValueError("Bounds must be a nonempty [left, top, right, bottom] rectangle.")
|
|
return self
|
|
|
|
|
|
class ReadControlRequest(GestureModel):
|
|
selector: ControlSelector
|
|
|
|
|
|
ControlAction = Literal[
|
|
"click",
|
|
"long_click",
|
|
"focus",
|
|
"show_on_screen",
|
|
"set_progress",
|
|
"scroll_forward",
|
|
"scroll_backward",
|
|
"scroll_up",
|
|
"scroll_down",
|
|
"scroll_left",
|
|
"scroll_right",
|
|
"set_selection",
|
|
"replace_selection",
|
|
"set_text",
|
|
"ime_enter",
|
|
]
|
|
|
|
|
|
class ControlActionRequest(ReadControlRequest):
|
|
action: ControlAction
|
|
expected_token: Annotated[str, Field(pattern=r"^[a-f0-9]{64}$")] | None = None
|
|
value: float | None = None
|
|
# Deliberately do not strip user text: leading/trailing whitespace is input data.
|
|
text: Annotated[str, StringConstraints(strip_whitespace=False, max_length=65536)] | None = None
|
|
selection_start: Annotated[int, Field(strict=True, ge=0)] | None = None
|
|
selection_end: Annotated[int, Field(strict=True, ge=0)] | None = None
|
|
ensure_state: "ControlCondition | None" = Field(
|
|
default=None,
|
|
description="Desired observed state: skip an already satisfied action; otherwise write ONCE and verify by read-only polling. Never infer state from an icon or missing attributes.",
|
|
)
|
|
verify_timeout_ms: Annotated[int, Field(strict=True, ge=1, le=3000)] = 1500
|
|
|
|
@model_validator(mode="after")
|
|
def action_arguments(self):
|
|
if (self.value is not None) != (self.action == "set_progress"):
|
|
raise ValueError("Only set_progress requires a finite value.")
|
|
if (self.text is not None) != (self.action in {"set_text", "replace_selection"}):
|
|
raise ValueError(
|
|
"Only set_text/replace_selection require text (empty text is allowed)."
|
|
)
|
|
selection = self.action == "set_selection"
|
|
if (self.selection_start is not None) != selection or (
|
|
self.selection_end is not None
|
|
) != selection:
|
|
raise ValueError(
|
|
"Only set_selection requires selection_start and selection_end (UTF-16)."
|
|
)
|
|
if (
|
|
self.selection_start is not None
|
|
and self.selection_end is not None
|
|
and self.selection_end < self.selection_start
|
|
):
|
|
raise ValueError("selection_end must not precede selection_start.")
|
|
return self
|
|
|
|
|
|
class TouchFrame(GestureModel):
|
|
at_ms: Annotated[int, Field(strict=True, ge=0, le=10000)]
|
|
points: Annotated[list[Point], Field(min_length=1, max_length=5)]
|
|
|
|
|
|
class TouchGestureRequest(GestureModel):
|
|
frames: Annotated[list[TouchFrame], Field(min_length=2, max_length=32)]
|
|
coordinate_space: CoordinateSpace = "normalized"
|
|
target_description: ShortText
|
|
require_visible: ShortText | None = None
|
|
region: Region | None = Field(
|
|
default=None,
|
|
description="Observed gesture surface bounds in coordinate_space; every finger path must stay inside it.",
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def continuous_timeline(self):
|
|
if self.frames[0].at_ms != 0:
|
|
raise ValueError("The first frame must start at 0ms.")
|
|
count = len(self.frames[0].points)
|
|
for before, after in zip(self.frames, self.frames[1:]):
|
|
if after.at_ms <= before.at_ms:
|
|
raise ValueError("Frame times must strictly increase.")
|
|
if len(after.points) != count:
|
|
raise ValueError("Keep the same fingers down throughout all frames.")
|
|
return self
|
|
|
|
|
|
class PinchRequest(GestureModel):
|
|
center: Point
|
|
start_span: Annotated[int, Field(strict=True, gt=0, le=20000)]
|
|
end_span: Annotated[int, Field(strict=True, gt=0, le=20000)]
|
|
angle_degrees: Annotated[float, Field(ge=-360, le=360)] = 0
|
|
rotation_degrees: Annotated[float, Field(ge=-360, le=360)] = 0
|
|
duration_ms: Duration = 800
|
|
coordinate_space: CoordinateSpace = "normalized"
|
|
target_description: ShortText
|
|
require_visible: ShortText | None = None
|
|
region: Region | None = Field(
|
|
default=None,
|
|
description="Observed zoom surface bounds in coordinate_space; not the whole screen or date rail. Both fingers must remain inside.",
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def meaningful_pinch(self):
|
|
if self.start_span == self.end_span and self.rotation_degrees == 0:
|
|
raise ValueError("Change the span or rotation; an unchanged pinch is not a zoom.")
|
|
return self
|
|
|
|
|
|
class WaitUntilRequest(ReadControlRequest):
|
|
state: Literal[
|
|
"visible",
|
|
"absent",
|
|
"enabled",
|
|
"disabled",
|
|
"checked",
|
|
"unchecked",
|
|
"selected",
|
|
"unselected",
|
|
"text",
|
|
"content_desc",
|
|
"value",
|
|
"stable",
|
|
]
|
|
expected_text: (
|
|
Annotated[str, StringConstraints(strip_whitespace=False, max_length=4096)] | None
|
|
) = None
|
|
expected_value: float | None = None
|
|
tolerance: Annotated[float, Field(ge=0)] = 0
|
|
stable_ms: Annotated[int, Field(strict=True, ge=0, le=5000)] = 300
|
|
timeout_ms: Annotated[int, Field(strict=True, ge=1, le=30000)] = 5000
|
|
poll_interval_ms: Annotated[int, Field(strict=True, ge=50, le=1000)] = 200
|
|
|
|
@model_validator(mode="after")
|
|
def expected_arguments(self):
|
|
if (self.expected_text is not None) != (self.state in {"text", "content_desc"}):
|
|
raise ValueError("Only state=text/content_desc requires expected_text (exact match).")
|
|
if (self.expected_value is not None) != (self.state == "value"):
|
|
raise ValueError("Only state=value requires expected_value.")
|
|
if self.state != "value" and self.tolerance != 0:
|
|
raise ValueError("tolerance only applies to state=value.")
|
|
if self.state == "stable" and self.stable_ms > self.timeout_ms:
|
|
raise ValueError("stable_ms must fit within timeout_ms.")
|
|
return self
|
|
|
|
|
|
class ControlCondition(GestureModel):
|
|
state: Literal[
|
|
"checked", "unchecked", "selected", "unselected", "text", "content_desc", "value"
|
|
]
|
|
expected_text: (
|
|
Annotated[str, StringConstraints(strip_whitespace=False, max_length=4096)] | None
|
|
) = None
|
|
expected_value: float | None = None
|
|
tolerance: Annotated[float, Field(ge=0)] = 0
|
|
|
|
@model_validator(mode="after")
|
|
def expected_arguments(self):
|
|
# Reuse the wait predicate contract without duplicating its semantics.
|
|
WaitUntilRequest(selector={"class_name": "condition"}, **self.model_dump())
|
|
return self
|
|
|
|
|
|
ControlActionRequest.model_rebuild()
|
|
|
|
|
|
class ScrollToRequest(GestureModel):
|
|
container: ControlSelector
|
|
target: ControlSelector
|
|
direction: Literal["forward", "backward", "up", "down", "left", "right"] = "forward"
|
|
max_scrolls: Annotated[int, Field(strict=True, ge=1, le=20)] = 10
|
|
timeout_ms: Annotated[int, Field(strict=True, ge=1, le=30000)] = 10000
|
|
settle_ms: Annotated[int, Field(strict=True, ge=50, le=3000)] = 300
|
|
|
|
@model_validator(mode="after")
|
|
def settle_fits(self):
|
|
if self.settle_ms > self.timeout_ms:
|
|
raise ValueError("settle_ms must fit within timeout_ms.")
|
|
return self
|
|
|
|
|
|
NATIVE_REQUESTS: dict[str, type[GestureModel]] = {
|
|
"pinch": PinchRequest,
|
|
"touch_gesture": TouchGestureRequest,
|
|
"read_control": ReadControlRequest,
|
|
"control_action": ControlActionRequest,
|
|
"wait_until": WaitUntilRequest,
|
|
"scroll_to": ScrollToRequest,
|
|
}
|
|
NATIVE_ACTIONS = frozenset(NATIVE_REQUESTS)
|
|
READ_ONLY_ACTIONS = frozenset({"read_control", "wait_until"})
|
|
|
|
|
|
def native_tools_for_capabilities(capabilities: frozenset[str]) -> frozenset[str]:
|
|
tools: set[str] = set()
|
|
if "touch_gesture" in capabilities:
|
|
tools.update({"pinch", "touch_gesture"})
|
|
if "inspect_control" in capabilities:
|
|
tools.update({"read_control", "wait_until"})
|
|
if "control_action" in capabilities:
|
|
tools.update({"control_action", "scroll_to"})
|
|
return frozenset(tools)
|
|
|
|
|
|
def plan_touch(request: TouchGestureRequest, width: int, height: int) -> dict:
|
|
frames = [
|
|
{
|
|
"at_ms": frame.at_ms,
|
|
"points": [
|
|
pixel_point(p, request.coordinate_space, width, height) for p in frame.points
|
|
],
|
|
}
|
|
for frame in request.frames
|
|
]
|
|
region = None
|
|
if request.region is not None:
|
|
region = pixel_region(request.region, request.coordinate_space, width, height)
|
|
require_in_region([p for frame in frames for p in frame["points"]], region)
|
|
# Reject trajectories that merge two fingers, including between keyframes.
|
|
for before, after in zip(frames, frames[1:]):
|
|
for first in range(len(before["points"])):
|
|
for second in range(first):
|
|
a = [before["points"][first][k] - before["points"][second][k] for k in (0, 1)]
|
|
b = [after["points"][first][k] - after["points"][second][k] - a[k] for k in (0, 1)]
|
|
length = sum(v * v for v in b)
|
|
t = max(0, min(1, -sum(a[k] * b[k] for k in (0, 1)) / length)) if length else 0
|
|
if sum((a[k] + t * b[k]) ** 2 for k in (0, 1)) < 1:
|
|
raise ValueError(
|
|
"Finger paths must remain distinct; use rotation keyframes instead of crossing fingers."
|
|
)
|
|
return {
|
|
"action": "touch_gesture",
|
|
"frames": frames,
|
|
"coordinate_space": "pixel",
|
|
"screen_size": [width, height],
|
|
"target_description": request.target_description,
|
|
**({"region": region} if region is not None else {}),
|
|
**({"require_visible": request.require_visible} if request.require_visible else {}),
|
|
}
|
|
|
|
|
|
def plan_pinch(request: PinchRequest, width: int, height: int) -> dict:
|
|
# A span in normalized space uses the short display edge, preserving angles/aspect.
|
|
center = pixel_point(request.center, request.coordinate_space, width, height)
|
|
factor = min(width, height) / 1000 if request.coordinate_space == "normalized" else 1
|
|
segments = max(1, math.ceil(abs(request.rotation_degrees) / 15))
|
|
if request.duration_ms < segments:
|
|
raise ValueError("duration_ms is too short for the requested rotation.")
|
|
frames = []
|
|
for index in range(segments + 1):
|
|
ratio = index / segments
|
|
radius = (request.start_span + (request.end_span - request.start_span) * ratio) * factor / 2
|
|
angle = math.radians(request.angle_degrees + request.rotation_degrees * ratio)
|
|
offset = [math.cos(angle) * radius, math.sin(angle) * radius]
|
|
frames.append(
|
|
{
|
|
"at_ms": round(request.duration_ms * ratio),
|
|
"points": [
|
|
[round(center[k] + sign * offset[k]) for k in (0, 1)] for sign in (-1, 1)
|
|
],
|
|
}
|
|
)
|
|
planned = plan_touch(
|
|
TouchGestureRequest(
|
|
frames=frames,
|
|
coordinate_space="pixel",
|
|
target_description=request.target_description,
|
|
require_visible=request.require_visible,
|
|
region=pixel_region(request.region, request.coordinate_space, width, height)
|
|
if request.region is not None
|
|
else None,
|
|
),
|
|
width,
|
|
height,
|
|
)
|
|
planned["action"] = "pinch"
|
|
planned["pinch"] = {
|
|
"start_span": request.start_span,
|
|
"end_span": request.end_span,
|
|
"rotation_degrees": request.rotation_degrees,
|
|
}
|
|
return planned
|
|
|
|
|
|
def validate_native_action(name: str, args: dict, width: int, height: int) -> GestureModel:
|
|
request = NATIVE_REQUESTS[name].model_validate(args)
|
|
if isinstance(request, PinchRequest):
|
|
plan_pinch(request, width, height)
|
|
elif isinstance(request, TouchGestureRequest):
|
|
plan_touch(request, width, height)
|
|
return request
|