from datetime import datetime, UTC from time import monotonic import os import platform import socket import subprocess import time import httpx try: import psutil except ModuleNotFoundError: # pragma: no cover - optional runtime dependency fallback psutil = None class SystemService: _last_net_bytes_sent: int | None = None _last_net_bytes_recv: int | None = None _last_net_sample_at: float | None = None _weather_cache: dict | None = None _weather_cached_at: float | None = None _weather_cached_location: str | None = None _weather_cache_ttl_seconds: float = 10 * 60 # 10 minutes def __init__(self): # Import settings here to avoid circular imports from app.config import settings self._settings = settings def _get_network_rates(self) -> tuple[float, float]: counters = psutil.net_io_counters() now = monotonic() if ( self.__class__._last_net_sample_at is None or self.__class__._last_net_bytes_sent is None or self.__class__._last_net_bytes_recv is None ): self.__class__._last_net_bytes_sent = counters.bytes_sent self.__class__._last_net_bytes_recv = counters.bytes_recv self.__class__._last_net_sample_at = now return 0.0, 0.0 elapsed = max(now - self.__class__._last_net_sample_at, 1e-6) upload_bps = max(counters.bytes_sent - self.__class__._last_net_bytes_sent, 0) / elapsed download_bps = max(counters.bytes_recv - self.__class__._last_net_bytes_recv, 0) / elapsed self.__class__._last_net_bytes_sent = counters.bytes_sent self.__class__._last_net_bytes_recv = counters.bytes_recv self.__class__._last_net_sample_at = now return round(upload_bps, 1), round(download_bps, 1) def _get_gpu_status(self) -> dict: empty = { 'gpu_name': None, 'gpu_memory_total_mb': None, 'gpu_memory_used_mb': None, 'gpu_util_percent': None, } try: result = subprocess.run( [ 'nvidia-smi', '--query-gpu=name,memory.total,memory.used,utilization.gpu', '--format=csv,noheader,nounits', ], capture_output=True, text=True, encoding='utf-8', timeout=2, check=False, ) except (FileNotFoundError, subprocess.SubprocessError, OSError): return empty if result.returncode != 0 or not result.stdout.strip(): return empty first_line = result.stdout.strip().splitlines()[0] parts = [part.strip() for part in first_line.split(',')] if len(parts) < 4: return empty def parse_number(value: str) -> float | None: try: return float(value) except (TypeError, ValueError): return None return { 'gpu_name': parts[0] or None, 'gpu_memory_total_mb': parse_number(parts[1]), 'gpu_memory_used_mb': parse_number(parts[2]), 'gpu_util_percent': parse_number(parts[3]), } def get_status(self) -> dict: if psutil is None: return { 'cpu_percent': 0.0, 'memory_percent': 0.0, 'disk_percent': 0.0, 'disk_used_gb': 0.0, 'disk_total_gb': 0.0, 'network_upload_bps': 0.0, 'network_download_bps': 0.0, 'system_name': platform.system(), 'system_version': platform.version(), 'hostname': socket.gethostname(), 'uptime_seconds': 0.0, 'gpu_name': None, 'gpu_memory_total_mb': None, 'gpu_memory_used_mb': None, 'gpu_util_percent': None, 'timestamp': datetime.now(UTC).isoformat(), } cpu_percent = psutil.cpu_percent(interval=None) memory = psutil.virtual_memory() disk = psutil.disk_usage('/') upload_bps, download_bps = self._get_network_rates() gpu_status = self._get_gpu_status() boot_time = psutil.boot_time() now_ts = datetime.now(UTC).timestamp() return { 'cpu_percent': round(cpu_percent, 1), 'memory_percent': round(memory.percent, 1), 'disk_percent': round(disk.percent, 1), 'disk_used_gb': round(disk.used / (1024 ** 3), 1), 'disk_total_gb': round(disk.total / (1024 ** 3), 1), 'network_upload_bps': upload_bps, 'network_download_bps': download_bps, 'system_name': platform.system(), 'system_version': platform.version(), 'hostname': socket.gethostname(), 'uptime_seconds': round(max(now_ts - boot_time, 0.0), 1), **gpu_status, 'timestamp': datetime.now(UTC).isoformat(), } async def _fetch_weather(self, location: str) -> dict: try: timeout = httpx.Timeout(10.0, connect=5.0) async with httpx.AsyncClient(timeout=timeout) as client: response = await client.get(f'https://wttr.in/{location}', params={'format': 'j1'}) response.raise_for_status() payload = response.json() current = (payload.get('current_condition') or [{}])[0] weather_code = current.get('weatherCode') temp = current.get('temp_C') parsed_code = int(weather_code) if weather_code is not None and str(weather_code).isdigit() else None if parsed_code is None or temp in (None, ''): return {'weather_code': None, 'weather_summary': 'Weather unavailable'} label = self._weather_code_label(parsed_code) return { 'weather_code': parsed_code, 'weather_summary': f'{label} {temp}°C', } except (httpx.HTTPError, ValueError, TypeError): return {'weather_code': None, 'weather_summary': 'Weather unavailable'} @staticmethod def _weather_code_label(code: int | None) -> str: if code == 0: return 'Clear' if code in {1, 2}: return 'Partly Cloudy' if code == 3: return 'Overcast' if code in {45, 48}: return 'Fog' if code in {51, 53, 55, 56, 57}: return 'Drizzle' if code in {61, 63, 65, 66, 67, 80, 81, 82}: return 'Rain' if code in {71, 73, 75, 77, 85, 86}: return 'Snow' if code in {95, 96, 99}: return 'Thunderstorm' return 'Weather' async def get_config(self) -> dict: """Get public system configuration.""" location = self._settings.LOCATION now = time.time() cached_weather = self.__class__._weather_cache cached_at = self.__class__._weather_cached_at cached_location = self.__class__._weather_cached_location cache_is_valid = ( cached_weather is not None and cached_at is not None and cached_location == location and (now - cached_at) < self.__class__._weather_cache_ttl_seconds ) if cache_is_valid: return { 'location': location, **cached_weather, 'weather_cached': True, 'weather_cached_at': cached_at, } weather = await self._fetch_weather(location) # If fetch failed but we have *any* last known weather for same location, return it to avoid UI flicker. if ( (weather.get('weather_code') is None) and cached_weather is not None and cached_location == location ): return { 'location': location, **cached_weather, 'weather_cached': True, 'weather_cached_at': cached_at, 'weather_stale': True, } # Update cache on successful/meaningful payload (or keep "unavailable" if never succeeded). self.__class__._weather_cache = weather self.__class__._weather_cached_at = now self.__class__._weather_cached_location = location return { 'location': location, **weather, 'weather_cached': False, 'weather_cached_at': now, }