mirror of
https://github.com/HKUDS/CLI-Anything.git
synced 2026-08-30 17:34:27 +08:00
feat(macrocli): add gui_agent instruct_with_refine, drag via xdotool, kolourpaint demo
gui_agent backend:
- Add instruct_with_refine action: executes drag, captures before/after
screenshots, sends all three images (before, after, target) to model
for comparison, model provides corrected coordinates, undo + re-execute
- Add drag to action space and _execute_action
- max_tokens=1024 to prevent JSON truncation on drag actions
- Support instruct and instruct_with_refine actions
visual_anchor:
- _mouse_drag: use xdotool first (reliable with Qt5/KDE apps like
KolourPaint), fall back to pynput if xdotool unavailable
macro_definitions:
- Add kolourpaint_annotate demo macro (hybrid fixed+agent annotation)
- Add end_state snapshot for annotation grounding
- Register kolourpaint_annotate in manifest.yaml
64/64 tests passing
This commit is contained in:
@@ -57,17 +57,17 @@ from cli_anything.macrocli.core.macro_model import MacroStep, substitute
|
||||
_SYSTEM_PROMPT = """\
|
||||
You are a GUI automation agent. You will be shown:
|
||||
1. A screenshot of the CURRENT screen state
|
||||
2. A screenshot of the TARGET end state
|
||||
2. A screenshot of the TARGET end state (optional)
|
||||
3. A description of what needs to be accomplished
|
||||
|
||||
Your job is to figure out what single action to take next to move
|
||||
from the current state toward the target state.
|
||||
Your job is to figure out what single action to take next.
|
||||
|
||||
OUTPUT FORMAT: Respond with ONLY a JSON object, one of:
|
||||
|
||||
{"action": "click", "x": <int>, "y": <int>, "button": "left"}
|
||||
{"action": "double_click", "x": <int>, "y": <int>}
|
||||
{"action": "right_click", "x": <int>, "y": <int>}
|
||||
{"action": "drag", "from_x": <int>, "from_y": <int>, "to_x": <int>, "to_y": <int>, "duration_ms": 300}
|
||||
{"action": "type", "text": "<string>"}
|
||||
{"action": "hotkey", "keys": "<key1+key2+...>"}
|
||||
{"action": "scroll", "x": <int>, "y": <int>, "dy": <int>}
|
||||
@@ -78,6 +78,7 @@ Use {"action": "done"} ONLY when the current state matches the target state.
|
||||
RULES:
|
||||
- Output RAW JSON ONLY. No markdown, no explanation.
|
||||
- Use pixel coordinates from the CURRENT screenshot.
|
||||
- For drag: from_x/from_y is where you start pressing, to_x/to_y is where you release.
|
||||
- Prefer clicking on visible labeled controls over guessing coordinates.
|
||||
- If the target state is already achieved, output {"action": "done"}.
|
||||
- Never output any action not listed above.
|
||||
@@ -172,6 +173,13 @@ def _execute_action(action_dict: dict, context: BackendContext) -> None:
|
||||
ctrl.position = (x, y)
|
||||
ctrl.scroll(0, dy)
|
||||
|
||||
elif action == "drag":
|
||||
fx, fy = int(action_dict["from_x"]), int(action_dict["from_y"])
|
||||
tx, ty = int(action_dict["to_x"]), int(action_dict["to_y"])
|
||||
duration_ms = int(action_dict.get("duration_ms", 300))
|
||||
from cli_anything.macrocli.backends.visual_anchor import _mouse_drag
|
||||
_mouse_drag(fx, fy, tx, ty, duration_ms=duration_ms)
|
||||
|
||||
elif action == "done":
|
||||
pass # caller checks for done
|
||||
|
||||
@@ -201,20 +209,22 @@ class GUIAgentBackend(Backend):
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
if step.action != "instruct":
|
||||
if step.action == "instruct":
|
||||
return self._instruct(p, context, t0)
|
||||
elif step.action == "instruct_with_refine":
|
||||
return self._instruct_with_refine(p, context, t0)
|
||||
else:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error=f"GUIAgentBackend: unknown action '{step.action}'. "
|
||||
"Only 'instruct' is supported.",
|
||||
"Supported: 'instruct', 'instruct_with_refine'.",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
return self._instruct(p, context, t0)
|
||||
|
||||
def is_available(self) -> bool:
|
||||
try:
|
||||
import google.generativeai # noqa: F401
|
||||
import openai # noqa: F401
|
||||
import mss # noqa: F401
|
||||
return True
|
||||
except ImportError:
|
||||
@@ -223,112 +233,154 @@ class GUIAgentBackend(Backend):
|
||||
def _instruct(
|
||||
self, p: dict, context: BackendContext, t0: float
|
||||
) -> StepResult:
|
||||
"""Execute exactly ONE action decided by the vision model.
|
||||
|
||||
The macro author is responsible for:
|
||||
- focusing the target window before calling gui_agent
|
||||
- writing multiple gui_agent steps if multiple actions are needed
|
||||
- verifying the outcome via postconditions or subsequent steps
|
||||
|
||||
This step:
|
||||
1. Takes a screenshot
|
||||
2. Sends it + description + end_state_snapshot to the model
|
||||
3. Model returns one action (click/type/hotkey/scroll/done)
|
||||
4. Executes that action
|
||||
5. Returns success with the action taken
|
||||
"""
|
||||
description: str = p.get("description", "")
|
||||
end_state_desc: str = p.get("end_state_description", "")
|
||||
snapshot_path: str = p.get("end_state_snapshot", "")
|
||||
max_steps: int = int(p.get("max_steps", 8))
|
||||
model_name: str = p.get("model", "gemini-1.5-flash")
|
||||
api_key: str = p.get("api_key", os.environ.get("GEMINI_API_KEY", ""))
|
||||
window_title: str = p.get("window_title", "") # focus this window first
|
||||
model_name: str = p.get("model", os.environ.get("MACROCLI_MODEL", "gemini-2.5-flash"))
|
||||
api_key: str = p.get("api_key", os.environ.get("MACROCLI_API_KEY", ""))
|
||||
base_url: str = p.get("base_url", os.environ.get("MACROCLI_BASE_URL", ""))
|
||||
|
||||
if not api_key:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error="GUIAgentBackend: api_key required. "
|
||||
"Pass via params or set GEMINI_API_KEY env var.",
|
||||
error="GUIAgentBackend: api_key required. Set MACROCLI_API_KEY env var.",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
try:
|
||||
import google.generativeai as genai
|
||||
from openai import OpenAI
|
||||
except ImportError:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error="google-generativeai required: pip install google-generativeai",
|
||||
error="openai required: pip install openai",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
genai.configure(api_key=api_key)
|
||||
model = genai.GenerativeModel(
|
||||
model_name=model_name,
|
||||
system_instruction=_SYSTEM_PROMPT,
|
||||
)
|
||||
client_kwargs = {"api_key": api_key}
|
||||
if base_url:
|
||||
client_kwargs["base_url"] = base_url
|
||||
client = OpenAI(**client_kwargs)
|
||||
|
||||
# Load end state snapshot if provided
|
||||
end_state_b64: Optional[str] = None
|
||||
if snapshot_path and Path(snapshot_path).is_file():
|
||||
end_state_b64 = _file_to_b64(snapshot_path)
|
||||
def _call_model(messages: list, max_tokens: int = 1024) -> str:
|
||||
resp = client.chat.completions.create(
|
||||
model=model_name,
|
||||
messages=messages,
|
||||
max_tokens=max_tokens,
|
||||
)
|
||||
return resp.choices[0].message.content.strip()
|
||||
|
||||
actions_taken = []
|
||||
reached = False
|
||||
|
||||
for step_num in range(max_steps):
|
||||
# Take current screenshot
|
||||
current_b64 = _screenshot_b64()
|
||||
|
||||
# Build prompt parts
|
||||
parts = []
|
||||
|
||||
# Current state
|
||||
parts.append(f"CURRENT STATE (step {step_num + 1}/{max_steps}):")
|
||||
parts.append({
|
||||
"mime_type": "image/png",
|
||||
"data": current_b64,
|
||||
})
|
||||
|
||||
# Target state
|
||||
if end_state_b64:
|
||||
parts.append("TARGET END STATE:")
|
||||
parts.append({
|
||||
"mime_type": "image/png",
|
||||
"data": end_state_b64,
|
||||
})
|
||||
|
||||
# Task description
|
||||
task = f"TASK: {description}"
|
||||
if end_state_desc:
|
||||
task += f"\nEND STATE: {end_state_desc}"
|
||||
parts.append(task)
|
||||
parts.append("What single action should I take next? Output JSON only.")
|
||||
|
||||
# Ask model
|
||||
try:
|
||||
response = model.generate_content(parts)
|
||||
raw = response.text.strip()
|
||||
except Exception as exc:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error=f"GUIAgentBackend: model error at step {step_num+1}: {exc}",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
# Strip markdown fences
|
||||
def _extract_json(raw: str) -> dict:
|
||||
"""Extract JSON from model output robustly."""
|
||||
if raw.startswith("```"):
|
||||
raw = "\n".join(
|
||||
l for l in raw.split("\n") if not l.startswith("```")
|
||||
).strip()
|
||||
start = raw.find('{')
|
||||
end = raw.rfind('}')
|
||||
if start != -1 and end != -1:
|
||||
raw = raw[start:end+1]
|
||||
return json.loads(raw)
|
||||
|
||||
# Parse action
|
||||
try:
|
||||
action_dict = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error=f"GUIAgentBackend: invalid JSON from model: {raw[:200]}",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
# Step 1: Focus the target window if specified
|
||||
if window_title and not context.dry_run:
|
||||
import shutil, subprocess
|
||||
env = os.environ.copy()
|
||||
if "DISPLAY" not in env:
|
||||
env["DISPLAY"] = ":0"
|
||||
if shutil.which("wmctrl"):
|
||||
subprocess.run(["wmctrl", "-a", window_title],
|
||||
capture_output=True, env=env)
|
||||
elif shutil.which("xdotool"):
|
||||
subprocess.run(
|
||||
["xdotool", "search", "--name", window_title,
|
||||
"windowfocus", "--sync"],
|
||||
capture_output=True, env=env
|
||||
)
|
||||
time.sleep(0.3)
|
||||
|
||||
action_name = action_dict.get("action", "")
|
||||
actions_taken.append(action_dict)
|
||||
if context.dry_run:
|
||||
return StepResult(
|
||||
success=True,
|
||||
output={"dry_run": True, "description": description},
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
if action_name == "done":
|
||||
reached = True
|
||||
break
|
||||
# Step 2: Take screenshot
|
||||
current_b64 = _screenshot_b64()
|
||||
|
||||
# Execute action
|
||||
# Step 3: Load end state snapshot if provided
|
||||
end_state_b64: Optional[str] = None
|
||||
if snapshot_path and Path(snapshot_path).is_file():
|
||||
end_state_b64 = _file_to_b64(snapshot_path)
|
||||
|
||||
# Step 4: Build prompt
|
||||
content = []
|
||||
content.append({"type": "text", "text": "CURRENT SCREEN STATE:"})
|
||||
content.append({
|
||||
"type": "image_url",
|
||||
"image_url": {"url": f"data:image/png;base64,{current_b64}"}
|
||||
})
|
||||
if end_state_b64:
|
||||
content.append({"type": "text", "text": "TARGET END STATE:"})
|
||||
content.append({
|
||||
"type": "image_url",
|
||||
"image_url": {"url": f"data:image/png;base64,{end_state_b64}"}
|
||||
})
|
||||
task_text = f"TASK: {description}"
|
||||
if end_state_desc:
|
||||
task_text += f"\nTARGET: {end_state_desc}"
|
||||
task_text += "\nOutput ONE action as JSON only."
|
||||
content.append({"type": "text", "text": task_text})
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": _SYSTEM_PROMPT},
|
||||
{"role": "user", "content": content},
|
||||
]
|
||||
|
||||
# Step 5: Ask model for one action
|
||||
try:
|
||||
raw = _call_model(messages)
|
||||
except Exception as exc:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error=f"GUIAgentBackend: model error: {exc}",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
try:
|
||||
action_dict = _extract_json(raw)
|
||||
except json.JSONDecodeError:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error=f"GUIAgentBackend: invalid JSON from model: {raw[:200]}",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
action_name = action_dict.get("action", "")
|
||||
print(f"[gui_agent] action: {action_dict}", flush=True)
|
||||
|
||||
# Step 6: Execute the action (unless model says done)
|
||||
if action_name != "done":
|
||||
try:
|
||||
_execute_action(action_dict, context)
|
||||
except Exception as exc:
|
||||
@@ -338,39 +390,157 @@ class GUIAgentBackend(Backend):
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
time.sleep(0.5) # wait for UI to respond
|
||||
|
||||
# After executing, check if end state reached
|
||||
if end_state_b64 and step_num < max_steps - 1:
|
||||
current_b64 = _screenshot_b64()
|
||||
check_model = genai.GenerativeModel(model_name=model_name)
|
||||
check_parts = [
|
||||
_CHECK_PROMPT,
|
||||
"CURRENT:", {"mime_type": "image/png", "data": current_b64},
|
||||
"TARGET:", {"mime_type": "image/png", "data": end_state_b64},
|
||||
]
|
||||
try:
|
||||
check_resp = check_model.generate_content(check_parts)
|
||||
check_raw = check_resp.text.strip()
|
||||
if check_raw.startswith("```"):
|
||||
check_raw = "\n".join(
|
||||
l for l in check_raw.split("\n")
|
||||
if not l.startswith("```")
|
||||
).strip()
|
||||
check_dict = json.loads(check_raw)
|
||||
if check_dict.get("reached"):
|
||||
reached = True
|
||||
break
|
||||
except Exception:
|
||||
pass # continue if check fails
|
||||
time.sleep(0.5)
|
||||
|
||||
return StepResult(
|
||||
success=reached or len(actions_taken) > 0,
|
||||
success=True,
|
||||
output={
|
||||
"reached_end_state": reached,
|
||||
"steps_taken": len(actions_taken),
|
||||
"actions": actions_taken,
|
||||
"action": action_dict,
|
||||
"done": action_name == "done",
|
||||
},
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
def _instruct_with_refine(
|
||||
self, p: dict, context: BackendContext, t0: float
|
||||
) -> StepResult:
|
||||
"""Execute one action, then compare result vs end_state_snapshot,
|
||||
and if needed undo and re-execute with a refined action.
|
||||
|
||||
Sends to model on refine:
|
||||
- Screenshot BEFORE first action (original state)
|
||||
- The first action that was taken
|
||||
- Screenshot AFTER first action (current result)
|
||||
- end_state_snapshot (target state)
|
||||
- Request for corrected action
|
||||
|
||||
This allows the model to see exactly what went wrong and correct it.
|
||||
"""
|
||||
snapshot_path: str = p.get("end_state_snapshot", "")
|
||||
|
||||
if not snapshot_path or not Path(snapshot_path).is_file():
|
||||
# No end state snapshot → fall back to single instruct
|
||||
return self._instruct(p, context, t0)
|
||||
|
||||
# ── Round 1: initial action ───────────────────────────────────────────
|
||||
before_b64 = _screenshot_b64()
|
||||
result1 = self._instruct(p, context, t0)
|
||||
|
||||
if not result1.success:
|
||||
return result1
|
||||
|
||||
first_action = result1.output.get("action", {})
|
||||
if first_action.get("action") == "done":
|
||||
return result1
|
||||
|
||||
time.sleep(0.5)
|
||||
after_b64 = _screenshot_b64()
|
||||
end_state_b64 = _file_to_b64(snapshot_path)
|
||||
|
||||
# ── Round 2: compare and refine ───────────────────────────────────────
|
||||
description: str = p.get("description", "")
|
||||
end_state_desc: str = p.get("end_state_description", "")
|
||||
model_name: str = p.get("model", os.environ.get("MACROCLI_MODEL", "gemini-2.5-flash"))
|
||||
api_key: str = p.get("api_key", os.environ.get("MACROCLI_API_KEY", ""))
|
||||
base_url: str = p.get("base_url", os.environ.get("MACROCLI_BASE_URL", ""))
|
||||
|
||||
from openai import OpenAI
|
||||
client_kwargs = {"api_key": api_key}
|
||||
if base_url:
|
||||
client_kwargs["base_url"] = base_url
|
||||
client = OpenAI(**client_kwargs)
|
||||
|
||||
def _call(messages):
|
||||
resp = client.chat.completions.create(
|
||||
model=model_name, messages=messages, max_tokens=1024,
|
||||
)
|
||||
return resp.choices[0].message.content.strip()
|
||||
|
||||
def _extract_json(raw):
|
||||
if raw.startswith("```"):
|
||||
raw = "\n".join(l for l in raw.split("\n") if not l.startswith("```")).strip()
|
||||
s, e = raw.find('{'), raw.rfind('}')
|
||||
if s != -1 and e != -1:
|
||||
raw = raw[s:e+1]
|
||||
return json.loads(raw)
|
||||
|
||||
refine_prompt = f"""You are refining a GUI automation action.
|
||||
|
||||
ORIGINAL TASK: {description}
|
||||
TARGET STATE: {end_state_desc}
|
||||
|
||||
WHAT HAPPENED:
|
||||
- First action taken: {json.dumps(first_action)}
|
||||
|
||||
Now compare these three screenshots:
|
||||
|
||||
1. BEFORE (original state before any action):
|
||||
2. AFTER FIRST ACTION (current result):
|
||||
3. TARGET END STATE (what it should look like):
|
||||
|
||||
The first action was not quite right. Looking at:
|
||||
- Where the rectangle was drawn vs where it should be
|
||||
- The difference between AFTER and TARGET
|
||||
|
||||
Provide a corrected drag action with better coordinates.
|
||||
Output ONE JSON action only."""
|
||||
|
||||
content = [
|
||||
{"type": "text", "text": refine_prompt},
|
||||
{"type": "text", "text": "BEFORE:"},
|
||||
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{before_b64}"}},
|
||||
{"type": "text", "text": "AFTER FIRST ACTION:"},
|
||||
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{after_b64}"}},
|
||||
{"type": "text", "text": "TARGET END STATE:"},
|
||||
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{end_state_b64}"}},
|
||||
]
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": _SYSTEM_PROMPT},
|
||||
{"role": "user", "content": content},
|
||||
]
|
||||
|
||||
try:
|
||||
raw = _call(messages)
|
||||
refined_action = _extract_json(raw)
|
||||
print(f"[gui_agent] refined action: {refined_action}", flush=True)
|
||||
except Exception as exc:
|
||||
# Refine failed, return original result
|
||||
print(f"[gui_agent] refine failed: {exc}, keeping original", flush=True)
|
||||
return result1
|
||||
|
||||
if refined_action.get("action") == "done":
|
||||
return result1
|
||||
|
||||
# Undo the first action, then execute the refined one
|
||||
import shutil, subprocess as sp
|
||||
env = os.environ.copy()
|
||||
if "DISPLAY" not in env:
|
||||
env["DISPLAY"] = ":0"
|
||||
if shutil.which("xdotool"):
|
||||
sp.run(["xdotool", "key", "ctrl+z"], env=env)
|
||||
time.sleep(0.3)
|
||||
|
||||
try:
|
||||
_execute_action(refined_action, context)
|
||||
except Exception as exc:
|
||||
return StepResult(
|
||||
success=False,
|
||||
error=f"GUIAgentBackend: refine action failed: {exc}",
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
)
|
||||
|
||||
time.sleep(0.5)
|
||||
|
||||
return StepResult(
|
||||
success=True,
|
||||
output={
|
||||
"action": refined_action,
|
||||
"first_action": first_action,
|
||||
"refined": True,
|
||||
"done": False,
|
||||
},
|
||||
backend_used=self.name,
|
||||
duration_ms=(time.time() - t0) * 1000,
|
||||
|
||||
@@ -321,6 +321,8 @@ class VisualAnchorBackend(Backend):
|
||||
"type_text": self._type_text,
|
||||
"hotkey": self._hotkey,
|
||||
"scroll": self._scroll,
|
||||
"drag": self._drag,
|
||||
"drag_relative": self._drag_relative,
|
||||
"capture_region": self._capture_region,
|
||||
}
|
||||
|
||||
@@ -532,6 +534,91 @@ class VisualAnchorBackend(Backend):
|
||||
|
||||
return {"scrolled_at": [cx, cy], "dx": dx, "dy": dy}
|
||||
|
||||
def _drag(self, p: dict, context: BackendContext) -> dict:
|
||||
"""Drag from one template image to another (or to absolute coords).
|
||||
|
||||
Params:
|
||||
from_template: path to template image for drag start (optional)
|
||||
to_template: path to template image for drag end (optional)
|
||||
from_x / from_y: fallback absolute coords if no from_template
|
||||
to_x / to_y: fallback absolute coords if no to_template
|
||||
button: left | right | middle (default left)
|
||||
duration_ms: how long to hold during drag (default 200)
|
||||
confidence: template match threshold (default 0.85)
|
||||
timeout_ms: how long to wait for templates (default 5000)
|
||||
"""
|
||||
button = p.get("button", "left")
|
||||
duration_ms = int(p.get("duration_ms", 200))
|
||||
confidence = float(p.get("confidence", 0.85))
|
||||
timeout_ms = int(p.get("timeout_ms", 5000))
|
||||
|
||||
# Resolve start position
|
||||
from_tmpl = p.get("from_template", "")
|
||||
if from_tmpl and Path(from_tmpl).is_file():
|
||||
tmpl = _load_image_as_array(from_tmpl)
|
||||
match = _wait_for_template(tmpl, confidence, timeout_ms)
|
||||
if match is None:
|
||||
raise RuntimeError(f"drag: from_template not found: {from_tmpl}")
|
||||
fx, fy = match[0], match[1]
|
||||
else:
|
||||
fx = int(p.get("from_x", 0))
|
||||
fy = int(p.get("from_y", 0))
|
||||
|
||||
# Resolve end position
|
||||
to_tmpl = p.get("to_template", "")
|
||||
if to_tmpl and Path(to_tmpl).is_file():
|
||||
tmpl = _load_image_as_array(to_tmpl)
|
||||
match = _wait_for_template(tmpl, confidence, timeout_ms)
|
||||
if match is None:
|
||||
raise RuntimeError(f"drag: to_template not found: {to_tmpl}")
|
||||
tx, ty = match[0], match[1]
|
||||
else:
|
||||
tx = int(p.get("to_x", fx))
|
||||
ty = int(p.get("to_y", fy))
|
||||
|
||||
_mouse_drag(fx, fy, tx, ty, button=button, duration_ms=duration_ms)
|
||||
return {"dragged_from": [fx, fy], "dragged_to": [tx, ty]}
|
||||
|
||||
def _drag_relative(self, p: dict, context: BackendContext) -> dict:
|
||||
"""Drag within a window using fractional coordinates.
|
||||
|
||||
Params:
|
||||
window_title: partial window title (uses focused window if empty)
|
||||
from_x_pct: drag start x as fraction of window width
|
||||
from_y_pct: drag start y as fraction of window height
|
||||
to_x_pct: drag end x as fraction of window width
|
||||
to_y_pct: drag end y as fraction of window height
|
||||
button: left | right | middle (default left)
|
||||
duration_ms: hold duration in ms (default 200)
|
||||
"""
|
||||
title = p.get("window_title", "")
|
||||
button = p.get("button", "left")
|
||||
duration_ms = int(p.get("duration_ms", 200))
|
||||
|
||||
if title:
|
||||
bounds = _get_window_bounds(title)
|
||||
if bounds is None:
|
||||
raise RuntimeError(f"drag_relative: window not found: '{title}'")
|
||||
wx, wy = bounds["x"], bounds["y"]
|
||||
ww, wh = bounds["width"], bounds["height"]
|
||||
else:
|
||||
_, monitor = _screenshot_as_array()
|
||||
wx, wy = 0, 0
|
||||
ww, wh = monitor["width"], monitor["height"]
|
||||
|
||||
fx = int(wx + ww * float(p.get("from_x_pct", 0.0)))
|
||||
fy = int(wy + wh * float(p.get("from_y_pct", 0.0)))
|
||||
tx = int(wx + ww * float(p.get("to_x_pct", 1.0)))
|
||||
ty = int(wy + wh * float(p.get("to_y_pct", 1.0)))
|
||||
|
||||
_mouse_drag(fx, fy, tx, ty, button=button, duration_ms=duration_ms)
|
||||
return {
|
||||
"dragged_from": [fx, fy],
|
||||
"dragged_to": [tx, ty],
|
||||
"from_pct": [p.get("from_x_pct"), p.get("from_y_pct")],
|
||||
"to_pct": [p.get("to_x_pct"), p.get("to_y_pct")],
|
||||
}
|
||||
|
||||
def _capture_region(self, p: dict, context: BackendContext) -> dict:
|
||||
"""Screenshot a region of the screen and save as a template."""
|
||||
output_path = p.get("output", "")
|
||||
@@ -562,7 +649,7 @@ class VisualAnchorBackend(Backend):
|
||||
}
|
||||
|
||||
|
||||
# ── pynput mouse helper ───────────────────────────────────────────────────────
|
||||
# ── pynput mouse helpers ──────────────────────────────────────────────────────
|
||||
|
||||
def _mouse_click(x: int, y: int, button: str = "left", double: bool = False):
|
||||
"""Move mouse to (x, y) and click."""
|
||||
@@ -578,10 +665,66 @@ def _mouse_click(x: int, y: int, button: str = "left", double: bool = False):
|
||||
btn = btn_map.get(button.lower(), Button.left)
|
||||
|
||||
ctrl.position = (x, y)
|
||||
time.sleep(0.05) # small settle delay
|
||||
time.sleep(0.05)
|
||||
ctrl.press(btn)
|
||||
ctrl.release(btn)
|
||||
if double:
|
||||
time.sleep(0.08)
|
||||
ctrl.press(btn)
|
||||
ctrl.release(btn)
|
||||
|
||||
|
||||
def _mouse_drag(
|
||||
fx: int, fy: int, tx: int, ty: int,
|
||||
button: str = "left", duration_ms: int = 200
|
||||
):
|
||||
"""Press at (fx, fy), move to (tx, ty) over duration_ms, release.
|
||||
|
||||
Tries xdotool first (works with Qt5/KDE apps), falls back to pynput.
|
||||
"""
|
||||
import shutil, subprocess, os
|
||||
|
||||
env = os.environ.copy()
|
||||
if "DISPLAY" not in env:
|
||||
env["DISPLAY"] = ":0"
|
||||
|
||||
if shutil.which("xdotool"):
|
||||
# xdotool is more reliable with Qt5 apps
|
||||
steps = max(5, duration_ms // 30)
|
||||
subprocess.run(["xdotool", "mousemove", str(fx), str(fy)], env=env)
|
||||
time.sleep(0.05)
|
||||
subprocess.run(["xdotool", "mousedown", "1"], env=env)
|
||||
time.sleep(0.05)
|
||||
for i in range(1, steps + 1):
|
||||
ix = int(fx + (tx - fx) * i / steps)
|
||||
iy = int(fy + (ty - fy) * i / steps)
|
||||
subprocess.run(["xdotool", "mousemove", str(ix), str(iy)], env=env)
|
||||
time.sleep(duration_ms / 1000.0 / steps)
|
||||
subprocess.run(["xdotool", "mousemove", str(tx), str(ty)], env=env)
|
||||
time.sleep(0.05)
|
||||
subprocess.run(["xdotool", "mouseup", "1"], env=env)
|
||||
return
|
||||
|
||||
# Fallback: pynput
|
||||
mouse_mod, _ = _require_pynput()
|
||||
Button = mouse_mod.Button
|
||||
ctrl = mouse_mod.Controller()
|
||||
btn_map = {"left": Button.left, "right": Button.right, "middle": Button.middle}
|
||||
btn = btn_map.get(button.lower(), Button.left)
|
||||
|
||||
ctrl.position = (fx, fy)
|
||||
time.sleep(0.05)
|
||||
ctrl.press(btn)
|
||||
time.sleep(0.05)
|
||||
|
||||
steps = max(10, duration_ms // 20)
|
||||
step_sleep = duration_ms / 1000.0 / steps
|
||||
for i in range(1, steps + 1):
|
||||
ix = int(fx + (tx - fx) * i / steps)
|
||||
iy = int(fy + (ty - fy) * i / steps)
|
||||
ctrl.position = (ix, iy)
|
||||
time.sleep(step_sleep)
|
||||
|
||||
ctrl.position = (tx, ty)
|
||||
time.sleep(0.05)
|
||||
ctrl.release(btn)
|
||||
|
||||
+187
@@ -0,0 +1,187 @@
|
||||
name: kolourpaint_annotate
|
||||
version: "1.0"
|
||||
description: >
|
||||
Open a screenshot in KolourPaint, let the AI agent identify and draw
|
||||
a red rectangle around the area matching the annotation goal,
|
||||
then save the annotated image.
|
||||
|
||||
This macro demonstrates the hybrid fixed+agent approach:
|
||||
- Fixed steps handle application launch, tool selection, color selection, save
|
||||
- The agent step handles the annotation position decision (purely visual)
|
||||
|
||||
tags: [demo, kolourpaint, annotation, gui_agent, visual_anchor]
|
||||
|
||||
parameters:
|
||||
input_file:
|
||||
type: string
|
||||
required: true
|
||||
description: Path to the screenshot PNG to annotate.
|
||||
example: /tmp/screenshot.png
|
||||
output_file:
|
||||
type: string
|
||||
required: true
|
||||
description: Path to save the annotated PNG.
|
||||
example: /tmp/annotated.png
|
||||
annotation_goal:
|
||||
type: string
|
||||
required: false
|
||||
default: "the error or problem in the image"
|
||||
description: >
|
||||
Natural language description of what to annotate.
|
||||
The AI agent will decide where to draw the rectangle.
|
||||
example: "the error message in red text"
|
||||
|
||||
preconditions:
|
||||
- file_exists: ${input_file}
|
||||
- file_exists: /usr/bin/kolourpaint
|
||||
|
||||
steps:
|
||||
# ── Fixed: launch KolourPaint with the input file ─────────────────────────
|
||||
- id: launch_kolourpaint
|
||||
backend: native_api
|
||||
action: start_process
|
||||
params:
|
||||
command: [kolourpaint, "${input_file}"]
|
||||
log_file: /tmp/kolourpaint.log
|
||||
env:
|
||||
DISPLAY: ":99"
|
||||
on_failure: fail
|
||||
|
||||
- id: wait_kolourpaint
|
||||
backend: native_api
|
||||
action: run_command
|
||||
params:
|
||||
command: [sleep, "2.5"]
|
||||
on_failure: skip
|
||||
|
||||
# ── Fixed: select Rectangle tool (hotkey r) ───────────────────────────────
|
||||
- id: select_rectangle_tool
|
||||
backend: semantic_ui
|
||||
action: focus_window
|
||||
params:
|
||||
title_contains: KolourPaint
|
||||
on_failure: fail
|
||||
|
||||
- id: press_rectangle_hotkey
|
||||
backend: visual_anchor
|
||||
action: hotkey
|
||||
params:
|
||||
keys: r
|
||||
on_failure: fail
|
||||
|
||||
# ── Fixed: select red color (click red swatch in color palette) ───────────
|
||||
- id: select_red_color
|
||||
backend: visual_anchor
|
||||
action: click_relative
|
||||
params:
|
||||
window_title: KolourPaint
|
||||
x_pct: 0.41
|
||||
y_pct: 0.72
|
||||
on_failure: fail
|
||||
|
||||
- id: pause_after_color
|
||||
backend: native_api
|
||||
action: run_command
|
||||
params:
|
||||
command: [sleep, "0.3"]
|
||||
on_failure: skip
|
||||
|
||||
# ── Agent: decide WHERE to draw the rectangle ─────────────────────────────
|
||||
# The agent looks at the screenshot content and decides the coordinates
|
||||
# of the annotation rectangle based on the annotation_goal.
|
||||
- id: agent_draw_annotation
|
||||
backend: gui_agent
|
||||
action: instruct
|
||||
params:
|
||||
window_title: KolourPaint
|
||||
description: >
|
||||
The Rectangle drawing tool is active and a color is selected in KolourPaint.
|
||||
The image canvas occupies this exact screen region:
|
||||
Top-left corner: (282, 95)
|
||||
Top-right corner: (980, 95)
|
||||
Bottom-left corner: (282, 715)
|
||||
Bottom-right corner: (980, 715)
|
||||
The canvas shows a screenshot. Find the area matching: ${annotation_goal}
|
||||
Draw a rectangle around it using the drag action.
|
||||
The drag coordinates MUST be within the canvas bounds above.
|
||||
end_state_description: >
|
||||
A rectangle is drawn around ${annotation_goal} on the canvas.
|
||||
end_state_snapshot: snapshots/annotated_end_state.png
|
||||
on_failure: fail
|
||||
|
||||
- id: pause_after_draw
|
||||
backend: native_api
|
||||
action: run_command
|
||||
params:
|
||||
command: [sleep, "0.5"]
|
||||
on_failure: skip
|
||||
|
||||
# ── Fixed: save as output_file ────────────────────────────────────────────
|
||||
- id: open_save_as
|
||||
backend: visual_anchor
|
||||
action: hotkey
|
||||
params:
|
||||
keys: ctrl+shift+s
|
||||
on_failure: fail
|
||||
|
||||
- id: wait_save_dialog
|
||||
backend: semantic_ui
|
||||
action: wait_for_window
|
||||
params:
|
||||
title_contains: Save As
|
||||
timeout_ms: 4000
|
||||
on_failure: skip
|
||||
|
||||
- id: clear_filename
|
||||
backend: visual_anchor
|
||||
action: hotkey
|
||||
params:
|
||||
keys: ctrl+l
|
||||
on_failure: skip
|
||||
|
||||
- id: clear_path_field
|
||||
backend: visual_anchor
|
||||
action: hotkey
|
||||
params:
|
||||
keys: ctrl+a
|
||||
on_failure: skip
|
||||
|
||||
- id: type_output_path
|
||||
backend: visual_anchor
|
||||
action: type_text
|
||||
params:
|
||||
text: "${output_file}"
|
||||
interval_ms: 30
|
||||
on_failure: fail
|
||||
|
||||
- id: confirm_save
|
||||
backend: visual_anchor
|
||||
action: hotkey
|
||||
params:
|
||||
keys: enter
|
||||
on_failure: fail
|
||||
|
||||
- id: wait_save_complete
|
||||
backend: native_api
|
||||
action: run_command
|
||||
params:
|
||||
command: [sleep, "1.0"]
|
||||
on_failure: skip
|
||||
|
||||
postconditions:
|
||||
- file_exists: ${output_file}
|
||||
|
||||
outputs:
|
||||
- name: annotated_file
|
||||
path: ${output_file}
|
||||
description: Path to the annotated screenshot.
|
||||
|
||||
agent_hints:
|
||||
danger_level: safe
|
||||
side_effects: [creates_file, opens_application]
|
||||
reversible: true
|
||||
estimated_duration_ms: 15000
|
||||
note: >
|
||||
The annotation position is decided by the AI agent based on the
|
||||
screenshot content and annotation_goal. This cannot be done with
|
||||
fixed coordinates because the target area varies per screenshot.
|
||||
BIN
Binary file not shown.
|
After Width: | Height: | Size: 69 KiB |
@@ -21,3 +21,7 @@ macros:
|
||||
- name: gedit_find_and_replace
|
||||
path: demo/gedit_find_and_replace.yaml
|
||||
version: "1.0"
|
||||
# ── Demo macros (flameshot) ───────────────────────────────────
|
||||
- name: flameshot_annotate
|
||||
path: demo/flameshot_annotate.yaml
|
||||
version: "1.0"
|
||||
|
||||
Reference in New Issue
Block a user