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.
285 lines
12 KiB
Python
285 lines
12 KiB
Python
"""Optional, task-local feedback for numeric adjustments; never retries an input."""
|
|
|
|
import asyncio
|
|
import json
|
|
import math
|
|
import re
|
|
from collections.abc import Awaitable, Callable
|
|
from asyncio import sleep
|
|
from time import monotonic
|
|
from typing import Annotated, Literal
|
|
|
|
from pydantic import Field, model_validator
|
|
|
|
from artemis.mcp.action_parameters import GestureModel
|
|
from artemis.mcp.action_types import ActionCode, ActionResult
|
|
from artemis.mcp.native_actions import ControlSelector
|
|
from artemis.mcp.native_executor import NativeCaller, execute_native
|
|
|
|
|
|
class ValueFeedback(GestureModel):
|
|
selector: ControlSelector = Field(
|
|
description="Unique, observed persistent value readout, not a transient drag tooltip. Copy its stable selector, excluding changing text."
|
|
)
|
|
value_format: Literal["range", "number", "clock"] = Field(
|
|
description="range reads native range.current; number requires an entire numeric text; clock requires exactly one HH:MM:SS in readable text and returns seconds since midnight."
|
|
)
|
|
target_value: float
|
|
tolerance: Annotated[float, Field(ge=0)]
|
|
stable_ms: Annotated[int, Field(strict=True, ge=100, le=5000)] = 1000
|
|
timeout_ms: Annotated[int, Field(strict=True, ge=200, le=10000)] = 3000
|
|
|
|
@model_validator(mode="after")
|
|
def bounded_observation(self):
|
|
if self.stable_ms >= self.timeout_ms:
|
|
raise ValueError("stable_ms must be smaller than timeout_ms.")
|
|
return self
|
|
|
|
|
|
def read_value(node: dict, value_format: str) -> float:
|
|
if node.get("password") is True or node.get("visible") is not True:
|
|
raise ValueError("The value must be visible, readable and non-password.")
|
|
if value_format == "range":
|
|
native_range = node.get("range")
|
|
if not isinstance(native_range, dict):
|
|
raise ValueError("No native numeric range; do not guess a value from the gesture.")
|
|
values = [native_range.get(k) for k in ("min", "current", "max")]
|
|
if any(type(v) not in (int, float) or not math.isfinite(v) for v in values):
|
|
raise ValueError("Invalid native numeric range.")
|
|
low, current, high = values
|
|
if not low <= current <= high:
|
|
raise ValueError("Invalid native numeric range bounds.")
|
|
return float(current)
|
|
text = node.get("text")
|
|
if not isinstance(text, str) or len(text) > 4096:
|
|
raise ValueError("The selected node has no bounded readable text.")
|
|
if value_format == "clock":
|
|
matches = re.findall(r"(?<![\d:])([01]\d|2[0-3]):([0-5]\d):([0-5]\d)(?![\d:])", text)
|
|
if len(matches) != 1:
|
|
raise ValueError(
|
|
"clock requires exactly one unambiguous HH:MM:SS; select its persistent readout."
|
|
)
|
|
h, m, s = map(int, matches[0])
|
|
return float(h * 3600 + m * 60 + s)
|
|
if not re.fullmatch(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?", text.strip()):
|
|
raise ValueError("number requires an entire numeric text, without units or other numbers.")
|
|
value = float(text)
|
|
if not math.isfinite(value):
|
|
raise ValueError("The observed value must be finite.")
|
|
return value
|
|
|
|
|
|
def strategy_key(planned: list[dict]) -> str:
|
|
"""Ignore tiny coordinate jitter; a new scale, axis or meaningful span is a new strategy."""
|
|
strategy = []
|
|
for step in planned:
|
|
if step["action"] == "drag":
|
|
x1, y1, x2, y2 = step["coordinates"]
|
|
dx, dy = x2 - x1, y2 - y1
|
|
calibration = step.get("calibration", {})
|
|
strategy.append(
|
|
[
|
|
"drag",
|
|
"x" if abs(dx) >= abs(dy) else "y",
|
|
1 if (dx if abs(dx) >= abs(dy) else dy) > 0 else -1,
|
|
math.hypot(dx, dy),
|
|
step["duration_ms"],
|
|
round(calibration.get("value_per_pixel", 0), 8),
|
|
step.get("region"),
|
|
]
|
|
)
|
|
else:
|
|
strategy.append(step)
|
|
return json.dumps(strategy, sort_keys=True, ensure_ascii=False)
|
|
|
|
|
|
def same_strategy(left: str, right: str) -> bool:
|
|
"""Compare relative changes; crossing an arbitrary pixel bucket is not a new strategy."""
|
|
before, after = json.loads(left), json.loads(right)
|
|
if len(before) != len(after):
|
|
return False
|
|
for old, new in zip(before, after, strict=True):
|
|
if isinstance(old, list) and isinstance(new, list) and old[0] == new[0] == "drag":
|
|
if old[1:3] != new[1:3] or old[6] != new[6]:
|
|
return False
|
|
if any(not 0.5 < new[i] / old[i] < 2 for i in (3, 4)):
|
|
return False
|
|
if not math.isclose(old[5], new[5], rel_tol=0.05):
|
|
return False
|
|
elif old != new:
|
|
return False
|
|
return True
|
|
|
|
|
|
class ValueFeedbackRunner:
|
|
"""Bounded memory owned by one device actuator/task, never persisted or shared."""
|
|
|
|
def __init__(self):
|
|
self.history: dict[str, tuple[str, float, int]] = {}
|
|
|
|
async def _observe(
|
|
self, request: ValueFeedback, native: NativeCaller, *, stable: bool
|
|
) -> tuple[float, dict]:
|
|
deadline = monotonic() + request.timeout_ms / 1000
|
|
previous = None
|
|
since = monotonic()
|
|
samples = 0
|
|
async with asyncio.timeout(request.timeout_ms / 1000):
|
|
while monotonic() < deadline:
|
|
receipt = await execute_native(
|
|
"read_control",
|
|
{"selector": request.selector.model_dump(exclude_none=True)},
|
|
1,
|
|
1,
|
|
native,
|
|
)
|
|
if monotonic() >= deadline:
|
|
raise TimeoutError("The value observation arrived after the feedback deadline.")
|
|
if not receipt["success"]:
|
|
if stable and receipt.get("code") in {
|
|
"NOT_FOUND",
|
|
"READ_FAILED",
|
|
"UNAVAILABLE",
|
|
}:
|
|
previous = None
|
|
since = monotonic()
|
|
await sleep(min(0.1, max(0, deadline - monotonic())))
|
|
continue
|
|
raise ValueError(
|
|
f"Value read failed: {receipt.get('code')}; no value inferred."
|
|
)
|
|
value = read_value(receipt["node"], request.value_format)
|
|
samples += 1
|
|
evidence = {
|
|
"node": receipt["node"],
|
|
"samples": samples,
|
|
"stable_for_ms": round((monotonic() - since) * 1000)
|
|
if stable and value == previous
|
|
else 0,
|
|
}
|
|
if not stable:
|
|
return value, evidence
|
|
if value != previous:
|
|
previous, since = value, monotonic()
|
|
elif monotonic() - since >= request.stable_ms / 1000:
|
|
return value, evidence
|
|
await sleep(min(0.1, max(0, deadline - monotonic())))
|
|
raise TimeoutError("The value did not settle within the feedback deadline.")
|
|
|
|
async def run(
|
|
self,
|
|
request: ValueFeedback | None,
|
|
planned: list[dict],
|
|
action: str,
|
|
native: NativeCaller,
|
|
operation: Callable[[], Awaitable[ActionResult]],
|
|
) -> ActionResult:
|
|
if request is None:
|
|
return await operation()
|
|
key = json.dumps(
|
|
[
|
|
request.selector.model_dump(),
|
|
request.value_format,
|
|
request.target_value,
|
|
request.tolerance,
|
|
],
|
|
sort_keys=True,
|
|
)
|
|
strategy = strategy_key(planned)
|
|
try:
|
|
before, before_observation = await self._observe(request, native, stable=False)
|
|
except (ValueError, TimeoutError) as exc:
|
|
self.history.pop(key, None)
|
|
return ActionResult.failure(
|
|
action,
|
|
f"Feedback unavailable; no input executed: {exc}",
|
|
code=ActionCode.TARGET_NOT_FOUND,
|
|
metadata={
|
|
"execution": "not_started",
|
|
"side_effect": "none",
|
|
"verification": "unavailable",
|
|
},
|
|
)
|
|
previous = self.history.get(key)
|
|
stalled = (
|
|
previous[2]
|
|
if previous and previous[1] == before and same_strategy(previous[0], strategy)
|
|
else 0
|
|
)
|
|
feedback = {
|
|
"selector": request.selector.model_dump(exclude_none=True),
|
|
"before_value": before,
|
|
"before_observation": before_observation,
|
|
"target_value": request.target_value,
|
|
"tolerance": request.tolerance,
|
|
"value_format": request.value_format,
|
|
}
|
|
if stalled >= 2:
|
|
return ActionResult.failure(
|
|
action,
|
|
"NO_PROGRESS: two adjustments did not reduce the actual residual. No further input executed. Re-observe the gesture surface and scale; zoom, change the meaningful drag span or use a native setter. Do not repeat the same small drag.",
|
|
metadata={
|
|
"execution": "not_started",
|
|
"side_effect": "none",
|
|
"verification": "strategy_change_required",
|
|
"value_feedback": {**feedback, "consecutive_no_progress": stalled},
|
|
},
|
|
)
|
|
result = await operation()
|
|
if not result.ok:
|
|
self.history.pop(key, None)
|
|
return result
|
|
try:
|
|
after, after_observation = await self._observe(request, native, stable=True)
|
|
except (ValueError, TimeoutError) as exc:
|
|
self.history.pop(key, None)
|
|
return result.model_copy(
|
|
update={
|
|
"ok": False,
|
|
"code": ActionCode.TIMEOUT
|
|
if isinstance(exc, TimeoutError)
|
|
else ActionCode.TARGET_NOT_FOUND,
|
|
"message": f"Input completed; value feedback unavailable: {exc}. Do not replay the input blindly.",
|
|
"metadata": {
|
|
**result.metadata,
|
|
"execution": "completed",
|
|
"verification": "unavailable",
|
|
"value_feedback": feedback,
|
|
},
|
|
}
|
|
)
|
|
residual = after - request.target_value
|
|
matched = abs(residual) <= request.tolerance
|
|
improved = abs(residual) < abs(before - request.target_value) - 1e-9
|
|
stalled = 0 if matched or improved else stalled + 1
|
|
if len(self.history) >= 32:
|
|
self.history.pop(next(iter(self.history)))
|
|
self.history[key] = (strategy, after, stalled)
|
|
feedback.update(
|
|
after_observation=after_observation,
|
|
after_value=after,
|
|
residual=residual,
|
|
changed=after != before,
|
|
improved=improved,
|
|
within_tolerance=matched,
|
|
consecutive_no_progress=stalled,
|
|
)
|
|
return result.model_copy(
|
|
update={
|
|
"ok": matched,
|
|
"code": ActionCode.OK if matched else ActionCode.DEVICE_ERROR,
|
|
"message": f"Input completed. Observed value={after:g}, target={request.target_value:g}, residual={residual:g}, tolerance={request.tolerance:g}. "
|
|
+ (
|
|
"Value condition met; complete task acceptance separately."
|
|
if matched
|
|
else "Value condition not met. Use this observed residual to change strategy; never infer success from dispatch."
|
|
),
|
|
"metadata": {
|
|
**result.metadata,
|
|
"execution": "completed",
|
|
"verification": "value_matched" if matched else "value_not_reached",
|
|
"value_feedback": feedback,
|
|
},
|
|
}
|
|
)
|