Spaces:
Running
Running
File size: 6,088 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 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | import aiohttp
from typing import Optional, Dict, Any
from openspace.utils.logging import Logger
from .config import get_client_base_url
logger = Logger.get_logger(__name__)
class SystemInfoClient:
"""
This client provides simple methods to get:
- Platform info (OS, architecture, version, etc.)
- Screen size
- Cursor position
"""
def __init__(
self,
base_url: Optional[str] = None,
timeout: int = 10
):
"""
Initialize system info client.
Args:
base_url: Base URL of the local server
(default: read from local_server/config.json or env LOCAL_SERVER_URL)
timeout: Request timeout in seconds
"""
# Get base_url: priority is explicit > env > config file
if base_url is None:
base_url = get_client_base_url()
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self._session: Optional[aiohttp.ClientSession] = None
self._cached_info: Optional[Dict[str, Any]] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Get or create aiohttp session."""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.timeout)
)
return self._session
async def get_system_info(self, use_cache: bool = True) -> Optional[Dict[str, Any]]:
"""
Get comprehensive system information.
Returns information including:
- system: OS name (Linux, Darwin, Windows)
- release: OS release version
- version: Detailed version string
- machine: Architecture (x86_64, arm64, etc.)
- processor: Processor type
- Additional platform-specific info
Args:
use_cache: Whether to use cached info (default: True)
"""
# Check cache
if use_cache and self._cached_info:
logger.debug("Using cached system info")
return self._cached_info
try:
session = await self._get_session()
url = f"{self.base_url}/platform"
async with session.get(url) as response:
if response.status == 200:
info = await response.json()
# Cache the result
if use_cache:
self._cached_info = info
logger.debug(f"System info retrieved: {info.get('system')}")
return info
else:
error_text = await response.text()
logger.error(f"Failed to get system info: HTTP {response.status} - {error_text}")
return None
except Exception as e:
logger.error(f"Failed to get system info: {e}")
return None
async def get_screen_size(self) -> Optional[Dict[str, int]]:
"""
Get screen size.
Returns:
Dict with 'width' and 'height', or None on failure
"""
try:
session = await self._get_session()
url = f"{self.base_url}/screen_size"
async with session.get(url) as response:
if response.status == 200:
size = await response.json()
logger.debug(f"Screen size: {size.get('width')}x{size.get('height')}")
return {
"width": size.get("width"),
"height": size.get("height")
}
else:
error_text = await response.text()
logger.error(f"Failed to get screen size: HTTP {response.status} - {error_text}")
return None
except Exception as e:
logger.error(f"Failed to get screen size: {e}")
return None
async def get_cursor_position(self) -> Optional[Dict[str, int]]:
"""
Get current cursor position.
Returns:
Dict with 'x' and 'y', or None on failure
"""
try:
session = await self._get_session()
url = f"{self.base_url}/cursor_position"
async with session.get(url) as response:
if response.status == 200:
pos = await response.json()
return {
"x": pos.get("x"),
"y": pos.get("y")
}
else:
error_text = await response.text()
logger.error(f"Failed to get cursor position: HTTP {response.status} - {error_text}")
return None
except Exception as e:
logger.error(f"Failed to get cursor position: {e}")
return None
def clear_cache(self):
"""Clear cached system information."""
self._cached_info = None
logger.debug("System info cache cleared")
async def close(self):
"""Close the HTTP session."""
if self._session and not self._session.closed:
await self._session.close()
logger.debug("System info client session closed")
async def __aenter__(self):
"""Context manager entry."""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
await self.close()
return False
async def get_system_info(base_url: Optional[str] = None) -> Optional[Dict[str, Any]]:
async with SystemInfoClient(base_url=base_url) as client:
return await client.get_system_info(use_cache=False)
async def get_screen_size(base_url: Optional[str] = None) -> Optional[Dict[str, int]]:
async with SystemInfoClient(base_url=base_url) as client:
return await client.get_screen_size() |