360 lines
11 KiB
Python
360 lines
11 KiB
Python
import json
|
|
|
|
from fabric.core.service import Property, Service, Signal
|
|
from fabric.hyprland.service import Hyprland
|
|
from loguru import logger
|
|
|
|
from services.custom_notification import CachedNotifications
|
|
|
|
notification_service = CachedNotifications()
|
|
|
|
|
|
class ModusService(Service):
|
|
@Signal
|
|
def bluetooth_changed(self, new_bluetooth: str) -> None: ...
|
|
|
|
@Signal
|
|
def volume_changed(self, new_volume: int) -> None: ...
|
|
|
|
@Signal
|
|
def wlan_changed(self, new_wlan: str) -> None: ...
|
|
|
|
@Signal
|
|
def battery_changed(self, new_battery: str) -> None: ...
|
|
|
|
@Signal
|
|
def dock_apps_changed(self, new_dock_apps: str) -> None: ...
|
|
|
|
@Signal
|
|
def dont_disturb_changed(self, value: bool) -> None: ...
|
|
|
|
@Signal
|
|
def current_active_app_name_changed(self, value: str) -> None: ...
|
|
|
|
@Signal
|
|
def current_workspace_changed(self, value: str) -> None: ...
|
|
|
|
@Signal
|
|
def music_changed(self, value: str) -> None: ...
|
|
|
|
@Signal
|
|
def current_dropdown_changed(self, value: str) -> None: ...
|
|
|
|
@Signal
|
|
def dropdowns_hide_changed(self, value: bool) -> None: ...
|
|
|
|
@Signal
|
|
def dock_width_changed(self, value: int) -> None: ...
|
|
|
|
@Signal
|
|
def dock_height_changed(self, value: int) -> None: ...
|
|
|
|
@Signal
|
|
def dock_hidden_changed(self, value: bool) -> None: ...
|
|
|
|
@Signal
|
|
def show_notificationcenter_changed(self, value: bool) -> None: ...
|
|
|
|
@Signal
|
|
def notification_count_changed(self, value: int) -> None: ...
|
|
|
|
@Property(str, flags="read-write")
|
|
def current_active_app_name(self) -> str:
|
|
return self._current_active_app_name
|
|
|
|
@Property(str, flags="read-write")
|
|
def current_workspace(self) -> str:
|
|
return self._current_workspace
|
|
|
|
@Property(str, flags="read-write")
|
|
def bluetooth(self) -> str:
|
|
return self._bluetooth
|
|
|
|
@Property(str, flags="read-write")
|
|
def wlan(self) -> str:
|
|
return self._wlan
|
|
|
|
@Property(str, flags="read-write")
|
|
def battery(self) -> str:
|
|
return self._battery
|
|
|
|
@Property(int, flags="read-write")
|
|
def volume(self) -> int:
|
|
return self._volume
|
|
|
|
@Property(str, flags="read-write")
|
|
def dock_apps(self) -> str:
|
|
return self._dock_apps
|
|
|
|
@Property(bool, flags="read-write", default_value=False)
|
|
def dont_disturb(self) -> bool:
|
|
return self._dont_disturb
|
|
|
|
@Property(str, flags="read-write")
|
|
def music(self) -> str:
|
|
return self._music
|
|
|
|
@Property(str, flags="read-write")
|
|
def current_dropdown(self) -> str:
|
|
return self._current_dropdown
|
|
|
|
@Property(bool, flags="read-write", default_value=False)
|
|
def dropdowns_hide(self) -> bool:
|
|
return self._dropdowns_hide
|
|
|
|
@Property(int, flags="read-write")
|
|
def dock_width(self) -> int:
|
|
return self._dock_width
|
|
|
|
@Property(int, flags="read-write")
|
|
def dock_height(self) -> int:
|
|
return self._dock_height
|
|
|
|
@Property(bool, flags="read-write", default_value=False)
|
|
def dock_hidden(self) -> bool:
|
|
return self._dock_hidden
|
|
|
|
@Property(bool, flags="read-write", default_value=False)
|
|
def show_notificationcenter(self) -> bool:
|
|
return self._show_notificationcenter
|
|
|
|
@current_active_app_name.setter
|
|
def current_active_app_name(self, value: str):
|
|
if value != self._current_active_app_name:
|
|
self._current_active_app_name = value
|
|
self.current_active_app_name_changed(value)
|
|
|
|
@current_workspace.setter
|
|
def current_workspace(self, value: str):
|
|
if value != self._current_workspace:
|
|
self._current_workspace = value
|
|
self.current_workspace_changed(value)
|
|
|
|
@volume.setter
|
|
def volume(self, value: int):
|
|
if value != self._volume:
|
|
self._name = value
|
|
self.volume_changed(value)
|
|
|
|
@wlan.setter
|
|
def wlan(self, value: str):
|
|
if value != self._wlan:
|
|
self._wlan = value
|
|
self.wlan_changed(value)
|
|
|
|
@battery.setter
|
|
def battery(self, value: str):
|
|
if value != self._battery:
|
|
self._battery = value
|
|
self.battery_changed(value)
|
|
|
|
@bluetooth.setter
|
|
def bluetooth(self, value: str):
|
|
if value != self._bluetooth:
|
|
self._bluetooth = value
|
|
self.bluetooth_changed(value)
|
|
|
|
@dock_apps.setter
|
|
def dock_apps(self, value: str):
|
|
if value != self._dock_apps:
|
|
self._dock_apps = value
|
|
self.dock_apps_changed(value)
|
|
|
|
@dont_disturb.setter
|
|
def dont_disturb(self, value: bool):
|
|
if value != self._dont_disturb:
|
|
self._dont_disturb = value
|
|
self.dont_disturb_changed(value)
|
|
|
|
@music.setter
|
|
def music(self, value: str):
|
|
if value != self._music:
|
|
self._music = value
|
|
self.music_changed(value)
|
|
|
|
@current_dropdown.setter
|
|
def current_dropdown(self, value: str):
|
|
if value != self._current_dropdown:
|
|
self._current_dropdown = value
|
|
self.current_dropdown_changed(value)
|
|
|
|
@dropdowns_hide.setter
|
|
def dropdowns_hide(self, value: bool):
|
|
if value != self._dropdowns_hide:
|
|
self._dropdowns_hide = value
|
|
self.dropdowns_hide_changed(value)
|
|
|
|
@dock_width.setter
|
|
def dock_width(self, value: int):
|
|
if value != self._dock_width:
|
|
self._dock_width = value
|
|
self.dock_width_changed(value)
|
|
|
|
@dock_height.setter
|
|
def dock_height(self, value: int):
|
|
if value != self._dock_height:
|
|
self._dock_height = value
|
|
self.dock_height_changed(value)
|
|
|
|
@dock_hidden.setter
|
|
def dock_hidden(self, value: bool):
|
|
if value != self._dock_hidden:
|
|
self._dock_hidden = value
|
|
self.dock_hidden_changed(value)
|
|
|
|
@show_notificationcenter.setter
|
|
def show_notificationcenter(self, value: bool):
|
|
if value != self._show_notificationcenter:
|
|
self._show_notificationcenter = value
|
|
self.show_notificationcenter_changed(value)
|
|
|
|
def sc(self, signal_name: str, callback: callable, def_value="..."):
|
|
self.connect(signal_name, callback)
|
|
# Return current property value instead of default
|
|
if signal_name == "bluetooth-changed":
|
|
return self.bluetooth if self.bluetooth else "Off"
|
|
elif signal_name == "wlan-changed":
|
|
return self.wlan if self.wlan else "No Connection"
|
|
elif signal_name == "battery-changed":
|
|
return self.battery if self.battery else "Unknown"
|
|
elif signal_name == "music-changed":
|
|
return self.music if self.music else ""
|
|
else:
|
|
return def_value
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._volume = 0
|
|
self._wlan = ""
|
|
self._battery = ""
|
|
self._bluetooth = ""
|
|
self._dock_apps = ""
|
|
self._dont_disturb = False
|
|
self._current_active_app_name = "Finder" # Changed from "Hyprland" to "Finder"
|
|
self._current_workspace = "1"
|
|
self._music = ""
|
|
self._current_dropdown = None
|
|
self._dropdowns_hide = False
|
|
self._dock_hidden = False
|
|
self._show_notificationcenter = False
|
|
|
|
self._dock_width = 0
|
|
self._dock_height = 0
|
|
|
|
# Initialize Hyprland connection for workspace and window monitoring
|
|
self._setup_workspace_monitoring()
|
|
self._setup_active_window_monitoring()
|
|
|
|
def _setup_workspace_monitoring(self):
|
|
"""Setup Hyprland connection and workspace monitoring"""
|
|
try:
|
|
self._hyprland_connection = Hyprland()
|
|
|
|
# Get initial workspace
|
|
workspace_data = self._hyprland_connection.send_command(
|
|
"j/activeworkspace"
|
|
).reply
|
|
active_workspace = json.loads(workspace_data.decode("utf-8"))["name"]
|
|
self._current_workspace = str(active_workspace)
|
|
|
|
# Connect to workspace change events
|
|
self._hyprland_connection.connect(
|
|
"event::workspace", self._on_workspace_changed
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"[ModusService] Failed to setup workspace monitoring: {e}")
|
|
self._current_workspace = "1"
|
|
|
|
def _setup_active_window_monitoring(self):
|
|
"""Setup active window monitoring"""
|
|
try:
|
|
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}")
|
|
|
|
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:
|
|
return
|
|
|
|
window_data = self._hyprland_connection.send_command("j/activewindow").reply
|
|
if not window_data:
|
|
self.current_active_app_name = "Finder"
|
|
return
|
|
|
|
window_info = json.loads(window_data.decode("utf-8"))
|
|
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]
|
|
else:
|
|
name = "Finder"
|
|
|
|
self.current_active_app_name = name
|
|
|
|
except Exception as e:
|
|
logger.error(f"[ModusService] Error updating active window: {e}")
|
|
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)
|
|
except Exception as e:
|
|
logger.error(f"[ModusService] Error processing workspace change: {e}")
|
|
|
|
def remove_notification(self, id: int):
|
|
notification_service.remove_cached_notification(id)
|
|
self.notification_count_changed(self.notification_count)
|
|
|
|
def clear_all_notifications(self):
|
|
notification_service.clear_all_cached_notifications()
|
|
self.notification_count_changed(self.notification_count)
|
|
|
|
def get_cached_notifications(self):
|
|
return notification_service.cached_notifications
|
|
|
|
def get_deserialized_with_ids(self):
|
|
return [
|
|
(notif._notification, notif.cache_id)
|
|
for notif in notification_service.cached_notifications
|
|
]
|
|
|
|
@property
|
|
def notification_count(self) -> int:
|
|
return notification_service.count
|
|
|
|
def toggle_dnd(self):
|
|
notification_service.toggle_dnd()
|
|
self.dont_disturb_changed(notification_service.dont_disturb)
|
|
|
|
|
|
global modus_service
|
|
try:
|
|
modus_service = ModusService()
|
|
except Exception as e:
|
|
logger.error("[Main] Failed to create ModusShellService:", e)
|