318 lines
12 KiB
Python
318 lines
12 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,
|
|
pixel_point,
|
|
)
|
|
from artemis.mcp.native_actions import (
|
|
ControlActionRequest,
|
|
PinchRequest,
|
|
TouchGestureRequest,
|
|
WaitUntilRequest,
|
|
plan_pinch,
|
|
plan_touch,
|
|
)
|
|
|
|
|
|
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."
|
|
)
|
|
|
|
|
|
class DragRequest(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
|
|
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 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(DragRequest):
|
|
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)
|
|
|
|
|
|
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: DragRequest, 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",
|
|
}
|
|
if start == end:
|
|
raise ValueError("Drag has no pixel movement; zoom further or choose a different endpoint.")
|
|
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,
|
|
**({"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
|