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.
357 lines
14 KiB
Python
357 lines
14 KiB
Python
"""Validated, shared plans for short gesture sequences and calibrated drags.
|
|
|
|
Planning is pure: every member is checked before the first device input. Scale
|
|
values are supplied from observations by the caller, never inferred by the driver.
|
|
"""
|
|
|
|
import math
|
|
from typing import Annotated, Literal
|
|
|
|
from pydantic import Field, StringConstraints, model_validator
|
|
|
|
from artemis.mcp.action_parameters import (
|
|
CoordinateSpace,
|
|
Duration,
|
|
GestureModel,
|
|
Point,
|
|
Region,
|
|
pixel_point,
|
|
pixel_region,
|
|
require_in_region,
|
|
)
|
|
from artemis.mcp.native_actions import (
|
|
ControlActionRequest,
|
|
PinchRequest,
|
|
TouchGestureRequest,
|
|
WaitUntilRequest,
|
|
plan_pinch,
|
|
plan_touch,
|
|
)
|
|
from artemis.mcp.value_feedback import ValueFeedback
|
|
|
|
|
|
class ScaleAnchor(GestureModel):
|
|
position: float = Field(
|
|
description="Observed axis coordinate in the selected coordinate space."
|
|
)
|
|
value: float = Field(description="Observed scale value, in one consistent numeric unit.")
|
|
|
|
|
|
class DragCalibration(GestureModel):
|
|
axis: Literal["x", "y"]
|
|
mode: Literal["thumb", "pan"] = Field(
|
|
description="thumb moves a handle to a value; pan moves a scale under a fixed indicator."
|
|
)
|
|
anchors: list[ScaleAnchor] = Field(min_length=2, max_length=2)
|
|
target_value: float
|
|
indicator_position: float | None = Field(
|
|
default=None, description="Required for pan: the fixed indicator's observed axis position."
|
|
)
|
|
tolerance: float | None = Field(
|
|
default=None,
|
|
ge=0,
|
|
description="Allowed value error. Reject a plan whose pixel rounding exceeds it; this is geometric error only, not proof of application precision.",
|
|
)
|
|
min_movement_px: float | None = Field(
|
|
default=None,
|
|
gt=0,
|
|
description="Previously measured minimum effective drag distance in pixels. If this calibrated move is smaller, refuse input and zoom first. Never guess a device-wide threshold.",
|
|
)
|
|
|
|
|
|
class DragGeometry(GestureModel):
|
|
start: Point
|
|
target_description: str = Field(min_length=1)
|
|
end: Point | None = Field(
|
|
default=None, description="Exact endpoint; MUST be null when calibration is supplied."
|
|
)
|
|
coordinate_space: CoordinateSpace = "normalized"
|
|
duration_ms: Duration = 300
|
|
region: Region | None = Field(
|
|
default=None,
|
|
description="Observed gesture surface rectangle in coordinate_space, excluding date/navigation rails and unrelated overlays.",
|
|
)
|
|
calibration: DragCalibration | None = Field(
|
|
default=None, description="Computes the endpoint from observed anchors; then set end=null."
|
|
)
|
|
require_visible: str | None = Field(
|
|
default=None,
|
|
min_length=1,
|
|
max_length=512,
|
|
description="Exact previously observed resource_id of a panel control. Check live visibility before dragging; refuse input if absent. Use for auto-hiding controls.",
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def endpoint_or_calibration(self):
|
|
if (self.end is None) == (self.calibration is None):
|
|
raise ValueError(
|
|
"Supply exactly one of end or calibration. Calibration computes the endpoint: set end=null (or omit end) when calibration is supplied. Do not repeat the rejected combination."
|
|
)
|
|
return self
|
|
|
|
|
|
class DragRequest(DragGeometry):
|
|
feedback: ValueFeedback | None = None
|
|
|
|
|
|
class ClickStep(GestureModel):
|
|
action: Literal["click"]
|
|
target: Point | Annotated[str, Field(min_length=1, max_length=512)] = Field(
|
|
description="Prefer an exact observed resource_id: resolve its live bounds before tapping, refusing absent/ambiguous targets. Otherwise use [x,y] in coordinate_space; never mix screenshot pixels with normalized coordinates."
|
|
)
|
|
target_description: str = Field(min_length=1, max_length=512)
|
|
coordinate_space: CoordinateSpace = "normalized"
|
|
times: int = Field(
|
|
default=1,
|
|
strict=True,
|
|
ge=1,
|
|
le=10,
|
|
description="Consecutive taps within this ONE member, e.g. 10 for a scale +/- button; no model turns between taps.",
|
|
)
|
|
delay_ms: int = Field(default=100, strict=True, ge=0, le=1000)
|
|
unless_visible: str | None = Field(
|
|
default=None,
|
|
min_length=1,
|
|
max_length=512,
|
|
description="Exact previously observed resource_id inside a toggled panel (not the opener). Skip this opener click when that control is already visible; otherwise click. Avoid closing a still-open panel.",
|
|
)
|
|
|
|
|
|
class DragStep(DragGeometry):
|
|
action: Literal["drag"]
|
|
|
|
|
|
class WaitStep(GestureModel):
|
|
action: Literal["wait"]
|
|
time_in_ms: int = Field(strict=True, ge=0, le=10000)
|
|
|
|
|
|
class EnsureVisibleStep(GestureModel):
|
|
action: Literal["ensure_visible"]
|
|
target: str = Field(
|
|
min_length=1,
|
|
max_length=512,
|
|
description="Panel control resource_id copied VERBATIM from observed elements, including the ':id/' separator. Never rewrite it as ':id:' or guess an ID.",
|
|
)
|
|
opener: str = Field(
|
|
min_length=1,
|
|
max_length=512,
|
|
description="UI opener resource_id copied VERBATIM from observed elements, including ':id/' (not ':id:'). Tap ONCE only if target is absent; never use a business submission as an opener.",
|
|
)
|
|
target_description: str = Field(min_length=1)
|
|
timeout_ms: int = Field(
|
|
default=1500,
|
|
strict=True,
|
|
ge=100,
|
|
le=3000,
|
|
description="Bounded live-UI wait AFTER opening, without extra model turns or repeated opener taps.",
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def distinct_opener(self):
|
|
if self.target == self.opener:
|
|
raise ValueError(
|
|
"ensure_visible target must be inside the panel, distinct from opener."
|
|
)
|
|
return self
|
|
|
|
|
|
class PinchStep(PinchRequest):
|
|
action: Literal["pinch"]
|
|
|
|
|
|
class TouchGestureStep(TouchGestureRequest):
|
|
action: Literal["touch_gesture"]
|
|
|
|
|
|
class SequenceControlRequest(ControlActionRequest):
|
|
text: Annotated[str, StringConstraints(strip_whitespace=False, max_length=1024)] | None = None
|
|
|
|
|
|
class ControlActionStep(GestureModel):
|
|
action: Literal["control_action"]
|
|
request: SequenceControlRequest
|
|
|
|
|
|
class WaitUntilStep(WaitUntilRequest):
|
|
action: Literal["wait_until"]
|
|
timeout_ms: int = Field(default=1500, strict=True, ge=1, le=3000)
|
|
|
|
|
|
GestureStep = (
|
|
ClickStep
|
|
| DragStep
|
|
| WaitStep
|
|
| EnsureVisibleStep
|
|
| PinchStep
|
|
| TouchGestureStep
|
|
| ControlActionStep
|
|
| WaitUntilStep
|
|
)
|
|
|
|
|
|
class SequenceRequest(GestureModel):
|
|
steps: list[GestureStep] = Field(min_length=1, max_length=10)
|
|
feedback: ValueFeedback | None = None
|
|
|
|
|
|
def validate_sequence(payload: dict, limit: int) -> SequenceRequest:
|
|
steps = payload.get("steps")
|
|
if isinstance(steps, list) and len(steps) > limit:
|
|
raise ValueError(
|
|
f"Sequence has {len(steps)} members; this task allows at most {limit}, including waits. "
|
|
"No input executed. Split into complete short open/wait/operate groups; "
|
|
"do not fall back to separate model turns for auto-hiding targets."
|
|
)
|
|
return SequenceRequest.model_validate(payload)
|
|
|
|
|
|
def plan_drag(request: DragGeometry, width: int, height: int) -> dict:
|
|
start = pixel_point(request.start, request.coordinate_space, width, height)
|
|
calibration = None
|
|
if request.calibration is None:
|
|
if request.end is None:
|
|
raise ValueError("Drag requires end or calibration.")
|
|
end = pixel_point(request.end, request.coordinate_space, width, height)
|
|
else:
|
|
scale = request.calibration
|
|
axis = 0 if scale.axis == "x" else 1
|
|
size = width if axis == 0 else height
|
|
|
|
def position(raw: float) -> float:
|
|
upper = 1000 if request.coordinate_space == "normalized" else size - 1
|
|
if not 0 <= raw <= upper:
|
|
raise ValueError("Calibration positions must be inside the current screen.")
|
|
return (
|
|
min(size - 1, raw * size / 1000)
|
|
if request.coordinate_space == "normalized"
|
|
else raw
|
|
)
|
|
|
|
a, b = scale.anchors
|
|
p1, p2 = position(a.position), position(b.position)
|
|
if p1 == p2 or a.value == b.value:
|
|
raise ValueError("Calibration needs two distinct positions and values.")
|
|
slope = (b.value - a.value) / (p2 - p1)
|
|
if not math.isfinite(slope) or slope == 0:
|
|
raise ValueError("Calibration slope must be finite and nonzero.")
|
|
target_position = p1 + (scale.target_value - a.value) / slope
|
|
if not math.isfinite(target_position):
|
|
raise ValueError("Calibrated target must be finite.")
|
|
if scale.mode == "thumb":
|
|
if not min(a.value, b.value) <= scale.target_value <= max(a.value, b.value):
|
|
raise ValueError("Thumb target must be within the observed calibration range.")
|
|
endpoint = target_position
|
|
else:
|
|
if scale.indicator_position is None:
|
|
raise ValueError("Pan calibration requires indicator_position.")
|
|
endpoint = start[axis] + position(scale.indicator_position) - target_position
|
|
if not 0 <= endpoint <= size - 1:
|
|
raise ValueError(
|
|
"Calibrated endpoint is outside the screen; zoom or use a smaller move."
|
|
)
|
|
end = list(start)
|
|
end[axis] = round(endpoint)
|
|
calibration = {
|
|
**scale.model_dump(mode="json"),
|
|
"anchor_coordinate_space": request.coordinate_space,
|
|
"value_per_pixel": slope,
|
|
"target_position_pixel": target_position,
|
|
"rounding_error_value": abs(end[axis] - endpoint) * abs(slope),
|
|
"verification": "pending_observation",
|
|
}
|
|
calibration["movement_px"] = abs(end[axis] - start[axis])
|
|
if scale.tolerance is not None:
|
|
calibration["pixel_error_budget"] = scale.tolerance / abs(slope)
|
|
if calibration["rounding_error_value"] > scale.tolerance:
|
|
raise ValueError(
|
|
"Pixel rounding exceeds the requested tolerance; zoom and re-observe anchors before input."
|
|
)
|
|
if scale.min_movement_px is not None and calibration["movement_px"] < scale.min_movement_px:
|
|
raise ValueError(
|
|
"Planned movement is below the observed minimum effective distance; zoom and re-observe anchors. No input executed."
|
|
)
|
|
if start == end:
|
|
raise ValueError("Drag has no pixel movement; zoom further or choose a different endpoint.")
|
|
region = None
|
|
if request.region is not None:
|
|
region = pixel_region(request.region, request.coordinate_space, width, height)
|
|
require_in_region([start, end], region)
|
|
return {
|
|
"action": "drag",
|
|
"coordinates": [*start, *end],
|
|
"coordinate_space": "pixel",
|
|
"screen_size": [width, height],
|
|
"normalized_coordinates": [
|
|
round(value * 1000 / size)
|
|
for value, size in zip([*start, *end], [width, height, width, height], strict=True)
|
|
],
|
|
"duration_ms": request.duration_ms,
|
|
"target_description": request.target_description,
|
|
**({"region": region} if region is not None else {}),
|
|
**({"require_visible": request.require_visible} if request.require_visible else {}),
|
|
**({"calibration": calibration} if calibration else {}),
|
|
}
|
|
|
|
|
|
def plan_sequence(request: SequenceRequest, width: int, height: int, limit: int) -> list[dict]:
|
|
if len(request.steps) > limit:
|
|
raise ValueError(f"Sequence exceeds the configured fast-action limit of {limit} steps.")
|
|
planned = []
|
|
for step in request.steps:
|
|
if isinstance(step, DragStep):
|
|
planned.append(plan_drag(step, width, height))
|
|
elif isinstance(step, (PinchStep, TouchGestureStep)):
|
|
geometry = (
|
|
plan_pinch(step, width, height)
|
|
if isinstance(step, PinchStep)
|
|
else plan_touch(step, width, height)
|
|
)
|
|
planned.append(
|
|
{
|
|
"action": step.action,
|
|
"args": step.model_dump(mode="json", exclude={"action"}, exclude_none=True),
|
|
"plan": geometry,
|
|
}
|
|
)
|
|
elif isinstance(step, ControlActionStep):
|
|
planned.append(
|
|
{
|
|
"action": step.action,
|
|
"args": step.request.model_dump(mode="json", exclude_none=True),
|
|
}
|
|
)
|
|
elif isinstance(step, WaitUntilStep):
|
|
planned.append(
|
|
{
|
|
"action": step.action,
|
|
"args": step.model_dump(mode="json", exclude={"action"}, exclude_none=True),
|
|
}
|
|
)
|
|
elif isinstance(step, ClickStep):
|
|
planned.append(
|
|
{
|
|
"action": "click",
|
|
**(
|
|
{"target_resource_id": step.target}
|
|
if isinstance(step.target, str)
|
|
else {
|
|
"coordinates": pixel_point(
|
|
step.target, step.coordinate_space, width, height
|
|
)
|
|
}
|
|
),
|
|
"coordinate_space": "pixel",
|
|
"target_description": step.target_description,
|
|
**(
|
|
{"times": step.times, "delay_ms": step.delay_ms}
|
|
if step.times != 1 or step.delay_ms != 100
|
|
else {}
|
|
),
|
|
**({"unless_visible": step.unless_visible} if step.unless_visible else {}),
|
|
}
|
|
)
|
|
else:
|
|
planned.append(step.model_dump(mode="json"))
|
|
return planned
|