Coverage for integrations / remote_desktop / engine_selector.py: 88.1%

1""" 

2Engine Selector — Auto-selects RustDesk or Sunshine/Moonlight based on context. 

3 

4Selection logic: 

5 - File transfer needed → RustDesk (Sunshine has no file transfer) 

6 - VLM agent / high-FPS needed → Sunshine+Moonlight (hardware encoding, <10ms) 

7 - Gaming → Sunshine+Moonlight 

8 - General remote support → RustDesk (full-featured AnyDesk replacement) 

9 - Fallback → HARTOS native (frame_capture + transport) when neither installed 

10 

11Also provides unified status across all engines. 

12""" 

13 

14import logging 

15from enum import Enum 

16from typing import Dict, List, Optional 

17 

18logger = logging.getLogger('hevolve.remote_desktop') 

19 

20 

21class Engine(Enum): 

22 RUSTDESK = 'rustdesk' 

23 SUNSHINE = 'sunshine' # Host side 

24 MOONLIGHT = 'moonlight' # Viewer side 

25 NATIVE = 'native' # HARTOS built-in fallback 

26 

27 

28class UseCase(Enum): 

29 REMOTE_SUPPORT = 'remote_support' 

30 FILE_TRANSFER = 'file_transfer' 

31 VLM_COMPUTER_USE = 'vlm_computer_use' 

32 GAMING = 'gaming' 

33 PERIPHERAL_FORWARD = 'peripheral_forward' 

34 SCREEN_CAST = 'screen_cast' 

35 GENERAL = 'general' 

36 

37 

38# ── Engine availability cache ─────────────────────────────────── 

39 

40_availability_cache: Optional[Dict[str, bool]] = None 

41 

42 

43def _detect_engines() -> Dict[str, bool]: 

44 """Detect which remote desktop engines are installed.""" 

45 global _availability_cache 

46 if _availability_cache is not None: 

47 return _availability_cache 

48 

49 result = {} 

50 

51 try: 

52 from integrations.remote_desktop.rustdesk_bridge import get_rustdesk_bridge 

53 result['rustdesk'] = get_rustdesk_bridge().available 

54 except Exception: 

55 result['rustdesk'] = False 

56 

57 try: 

58 from integrations.remote_desktop.sunshine_bridge import ( 

59 get_sunshine_bridge, get_moonlight_bridge, 

60 ) 

61 result['sunshine'] = get_sunshine_bridge().available 

62 result['moonlight'] = get_moonlight_bridge().available 

63 except Exception: 

64 result['sunshine'] = False 

65 result['moonlight'] = False 

66 

67 # Native fallback is always available 

68 result['native'] = True 

69 

70 _availability_cache = result 

71 logger.info(f"Remote desktop engines: {result}") 

72 return result 

73 

74 

75def reset_cache() -> None: 

76 """Reset engine detection cache (e.g., after install).""" 

77 global _availability_cache 

78 _availability_cache = None 

79 

80 

81def select_engine(use_case: UseCase = UseCase.GENERAL, 

82 role: str = 'viewer', 

83 prefer: Optional[Engine] = None) -> Engine: 

84 """Select the best remote desktop engine for the given context. 

85 

86 Args: 

87 use_case: What the remote desktop will be used for 

88 role: 'host' (sharing screen) or 'viewer' (connecting) 

89 prefer: User preference override 

90 

91 Returns: 

92 Best available Engine for the context. 

93 """ 

94 engines = _detect_engines() 

95 

96 # User preference takes priority 

97 if prefer: 

98 if prefer == Engine.RUSTDESK and engines.get('rustdesk'): 

99 return Engine.RUSTDESK 

100 if prefer == Engine.SUNSHINE and engines.get('sunshine'): 

101 return Engine.SUNSHINE 

102 if prefer == Engine.MOONLIGHT and engines.get('moonlight'): 

103 return Engine.MOONLIGHT 

104 

105 # Use-case-based selection 

106 if use_case == UseCase.FILE_TRANSFER: 

107 # RustDesk has file transfer; Sunshine does not 

108 if engines.get('rustdesk'): 

109 return Engine.RUSTDESK 

110 return Engine.NATIVE 

111 

112 if use_case in (UseCase.VLM_COMPUTER_USE, UseCase.GAMING): 

113 # Sunshine+Moonlight have best streaming quality 

114 if role == 'host' and engines.get('sunshine'): 

115 return Engine.SUNSHINE 

116 if role == 'viewer' and engines.get('moonlight'): 

117 return Engine.MOONLIGHT 

118 # Fall back to RustDesk 

119 if engines.get('rustdesk'): 

120 return Engine.RUSTDESK 

121 return Engine.NATIVE 

122 

123 if use_case == UseCase.REMOTE_SUPPORT: 

124 # RustDesk is the full AnyDesk replacement 

125 if engines.get('rustdesk'): 

126 return Engine.RUSTDESK 

127 return Engine.NATIVE 

128 

129 # General: prefer RustDesk (most complete) 

130 if engines.get('rustdesk'): 

131 return Engine.RUSTDESK 

132 # Then Sunshine/Moonlight 

133 if role == 'host' and engines.get('sunshine'): 

134 return Engine.SUNSHINE 

135 if role == 'viewer' and engines.get('moonlight'): 

136 return Engine.MOONLIGHT 

137 

138 return Engine.NATIVE 

139 

140 

