Spaces:
Sleeping
Sleeping
File size: 5,408 Bytes
399b80c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | from typing import Dict, Any, Union
from openspace.grounding.core.types import BackendType, SessionConfig
from openspace.grounding.core.provider import Provider
from openspace.grounding.core.session import BaseSession
from openspace.config import get_config
from openspace.config.utils import get_config_value
from openspace.platforms import get_local_server_config
from openspace.utils.logging import Logger
from .transport.connector import GUIConnector
from .transport.local_connector import LocalGUIConnector
from .session import GUISession
logger = Logger.get_logger(__name__)
class GUIProvider(Provider):
"""
Provider for GUI desktop environment.
Manages communication with desktop_env through HTTP API or local in-process execution.
Supports two modes:
- "local": Execute GUI operations directly in-process (no server needed)
- "server": Connect to a running local_server via HTTP API
Supports automatic default session creation:
- If no session exists, a default session will be created on first use
- Default session uses configuration from config file or environment
"""
DEFAULT_SID = BackendType.GUI.value
def __init__(self, config: Dict[str, Any] = None):
"""
Initialize GUI provider.
Args:
config: Provider configuration
"""
super().__init__(BackendType.GUI, config)
self.connectors: Dict[str, Union[GUIConnector, LocalGUIConnector]] = {}
async def initialize(self) -> None:
"""
Initialize the provider and create default session.
"""
if not self.is_initialized:
logger.info("Initializing GUI provider")
# Auto-create default session
await self.create_session(SessionConfig(
session_name=self.DEFAULT_SID,
backend_type=BackendType.GUI,
connection_params={}
))
self.is_initialized = True
async def create_session(self, session_config: SessionConfig) -> BaseSession:
"""
Create GUI session.
Args:
session_config: Session configuration
Returns:
GUISession instance
"""
# Load GUI backend configuration
gui_config = get_config().get_backend_config("gui")
# Determine execution mode: "local" or "server"
mode = getattr(gui_config, "mode", "local")
# Extract connection parameters
conn_params = session_config.connection_params
timeout = get_config_value(conn_params, 'timeout', gui_config.timeout)
retry_times = get_config_value(conn_params, 'retry_times', gui_config.max_retries)
retry_interval = get_config_value(conn_params, 'retry_interval', gui_config.retry_interval)
# Build pkgs_prefix with failsafe setting
failsafe_str = "True" if gui_config.failsafe else "False"
pkgs_prefix = get_config_value(
conn_params,
'pkgs_prefix',
gui_config.pkgs_prefix.format(failsafe=failsafe_str, command="{command}")
)
if mode == "local":
# ---------- LOCAL MODE ----------
logger.info("GUI backend using LOCAL mode (no server required)")
connector = LocalGUIConnector(
timeout=timeout,
retry_times=retry_times,
retry_interval=retry_interval,
pkgs_prefix=pkgs_prefix,
)
else:
# ---------- SERVER MODE ----------
logger.info("GUI backend using SERVER mode (connecting to local_server)")
local_server_config = get_local_server_config()
vm_ip = get_config_value(conn_params, 'vm_ip', local_server_config['host'])
server_port = get_config_value(conn_params, 'server_port', local_server_config['port'])
connector = GUIConnector(
vm_ip=vm_ip,
server_port=server_port,
timeout=timeout,
retry_times=retry_times,
retry_interval=retry_interval,
pkgs_prefix=pkgs_prefix,
)
# Create session
session = GUISession(
connector=connector,
session_id=session_config.session_name,
backend_type=BackendType.GUI,
config=session_config,
)
# Store connector and session
self.connectors[session_config.session_name] = connector
self._sessions[session_config.session_name] = session
logger.info(f"Created GUI session: {session_config.session_name} (mode={mode})")
return session
async def close_session(self, session_name: str) -> None:
"""
Close GUI session.
Args:
session_name: Name of the session to close
"""
if session_name in self._sessions:
session = self._sessions[session_name]
await session.disconnect()
del self._sessions[session_name]
if session_name in self.connectors:
connector = self.connectors[session_name]
await connector.disconnect()
del self.connectors[session_name]
logger.info(f"Closed GUI session: {session_name}") |