|
1 | 1 | import json |
2 | 2 | import time |
3 | 3 | from typing import Any, Callable |
4 | | -from computers.kernel_computer import KernelComputer |
| 4 | +from computers.kernel_computer import ( |
| 5 | + KernelComputer, |
| 6 | + _describe_action, |
| 7 | + _describe_batch_actions, |
| 8 | +) |
5 | 9 | from utils import ( |
6 | 10 | create_response, |
7 | 11 | show_image, |
@@ -186,56 +190,6 @@ def _extract_prompt_text(self, item: dict[str, Any]) -> str | None: |
186 | 190 | parts.append(text) |
187 | 191 | return " ".join(parts) if parts else None |
188 | 192 |
|
189 | | - def _describe_action(self, action_type: str, action_args: dict[str, Any]) -> str: |
190 | | - if action_type == "click": |
191 | | - x = int(action_args.get("x", 0)) |
192 | | - y = int(action_args.get("y", 0)) |
193 | | - button = action_args.get("button", "left") |
194 | | - if button in ("", "left"): |
195 | | - return f"click({x}, {y})" |
196 | | - return f"click({x}, {y}, {button})" |
197 | | - if action_type == "double_click": |
198 | | - return f"double_click({int(action_args.get('x', 0))}, {int(action_args.get('y', 0))})" |
199 | | - if action_type == "type": |
200 | | - text = str(action_args.get("text", "")) |
201 | | - if len(text) > 60: |
202 | | - text = f"{text[:57]}..." |
203 | | - return f"type({text!r})" |
204 | | - if action_type == "keypress": |
205 | | - keys = action_args.get("keys", []) |
206 | | - hold_keys = action_args.get("hold_keys", []) |
207 | | - if hold_keys: |
208 | | - return f"keypress(hold={hold_keys}, keys={keys})" |
209 | | - return f"keypress({keys})" |
210 | | - if action_type == "scroll": |
211 | | - return ( |
212 | | - f"scroll({int(action_args.get('x', 0))}, {int(action_args.get('y', 0))}, " |
213 | | - f"dx={int(action_args.get('scroll_x', 0))}, dy={int(action_args.get('scroll_y', 0))})" |
214 | | - ) |
215 | | - if action_type == "move": |
216 | | - return f"move({int(action_args.get('x', 0))}, {int(action_args.get('y', 0))})" |
217 | | - if action_type == "drag": |
218 | | - return "drag(...)" |
219 | | - if action_type == "wait": |
220 | | - return f"wait({int(action_args.get('ms', 1000))}ms)" |
221 | | - if action_type == "goto": |
222 | | - return f"goto({action_args.get('url', '')!r})" |
223 | | - if action_type == "back": |
224 | | - return "back()" |
225 | | - if action_type == "url": |
226 | | - return "url()" |
227 | | - if action_type == "screenshot": |
228 | | - return "screenshot()" |
229 | | - return action_type |
230 | | - |
231 | | - def _describe_batch_actions(self, actions: list[dict[str, Any]]) -> str: |
232 | | - pieces: list[str] = [] |
233 | | - for action in actions: |
234 | | - action_type = str(action.get("type", "unknown")) |
235 | | - action_args = {k: v for k, v in action.items() if k != "type"} |
236 | | - pieces.append(self._describe_action(action_type, action_args)) |
237 | | - return "batch[" + " -> ".join(pieces) + "]" |
238 | | - |
239 | 193 | def _batch_terminal_read_action(self, actions: list[dict[str, Any]]) -> str: |
240 | 194 | if not actions: |
241 | 195 | return "" |
@@ -269,7 +223,7 @@ def handle_item(self, item): |
269 | 223 | typed_actions = [a for a in actions if isinstance(a, dict)] |
270 | 224 | payload = { |
271 | 225 | "action_type": "batch", |
272 | | - "description": self._describe_batch_actions(typed_actions), |
| 226 | + "description": _describe_batch_actions(typed_actions), |
273 | 227 | "action": {"type": "batch", "actions": typed_actions}, |
274 | 228 | } |
275 | 229 | if elapsed_ms is not None: |
@@ -315,14 +269,14 @@ def handle_item(self, item): |
315 | 269 | if len(typed_actions) == 1: |
316 | 270 | action_type = str(typed_actions[0].get("type", "unknown")) |
317 | 271 | action_payload: dict[str, Any] = typed_actions[0] |
318 | | - description = self._describe_action( |
| 272 | + description = _describe_action( |
319 | 273 | action_type, |
320 | 274 | {k: v for k, v in typed_actions[0].items() if k != "type"}, |
321 | 275 | ) |
322 | 276 | else: |
323 | 277 | action_type = "batch" |
324 | 278 | action_payload = {"type": "batch", "actions": typed_actions} |
325 | | - description = self._describe_batch_actions(typed_actions) |
| 279 | + description = _describe_batch_actions(typed_actions) |
326 | 280 |
|
327 | 281 | payload = { |
328 | 282 | "action_type": action_type, |
|
0 commit comments