sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor(services): convert weather + gamemode to Fabric services

Browse Source 
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.
This commit is contained in:
sankalp committed 2026-07-14 00:37:52 +05:45
1 parent f3f17bb5fc
commit 112f4b1039
4 files changed
+280 -321
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/services/gamemode.py
+80 -109
View File
@@ -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)
src/shared/window/battery_widget.py
+9 -23
View File
@@ -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):
src/window/desktop/constants.py
-1
View File
@@ -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 = {
src/window/desktop/widget.py
+191 -188
View File
@@ -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"&current_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}"
"&current_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()