141def get_all_status() -> dict: 

142 """Get status of all remote desktop engines.""" 

143 status = {'engines': {}} 

144 

145 try: 

146 from integrations.remote_desktop.rustdesk_bridge import get_rustdesk_bridge 

147 status['engines']['rustdesk'] = get_rustdesk_bridge().get_status() 

148 except Exception as e: 

149 status['engines']['rustdesk'] = {'available': False, 'error': str(e)} 

150 

151 try: 

152 from integrations.remote_desktop.sunshine_bridge import ( 

153 get_sunshine_bridge, get_moonlight_bridge, 

154 ) 

155 status['engines']['sunshine'] = get_sunshine_bridge().get_status() 

156 status['engines']['moonlight'] = get_moonlight_bridge().get_status() 

157 except Exception as e: 

158 status['engines']['sunshine'] = {'available': False, 'error': str(e)} 

159 status['engines']['moonlight'] = {'available': False, 'error': str(e)} 

160 

161 status['engines']['native'] = { 

162 'available': True, 

163 'engine': 'native', 

164 'description': 'HARTOS built-in (frame_capture + transport)', 

165 } 

166 

167 # Recommend engines based on availability 

168 engines = _detect_engines() 

169 recommendations = [] 

170 if not engines.get('rustdesk'): 

171 try: 

172 from integrations.remote_desktop.rustdesk_bridge import RustDeskBridge 

173 recommendations.append({ 

174 'engine': 'rustdesk', 

175 'reason': 'General remote desktop (AnyDesk replacement)', 

176 'install': RustDeskBridge().get_install_command(), 

177 }) 

178 except Exception: 

179 pass 

180 if not engines.get('sunshine'): 

181 try: 

182 from integrations.remote_desktop.sunshine_bridge import SunshineBridge 

183 recommendations.append({ 

184 'engine': 'sunshine', 

185 'reason': 'High-fidelity streaming (gaming, VLM, creative)', 

186 'install': SunshineBridge().get_install_command(), 

187 }) 

188 except Exception: 

189 pass 

190 if not engines.get('moonlight'): 

191 try: 

192 from integrations.remote_desktop.sunshine_bridge import MoonlightBridge 

193 recommendations.append({ 

194 'engine': 'moonlight', 

195 'reason': 'Viewer for Sunshine streams (4K@120fps)', 

196 'install': MoonlightBridge().get_install_command(), 

197 }) 

198 except Exception: 

199 pass 

200 

201 status['install_recommendations'] = recommendations 

202 return status 

203 

204 

205def get_available_engines() -> List[Engine]: 

206 """Get list of available engines.""" 

207 engines = _detect_engines() 

208 result = [] 

209 if engines.get('rustdesk'): 

210 result.append(Engine.RUSTDESK) 

211 if engines.get('sunshine'): 

212 result.append(Engine.SUNSHINE) 

213 if engines.get('moonlight'): 

214 result.append(Engine.MOONLIGHT) 

215 result.append(Engine.NATIVE) 

216 return result 

217 

218 

219def recommend_engine_switch(current_engine: str, 

220 context: Optional[Dict] = None) -> Optional[Dict]: 

221 """AI-native: Recommend switching to a better engine based on context. 

222 

223 Args: 

224 current_engine: Currently active engine name (e.g., 'rustdesk') 

225 context: Session context — {mode, fps, latency_ms, use_case, ...} 

226 

227 Returns: 

228 {recommend: str, reason: str, current: str} or None. 

229 """ 

230 if context is None: 

231 context = {} 

232 

233 engines = _detect_engines() 

234 mode = context.get('mode', 'full_control') 

235 use_case = context.get('use_case', 'general') 

236 

237 # File transfer on Moonlight/Native → suggest RustDesk 

238 if mode == 'file_transfer' and current_engine != 'rustdesk': 

239 if engines.get('rustdesk'): 

240 return { 

241 'recommend': 'rustdesk', 

242 'reason': 'RustDesk has native file transfer (drag-and-drop)', 

243 'current': current_engine, 

244 } 

245 

246 # Gaming/VLM on RustDesk → suggest Moonlight 

247 if use_case in ('gaming', 'vlm_computer_use') and current_engine == 'rustdesk': 

248 if engines.get('moonlight'): 

249 return { 

250 'recommend': 'moonlight', 

251 'reason': 'Moonlight offers hardware-decoded 4K@120fps with <10ms latency', 

252 'current': current_engine, 

253 } 

254 

255 # High latency on WAMP relay → suggest installing engines 

256 latency = context.get('latency_ms', 0) 

257 if latency > 200 and current_engine == 'native': 

258 if engines.get('rustdesk'): 

259 return { 

260 'recommend': 'rustdesk', 

261 'reason': f'High latency ({latency}ms). RustDesk has better NAT traversal.', 

262 'current': current_engine, 

263 } 

264 

265 # Native fallback when engines available → suggest upgrade 

266 if current_engine == 'native': 

267 if engines.get('rustdesk'): 

268 return { 

269 'recommend': 'rustdesk', 

270 'reason': 'RustDesk provides better quality, clipboard, and file transfer', 

271 'current': current_engine, 

272 } 

273 if engines.get('moonlight'): 

274 return { 

275 'recommend': 'moonlight', 

276 'reason': 'Moonlight provides hardware-accelerated streaming', 

277 'current': current_engine, 

278 } 

279 

280 return None