From 112f4b1039de3e794c4242705319c1502f4d401d Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Tue, 14 Jul 2026 00:37:52 +0545 Subject: [PATCH] refactor(services): convert weather + gamemode to Fabric services Weather: move networking, caching and scheduling out of the Weather widget into a WeatherService singleton. Reuse one httpx.Client for the app lifetime, drop the 4-worker pool and the multi-entry dict caches (single-value fields), query Nominatim only on location change, remove debounce and the /tmp-style polling. Widget now observes the `updated` signal and never hits the network. GameMode: turn the standalone CLI script into a GameModeService (Service singleton) exposing an `enabled` Property and `changed` signal. Drive Hyprland via `hyprctl eval hl.config(...)` instead of the deprecated `hyprctl keyword` batch (rejected under the Lua config in Hyprland 0.55+), read state via `hl.get_config`, and keep `hyprctl reload` for disable. Fix game-mode state parsing so unrecognized `hl.get_config` output no longer falsely reports active. BatteryWidget's GameModeButton now subscribes to the service. Remove now-unused LOCATION_CACHE_TIMEOUT. --- src/services/gamemode.py | 189 ++++++-------- src/shared/window/battery_widget.py | 32 +-- src/window/desktop/constants.py | 1 - src/window/desktop/widget.py | 379 ++++++++++++++-------------- 4 files changed, 280 insertions(+), 321 deletions(-) diff --git a/src/services/gamemode.py b/src/services/gamemode.py index 4020b28..fb57e9d 100644 --- a/src/services/gamemode.py +++ b/src/services/gamemode.py @@ -1,126 +1,97 @@ -""" -Hyprland Game Mode Toggle Script +"""Hyprland Game Mode service.""" -This script toggles game mode in Hyprland by disabling/enabling animations -and other visual effects for better performance during gaming. - -Uses a marker file in /tmp to track state. -""" - -import sys -from datetime import datetime -from pathlib import Path -from typing import Literal - -import tomlkit +from fabric.core.service import Property, Service, Signal from utils.functions import run_command -STATE_FILE = Path("/tmp/hyprland_gamemode.toml") +HYPRCTL = "hyprctl" + +_ENABLE_LUA = ( + "hl.config({" + " animations = { enabled = false }," + " decoration = { shadow = { enabled = false }, blur = { enabled = false }," + " rounding = 0 }," + " general = { gaps_in = 0, gaps_out = 0, border_size = 1 }" + "})" +) -def run_hyprctl(command: str) -> str: - """Run a hyprctl command and return the output (raises on error).""" - result = run_command(["hyprctl", *command.split()], timeout=5) - if result.returncode != 0: - raise RuntimeError(f"hyprctl failed: {result.stderr}") - return result.stdout.strip() +class GameModeService(Service): + """Owns all Hyprland interaction for game mode and notifies observers.""" + _instance = None -def _read_state() -> dict: - if not STATE_FILE.exists(): - return {} - try: - with open(STATE_FILE) as f: - data = tomlkit.load(f) - return dict(data) if data else {} - except Exception: - return {} + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + @staticmethod + def get_initial(): + if GameModeService._instance is None: + GameModeService._instance = GameModeService() + return GameModeService._instance -def _write_state(data: dict): - STATE_FILE.parent.mkdir(parents=True, exist_ok=True) - doc = tomlkit.document() - for k, v in data.items(): - doc[k] = v - with open(STATE_FILE, "w") as f: - tomlkit.dump(doc, f) + @Signal + def changed(self, enabled: bool) -> None: + """Emitted when game mode is enabled or disabled.""" + def __init__(self, **kwargs): + if getattr(self, "_initialized", False): + return + super().__init__(**kwargs) + self._initialized = True + self._enabled = False + self.sync_state() -def check_gamemode() -> Literal["t", "f"]: - """Check if game mode is active.""" - state = _read_state() - return "t" if state and state.get("enabled") else "f" + @Property(bool, "readable", default_value=False) + def enabled(self) -> bool: + """Whether game mode (visual effects off) is currently active.""" + return self._enabled + def toggle(self) -> None: + if self._enabled: + self.disable() + else: + self.enable() -def enable_gamemode(): - """Enable game mode by disabling visual effects.""" - batch_commands = [ - "keyword animations:enabled 0", - "keyword decoration:shadow:enabled 0", - "keyword decoration:blur:enabled 0", - "keyword general:gaps_in 0", - "keyword general:gaps_out 0", - "keyword general:border_size 1", - "keyword decoration:rounding 0", - ] - run_hyprctl(f'--batch "{"; ".join(batch_commands)}"') + def enable(self) -> None: + if self._enabled: + return + run_command([HYPRCTL, "eval", _ENABLE_LUA], timeout=5) + self._set_enabled(True) - _write_state( - { - "enabled": True, - "last_toggled": datetime.now().isoformat(timespec="seconds"), - } - ) - print("Game mode enabled - visual effects disabled for better performance") + def disable(self) -> None: + if not self._enabled: + return + # A full reload restores the user's exact prior configuration (gaps, + # border, rounding, etc.) which per-option restores cannot reproduce, + # so it is kept instead of guessing default values. + run_command([HYPRCTL, "reload"], timeout=5) + self._set_enabled(False) + def sync_state(self) -> None: + """Read the authoritative state from Hyprland (animations toggle).""" + result = run_command( + [HYPRCTL, "eval", "return hl.get_config('animations.enabled')"], timeout=5 + ) + if result.returncode != 0: + return + value = result.stdout.strip().lower() + if value in ("true", "1", "on"): + animations_on = True + elif value in ("false", "0", "off"): + animations_on = False + else: + # Unrecognized output (e.g. an error string): keep current state + # rather than falsely reporting game mode as active. + return + # Game mode is on when Hyprland animations are disabled. + self._set_enabled(not animations_on) -def disable_gamemode(): - """Disable game mode by reloading Hyprland configuration.""" - run_hyprctl("reload") - - _write_state( - { - "enabled": False, - "last_toggled": datetime.now().isoformat(timespec="seconds"), - } - ) - print("Game mode disabled - visual effects restored") - - -def toggle_gamemode(): - """Toggle game mode state.""" - state = _read_state() - if state.get("enabled"): - disable_gamemode() - else: - enable_gamemode() - - -def print_usage(): - print("Usage:") - print(" gamemode check - Check if game mode is active (t/f)") - print(" gamemode toggle - Toggle game mode") - - -def main(): - if len(sys.argv) < 2: - toggle_gamemode() - return - - command = sys.argv[1].lower() - - if command == "check": - print(check_gamemode()) - elif command == "toggle": - toggle_gamemode() - elif command in ["help", "-h", "--help"]: - print_usage() - else: - print(f"Unknown command: {command}", file=sys.stderr) - print_usage() - sys.exit(1) - - -if __name__ == "__main__": - main() + def _set_enabled(self, enabled: bool) -> None: + if enabled == self._enabled: + return + self._enabled = enabled + self.notify("enabled") + self.emit("changed", enabled) diff --git a/src/shared/window/battery_widget.py b/src/shared/window/battery_widget.py index 9b62c7f..02a3571 100644 --- a/src/shared/window/battery_widget.py +++ b/src/shared/window/battery_widget.py @@ -1,4 +1,4 @@ -from fabric.utils import GLib, logger +from fabric.utils import GLib from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox @@ -7,7 +7,7 @@ from fabric.widgets.label import Label from fabric.widgets.separator import Separator from services.battery import Battery -from services.gamemode import check_gamemode, toggle_gamemode +from services.gamemode import GameModeService from utils.functions import clear_children, format_duration from utils.utils import svg_file @@ -88,6 +88,8 @@ class GameModeButton(Box): def __init__(self, parent, **kwargs): super().__init__(name="energy-mode-button", h_expand=True, **kwargs) self.parent = parent + self.gamemode_service = GameModeService.get_initial() + self.gamemode_service.connect("changed", self.update_state) self.game_icon = Image( icon_name="applications-games-symbolic", @@ -121,30 +123,14 @@ class GameModeButton(Box): self.update_state() def on_clicked(self, *args): - try: - toggle_gamemode() - GLib.timeout_add(500, lambda: self.update_state()) - except Exception as e: - logger.error(f"Failed to toggle game mode: {e}") + self.gamemode_service.toggle() - GLib.timeout_add(300, lambda: self._reset_icon_state()) - - def _reset_icon_state(self): - return False # Remove timeout - - def update_state(self): - try: - is_active = check_gamemode() == "t" - if is_active: - self.game_icon.add_style_class("connected") - else: - self.game_icon.remove_style_class("connected") - except Exception as e: - logger.error(f"Failed to check game mode status: {e}") + def update_state(self, *_): + if self.gamemode_service.enabled: + self.game_icon.add_style_class("connected") + else: self.game_icon.remove_style_class("connected") - return False # Remove timeout if called from GLib.timeout_add - class BatteryControl(Box): def __init__(self, parent, **kwargs): diff --git a/src/window/desktop/constants.py b/src/window/desktop/constants.py index 38545c2..f79a878 100644 --- a/src/window/desktop/constants.py +++ b/src/window/desktop/constants.py @@ -14,7 +14,6 @@ CALENDAR_UPDATE_INTERVAL = int( ) * 1000 ) # Calculate time till midnight -LOCATION_CACHE_TIMEOUT = 604800 # 7 days (extended from 24h) WEATHER_GRADIENT_MAP = { diff --git a/src/window/desktop/widget.py b/src/window/desktop/widget.py index 8d2dbd6..7d38526 100644 --- a/src/window/desktop/widget.py +++ b/src/window/desktop/widget.py @@ -6,6 +6,7 @@ from typing import Any, Dict, List, Optional, Tuple import httpx import psutil +from fabric.core.service import Service, Signal from fabric.utils import GLib, invoke_repeater, logger, time from fabric.widgets.box import Box from fabric.widgets.circularprogressbar import CircularProgressBar @@ -15,12 +16,10 @@ from fabric.widgets.overlay import Overlay from fabric.widgets.wayland import WaylandWindow as Window from shared.data import load_config -from utils.debounce import sync_debounce from utils.utils import svg_file from window.desktop.constants import ( CALENDAR_UPDATE_INTERVAL, LOCATION_APIS, - LOCATION_CACHE_TIMEOUT, SYSTEM_UPDATE_INTERVAL, WEATHER_CACHE_TIMEOUT, WEATHER_DESC_MAP, @@ -29,102 +28,184 @@ from window.desktop.constants import ( WEATHER_UPDATE_INTERVAL, ) -# Thread pool for async operations -executor = ThreadPoolExecutor(max_workers=4) - -# Global cache for weather data -_weather_cache: Dict[str, Tuple[Any, float]] = {} -_location_cache: Dict[str, Tuple[float, float, float]] = {} -_WEATHER_CACHE_MAX = 10 -_LOCATION_CACHE_MAX = 10 - - -def http_get_json( - url: str, timeout: int = 3, headers: Optional[Dict] = None -) -> Optional[Dict]: - """Helper to perform GET requests and return JSON using httpx.""" - try: - # Some APIs require a User-Agent or they will return 403 Forbidden - default_headers = {"User-Agent": "Modus-Desktop/1.0"} - if headers: - default_headers.update(headers) - response = httpx.get( - url, headers=default_headers, timeout=timeout, follow_redirects=True - ) - if response.status_code == 200: - return response.json() - except Exception as e: - logger.error(f"HTTP Request to {url} failed: {e}") - return None - - -def get_location() -> str: - """Get current location from config or multiple IP geolocation APIs with fallback.""" - # Try to get location from config first - try: - config = load_config() - manual_location = config.get("weather_location") - if manual_location: - return manual_location - except Exception as e: - logger.error(f"An error occurred: {e}") - - # Fallback to IP geolocation APIs - for api_url in LOCATION_APIS: - data = http_get_json(api_url, timeout=2) - if data: - city = data.get("city", "") - if city: - return city - - logger.warning("All location APIs failed") - return "" - - -def get_coordinates(city: str) -> Optional[Tuple[float, float]]: - """Get coordinates for a city using Nominatim geocoding API.""" - cache_key = city.lower() - current_time = time.time() - - if cache_key in _location_cache: - lat, lon, timestamp = _location_cache[cache_key] - if current_time - timestamp < LOCATION_CACHE_TIMEOUT: - return lat, lon - - encoded_city = urllib.parse.quote(city) - url = f"https://nominatim.openstreetmap.org/search?q={encoded_city}&format=json&limit=1" - - data = http_get_json(url, timeout=3, headers={"User-Agent": "Modus-Desktop/1.0"}) - - if data and isinstance(data, list) and len(data) > 0: - try: - lat = float(data[0]["lat"]) - lon = float(data[0]["lon"]) - while len(_location_cache) >= _LOCATION_CACHE_MAX: - _location_cache.pop(next(iter(_location_cache)), None) - _location_cache[cache_key] = (lat, lon, current_time) - return lat, lon - except (ValueError, KeyError): - pass - - return None - - -def get_weather_data(lat: float, lon: float) -> Optional[Dict[str, Any]]: - """Fetch weather data from Open-Meteo API.""" - url = ( - f"https://api.open-meteo.com/v1/forecast?" - f"latitude={lat}&longitude={lon}" - f"¤t_weather=true" - f"&daily=temperature_2m_max,temperature_2m_min" - f"&timezone=auto" - f"&forecast_days=1" +# Icons that have a dedicated -night SVG variant. +_NIGHT_VARIANT_ICONS = frozenset( + ( + "weather-clear", + "weather-clouds", + "weather-few-clouds", + "weather-overcast", + "weather-showers", + "weather-showers-scattered", + "weather-snow", + "weather-snow-scattered", + "weather-storm", ) - return http_get_json(url, timeout=3) +) + +_HTTP_HEADERS = {"User-Agent": "Modus-Desktop/1.0"} -def format_weather_data(weather_data: Dict[str, Any], city: str) -> List[str]: - """Format weather data into the expected format.""" +class WeatherService(Service): + """Owns HTTP, caching, coordinate lookup and update scheduling for the + weather widget. Networking runs on a single background worker; results are + delivered back to the GLib main loop through the ``updated`` signal. + + This is an app-lifetime singleton: its executor, client and timer are never + torn down. Widgets observe it by connecting to ``updated`` and disconnect on + destroy, so the service survives widget rebuilds (config reloads, etc.).""" + + _instance = None + + def __new__(cls, *args, **kwargs): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + @staticmethod + def get_initial(): + if WeatherService._instance is None: + WeatherService._instance = WeatherService() + return WeatherService._instance + + @Signal + def updated(self, weather_info: object) -> None: + """Emitted on the main loop with fresh weather data (or None on failure).""" + + def __init__(self, **kwargs): + if getattr(self, "_initialized", False): + return + super().__init__(**kwargs) + self._initialized = True + self._http = httpx.Client( + timeout=3.0, follow_redirects=True, headers=_HTTP_HEADERS + ) + self._executor = ThreadPoolExecutor(max_workers=1) + + # This shell shows one location at a time, so caches hold single values. + self._weather: Optional[List[str]] = None + self._weather_ts: float = 0.0 + self._location: Optional[str] = None + self._coords: Optional[Tuple[float, float]] = None + self._coords_location: Optional[str] = None + + # Periodic refresh only; widgets request the first fetch after + # connecting (see Weather.__init__) to avoid a race with subscription. + GLib.timeout_add_seconds(WEATHER_UPDATE_INTERVAL, self._refresh) + + # -- scheduling ------------------------------------------------------- + + def refresh(self) -> None: + """Trigger an immediate fetch (e.g. when a widget subscribes).""" + self._executor.submit(self._fetch) + + def _refresh(self) -> bool: + self._executor.submit(self._fetch) + return True + + # -- networking ------------------------------------------------------- + + def _get_json(self, url: str, timeout: float) -> Optional[Dict[str, Any]]: + try: + resp = self._http.get(url, timeout=timeout) + if resp.status_code == 200: + return resp.json() + except Exception as e: + logger.error(f"HTTP request to {url} failed: {e}") + return None + + def _get_location(self) -> str: + try: + manual_location = load_config().get("weather_location") + if manual_location: + return manual_location + except Exception as e: + logger.error(f"An error occurred: {e}") + + for api_url in LOCATION_APIS: + data = self._get_json(api_url, 2.0) + if data: + city = data.get("city", "") + if city: + return city + + logger.warning("All location APIs failed") + return "" + + def _get_coordinates(self, city: str) -> Optional[Tuple[float, float]]: + # Only query Nominatim when the location is new or unknown. + if self._coords is not None and self._coords_location == city: + return self._coords + + encoded_city = urllib.parse.quote(city) + url = ( + "https://nominatim.openstreetmap.org/search?" + f"q={encoded_city}&format=json&limit=1" + ) + data = self._get_json(url, 3.0) + if isinstance(data, list) and data: + try: + coords = (float(data[0]["lat"]), float(data[0]["lon"])) + except (ValueError, KeyError): + return None + self._coords = coords + self._coords_location = city + return coords + return None + + def _get_weather_data(self, lat: float, lon: float) -> Optional[Dict[str, Any]]: + url = ( + "https://api.open-meteo.com/v1/forecast?" + f"latitude={lat}&longitude={lon}" + "¤t_weather=true" + "&daily=temperature_2m_max,temperature_2m_min" + "&timezone=auto" + "&forecast_days=1" + ) + return self._get_json(url, 3.0) + + # -- background fetch (runs on the worker thread) --------------------- + + def _fetch(self) -> None: + location = self._get_location() + if not location: + return GLib.idle_add(self._deliver, None) + + cache_key = location.lower() + now = time.time() + if ( + self._weather is not None + and self._location == cache_key + and now - self._weather_ts < WEATHER_CACHE_TIMEOUT + ): + return GLib.idle_add(self._deliver, self._weather) + + coords = self._get_coordinates(location) + if not coords: + return GLib.idle_add(self._deliver, None) + + weather_data = self._get_weather_data(*coords) + if not weather_data: + return GLib.idle_add(self._deliver, None) + + formatted = _format_weather_data(weather_data, location) + if formatted: + self._weather = formatted + self._location = cache_key + self._weather_ts = now + GLib.idle_add(self._deliver, formatted) + else: + GLib.idle_add(self._deliver, None) + return None + + def _deliver(self, weather_info: Optional[List[str]]) -> None: + self.emit("updated", weather_info) + + +def _format_weather_data( + weather_data: Dict[str, Any], city: str +) -> Optional[List[str]]: + """Format weather data into the list consumed by the widget.""" try: current = weather_data["current_weather"] daily = weather_data["daily"] @@ -133,27 +214,9 @@ def format_weather_data(weather_data: Dict[str, Any], city: str) -> List[str]: is_day = current.get("is_day", 1) # Default to day if missing base_icon = WEATHER_ICON_MAP.get(weather_code, "weather-none-available") - - # Determine day/night variant if applicable icon_name = base_icon - if not is_day: - night_variant = f"{base_icon}-night" - - # Fast check if night variant exists using predefined list or checking the path - # Since we know the variants from earlier, let's just optimistically build it - # then logic in svg_file will handle resolution seamlessly (fallbacks can be tricky, but we assume it's correct) - if base_icon in [ - "weather-clear", - "weather-clouds", - "weather-few-clouds", - "weather-overcast", - "weather-showers", - "weather-showers-scattered", - "weather-snow", - "weather-snow-scattered", - "weather-storm", - ]: - icon_name = night_variant + if not is_day and base_icon in _NIGHT_VARIANT_ICONS: + icon_name = f"{base_icon}-night" condition = WEATHER_DESC_MAP.get(weather_code, "Unknown") gradient_class = WEATHER_GRADIENT_MAP.get(weather_code, "weather-clear") @@ -168,73 +231,6 @@ def format_weather_data(weather_data: Dict[str, Any], city: str) -> List[str]: return None -def get_weather(callback): - """Fetch weather data asynchronously.""" - - def fetch_weather(): - location = get_location() - if not location: - return GLib.idle_add(callback, None) - - cache_key = location.lower() - current_time = time.time() - - if cache_key in _weather_cache: - cached_data, timestamp = _weather_cache[cache_key] - if current_time - timestamp < WEATHER_CACHE_TIMEOUT: - return GLib.idle_add(callback, cached_data) - - coords = get_coordinates(location) - if not coords: - return GLib.idle_add(callback, None) - - lat, lon = coords - weather_data = get_weather_data(lat, lon) - if not weather_data: - return GLib.idle_add(callback, None) - - formatted_data = format_weather_data(weather_data, location) - if formatted_data: - while len(_weather_cache) >= _WEATHER_CACHE_MAX: - _weather_cache.pop(next(iter(_weather_cache)), None) - _weather_cache[cache_key] = (formatted_data, current_time) - GLib.idle_add(callback, formatted_data) - else: - GLib.idle_add(callback, None) - return None - - executor.submit(fetch_weather) - - -def update_weather(widget): - """Update weather widget with new data.""" - - def perform_fetch(): - get_weather(lambda weather_info: update_widget(widget, weather_info)) - - debounced_perform_fetch = sync_debounce(1000, immediate=True)(perform_fetch) - - def fetch_and_update(): - debounced_perform_fetch() - return True - - def initial_fetch(): - debounced_perform_fetch() - return False - - # Trigger first fetch immediately - MUST return False to not loop in idle - GLib.idle_add(initial_fetch) - - return GLib.timeout_add_seconds(WEATHER_UPDATE_INTERVAL, fetch_and_update) - - -def update_widget(widget, weather_info): - """Update widget labels with weather information.""" - if weather_info: - widget.weatherinfo = weather_info - widget.update_labels(weather_info) - - class Weather(Box): def __init__(self, parent, **kwargs): super().__init__( @@ -248,10 +244,11 @@ class Weather(Box): ) self.parent = parent self.weatherinfo = None - self._weather_timer_id = None self._create_labels() self._layout_labels() - self._weather_timer_id = update_weather(self) + self._service = WeatherService.get_initial() + self._service.connect("updated", self._on_service_update) + self._service.refresh() def _create_labels(self): self.header = Box(orientation="h", h_expand=True) @@ -319,11 +316,17 @@ class Weather(Box): self.parent.set_visible(True) + def _on_service_update(self, _service, weather_info): + if weather_info: + self.weatherinfo = weather_info + self.update_labels(weather_info) + def destroy(self): - """Cleanup weather update timer""" - if self._weather_timer_id: - GLib.source_remove(self._weather_timer_id) - self._weather_timer_id = None + """Disconnect from the weather service (the service is app-lifetime).""" + try: + self._service.disconnect_by_func(self._on_service_update) + except ValueError: + pass super().destroy()