From bc139feda70650be99019c4de71892749d054547 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Thu, 26 Feb 2026 05:54:58 +0545 Subject: [PATCH] refactor(services): drop auth service and tidy network/todo Remove unused auth module and associated screenshot/OQR code logic. Clean up network import statements and unused dependencies. Minor cleanup in todo service to eliminate trailing whitespace. New Features - None Enhancements - Refactored network module imports. - Streamlined todo service by removing dead code. Bug Fixes - None --- services/auth.py | 348 ----------------- services/battery.py | 357 ++++++++---------- services/brightness.py | 7 +- services/config.py | 162 ++++++++ services/custom_notification.py | 204 +++++----- services/modus.py | 27 +- services/mpris.py | 641 +++++++++++++++++++------------- services/network.py | 14 +- services/todo.py | 1 - 9 files changed, 843 insertions(+), 918 deletions(-) delete mode 100644 services/auth.py create mode 100644 services/config.py diff --git a/services/auth.py b/services/auth.py deleted file mode 100644 index 3a7b070..0000000 --- a/services/auth.py +++ /dev/null @@ -1,348 +0,0 @@ -import json -import os -import subprocess -import time -from datetime import datetime -from pathlib import Path -from urllib.parse import parse_qs, urlparse - -import pyotp -from PIL import Image -from pyzbar.pyzbar import decode - -import config.data as data - - -def get_otp_file_path(): - """Returns the path to the OTP file inside the cache directory.""" - cache_dir = Path(data.CACHE_DIR) / "otp" - # Create the directory if it doesn't exist - cache_dir.mkdir(parents=True, exist_ok=True) - - file_path = cache_dir / "otp_codes.json" - - # Create the file if it doesn't exist - if not file_path.exists(): - with open(file_path, "w") as f: - json.dump([], f) - print(f"File {file_path} created successfully!") - - return file_path - - -def capture_selected_area(filename="/tmp/screenshot.png"): - """ - Uses slurp to let the user select an area of the screen, - then captures that area using grim. - """ - try: - # slurp returns coordinates in format: x,y widthxheight (e.g. 100,200 300x400) - result = subprocess.run(["slurp"], check=True, capture_output=True, text=True) - geometry = result.stdout.strip() - if not geometry: - print("No area selected with slurp.") - return None - except subprocess.CalledProcessError as e: - print("Error selecting area with slurp:", e) - return None - - try: - # grim uses -g to capture a specific region - subprocess.run(["grim", "-g", geometry, filename], check=True) - except subprocess.CalledProcessError as e: - print("Error capturing screenshot with grim:", e) - return None - - return filename - - -def read_and_save_to_json(): - # Get the full path to the JSON file - json_file = get_otp_file_path() - - screenshot = capture_selected_area() - if screenshot is None: - print("Failed to capture the selected area.") - return False - - # Short delay to ensure the file is written - time.sleep(1) - - # Open the captured image - try: - img = Image.open(screenshot) - except Exception as e: - print("Error opening the image:", e) - return False - - # Decode QR Code(s) from the image - decoded_objects = decode(img) - if not decoded_objects: - print("No QR Code detected in the selected area.") - return False - - results = [] - - for obj in decoded_objects: - data = obj.data.decode("utf-8") - print("QR Code detected:", data) - - result_entry = {"timestamp": datetime.now().isoformat(), "qr_data": data} - - if data.startswith("otpauth://"): - parsed = urlparse(data) - query = parse_qs(parsed.query) - - # Extract the secret properly - secret = query.get("secret", [None])[0] - issuer_from_query = query.get("issuer", [None])[0] - - # Extract the label (path), which may contain issuer and account - label = parsed.path.lstrip("/") if parsed.path else "" - account_name = label - issuer_from_path = None - - if ":" in label: - parts = label.split(":", 1) - issuer_from_path = parts[0] - account_name = parts[1] - - # Prefer issuer from path if available, otherwise use from query - issuer = issuer_from_path or issuer_from_query - - # Extract the period (default is 30 seconds) - period = int(query.get("period", ["30"])[0]) - - # Create TOTP object with the correct interval - totp = pyotp.TOTP(secret, interval=period) - current_otp = totp.now() - print(f"Generated OTP: {current_otp} (valid for {period} seconds)") - - result_entry.update( - { - "type": "otp", - "secret": secret, - "issuer": issuer, - "account_name": account_name, - } - ) - else: - result_entry["type"] = "unknown" - print("Unrecognized format. Expected a URI like otpauth://") - return False - - results.append(result_entry) - - # Load existing data if the file exists - existing_data = [] - if os.path.exists(json_file): - try: - with open(json_file, "r") as f: - existing_data = json.load(f) - except json.JSONDecodeError: - print(f"Error reading the file {json_file}. Creating a new one.") - - # Append new results - existing_data.extend(results) - - # Save to JSON file - with open(json_file, "w") as f: - json.dump(existing_data, f, indent=4) - - print(f"OTP data saved to {json_file}") - return True - - -def CodeOTP(uri): - parsed = urlparse(uri) - query = parse_qs(parsed.query) - secret = query.get("secret", [None])[0] - - if secret is None: - return None - else: - totp = pyotp.TOTP(secret) - return totp.now() - - -# TOTP/OTP utility functions -def generate_totp(secret: str) -> str: - """Generate TOTP code from secret.""" - try: - return pyotp.TOTP(secret).now() - except Exception as e: - print(f"Error generating TOTP: {e}") - return None - - -def get_time_remaining() -> int: - """Get seconds remaining until next token refresh.""" - return 30 - (int(time.time()) % 30) - - -def get_time_remaining_with_blink() -> str: - """Get time remaining with blinking effect.""" - time_remaining = get_time_remaining() - current_second = int(time.time()) - should_blink = current_second % 2 == 0 - - if should_blink: - return f"{time_remaining}s" - else: - return f"{time_remaining}s" - - -def validate_base32_secret(secret: str) -> dict: - """Validate and clean Base32 secret.""" - import base64 - import re - - try: - # Clean up the secret - remove spaces, dashes, and convert to uppercase - clean_secret = secret.replace(" ", "").replace("-", "").replace("_", "").upper() - - # Remove any non-base32 characters - clean_secret = re.sub(r"[^A-Z2-7]", "", clean_secret) - - # Add padding if needed (Base32 requires padding to multiple of 8) - while len(clean_secret) % 8 != 0: - clean_secret += "=" - - # Validate Base32 format - try: - base64.b32decode(clean_secret) - except Exception as e: - return {"success": False, "error": f"Invalid Base32 secret: {str(e)}"} - - # Test if the secret can generate a valid TOTP - try: - test_totp = pyotp.TOTP(clean_secret) - test_code = test_totp.now() - if not test_code or len(test_code) != 6: - raise ValueError("Generated invalid TOTP code") - except Exception as e: - return {"success": False, "error": f"Cannot generate TOTP: {str(e)}"} - - return {"success": True, "secret": clean_secret} - except Exception as e: - return {"success": False, "error": f"Unexpected error: {str(e)}"} - - -def parse_otpauth_uri(uri: str, account_name: str = "") -> dict: - """Parse otpauth URI and extract account information.""" - try: - parsed = urlparse(uri) - if parsed.scheme != "otpauth" or parsed.netloc != "totp": - return { - "success": False, - "error": "Only otpauth://totp/ URIs are supported", - } - - if not account_name: - account_path = parsed.path.lstrip("/") - if ":" in account_path: - issuer, extracted_name = account_path.split(":", 1) - account_name = extracted_name - else: - account_name = account_path - - params = parse_qs(parsed.query) - secret = params.get("secret", [""])[0] - issuer = params.get("issuer", [""])[0] - algorithm = params.get("algorithm", ["SHA1"])[0] - digits = int(params.get("digits", ["6"])[0]) - period = int(params.get("period", ["30"])[0]) - - if not secret: - return {"success": False, "error": "No secret found in URI"} - - return { - "success": True, - "account_name": account_name, - "secret": secret, - "issuer": issuer, - "algorithm": algorithm, - "digits": digits, - "period": period, - } - except Exception as e: - return {"success": False, "error": f"Error parsing otpauth URI: {str(e)}"} - - -def scan_qr_and_add_account(account_name: str, secrets_file_path: str) -> dict: - """Scan QR code and add OTP account to secrets file.""" - try: - # Capture QR code from screen - screenshot_path = capture_selected_area() - if not screenshot_path: - return {"success": False, "error": "QR scan cancelled or failed"} - - # Decode QR code - try: - img = Image.open(screenshot_path) - decoded_objects = decode(img) - - if not decoded_objects: - return { - "success": False, - "error": "No QR code detected in selected area", - } - - # Process the first QR code found - qr_data = decoded_objects[0].data.decode("utf-8") - print(f"QR Code detected: {qr_data}") - - if qr_data.startswith("otpauth://"): - # Parse otpauth URI - result = parse_otpauth_uri(qr_data, account_name) - if not result["success"]: - return result - - # Load existing secrets - secrets = {} - if os.path.exists(secrets_file_path): - try: - with open(secrets_file_path, "r", encoding="utf-8") as f: - secrets = json.load(f) - except Exception as e: - print(f"Error loading secrets: {e}") - - # Add new account - secrets[result["account_name"]] = { - "secret": result["secret"], - "issuer": result["issuer"], - "algorithm": result["algorithm"], - "digits": result["digits"], - "period": result["period"], - } - - # Save secrets - try: - os.makedirs(os.path.dirname(secrets_file_path), exist_ok=True) - with open(secrets_file_path, "w", encoding="utf-8") as f: - json.dump(secrets, f, indent=2) - except Exception as e: - return { - "success": False, - "error": f"Error saving secrets: {str(e)}", - } - - display_name = ( - f"{result['issuer']} - {result['account_name']}" - if result["issuer"] - else result["account_name"] - ) - return { - "success": True, - "account_name": result["account_name"], - "display_name": display_name, - "message": f"Successfully added OTP account: {display_name}", - } - else: - return {"success": False, "error": "QR code is not an otpauth URI"} - - except Exception as e: - return {"success": False, "error": f"Error processing QR code: {str(e)}"} - - except Exception as e: - return {"success": False, "error": f"Error during QR scan: {str(e)}"} diff --git a/services/battery.py b/services/battery.py index a28cceb..10ce8ba 100644 --- a/services/battery.py +++ b/services/battery.py @@ -1,8 +1,10 @@ -import psutil -from gi.repository import GLib -from pydbus import SystemBus +from typing import Literal, Optional -from fabric.core import Property, Service, Signal +from fabric import Service, Signal +from gi.repository import Gio, GLib +from loguru import logger + +from utils.dbus_helper import GioDBusHelper DeviceState = { 0: "UNKNOWN", @@ -14,238 +16,173 @@ DeviceState = { 6: "PENDING_DISCHARGE", } +PowerProfile = { + "power-saver": "Power Saver", + "balanced": "Balanced", + "performance": "Performance", +} -class Battery(Service): - @staticmethod - def seconds_to_hours_minutes(seconds): - hours = seconds // 3600 - minutes = (seconds % 3600) // 60 - return f"{hours}h {minutes}m" if hours else f"{minutes}m" - @staticmethod - def get_battery_icon_level(percentage): - """Get battery icon level based on percentage""" - if percentage >= 90: - return "100" - elif percentage >= 80: - return "090" - elif percentage >= 70: - return "080" - elif percentage >= 60: - return "070" - elif percentage >= 50: - return "060" - elif percentage >= 40: - return "050" - elif percentage >= 30: - return "040" - elif percentage >= 20: - return "030" - elif percentage >= 10: - return "020" - else: - return "010" - - @staticmethod - def get_battery_icon_file(percentage, is_charging, base_path=""): - """Get battery icon file path""" - level = Battery.get_battery_icon_level(percentage) - suffix = "-charging" if is_charging else "" - return f"{base_path}battery/battery-{level}{suffix}.svg" - - @staticmethod - def get_profile_display_name(profile: str) -> str: - """Get user-friendly display name for power profile""" - profile_names = { - "power-saver": "Power Saver", - "powersave": "Power Saver", - "power_saver": "Power Saver", - "balanced": "Balanced", - "balance": "Balanced", - "performance": "Performance", - "performance-mode": "Performance", - } - return profile_names.get(profile, profile.title()) +class BatteryService(Service): + """Service to interact with UPower and Power Profiles via GIO D-Bus""" @Signal - def changed(self) -> None: ... + def changed(self) -> None: + """Signal emitted when battery changes.""" @Signal - def profile_changed(self, value: str) -> None: ... + def power_profile_changed(self) -> None: + """Signal emitted when power profile changes.""" - @Property(int, "readable") - def percentage(self): - if self._use_psutil_fallback: - if self._psutil_battery: - return int(self._psutil_battery.percent) - return 0 - return int(self._battery.Percentage) + _instance = None - @Property(str, "readable") - def temperature(self): - if self._use_psutil_fallback: - return "N/A" # psutil doesn't provide temperature - return ( - f"{self._battery.Temperature}°C" - if hasattr(self._battery, "Temperature") - else "N/A" + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + # UPower D-Bus configuration + self.bus_name = "org.freedesktop.UPower" + self.object_path = "/org/freedesktop/UPower/devices/DisplayDevice" + self.interface_name = "org.freedesktop.UPower.Device" + + self.dbus_helper = GioDBusHelper( + bus_type=Gio.BusType.SYSTEM, + bus_name=self.bus_name, + object_path=self.object_path, + interface_name=self.interface_name, ) - @Property(str, "readable") - def time_to_empty(self): - if self._use_psutil_fallback: - if self._psutil_battery and hasattr(self._psutil_battery, "secsleft"): - return self.seconds_to_hours_minutes(self._psutil_battery.secsleft) - return "N/A" - return self.seconds_to_hours_minutes(getattr(self._battery, "TimeToEmpty", 0)) + self.proxy = self.dbus_helper.proxy - @Property(str, "readable") - def time_to_full(self): - if self._use_psutil_fallback: - return "N/A" # psutil doesn't provide time to full - return self.seconds_to_hours_minutes(getattr(self._battery, "TimeToFull", 0)) + # Listen for PropertiesChanged signals from UPower + self.dbus_helper.listen_signal( + member="PropertiesChanged", + callback=self.handle_property_change, + ) - @Property(str, "readable") - def icon_name(self): - if self._use_psutil_fallback: - return "battery" # Generic icon name for psutil fallback - return self._battery.IconName + # Power Profiles D-Bus configuration + self.power_profile_bus_name = "net.hadess.PowerProfiles" + self.power_profile_object_path = "/net/hadess/PowerProfiles" + self.power_profile_interface_name = "net.hadess.PowerProfiles" - @Property(str, "readable") - def state(self): - if self._use_psutil_fallback: - if self._psutil_battery: - # psutil returns power_plugged boolean, convert to state - if self._psutil_battery.power_plugged: - if self._psutil_battery.percent >= 100: - return "FULLY_CHARGED" - else: - return "CHARGING" - else: - return "DISCHARGING" - return "UNKNOWN" - return DeviceState.get(self._battery.State, "UNKNOWN") - - @Property(str, "readable") - def capacity(self): - if self._use_psutil_fallback: - return "N/A" # psutil doesn't provide capacity info - return f"{int(self._battery.Capacity)}%" - - @Property(bool, "readable", default_value=False) - def is_present(self): - if self._use_psutil_fallback: - return self._psutil_battery is not None - return self._battery.IsPresent - - @Property(str, "readable") - def power_profile(self): - if hasattr(self, "_profile_proxy") and self._profile_proxy: - try: - return self._profile_proxy.ActiveProfile - except Exception: - return None - return None - - @Property(list, "readable") - def available_profiles(self): - if hasattr(self, "_profile_proxy") and self._profile_proxy: - try: - profiles = [] - for p in self._profile_proxy.Profiles: - if hasattr(p, "Profile"): - profiles.append(p.Profile) - elif isinstance(p, dict) and "Profile" in p: - profiles.append(p["Profile"]) - elif isinstance(p, str): - profiles.append(p) - return profiles - except Exception: - return [] - return [] - - def change_power_profile(self, profile: str) -> bool: - if not hasattr(self, "_profile_proxy") or not self._profile_proxy: - return False - - # Get available profiles using the same logic as available_profiles property - available_profiles = [] try: - for p in self._profile_proxy.Profiles: - if hasattr(p, "Profile"): - available_profiles.append(p.Profile) - elif isinstance(p, dict) and "Profile" in p: - available_profiles.append(p["Profile"]) - elif isinstance(p, str): - available_profiles.append(p) - except Exception: + self.power_profile_helper = GioDBusHelper( + bus_type=Gio.BusType.SYSTEM, + bus_name=self.power_profile_bus_name, + object_path=self.power_profile_object_path, + interface_name=self.power_profile_interface_name, + ) + + self.power_profile_proxy = self.power_profile_helper.proxy + + # Listen for PropertiesChanged signals from Power Profiles + self.power_profile_helper.listen_signal( + member="PropertiesChanged", + callback=self.handle_power_profile_change, + ) + + self._power_profiles_available = True + except Exception as e: + logger.warning(f"[Battery] Power Profiles daemon not available: {e}") + self.power_profile_helper = None + self.power_profile_proxy = None + self._power_profiles_available = False + + def get_property( + self, + property: Literal[ + "Percentage", + "Temperature", + "TimeToEmpty", + "TimeToFull", + "IconName", + "State", + "Capacity", + "IsPresent", + "Vendor", + ], + ): + try: + result = self.proxy.get_cached_property(property) + return result.unpack() if result is not None else None + except Exception as e: + logger.exception(f"[Battery] Error retrieving '{property}': {e}") + return None + + def get_power_profile(self) -> Optional[str]: + """Get the current active power profile.""" + if not self._power_profiles_available: + return None + + try: + result = self.power_profile_proxy.get_cached_property("ActiveProfile") + return result.unpack() if result is not None else None + except Exception as e: + logger.exception(f"[Battery] Error retrieving active power profile: {e}") + return None + + def set_power_profile( + self, profile: Literal["power-saver", "balanced", "performance"] + ) -> bool: + """Set the active power profile.""" + if not self._power_profiles_available: + logger.warning("[Battery] Power Profiles daemon not available") return False - if profile not in available_profiles: + if profile not in PowerProfile: return False try: - self._profile_proxy.ActiveProfile = profile - self.profile_changed.emit(profile) - self.changed.emit() + self.power_profile_helper.set_property( + interface_name=self.power_profile_interface_name, + property_name="ActiveProfile", + value_variant=GLib.Variant("s", profile), + ) return True - except Exception: + except Exception as e: + logger.exception( + f"[Battery] Error setting power profile to '{profile}': {e}" + ) return False - def __init__(self): - super().__init__() - self._bus = SystemBus() - self._use_psutil_fallback = False - self._psutil_battery = None - self._profile_proxy = None # Initialize to None first + def get_available_power_profiles(self) -> Optional[list]: + """Get list of available power profiles.""" + if not self._power_profiles_available: + return None - # Battery device try: - self._battery = self._bus.get( - "org.freedesktop.UPower", "/org/freedesktop/UPower/devices/battery_BAT0" + result = self.power_profile_proxy.get_cached_property("Profiles") + if result is not None: + profiles_data = result.unpack() + # Extract profile names from the array of dictionaries + profiles = [] + for profile_dict in profiles_data: + if "Profile" in profile_dict: + profiles.append(profile_dict["Profile"]) + return profiles + return None + except Exception as e: + logger.exception( + f"[Battery] Error retrieving available power profiles: {e}" ) - self._battery.onPropertiesChanged = self.handle_battery_change - except Exception: - # Fallback to psutil if UPower is not available - self._use_psutil_fallback = True - try: - self._psutil_battery = psutil.sensors_battery() - if self._psutil_battery is None: - return # No battery found - # Start periodic updates for psutil fallback - increased interval - GLib.timeout_add_seconds(10, self._update_psutil_battery) - except Exception: - return # psutil battery not available either + return None - # PowerProfiles - Initialize after other attributes - try: - self._profile_proxy = self._bus.get( - "net.hadess.PowerProfiles", "/net/hadess/PowerProfiles" - ) - # Use onPropertiesChanged for consistency with battery device - self._profile_proxy.onPropertiesChanged = ( - lambda _, changed, __: self._handle_profile_props_changed(changed) - ) - except Exception: - self._profile_proxy = None + def is_power_profiles_available(self) -> bool: + """Check if power profiles daemon is available.""" + return self._power_profiles_available - self.changed.emit() + def get_power_profile_display_name(self, profile: str) -> str: + """Get display name for a power profile.""" + return PowerProfile.get(profile, profile.title()) - def _update_psutil_battery(self): - """Update psutil battery data periodically""" - try: - self._psutil_battery = psutil.sensors_battery() - self.changed.emit() - except Exception: - pass # Continue trying - return True # Keep the timeout active + def handle_property_change(self, *_): + # You may filter which property changed by checking parameters[1] + self.emit("changed") - def _handle_profile_props_changed(self, changed): - """Internal handler for property changes that processes only the changed properties""" - if "ActiveProfile" in changed: - new_profile = changed["ActiveProfile"] - self.profile_changed.emit(new_profile) - self.changed.emit() - - def handle_battery_change(self, iface, changed, invalidated): - self.changed.emit() + def handle_power_profile_change(self, *_): + """Handle power profile property changes.""" + self.emit("power_profile_changed") diff --git a/services/brightness.py b/services/brightness.py index 1031444..0723e2f 100644 --- a/services/brightness.py +++ b/services/brightness.py @@ -1,10 +1,5 @@ -import os - -from gi.repository import GLib -from loguru import logger - from fabric.core.service import Property, Service, Signal -from fabric.utils import exec_shell_command_async, monitor_file +from fabric.utils import GLib, exec_shell_command_async, logger, monitor_file, os def exec_brightnessctl_async(args: str): diff --git a/services/config.py b/services/config.py new file mode 100644 index 0000000..fa57c9b --- /dev/null +++ b/services/config.py @@ -0,0 +1,162 @@ +""" +Centralized configuration service with file watching and dynamic reloads. + +This service encapsulates the logic for loading the application's JSON config, +watching for file changes, and notifying registered listeners when the config +changes. It is implemented as a singleton and intended to be reused by any +module that needs dynamic configuration. +""" + +from __future__ import annotations + +import json +from typing import Any, Callable, Dict, List, Optional + +from fabric.utils import Gio, GLib, get_relative_path, logger, os + + +class ConfigService: + """Singleton service handling config state and reload notifications.""" + + _instance: Optional["ConfigService"] = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__(self): + if getattr(self, "_initialized", False): + return + + self._initialized = True + self._config: Dict[str, Any] = {} + self._reload_callbacks: List[ + Callable[[Dict[str, Any], Dict[str, Any]], None] + ] = [] + self._config_file: str = get_relative_path("../config.json") + self._monitors: List[Gio.FileMonitor] = [] + self._reload_pending: bool = False + self.RELOAD_DELAY_MS: int = 100 + + self._load_config() + self._setup_monitors() + + def get(self, key: str, default: Any = None) -> Any: + return self._config.get(key, default) + + def get_all(self) -> Dict[str, Any]: + return self._config.copy() + + def has_changed(self, key: str, old_config: Dict[str, Any]) -> bool: + return self._config.get(key) != old_config.get(key) + + def register_reload_callback( + self, callback: Callable[[Dict[str, Any], Dict[str, Any]], None] + ) -> None: + if callback not in self._reload_callbacks: + self._reload_callbacks.append(callback) + + def unregister_reload_callback( + self, callback: Callable[[Dict[str, Any], Dict[str, Any]], None] + ) -> None: + if callback in self._reload_callbacks: + self._reload_callbacks.remove(callback) + + def stop(self) -> None: + for monitor in self._monitors: + try: + monitor.cancel() + except Exception: + pass + self._monitors.clear() + + def _load_config(self) -> None: + try: + if os.path.exists(self._config_file): + with open(self._config_file, "r") as f: + self._config = json.load(f) + else: + self._config = {} + except Exception as e: + logger.error(f"[ConfigService] Failed to load config: {e}") + self._config = {} + + def _setup_monitors(self) -> None: + files_to_watch = [self._config_file] + for file_path in files_to_watch: + if not os.path.exists(file_path): + continue + try: + gio_file = Gio.File.new_for_path(file_path) + monitor = gio_file.monitor_file(Gio.FileMonitorFlags.NONE, None) + monitor.connect("changed", self._on_file_changed, file_path) + self._monitors.append(monitor) + except Exception as e: + logger.error(f"[ConfigService] Failed to monitor {file_path}: {e}") + + def _on_file_changed(self, monitor, file, other_file, event_type, file_path: str): + if ( + event_type == Gio.FileMonitorEvent.CHANGES_DONE_HINT + and not self._reload_pending + ): + self._reload_pending = True + GLib.timeout_add(self.RELOAD_DELAY_MS, self._reload_config) + + def _reload_config(self) -> bool: + try: + old_config = self._config.copy() + self._load_config() + + for callback in list(self._reload_callbacks): + try: + callback(self._config, old_config) + except Exception as e: + logger.error(f"[ConfigService] Callback failed: {e}") + + self._reload_pending = False + return True + except Exception as e: + logger.error(f"[ConfigService] Failed to reload configuration: {e}") + self._reload_pending = False + return False + + +_service: Optional[ConfigService] = None + + +def start_config_service() -> ConfigService: + global _service + if _service is None: + _service = ConfigService() + return _service + + +def stop_config_service() -> None: + global _service + if _service is not None: + _service.stop() + _service = None + + +# Convenience helpers for concise usage +def config() -> ConfigService: + """Get the singleton service (concise alias).""" + return start_config_service() + + +def get_config(key: str, default: Any = None) -> Any: + """Get a single config value from the singleton service.""" + return start_config_service().get(key, default) + + +def get_all_config() -> Dict[str, Any]: + """Get all config values from the singleton service.""" + return start_config_service().get_all() + + +def on_config_change( + callback: Callable[[Dict[str, Any], Dict[str, Any]], None], +) -> None: + """Register a reload callback on the singleton service.""" + start_config_service().register_reload_callback(callback) diff --git a/services/custom_notification.py b/services/custom_notification.py index 7619887..4386ca1 100644 --- a/services/custom_notification.py +++ b/services/custom_notification.py @@ -1,18 +1,8 @@ # Standard library imports import json -import os import time from typing import List -# Fabric imports -import gi - -gi.require_version("Gtk", "3.0") -gi.require_version("GdkPixbuf", "2.0") - -from gi.repository import GdkPixbuf - -import config.data as data from fabric.core.service import Property, Service, Signal from fabric.notifications import ( Notification, @@ -21,8 +11,10 @@ from fabric.notifications import ( Notifications, ) -gi.require_version("Gtk", "3.0") -gi.require_version("GdkPixbuf", "2.0") +# Fabric imports +from fabric.utils import GdkPixbuf, os + +import config.data as data NOTIFICATION_CACHE_FILE = f"{data.CACHE_DIR}/notification_history.json" @@ -36,11 +28,11 @@ class CachedNotification(Service): Service.__init__(self, **kwargs) self._notification = Notification.deserialize(data) self._cache_id = data["cached-id"] # Set directly to private var - + # Store cache metadata for cleanup self.cache_metadata = data.get("cache_metadata", {}) self.timestamp = data.get("timestamp", int(time.time())) - + return self @Signal @@ -113,29 +105,33 @@ class CachedNotification(Service): get_cache_key, get_notification_image_cache_key, ) - + # Get better cache keys for icons app_icon_cache_key = None notification_image_cache_key = None - + if self.app_icon: app_icon_cache_key = get_cache_key(self.app_icon, (35, 35), self.app_name) - + # Only try to get notification image cache key if we can safely access image_pixbuf if self.id: try: # First check if we already have the cache key stored - if hasattr(self, 'cache_metadata') and self.cache_metadata: - notification_image_cache_key = self.cache_metadata.get('notification_image_cache_key') - + if hasattr(self, "cache_metadata") and self.cache_metadata: + notification_image_cache_key = self.cache_metadata.get( + "notification_image_cache_key" + ) + # If not, try to generate it safely - if not notification_image_cache_key and hasattr(self._notification, 'image_pixbuf'): + if not notification_image_cache_key and hasattr( + self._notification, "image_pixbuf" + ): try: # Check if image_pixbuf exists and can be accessed without loading from file - image_pixbuf = getattr(self._notification, 'image_pixbuf', None) + image_pixbuf = getattr(self._notification, "image_pixbuf", None) if image_pixbuf: - notification_image_cache_key = get_notification_image_cache_key( - self.id, image_pixbuf + notification_image_cache_key = ( + get_notification_image_cache_key(self.id, image_pixbuf) ) except (AttributeError, OSError, Exception): # If temp file is gone or any other error, just mark as None @@ -143,7 +139,7 @@ class CachedNotification(Service): except Exception: # If any error occurs during cache key generation, skip it pass - + return { "cached-id": self.cache_id, "id": self.id, @@ -164,8 +160,8 @@ class CachedNotification(Service): "app_icon_cache_key": app_icon_cache_key, "notification_image_cache_key": notification_image_cache_key, "has_cached_image": notification_image_cache_key is not None, - "cache_timestamp": int(time.time()) - } + "cache_timestamp": int(time.time()), + }, } def __init__(self, notification: Notification, cache_id: int, **kwargs): @@ -211,7 +207,7 @@ class CachedNotifications(Notifications): def dont_disturb(self) -> bool: """Return the pause status.""" return self._dont_disturb - + def set_dont_disturb(self, value: bool): """Set the pause status.""" self._dont_disturb = value @@ -224,10 +220,12 @@ class CachedNotifications(Notifications): self._dont_disturb = False self._count = 0 self._next_cache_id = 1 # Track next available cache ID - self._session_start_time = int(time.time()) # Track session start time for deduplication + self._session_start_time = int( + time.time() + ) # Track session start time for deduplication self.load_cached_notifications() - + # Connect to the notification_added signal to cache new notifications # Note: self here refers to the CachedNotifications service, which inherits from Notifications # So we connect to our own notification_added signal @@ -247,12 +245,10 @@ class CachedNotifications(Notifications): cached_notification = CachedNotification.create_from_dict(notification) cache_id = cached_notification.cache_id max_cache_id = max(max_cache_id, cache_id) - + handler_id = cached_notification.connect( "removed-from-cache", - lambda *args: self.remove_cached_notification( - notification_id=cache_id - ), + lambda *args: self.remove_cached_notification(notification_id=cache_id), ) self._signal_handlers[cache_id] = handler_id self._cached_notifications[cache_id] = cached_notification @@ -278,15 +274,15 @@ class CachedNotifications(Notifications): """Empty the notifications with enhanced cache cleanup""" # Clean up all cached files before clearing notifications from modules.notification.notification import cleanup_all_notification_caches - + for cached_notification in self._cached_notifications.values(): handler_id = self._signal_handlers.pop(cached_notification.cache_id, None) if handler_id: cached_notification.disconnect(handler_id) - + # Clear all notification caches (icons and images) cleanup_all_notification_caches() - + self._cached_notifications = {} self.cache_notifications() self._count = 0 @@ -297,55 +293,66 @@ class CachedNotifications(Notifications): def on_notification_added(self, service, notification_id: int) -> None: """Handle notification added and cache it with enhanced metadata - GUARANTEED STORAGE""" # Don't call super() - we're handling this ourselves - + # Import logger at the top of the function from loguru import logger - + notification = self.get_notification_from_id(notification_id) if not notification: - logger.error(f"CRITICAL: Failed to get notification with ID {notification_id}") + logger.error( + f"CRITICAL: Failed to get notification with ID {notification_id}" + ) return # Import here to avoid circular imports from config import data from modules.notification.notification import ( - preload_notification_assets, cache_notification_icon, cache_notification_image, get_cache_key, - get_notification_image_cache_key + get_notification_image_cache_key, + preload_notification_assets, ) # Check if this app should be ignored for history (don't cache) if notification.app_name in data.NOTIFICATION_IGNORED_APPS_HISTORY: # Don't cache notifications from ignored apps, but still allow popup display - logger.debug(f"Ignoring notification from {notification.app_name} (in ignore list)") + logger.debug( + f"Ignoring notification from {notification.app_name} (in ignore list)" + ) return # Check for duplicates using both notification ID and timestamp to avoid session restart issues existing_notification = None current_time = int(time.time()) - + for cached_notif in self._cached_notifications.values(): # Only consider it a duplicate if: # 1. Same notification ID AND - # 2. Notification was cached in the current session (after session start time) AND + # 2. Notification was cached in the current session (after session start time) AND # 3. Notification was cached recently (within last 5 minutes) - cached_time = getattr(cached_notif, 'timestamp', 0) + cached_time = getattr(cached_notif, "timestamp", 0) is_recent = (current_time - cached_time) < 300 # 5 minutes is_current_session = cached_time >= self._session_start_time - - if (cached_notif._notification.id == notification.id and - is_current_session and is_recent): + + if ( + cached_notif._notification.id == notification.id + and is_current_session + and is_recent + ): existing_notification = cached_notif break - + if existing_notification: - logger.debug(f"Notification ID {notification.id} already cached in current session, skipping") + logger.debug( + f"Notification ID {notification.id} already cached in current session, skipping" + ) return - logger.debug(f"Caching new notification: ID={notification.id}, App={notification.app_name}, Summary={notification.summary[:50]}...") + logger.debug( + f"Caching new notification: ID={notification.id}, App={notification.app_name}, Summary={notification.summary[:50]}..." + ) # GUARANTEED STORAGE: Always create and store notification to history first cache_id = self._next_cache_id @@ -355,17 +362,17 @@ class CachedNotifications(Notifications): cached_notification = CachedNotification( notification=notification, cache_id=cache_id ) - # Set cache_id directly since it's read-only property + # Set cache_id directly since it's read-only property cached_notification._cache_id = cache_id - + # Initialize cache metadata (will be populated below) cached_notification.cache_metadata = { "app_icon_cache_key": None, "notification_image_cache_key": None, "has_cached_image": False, - "cache_timestamp": int(time.time()) + "cache_timestamp": int(time.time()), } - + # IMMEDIATELY store to history before attempting any caching operations handler_id = cached_notification.connect( "removed-from-cache", @@ -373,83 +380,110 @@ class CachedNotifications(Notifications): ) self._signal_handlers[cache_id] = handler_id self._cached_notifications[cache_id] = cached_notification - + # Save to JSON file immediately - GUARANTEED STORAGE try: self.cache_notifications() logger.debug(f"GUARANTEED: Notification {cache_id} stored to history") except Exception as e: - logger.error(f"CRITICAL: Failed to save notification {cache_id} to history: {e}") - + logger.error( + f"CRITICAL: Failed to save notification {cache_id} to history: {e}" + ) + # Now attempt asset caching (failures here won't affect history storage) try: # Preload assets and store cache metadata preload_notification_assets(notification) - + # Store enhanced cache metadata app_icon_cache_key = None notification_image_cache_key = None - + if notification.app_icon: try: # Only cache at 35x35 to reduce disk usage - headers will scale this down - app_icon_cache_key = get_cache_key(notification.app_icon, (35, 35), notification.app_name) - cache_notification_icon(notification.app_icon, (35, 35), notification.app_name) - cached_notification.cache_metadata["app_icon_cache_key"] = app_icon_cache_key + app_icon_cache_key = get_cache_key( + notification.app_icon, (35, 35), notification.app_name + ) + cache_notification_icon( + notification.app_icon, (35, 35), notification.app_name + ) + cached_notification.cache_metadata["app_icon_cache_key"] = ( + app_icon_cache_key + ) logger.debug(f"Cached app icon (35x35) for notification {cache_id}") except Exception as e: - logger.warning(f"Failed to cache app icon for notification {cache_id}: {e}") - - if hasattr(notification, 'image_pixbuf'): + logger.warning( + f"Failed to cache app icon for notification {cache_id}: {e}" + ) + + if hasattr(notification, "image_pixbuf"): try: # Safely try to access image_pixbuf - image_pixbuf = getattr(notification, 'image_pixbuf', None) + image_pixbuf = getattr(notification, "image_pixbuf", None) if image_pixbuf: notification_image_cache_key = get_notification_image_cache_key( notification.id, image_pixbuf ) - cache_notification_image(notification.id, image_pixbuf, (35, 35)) - cached_notification.cache_metadata["notification_image_cache_key"] = notification_image_cache_key + cache_notification_image( + notification.id, image_pixbuf, (35, 35) + ) + cached_notification.cache_metadata[ + "notification_image_cache_key" + ] = notification_image_cache_key cached_notification.cache_metadata["has_cached_image"] = True - logger.debug(f"Cached notification image for notification {cache_id}") + logger.debug( + f"Cached notification image for notification {cache_id}" + ) except (AttributeError, OSError, Exception) as e: - logger.warning(f"Failed to cache notification image for notification {cache_id}: {e}") - + logger.warning( + f"Failed to cache notification image for notification {cache_id}: {e}" + ) + # Update cached notification with final metadata self._cached_notifications[cache_id] = cached_notification - + # Save updated metadata to JSON self.cache_notifications() - + except Exception as e: - logger.error(f"Asset caching failed for notification {cache_id}, but notification is still stored: {e}") + logger.error( + f"Asset caching failed for notification {cache_id}, but notification is still stored: {e}" + ) # Always emit signals regardless of caching success self.notify("count") self.emit("cached-notification-added", cached_notification) - - logger.debug(f"Successfully processed notification: Cache ID={cache_id}, Total cached={len(self._cached_notifications)}") + + logger.debug( + f"Successfully processed notification: Cache ID={cache_id}, Total cached={len(self._cached_notifications)}" + ) def remove_cached_notification(self, notification_id: int): """Remove the notification of given id with enhanced cache cleanup""" if notification_id in self._cached_notifications: cached_notification = self._cached_notifications.pop(notification_id) - + # Enhanced cache cleanup using stored metadata - if hasattr(cached_notification, 'cache_metadata'): + if hasattr(cached_notification, "cache_metadata"): cache_metadata = cached_notification.cache_metadata - + # Clean up specific cached files using stored keys - from modules.notification.notification import cleanup_notification_specific_caches + from modules.notification.notification import ( + cleanup_notification_specific_caches, + ) + cleanup_notification_specific_caches( app_icon_source=cached_notification.app_icon, - notification_image_cache_key=cache_metadata.get('notification_image_cache_key') + notification_image_cache_key=cache_metadata.get( + "notification_image_cache_key" + ), ) - + self.cache_notifications() # Update JSON self._count -= 1 self.notify("count") - + # Get the stored signal handler ID and disconnect it handler_id = self._signal_handlers.pop(notification_id, None) if handler_id: diff --git a/services/modus.py b/services/modus.py index f9e9444..cb1b042 100644 --- a/services/modus.py +++ b/services/modus.py @@ -2,7 +2,7 @@ import json from fabric.core.service import Property, Service, Signal from fabric.hyprland.service import Hyprland -from loguru import logger +from fabric.utils import logger from services.custom_notification import CachedNotifications @@ -269,23 +269,24 @@ class ModusService(Service): def _setup_active_window_monitoring(self): """Setup active window monitoring""" try: - if not hasattr(self, '_hyprland_connection') or not self._hyprland_connection: + if ( + not hasattr(self, "_hyprland_connection") + or not self._hyprland_connection + ): return - # Get initial active window self._update_active_window() - - # Note: The HyprlandActiveWindow widget from Fabric library - # should handle active window updates automatically. - # We just need to ensure the initial state is correct. - except Exception as e: - logger.error(f"[ModusService] Failed to setup active window monitoring: {e}") + logger.error( + f"[ModusService] Failed to setup active window monitoring: {e}" + ) def _update_active_window(self): - """Update the current active app name based on active window""" try: - if not hasattr(self, '_hyprland_connection') or not self._hyprland_connection: + if ( + not hasattr(self, "_hyprland_connection") + or not self._hyprland_connection + ): return window_data = self._hyprland_connection.send_command("j/activewindow").reply @@ -297,15 +298,12 @@ class ModusService(Service): wmclass = window_info.get("class", "") title = window_info.get("title", "") - # Handle the case when there's no active window if not title and not wmclass: self.current_active_app_name = "Finder" return - # Simple app name formatting without circular import name = wmclass if wmclass else title if name: - # Basic formatting: capitalize first letter and remove file extensions name = str(name).title() if "." in name: name = name.split(".")[-1] @@ -319,7 +317,6 @@ class ModusService(Service): self.current_active_app_name = "Finder" def _on_workspace_changed(self, obj, signal): - """Handle workspace change events from Hyprland""" try: workspace_name = json.loads(signal.data[0]) self.current_workspace = str(workspace_name) diff --git a/services/mpris.py b/services/mpris.py index 539a727..338dac8 100644 --- a/services/mpris.py +++ b/services/mpris.py @@ -1,303 +1,446 @@ -# Standard library imports -import contextlib +from typing import Literal, Optional -import gi - -# Fabric imports from fabric.core.service import Property, Service, Signal -from fabric.utils import bulk_connect -from gi.repository import GLib -from loguru import logger +from fabric.utils import Gio, GLib, logger +from fabric.utils.helpers import ( + bulk_connect, + clamp, + pascal_case_to_snake_case, + snake_case_to_kebab_case, +) +from gi.repository.GLib import Variant + +from utils.dbus_helper import GioDBusHelper + +MPRIS_MEDIAPLAYER_BUS_NAME = "org.mpris.MediaPlayer2" +MPRIS_MEDIAPLAYER_BUS_PATH = "/org/mpris/MediaPlayer2" +MPRIS_MEDIAPLAYER_PLAYER_BUS_NAME = "org.mpris.MediaPlayer2.Player" -class PlayerctlImportError(ImportError): - def __init__(self, *args): - super().__init__( - "Playerctl is not installed, please install it first", - *args, - ) - - -try: - gi.require_version("Playerctl", "2.0") - from gi.repository import Playerctl -except ValueError: - raise PlayerctlImportError +# TODO Shuffle is broken because there is no CanShuffle property, it relies on can_control but thats not all inclusive class MprisPlayer(Service): - """A service to manage a mpris player.""" - @Signal - def exit(self, value: bool) -> bool: ... + def closed(self, value: bool) -> bool: ... @Signal def changed(self) -> None: ... - def __init__( - self, - player: Playerctl.Player, - **kwargs, - ): - self._signal_connectors: dict = {} - self._player: Playerctl.Player = player - super().__init__(**kwargs) - for sn in ["playback-status", "loop-status", "shuffle"]: - self._signal_connectors[sn] = self._player.connect( - sn, - lambda *args, sn=sn: self.notifier(sn, args), + # TODO: why??? object + # TypeError: can't convert return value to desired type + @Signal + def seeked(self, position: int) -> object: ... + + @Property(str, "readable") + def player_name(self): + return self.bus_name.split(".", 3)[-1].split(".")[0] + + @Property(str, "readable") + def playback_status(self) -> Literal["Playing", "Paused", "Stopped"]: + # type: ignore + return self._dbus_helper.proxy.get_cached_property( + "PlaybackStatus" + ).get_string() + + @Property(str, "read-write", default_value="None") + def loop_status(self) -> Literal["None", "Track", "Playlist"]: + # type: ignore + return self._dbus_helper.proxy.get_cached_property("LoopStatus").get_string() + + @loop_status.setter + def loop_status(self, status: Literal["None", "Track", "Playlist"]) -> None: + if self._dbus_helper.proxy.get_cached_property("LoopStatus") is None: + return + + ( + self._dbus_helper.set_property( + MPRIS_MEDIAPLAYER_PLAYER_BUS_NAME, + "LoopStatus", + GLib.Variant("s", value=status), ) - - self._signal_connectors["exit"] = self._player.connect( - "exit", - self.on_player_exit, + if self.can_control + else None ) - self._signal_connectors["metadata"] = self._player.connect( - "metadata", - lambda *_: self.update_status(), - ) - GLib.idle_add(self.update_status_once) - def update_status(self): - # schedule each notifier asynchronously. - def notify_property(prop): - if self.get_property(prop) is not None: - self.notifier(prop) - - for prop in [ - "metadata", - "title", - "artist", - "arturl", - "length", - ]: - GLib.idle_add(lambda p=prop: (notify_property(p), False)) - for prop in [ - "can-seek", - "can-pause", - "can-shuffle", - "can-go-next", - "can-go-previous", - ]: - GLib.idle_add(lambda p=prop: (self.notifier(p), False)) - - def update_status_once(self): - # schedule notifier calls for each property - def notify_all(): - for prop in self.list_properties(): # type: ignore - self.notifier(prop.name) - return False - - GLib.idle_add(notify_all, priority=GLib.PRIORITY_DEFAULT_IDLE) - - def notifier(self, name: str, args=None): - def notify_and_emit(): - self.notify(name) - self.emit("changed") - return False - - GLib.idle_add(notify_and_emit, priority=GLib.PRIORITY_DEFAULT_IDLE) - - def on_player_exit(self, player): - for id in list(self._signal_connectors.values()): - with contextlib.suppress(Exception): - self._player.disconnect(id) - del self._signal_connectors - GLib.idle_add(lambda: (self.emit("exit", True), False)) - del self._player - - def toggle_shuffle(self, *_): - if self.can_shuffle: - # schedule the shuffle toggle in the GLib idle loop - GLib.idle_add(lambda: (setattr(self, "shuffle", not self.shuffle), False)) - # else do nothing - - def play_pause(self, *_): - if self.can_pause: - GLib.idle_add(lambda: (self._player.play_pause(), False)) - - def next(self, *_): - if self.can_go_next: - GLib.idle_add(lambda: (self._player.next(), False)) - - def previous(self, *_): - if self.can_go_previous: - GLib.idle_add(lambda: (self._player.previous(), False)) - - # Properties - @Property(str, "readable") - def player_name(self) -> int: - return self._player.get_property("player-name") # type: ignore - - @Property(int, "read-write", default_value=0) - def position(self) -> int: - return self._player.get_property("position") # type: ignore - - @position.setter - def position(self, new_pos: int): - self._player.set_position(new_pos) - - @Property(object, "readable") - def metadata(self) -> dict: - return self._player.get_property("metadata") # type: ignore - - @Property(str or None, "readable") - def arturl(self) -> str | None: - if "mpris:artUrl" in self.metadata.keys(): # type: ignore # noqa: SIM118 - return self.metadata["mpris:artUrl"] # type: ignore - return None - - @Property(str or None, "readable") - def length(self) -> str | None: - if "mpris:length" in self.metadata.keys(): # type: ignore # noqa: SIM118 - return self.metadata["mpris:length"] # type: ignore - return None - - @Property(str, "readable") - def artist(self) -> str: - artist = self._player.get_artist() # type: ignore - if isinstance(artist, (list, tuple)): - return ", ".join(artist) - return artist - - @Property(str, "readable") - def album(self) -> str: - return self._player.get_album() # type: ignore - - @Property(str, "readable") - def title(self): - return self._player.get_title() + # TODO: Playback rate??? (Rate) idk i dont really see why someone would use this ngl @Property(bool, "read-write", default_value=False) def shuffle(self) -> bool: - return self._player.get_property("shuffle") # type: ignore + if self._dbus_helper.proxy.get_cached_property("Shuffle"): + # type: ignore + return self._dbus_helper.proxy.get_cached_property("Shuffle").get_boolean() + return False @shuffle.setter - def shuffle(self, do_shuffle: bool): - self.notifier("shuffle") - return self._player.set_shuffle(do_shuffle) + def shuffle(self, is_shuffle: bool) -> None: + if self._dbus_helper.proxy.get_cached_property("Shuffle") is None: + return + + ( + self._dbus_helper.set_property( + MPRIS_MEDIAPLAYER_PLAYER_BUS_NAME, + "Shuffle", + GLib.Variant("b", value=is_shuffle), + ) + if self.can_control + else None + ) + + @Property(dict, "readable") + def metadata(self) -> dict: + prop: Variant | None = self._dbus_helper.proxy.get_cached_property("Metadata") # type: ignore + return dict(prop) if prop else {} # type: ignore + + # RELY ON METADATA + @Property(str, "readable") + def arturl(self) -> str: + return self.metadata.get("mpris:artUrl", "") + + @Property(int, "readable", default_value=0) + def length(self) -> int: + return self.metadata.get("mpris:length", 0) + + @Property(list, "readable") + def artist(self) -> list: + return self.metadata.get("xesam:artist", "") @Property(str, "readable") - def playback_status(self) -> str: - return { - Playerctl.PlaybackStatus.PAUSED: "paused", - Playerctl.PlaybackStatus.PLAYING: "playing", - Playerctl.PlaybackStatus.STOPPED: "stopped", - # type: ignore - }.get(self._player.get_property("playback_status"), "unknown") + def album(self) -> str: + return self.metadata.get("xesam:album", "") - @Property(str, "read-write") - def loop_status(self) -> str: - return { - Playerctl.LoopStatus.NONE: "none", - Playerctl.LoopStatus.TRACK: "track", - Playerctl.LoopStatus.PLAYLIST: "playlist", - }.get( - self._player.get_property("loop_status"), "unknown" - ) # type: ignore + @Property(str, "readable") + def title(self): + return self.metadata.get("xesam:title", "") - @loop_status.setter - def loop_status(self, status: str): - loop_status = { - "none": Playerctl.LoopStatus.NONE, - "track": Playerctl.LoopStatus.TRACK, - "playlist": Playerctl.LoopStatus.PLAYLIST, - }.get(status) - self._player.set_loop_status(loop_status) if loop_status else None + # END RELY ON METADATA + + @Property(float, "read-write") + def volume(self) -> float: + # type: ignore + return self._dbus_helper.proxy.get_cached_property("Volume").get_double() + + @volume.setter + def volume(self, volume: float) -> None: + ( + self._dbus_helper.set_property( + MPRIS_MEDIAPLAYER_PLAYER_BUS_NAME, + "Volume", + GLib.Variant("d", clamp(value=volume, min_value=0.0, max_value=1.0)), + ) + if self.can_control + else None + ) + + @Property(int, "read-write", default_value=0) + def position(self) -> int: + # type: ignore + return self._dbus_helper.proxy.get_cached_property("Position").get_int64() + + @position.setter + def position(self, new_pos: int) -> None: + self._dbus_helper.call_method( + self.bus_name, + MPRIS_MEDIAPLAYER_BUS_PATH, + MPRIS_MEDIAPLAYER_PLAYER_BUS_NAME, + "SetPosition", + GLib.Variant( + "(ox)", + (self.metadata["mpris:trackid"], new_pos), + ), + ) + + # TODO: consider MinimumRate, MaximumRate @Property(bool, "readable", default_value=False) def can_go_next(self) -> bool: - return self._player.get_property("can_go_next") # type: ignore + return self._dbus_helper.proxy.get_cached_property("CanGoNext").get_boolean() @Property(bool, "readable", default_value=False) def can_go_previous(self) -> bool: - return self._player.get_property("can_go_previous") # type: ignore + return self._dbus_helper.proxy.get_cached_property( + "CanGoPrevious" + ).get_boolean() @Property(bool, "readable", default_value=False) - def can_seek(self) -> bool: - return self._player.get_property("can_seek") # type: ignore + def can_play(self) -> bool: + return self._dbus_helper.proxy.get_cached_property("CanPlay").get_boolean() @Property(bool, "readable", default_value=False) def can_pause(self) -> bool: - return self._player.get_property("can_pause") # type: ignore + return self._dbus_helper.proxy.get_cached_property("CanPause").get_boolean() @Property(bool, "readable", default_value=False) - def can_shuffle(self) -> bool: - try: - self._player.set_shuffle(self._player.get_property("shuffle")) - return True - except Exception: - return False + def can_seek(self) -> bool: + return self._dbus_helper.proxy.get_cached_property("CanSeek").get_boolean() @Property(bool, "readable", default_value=False) - def can_loop(self) -> bool: + def can_control(self) -> bool: + return self._dbus_helper.proxy.get_cached_property("CanControl").get_boolean() + + def __init__(self, bus_name: str, **kwargs): + super().__init__(**kwargs) + self.bus_name: str = bus_name + self._dbus_helper: GioDBusHelper | None = None + + # Ahoy! + self.do_register() + + def do_register(self) -> None: # the bus id + self._dbus_helper = GioDBusHelper( + self.bus_name, + MPRIS_MEDIAPLAYER_BUS_PATH, + MPRIS_MEDIAPLAYER_PLAYER_BUS_NAME, + Gio.BusType.SESSION, + ) + + if not self._dbus_helper.proxy: + return + + bulk_connect( + self._dbus_helper.proxy, + { + "g-properties-changed": self._do_handle_properties_changed, + "g-signal": self._do_handle_signal_changed, + "notify::g-name-owner": self._on_name_owner_change, + }, + ) + # Update all properties to start + self.update_all_properties() + + def update_all_properties(self): + def on_update_finish(proxy, task): + try: + self._do_handle_properties_changed( + self._dbus_helper.proxy, proxy.call_finish(task)[0], "" + ) + except Exception: + logger.error(f"[MPRIS-{self.bus_name}] Failed to retrieve properties") + + self._dbus_helper.proxy.call( + "org.freedesktop.DBus.Properties.GetAll", + GLib.Variant.new_tuple( + GLib.Variant.new_string("org.mpris.MediaPlayer2.Player") + ), + Gio.DBusCallFlags.NONE, + -1, + None, + on_update_finish, + ) # User data + + def _do_handle_properties_changed( + self, proxy: Gio.DBusProxy, changed_properties, invalidated_properties: str + ): + for prop_name in set( + [ + snake_case_to_kebab_case(pascal_case_to_snake_case(x)) + for x in changed_properties.keys() + ] + ).intersection([prop.name for prop in self.get_properties()]): + self.notifier(prop_name) + + if prop_name == "metadata": + for sub_prop in ["arturl", "album", "artist", "length", "title"]: + self.notifier(sub_prop) + + def notifier(self, prop): + self.notify(prop) + self.changed() + + def _do_handle_signal_changed( + self, + proxy: Gio.DBusProxy, + sender_name: str, + signal_name: str, + params: tuple[GLib.Variant], + ): + # Only One Signal for Mpris + if signal_name == "Seeked": + self.seeked(params[0]) + + def _proxy_call(self, method_name: str, parameter: Optional[GLib.Variant]): + self._dbus_helper.proxy.call( + method_name, + parameter, + Gio.DBusCallFlags.NONE, + self._dbus_helper.proxy.get_default_timeout(), + None, + self._do_method_callback, + pascal_case_to_snake_case(method_name), + ) + + def _do_method_callback(self, _, res: Gio.AsyncResult, user_data): try: - self._player.set_shuffle(self._player.get_property("shuffle")) - return True - except Exception: - return False + self._dbus_helper.proxy.call_finish(res) + except Exception as e: + print(e) + logger.error( + f"[MPRIS-{self.bus_name}] Failed to invoke method: {user_data}" + ) + + def _on_name_owner_change(self, proxy: Gio.DBusProxy, _): + if not self._dbus_helper.proxy.get_name_owner(): + # proxy is automatically destroyed + self.closed(True) + + # Methods + def next(self): + self._proxy_call("Next", None) if self.can_go_next else None + + def previous(self): + self._proxy_call("Previous", None) if self.can_go_previous else None + + def pause(self): + self._proxy_call("Pause", None) if self.can_pause else None + + def play_pause(self): + ( + self._proxy_call("PlayPause", None) + if self.can_pause + else logger.error( + f"[MPRIS-{self.bus_name}] `play_pause` is not supported by this player" + ) + ) + + def stop(self): + ( + self._proxy_call("Stop", None) + if self.can_control + else logger.error( + f"[MPRIS-{self.bus_name}] `stop` is not supported by this player" + ) + ) + + def play(self): + self._proxy_call("Play", None) if self.can_play else None + + def seek(self, time_in_us: int): + ( + self._proxy_call("Seek", GLib.Variant.new_int64(time_in_us)) + if self.can_seek + else None + ) + + # No need for SetPositition, that is handled by the setter + + # Ignoring OpenUri class MprisPlayerManager(Service): - """A service to manage mpris players.""" + @Signal + def player_appeared(self, player: MprisPlayer) -> MprisPlayer: + logger.info(f"[MPRIS] Found Player: {player.bus_name}") + self._players[player.bus_name] = player + self.notify("players") + return player @Signal - def player_appeared(self, player: Playerctl.Player) -> Playerctl.Player: ... + def player_vanished(self, bus_name: str) -> str: + logger.info(f"[MPRIS] Lost Player: {bus_name}") + self._players.pop(bus_name) + self.notify("players") + return bus_name - @Signal - def player_vanished(self, player_name: str) -> str: ... + @Property(dict[str, MprisPlayer], "readable") + def players(self): + return self._players - def __init__( - self, - **kwargs, - ): - self._manager = Playerctl.PlayerManager.new() - self._signal_connections = [] - - # Track signal connections for cleanup - connections = bulk_connect( - self._manager, - { - "name-appeared": self.on_name_appeared, - "name-vanished": self.on_name_vanished, - }, - ) - # Store as (object, handler_id) tuples - for handler_id in connections: - self._signal_connections.append((self._manager, handler_id)) - - self.add_players() + def __init__(self, **kwargs): super().__init__(**kwargs) + self._players: dict[str, MprisPlayer] = {} + self._dbus_helper: GioDBusHelper | None = None + self._subscription_id: int | None = None + + # ahoy + self.do_register() + + def do_register(self): + self._dbus_helper = GioDBusHelper( + "org.freedesktop.DBus", + "/org/freedesktop/DBus", + "org.freedesktop.DBus", + Gio.BusType.SESSION, + ) + + self._subscription_id = self._dbus_helper.listen_signal( + "NameOwnerChanged", + self.on_name_owner_change, + "org.freedesktop.DBus", + ) + + # Defer player discovery to avoid blocking initialization + GLib.idle_add(self._get_available_players) def destroy(self): - """Clean up resources when the manager is destroyed.""" - # Disconnect all signal connections - for obj, handler_id in self._signal_connections: - try: - obj.disconnect(handler_id) - except Exception as e: - logger.warning(f"Failed to disconnect manager signal: {e}") - self._signal_connections.clear() - - # Clean up the manager - if hasattr(self, '_manager'): - del self._manager + """Unsubscribe bus signal and clear players.""" + try: + if self._subscription_id is not None and self._dbus_helper is not None: + self._dbus_helper.unsubscribe_signal(self._subscription_id) + self._subscription_id = None + except Exception: + pass - def on_name_appeared(self, manager, player_name: Playerctl.PlayerName): - logger.info(f"[MprisPlayer] {player_name.name} appeared") - new_player = Playerctl.Player.new_from_name(player_name) - manager.manage_player(new_player) - self.emit("player-appeared", new_player) # type: ignore + try: + # notify vanish for all remaining players + for bus_name in list(self._players.keys()): + try: + self.player_vanished(bus_name) + except Exception: + pass + self._players.clear() + except Exception: + pass - def on_name_vanished(self, manager, player_name: Playerctl.PlayerName): - logger.info(f"[MprisPlayer] {player_name.name} vanished") - self.emit("player-vanished", player_name.name) # type: ignore + def _list_names_callback( + self, + proxy: Gio.DBusProxy, + res: Gio.AsyncResult, + ): + try: + reply = proxy.call_finish(res) + for player in filter( + lambda x: x.startswith(MPRIS_MEDIAPLAYER_BUS_NAME), + reply.get_child_value(0), # type: ignore + ): + self.do_handle_new_player(player) - def add_players(self): - # type: ignore - for player in self._manager.get_property("player-names"): - self._manager.manage_player(Playerctl.Player.new_from_name(player)) # type: ignore + except Exception: + logger.error("[MPRIS MANAGER] Failed to ListNames") - @Property(object, "readable") - def players(self): - return self._manager.get_property("players") # type: ignore + def _get_available_players(self): + self._dbus_helper.proxy.call( + "ListNames", + GLib.Variant("()", ()), + Gio.DBusCallFlags.NONE, + -1, + None, + self._list_names_callback, + ) + + def on_name_owner_change( + self, + conn, + sender_name: str, + object_path: str, + interface_name: str, + signal_name: str, + user_data, + *params, + ) -> None: + name, old_owner, new_owner = user_data + if not name.startswith(MPRIS_MEDIAPLAYER_BUS_NAME): + return + + if old_owner == "" and new_owner != "": + self.do_handle_new_player(name) + + def do_handle_new_player(self, bus_name: str): + if bus_name in self._players.keys(): + return + + player = MprisPlayer(bus_name) + player.connect( + "closed", + lambda *_: self.player_vanished(bus_name), + ) + self.player_appeared(player) diff --git a/services/network.py b/services/network.py index 619e81d..01a8bc6 100644 --- a/services/network.py +++ b/services/network.py @@ -1,9 +1,15 @@ -from gi.repository import NM, GLib +from typing import Optional + import gi -from typing import List, Optional from fabric.core.service import Property, Service, Signal -from fabric.utils import bulk_connect, get_enum_member_name, snake_case_to_kebab_case -from loguru import logger +from fabric.utils import ( + GLib, + bulk_connect, + get_enum_member_name, + logger, + snake_case_to_kebab_case, +) +from gi.repository import NM gi.require_version("NM", "1.0") # Ensure the correct version is loaded diff --git a/services/todo.py b/services/todo.py index 876505e..233a23b 100644 --- a/services/todo.py +++ b/services/todo.py @@ -166,4 +166,3 @@ class TodoService(Service): # Global service instance todo_service = TodoService() -