Coverage for integrations / remote_desktop / recipe_hooks.py: 88.1%
59 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Recipe Hooks — Capture remote desktop sessions as recipes and replay them.
4Extends the Recipe Pattern:
5 - Capture: Record a remote desktop action sequence as recipe steps
6 - Replay: Execute a saved recipe on a remote device (CREATE once → REUSE forever)
8Integration points:
9 - create_recipe.py (CREATE mode) — hook into action execution to record remote actions
10 - reuse_recipe.py (REUSE mode) — replay recorded remote actions on connected device
11"""
12import json
13import logging
14import time
15from typing import Any, Dict, List, Optional
17logger = logging.getLogger('hevolve.remote_desktop')
20class RemoteDesktopRecipeBridge:
21 """Bridge between remote desktop sessions and the HARTOS recipe system."""
23 def __init__(self):
24 self._recording = False
25 self._recorded_actions: List[Dict[str, Any]] = []
26 self._session_id: Optional[str] = None
28 def start_recording(self, session_id: str) -> None:
29 """Start recording remote desktop actions as recipe steps."""
30 self._recording = True
31 self._session_id = session_id
32 self._recorded_actions = []
33 logger.info(f"Recording remote desktop actions for session {session_id[:8]}")
35 def stop_recording(self) -> List[Dict[str, Any]]:
36 """Stop recording and return captured actions."""
37 self._recording = False
38 actions = self._recorded_actions.copy()
39 logger.info(f"Stopped recording: {len(actions)} actions captured")
40 return actions
42 def record_action(self, action: Dict[str, Any]) -> None:
43 """Record a single remote desktop action (called by input_handler)."""
44 if not self._recording:
45 return
46 step = {
47 'type': 'remote_desktop_action',
48 'action': action,
49 'timestamp': time.time(),
50 'session_id': self._session_id,
51 }
52 self._recorded_actions.append(step)
54 def capture_session_as_recipe(self, session_id: str,
55 actions: List[Dict[str, Any]]) -> dict:
56 """Convert a sequence of remote desktop actions into a recipe.
58 Args:
59 session_id: The remote desktop session that generated these actions
60 actions: List of action dicts (from input_handler event format)
62 Returns:
63 Recipe dict compatible with HARTOS recipe format.
64 """
65 steps = []
66 for i, action in enumerate(actions):
67 step = {
68 'step_id': i + 1,
69 'action_type': action.get('type', 'unknown'),
70 'parameters': {k: v for k, v in action.items() if k != 'type'},
71 'description': _describe_action(action),
72 }
73 steps.append(step)
75 recipe = {
76 'recipe_type': 'remote_desktop',
77 'session_id': session_id,
78 'steps': steps,
79 'step_count': len(steps),
80 'created_at': time.time(),
81 }
82 logger.info(f"Captured {len(steps)} steps as recipe from session {session_id[:8]}")
83 return recipe
85 def replay_recipe_on_device(self, recipe: dict,
86 device_id: Optional[str] = None,
87 password: Optional[str] = None,
88 delay: float = 0.5) -> dict:
89 """Replay a saved recipe — execute steps on this or a remote device.
91 Args:
92 recipe: Recipe dict from capture_session_as_recipe()
93 device_id: Remote device to connect to (None = local)
94 password: Password for remote connection
95 delay: Delay between steps in seconds
97 Returns:
98 {'success': bool, 'steps_executed': int, 'errors': list}
99 """
100 steps = recipe.get('steps', [])
101 if not steps:
102 return {'success': True, 'steps_executed': 0, 'errors': []}
104 errors = []
105 executed = 0
107 try:
108 from integrations.remote_desktop.input_handler import InputHandler
109 handler = InputHandler()
110 except Exception as e:
111 return {'success': False, 'steps_executed': 0,
112 'errors': [f'InputHandler unavailable: {e}']}
114 # Connect to remote if device_id provided
115 if device_id:
116 try:
117 from integrations.remote_desktop.rustdesk_bridge import get_rustdesk_bridge
118 bridge = get_rustdesk_bridge()
119 if bridge.available:
120 ok, msg = bridge.connect(device_id, password=password)
121 if not ok:
122 return {'success': False, 'steps_executed': 0,
123 'errors': [f'Connection failed: {msg}']}
124 time.sleep(2) # Wait for connection
125 except Exception as e:
126 errors.append(f'Connection warning: {e}')
128 for step in steps:
129 try:
130 action = {
131 'type': step.get('action_type', 'unknown'),
132 **step.get('parameters', {}),
133 }
134 result = handler.handle_input_event(action)
135 executed += 1
136 if delay > 0:
137 time.sleep(delay)
138 except Exception as e:
139 errors.append(f"Step {step.get('step_id', '?')}: {e}")
141 return {
142 'success': len(errors) == 0,
143 'steps_executed': executed,
144 'total_steps': len(steps),
145 'errors': errors,
146 }
149def _describe_action(action: dict) -> str:
150 """Human-readable description of a remote desktop action."""
151 action_type = action.get('type', 'unknown')
152 if action_type == 'click':
153 return f"Click at ({action.get('x')}, {action.get('y')})"
154 elif action_type in ('rightclick', 'doubleclick', 'middleclick'):
155 return f"{action_type.title()} at ({action.get('x')}, {action.get('y')})"
156 elif action_type == 'type':
157 text = action.get('text', '')
158 preview = text[:30] + '...' if len(text) > 30 else text
159 return f"Type: {preview!r}"
160 elif action_type == 'key':
161 return f"Key press: {action.get('key')}"
162 elif action_type == 'hotkey':
163 return f"Hotkey: {action.get('key')}"
164 elif action_type == 'scroll':
165 return f"Scroll ({action.get('delta_x', 0)}, {action.get('delta_y', 0)})"
166 elif action_type == 'move':
167 return f"Move cursor to ({action.get('x')}, {action.get('y')})"
168 elif action_type == 'drag':
169 return f"Drag from ({action.get('x')}, {action.get('y')}) to ({action.get('end_x')}, {action.get('end_y')})"
170 return f"Action: {action_type}"
173# ── Singleton ──────────────────────────────────────────────────
175_bridge: Optional[RemoteDesktopRecipeBridge] = None
178def get_recipe_bridge() -> RemoteDesktopRecipeBridge:
179 global _bridge
180 if _bridge is None:
181 _bridge = RemoteDesktopRecipeBridge()
182 return _bridge