347 lines
14 KiB
Python
347 lines
14 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 = Field(
|
|
default=None,
|
|
description="Exact identity filter, NOT the text to read or wait for. Omit changing readouts; use the observed native_selector instead.",
|
|
)
|
|
content_desc: ShortText | None = None
|
|
class_name: ShortText | None = None
|
|
package_name: ShortText | None = None
|
|
bounds: Bounds | None = Field(
|
|
default=None,
|
|
description="Exact original Android screen PIXELS [left,top,right,bottom], only when needed to disambiguate. Never copy normalized 0..1000 UI Bounds or convert their rounded values back to pixels. Prefer the observed native_selector.",
|
|
)
|
|
|
|
@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 = Field(
|
|
description="Copy the current observation's native_selector object unchanged. Every supplied attribute is an exact AND filter. Do not add changing text/content_desc or normalized UI Bounds.",
|
|
)
|
|
|
|
|
|
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
|