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
This commit is contained in:
9 files changed
+843
-918
No files matched your search
@@ -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"<span alpha='30%'>{time_remaining}s</span>"
|
||||
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)}"}
|
||||
+147
-210
@@ -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")
|
||||
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
+119
-85
@@ -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:
|
||||
|
||||
+12
-15
@@ -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)
|
||||
|
||||
+392
-249
@@ -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)
|
||||
+10
-4
@@ -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
|
||||
|
||||
|
||||
@@ -166,4 +166,3 @@ class TodoService(Service):
|
||||
|
||||
# Global service instance
|
||||
todo_service = TodoService()
|
||||
|
||||
Reference in new issue
Block a user