chore(starting from new):
This commit is contained in:
82 files changed
-29339
No files matched your search
@@ -1,10 +0,0 @@
|
||||
from fabric.audio import Audio
|
||||
from .brightness import Brightness
|
||||
from .network import NetworkClient
|
||||
from .notification import *
|
||||
from .powerprofile import *
|
||||
|
||||
network_client = NetworkClient()
|
||||
brightness = Brightness()
|
||||
audio = Audio()
|
||||
notification_service = CustomNotifications()
|
||||
@@ -1,115 +0,0 @@
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
from fabric.core.service import Property, Service, Signal
|
||||
from fabric.utils import exec_shell_command, monitor_file
|
||||
from gi.repository import GLib
|
||||
from loguru import logger # Import loguru logger
|
||||
|
||||
|
||||
# Helper function to execute brightnessctl asynchronously
|
||||
def exec_brightnessctl_async(args: str):
|
||||
# Executes brightnessctl command asynchronously, ensuring no resource leaks.
|
||||
try:
|
||||
# Use subprocess.Popen to run the command without blocking
|
||||
process = subprocess.Popen(
|
||||
f"brightnessctl {args}",
|
||||
shell=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
|
||||
# Wait for the process to complete
|
||||
stdout, stderr = process.communicate()
|
||||
|
||||
if process.returncode != 0:
|
||||
logger.error(f"Error executing brightnessctl: {stderr.decode().strip()}")
|
||||
else:
|
||||
logger.debug(
|
||||
f"brightnessctl output: {stdout.decode().strip()}"
|
||||
) # Optional: Log the output
|
||||
except Exception as e:
|
||||
logger.exception(f"Exception in exec_brightnessctl_async: {e}")
|
||||
|
||||
|
||||
# Discover screen backlight device
|
||||
try:
|
||||
screen_device = str(exec_shell_command("ls -w1 /sys/class/backlight")).split("\n")[
|
||||
0
|
||||
]
|
||||
except IndexError:
|
||||
screen_device = None
|
||||
|
||||
|
||||
class NoBrightnessError(ImportError):
|
||||
# Raised when brightness-related dependencies are missing.
|
||||
def __init__(self, *args):
|
||||
super().__init__(
|
||||
"Brightness control not available or missing dependencies", *args
|
||||
)
|
||||
|
||||
|
||||
class Brightness(Service):
|
||||
# Service to manage screen brightness levels.
|
||||
|
||||
@Signal
|
||||
def screen(self, value: int) -> None:
|
||||
"""Signal emitted when screen brightness changes."""
|
||||
pass # Implement as needed for your application
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
|
||||
if not screen_device:
|
||||
raise NoBrightnessError()
|
||||
|
||||
# Path for screen backlight control
|
||||
self.screen_backlight_path = f"/sys/class/backlight/{screen_device}"
|
||||
|
||||
# Initialize maximum brightness level
|
||||
self.max_screen = self._read_max_brightness(self.screen_backlight_path)
|
||||
|
||||
# Monitor screen brightness file
|
||||
self.screen_monitor = monitor_file(f"{self.screen_backlight_path}/brightness")
|
||||
self.screen_monitor.connect(
|
||||
"changed",
|
||||
lambda _, file, *args: self.emit(
|
||||
"screen", round(int(file.load_bytes()[0].get_data()))
|
||||
),
|
||||
)
|
||||
|
||||
# Log the initialization of the service
|
||||
logger.info(f"Brightness service initialized for device: {screen_device}")
|
||||
|
||||
def _read_max_brightness(self, path: str) -> int:
|
||||
# Reads the maximum brightness value from the specified path.
|
||||
max_brightness_path = os.path.join(path, "max_brightness")
|
||||
if os.path.exists(max_brightness_path):
|
||||
with open(max_brightness_path) as f:
|
||||
return int(f.read().strip())
|
||||
return -1 # Return -1 if file doesn't exist, indicating an error.
|
||||
|
||||
@Property(int, "read-write")
|
||||
def screen_brightness(self) -> int:
|
||||
# Property to get or set the screen brightness.
|
||||
brightness_path = os.path.join(self.screen_backlight_path, "brightness")
|
||||
if os.path.exists(brightness_path):
|
||||
with open(brightness_path) as f:
|
||||
return int(f.read().strip())
|
||||
logger.warning(f"Brightness file does not exist: {brightness_path}")
|
||||
return -1 # Return -1 if file doesn't exist, indicating error.
|
||||
|
||||
@screen_brightness.setter
|
||||
def screen_brightness(self, value: int):
|
||||
# Setter for screen brightness property.
|
||||
if not (0 <= value <= self.max_screen):
|
||||
value = max(0, min(value, self.max_screen))
|
||||
|
||||
try:
|
||||
exec_brightnessctl_async(f"--device '{screen_device}' set {value}")
|
||||
self.emit("screen", int((value / self.max_screen) * 100))
|
||||
logger.info(f"Set screen brightness to {value} (out of {self.max_screen})")
|
||||
except GLib.Error as e:
|
||||
logger.error(f"Error setting screen brightness: {e.message}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Unexpected error setting screen brightness: {e}")
|
||||
@@ -1,334 +0,0 @@
|
||||
from typing import Any, List, Literal
|
||||
|
||||
import gi
|
||||
from fabric.core.service import Property, Service, Signal
|
||||
from fabric.utils import bulk_connect, exec_shell_command_async
|
||||
from gi.repository import Gio
|
||||
from loguru import logger
|
||||
|
||||
try:
|
||||
gi.require_version("NM", "1.0")
|
||||
from gi.repository import NM
|
||||
except ValueError:
|
||||
logger.error("Failed to start network manager")
|
||||
|
||||
|
||||
class Wifi(Service):
|
||||
"""A service to manage wifi devices"""
|
||||
|
||||
@Signal
|
||||
def changed(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def enabled(self) -> bool: ...
|
||||
|
||||
@Signal
|
||||
def scanning(self, is_scanning: bool) -> None: ...
|
||||
|
||||
def __init__(self, client: NM.Client, device: NM.DeviceWifi, **kwargs):
|
||||
self._client: NM.Client = client
|
||||
self._device: NM.DeviceWifi = device
|
||||
self._ap: NM.AccessPoint | None = None
|
||||
self._ap_signal: int | None = None
|
||||
super().__init__(**kwargs)
|
||||
|
||||
self._client.connect(
|
||||
"notify::wireless-enabled",
|
||||
lambda *args: self.notifier("enabled", args),
|
||||
)
|
||||
if self._device:
|
||||
bulk_connect(
|
||||
self._device,
|
||||
{
|
||||
"notify::active-access-point": lambda *args: self._activate_ap(),
|
||||
"access-point-added": lambda *args: self.emit("changed"),
|
||||
"access-point-removed": lambda *args: self.emit("changed"),
|
||||
"state-changed": lambda *args: self.ap_update(),
|
||||
},
|
||||
)
|
||||
self._activate_ap()
|
||||
|
||||
def ap_update(self):
|
||||
self.emit("changed")
|
||||
for sn in [
|
||||
"enabled",
|
||||
"internet",
|
||||
"strength",
|
||||
"frequency",
|
||||
"access-points",
|
||||
"ssid",
|
||||
"state",
|
||||
"icon-name",
|
||||
]:
|
||||
self.notify(sn)
|
||||
|
||||
def _activate_ap(self):
|
||||
if self._ap:
|
||||
self._ap.disconnect(self._ap_signal)
|
||||
self._ap = self._device.get_active_access_point()
|
||||
if not self._ap:
|
||||
return
|
||||
|
||||
self._ap_signal = self._ap.connect(
|
||||
"notify::strength", lambda *args: self.ap_update()
|
||||
) # type: ignore
|
||||
|
||||
def toggle_wifi(self):
|
||||
self._client.wireless_set_enabled(not self._client.wireless_get_enabled())
|
||||
|
||||
# def set_active_ap(self, ap):
|
||||
# self._device.access
|
||||
|
||||
def scan(self):
|
||||
"""Start scanning for WiFi networks and emit scanning signal"""
|
||||
if self._device:
|
||||
self.emit("scanning", True) # Emit signal that scanning has started
|
||||
self._device.request_scan_async(
|
||||
None,
|
||||
lambda device, result: [
|
||||
device.request_scan_finish(result),
|
||||
self.emit(
|
||||
"scanning", False
|
||||
), # Emit signal that scanning has stopped
|
||||
],
|
||||
)
|
||||
|
||||
def notifier(self, name: str, *args):
|
||||
self.notify(name)
|
||||
self.emit("changed")
|
||||
return
|
||||
|
||||
@Property(bool, "read-write", default_value=False)
|
||||
def enabled(self) -> bool: # type: ignore
|
||||
return bool(self._client.wireless_get_enabled())
|
||||
|
||||
@enabled.setter
|
||||
def enabled(self, value: bool):
|
||||
self._client.wireless_set_enabled(value)
|
||||
|
||||
@Property(int, "readable")
|
||||
def strength(self):
|
||||
return self._ap.get_strength() if self._ap else -1
|
||||
|
||||
@Property(str, "readable")
|
||||
def icon_name(self):
|
||||
if not self._ap:
|
||||
return "network-wireless-disabled-symbolic"
|
||||
|
||||
if self.internet == "activated":
|
||||
return {
|
||||
80: "network-wireless-signal-excellent-symbolic",
|
||||
60: "network-wireless-signal-good-symbolic",
|
||||
40: "network-wireless-signal-ok-symbolic",
|
||||
20: "network-wireless-signal-weak-symbolic",
|
||||
00: "network-wireless-signal-none-symbolic",
|
||||
}.get(
|
||||
min(80, 20 * round(self._ap.get_strength() / 20)),
|
||||
"network-wireless-no-route-symbolic",
|
||||
)
|
||||
if self.internet == "activating":
|
||||
return "network-wireless-acquiring-symbolic"
|
||||
|
||||
return "network-wireless-offline-symbolic"
|
||||
|
||||
@Property(int, "readable")
|
||||
def frequency(self):
|
||||
return self._ap.get_frequency() if self._ap else -1
|
||||
|
||||
@Property(int, "readable")
|
||||
def internet(self):
|
||||
return {
|
||||
NM.ActiveConnectionState.ACTIVATED: "activated",
|
||||
NM.ActiveConnectionState.ACTIVATING: "activating",
|
||||
NM.ActiveConnectionState.DEACTIVATING: "deactivating",
|
||||
NM.ActiveConnectionState.DEACTIVATED: "deactivated",
|
||||
}.get(
|
||||
self._device.get_active_connection().get_state(),
|
||||
"unknown",
|
||||
)
|
||||
|
||||
@Property(object, "readable")
|
||||
def access_points(self) -> List[object]:
|
||||
points: list[NM.AccessPoint] = self._device.get_access_points()
|
||||
|
||||
def make_ap_dict(ap: NM.AccessPoint):
|
||||
return {
|
||||
"bssid": ap.get_bssid(),
|
||||
# "address": ap.get_
|
||||
"last_seen": ap.get_last_seen(),
|
||||
"ssid": NM.utils_ssid_to_utf8(ap.get_ssid().get_data())
|
||||
if ap.get_ssid()
|
||||
else "Unknown",
|
||||
"active-ap": self._ap,
|
||||
"strength": ap.get_strength(),
|
||||
"frequency": ap.get_frequency(),
|
||||
"icon-name": {
|
||||
80: "network-wireless-signal-excellent-symbolic",
|
||||
60: "network-wireless-signal-good-symbolic",
|
||||
40: "network-wireless-signal-ok-symbolic",
|
||||
20: "network-wireless-signal-weak-symbolic",
|
||||
00: "network-wireless-signal-none-symbolic",
|
||||
}.get(
|
||||
min(80, 20 * round(ap.get_strength() / 20)),
|
||||
"network-wireless-no-route-symbolic",
|
||||
),
|
||||
}
|
||||
|
||||
return list(map(make_ap_dict, points))
|
||||
|
||||
@Property(str, "readable")
|
||||
def ssid(self):
|
||||
if not self._ap:
|
||||
return "Disconnected"
|
||||
ssid = self._ap.get_ssid().get_data()
|
||||
return NM.utils_ssid_to_utf8(ssid) if ssid else "Unknown"
|
||||
|
||||
@Property(int, "readable")
|
||||
def state(self):
|
||||
return {
|
||||
NM.DeviceState.UNMANAGED: "unmanaged",
|
||||
NM.DeviceState.UNAVAILABLE: "unavailable",
|
||||
NM.DeviceState.DISCONNECTED: "disconnected",
|
||||
NM.DeviceState.PREPARE: "prepare",
|
||||
NM.DeviceState.CONFIG: "config",
|
||||
NM.DeviceState.NEED_AUTH: "need_auth",
|
||||
NM.DeviceState.IP_CONFIG: "ip_config",
|
||||
NM.DeviceState.IP_CHECK: "ip_check",
|
||||
NM.DeviceState.SECONDARIES: "secondaries",
|
||||
NM.DeviceState.ACTIVATED: "activated",
|
||||
NM.DeviceState.DEACTIVATING: "deactivating",
|
||||
NM.DeviceState.FAILED: "failed",
|
||||
}.get(self._device.get_state(), "unknown")
|
||||
|
||||
|
||||
class Ethernet(Service):
|
||||
"""A service to manage ethernet devices"""
|
||||
|
||||
@Signal
|
||||
def changed(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def enabled(self) -> bool: ...
|
||||
|
||||
@Property(int, "readable")
|
||||
def speed(self) -> int:
|
||||
return self._device.get_speed()
|
||||
|
||||
@Property(str, "readable")
|
||||
def internet(self) -> str:
|
||||
return {
|
||||
NM.ActiveConnectionState.ACTIVATED: "activated",
|
||||
NM.ActiveConnectionState.ACTIVATING: "activating",
|
||||
NM.ActiveConnectionState.DEACTIVATING: "deactivating",
|
||||
NM.ActiveConnectionState.DEACTIVATED: "deactivated",
|
||||
}.get(
|
||||
self._device.get_active_connection().get_state(),
|
||||
"disconnected",
|
||||
)
|
||||
|
||||
@Property(str, "readable")
|
||||
def icon_name(self) -> str:
|
||||
network = self.internet
|
||||
if network == "activated":
|
||||
return "network-wired-symbolic"
|
||||
|
||||
elif network == "activating":
|
||||
return "network-wired-acquiring-symbolic"
|
||||
|
||||
elif self._device.get_connectivity != NM.ConnectivityState.FULL:
|
||||
return "network-wired-no-route-symbolic"
|
||||
|
||||
return "network-wired-disconnected-symbolic"
|
||||
|
||||
def __init__(self, client: NM.Client, device: NM.DeviceEthernet, **kwargs) -> None:
|
||||
super().__init__(**kwargs)
|
||||
self._client: NM.Client = client
|
||||
self._device: NM.DeviceEthernet = device
|
||||
|
||||
for names in (
|
||||
"active-connection",
|
||||
"icon-name",
|
||||
"internet",
|
||||
"speed",
|
||||
"state",
|
||||
):
|
||||
self._device.connect(f"notify::{names}", lambda *_: self.notifier(names))
|
||||
|
||||
self._device.connect("notify::speed", lambda *_: print(_))
|
||||
|
||||
def notifier(self, names):
|
||||
self.notify(names)
|
||||
self.emit("changed")
|
||||
|
||||
|
||||
class NetworkClient(Service):
|
||||
"""A service to manage network devices"""
|
||||
|
||||
@Signal
|
||||
def device_ready(self) -> None: ...
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self._client: NM.Client | None = None
|
||||
self.wifi_device: Wifi | None = None
|
||||
self.ethernet_device: Ethernet | None = None
|
||||
super().__init__(**kwargs)
|
||||
NM.Client.new_async(
|
||||
cancellable=None,
|
||||
callback=self._init_network_client,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def _init_network_client(self, client: NM.Client, task: Gio.Task, **kwargs):
|
||||
self._client = client
|
||||
wifi_device: NM.DeviceWifi | None = self._get_device(NM.DeviceType.WIFI) # type: ignore
|
||||
ethernet_device: NM.DeviceEthernet | None = self._get_device(
|
||||
NM.DeviceType.ETHERNET
|
||||
)
|
||||
|
||||
if wifi_device:
|
||||
self.wifi_device = Wifi(self._client, wifi_device)
|
||||
self.emit("device-ready")
|
||||
|
||||
if ethernet_device:
|
||||
self.ethernet_device = Ethernet(client=self._client, device=ethernet_device)
|
||||
self.emit("device-ready")
|
||||
|
||||
self.notify("primary-device")
|
||||
|
||||
def _get_device(self, device_type) -> Any:
|
||||
devices: List[NM.Device] = self._client.get_devices() # type: ignore
|
||||
return next(
|
||||
(
|
||||
x
|
||||
for x in devices
|
||||
if x.get_device_type() == device_type
|
||||
and x.get_active_connection() is not None
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
def _get_primary_device(self) -> Literal["wifi", "wired"] | None:
|
||||
if not self._client:
|
||||
return None
|
||||
|
||||
if self._client.get_primary_connection() is None:
|
||||
return "wifi"
|
||||
return (
|
||||
"wifi"
|
||||
if "wireless"
|
||||
in str(self._client.get_primary_connection().get_connection_type())
|
||||
else "wired"
|
||||
if "ethernet"
|
||||
in str(self._client.get_primary_connection().get_connection_type())
|
||||
else None
|
||||
)
|
||||
|
||||
def connect_wifi_bssid(self, bssid):
|
||||
# We are using nmcli here, idk im lazy
|
||||
exec_shell_command_async(
|
||||
f"nmcli device wifi connect {bssid}", lambda *args: print(args)
|
||||
)
|
||||
|
||||
@Property(str, "readable")
|
||||
def primary_device(self) -> Literal["wifi", "wired"] | None:
|
||||
return self._get_primary_device()
|
||||
@@ -1,226 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
from typing import List
|
||||
|
||||
from fabric import Signal
|
||||
from fabric.notifications import Notification, Notifications
|
||||
from loguru import logger
|
||||
import gi
|
||||
gi.require_version('GLib', '2.0')
|
||||
from gi.repository import GLib
|
||||
|
||||
import config.data as data
|
||||
|
||||
NOTIFICATION_CACHE_FILE = f"{data.CACHE_DIR}/notifications.json"
|
||||
|
||||
|
||||
class CustomNotifications(Notifications):
|
||||
"""A service to manage the notifications."""
|
||||
|
||||
@Signal
|
||||
def clear_all(self, value: bool) -> None:
|
||||
"""Signal emitted when notifications are emptied."""
|
||||
# Implement as needed for your application
|
||||
|
||||
@Signal
|
||||
def notification_count(self, value: int) -> None:
|
||||
"""Signal emitted when a new notification is added."""
|
||||
# Implement as needed for your application
|
||||
|
||||
@Signal
|
||||
def dnd(self, value: bool) -> None:
|
||||
"""Signal emitted when dnd is toggled."""
|
||||
# Implement as needed for your application
|
||||
|
||||
@property
|
||||
def count(self) -> int:
|
||||
"""Return the count of notifications."""
|
||||
return len(self.all_notifications)
|
||||
|
||||
@property
|
||||
def dont_disturb(self) -> bool:
|
||||
"""Return the pause status."""
|
||||
return self._dont_disturb
|
||||
|
||||
@dont_disturb.setter
|
||||
def dont_disturb(self, value: bool):
|
||||
"""Set the pause status."""
|
||||
self._dont_disturb = value
|
||||
self.emit("dnd", value)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.all_notifications = []
|
||||
self._count = 0 # Will be updated to highest ID when loading
|
||||
self.deserialized_notifications = []
|
||||
self._dont_disturb = False
|
||||
self._load_notifications()
|
||||
|
||||
def _load_notifications(self):
|
||||
"""Read notifications from the cache file."""
|
||||
if os.path.exists(NOTIFICATION_CACHE_FILE):
|
||||
try:
|
||||
with open(NOTIFICATION_CACHE_FILE, "r") as file:
|
||||
notifications = json.load(file)
|
||||
|
||||
def validate_with_id(notif):
|
||||
"""Helper to validate and return ID if valid."""
|
||||
try:
|
||||
self._deserialize_notification(notif)
|
||||
return (True, notif, notif.get("id", 0))
|
||||
except Exception as e:
|
||||
msg = f"[Notification] Invalid: {str(e)[:50]}"
|
||||
logger.error(msg)
|
||||
return (False, None, 0)
|
||||
|
||||
# Validate all notifications at once
|
||||
results = [validate_with_id(n) for n in notifications]
|
||||
|
||||
# Process results and find highest ID
|
||||
valid_notifications = []
|
||||
highest_id = self._count # Start with current count
|
||||
for is_valid, notif, notif_id in results:
|
||||
if is_valid:
|
||||
valid_notifications.append(notif)
|
||||
highest_id = max(highest_id, notif_id)
|
||||
|
||||
self.all_notifications = valid_notifications
|
||||
self._count = highest_id # Update to highest ID seen
|
||||
self._write_notifications(self.all_notifications)
|
||||
|
||||
except (json.JSONDecodeError, KeyError, ValueError, IndexError) as e:
|
||||
logger.error(f"[Notification] {e}")
|
||||
self.all_notifications = []
|
||||
self._count = 0
|
||||
|
||||
def remove_notification(self, id: int):
|
||||
"""Remove the notification of given id."""
|
||||
item = next((p for p in self.all_notifications if p["id"] == id), None)
|
||||
if item:
|
||||
self.all_notifications.remove(item)
|
||||
self._write_notifications(self.all_notifications)
|
||||
self.emit("notification_count", len(self.all_notifications))
|
||||
|
||||
# Emit clear_all signal if there are no notifications left
|
||||
if len(self.all_notifications) == 0:
|
||||
self.emit("clear_all", True)
|
||||
|
||||
def cache_notification(self, data: Notification):
|
||||
"""Cache the notification without widget_config and max_count."""
|
||||
|
||||
# First clean up any invalid notifications
|
||||
self._cleanup_invalid_notifications()
|
||||
|
||||
# Create the new notification
|
||||
self._count += 1
|
||||
new_id = self._count
|
||||
timestamp = GLib.get_real_time() / 1000000 # Convert microseconds to seconds
|
||||
serialized_data = data.serialize()
|
||||
serialized_data.update({
|
||||
"id": new_id,
|
||||
"app_name": data.app_name,
|
||||
"timestamp": timestamp
|
||||
})
|
||||
|
||||
# Append the new notification
|
||||
self.all_notifications.append(serialized_data)
|
||||
|
||||
self._write_notifications(self.all_notifications)
|
||||
self.emit("notification_count", len(self.all_notifications))
|
||||
|
||||
def _cleanup_invalid_notifications(self):
|
||||
"""Remove any invalid notifications."""
|
||||
|
||||
def validate_with_id(notif):
|
||||
"""Helper to validate and return result with ID."""
|
||||
try:
|
||||
self._deserialize_notification(notif)
|
||||
return (True, notif, None)
|
||||
except Exception as e:
|
||||
msg = f"[Notification] Removing invalid: {str(e)[:50]}"
|
||||
logger.debug(msg)
|
||||
return (False, None, notif.get("id", 0))
|
||||
|
||||
# Validate all notifications at once
|
||||
results = [validate_with_id(n) for n in self.all_notifications]
|
||||
|
||||
# Process results
|
||||
valid_notifications = []
|
||||
invalid_count = 0
|
||||
for is_valid, notif, invalid_id in results:
|
||||
if is_valid:
|
||||
valid_notifications.append(notif)
|
||||
else:
|
||||
invalid_count += 1
|
||||
self.emit("notification-closed", invalid_id, "dismissed-by-limit")
|
||||
|
||||
if invalid_count > 0:
|
||||
self.all_notifications = valid_notifications
|
||||
self._write_notifications(self.all_notifications)
|
||||
self.emit("notification_count", len(self.all_notifications))
|
||||
|
||||
def _deserialize_notification(self, notification):
|
||||
"""Deserialize a notification."""
|
||||
try:
|
||||
notif = Notification.deserialize(notification)
|
||||
# Store the timestamp in the notification object
|
||||
notif.timestamp = notification.get("timestamp", GLib.get_real_time() / 1000000)
|
||||
return notif
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"[Notification] Deserialize failed for: {notification}, Error: {e}"
|
||||
)
|
||||
raise
|
||||
|
||||
def clear_all_notifications(self):
|
||||
"""Empty the notifications."""
|
||||
logger.info("[Notification] Clearing all notifications")
|
||||
# Clear notifications but preserve the highest ID we've seen
|
||||
highest_id = self._count
|
||||
self.all_notifications = []
|
||||
self._write_notifications(self.all_notifications)
|
||||
self.emit("notification_count", 0)
|
||||
self.emit("clear_all", True)
|
||||
# Restore the ID counter so new notifications get unique IDs
|
||||
self._count = highest_id
|
||||
|
||||
def _write_notifications(self, data):
|
||||
"""Write the notifications to the cache file."""
|
||||
try:
|
||||
with open(NOTIFICATION_CACHE_FILE, "w") as f:
|
||||
json.dump(data, f, indent=4, ensure_ascii=False)
|
||||
logger.info("[Notification] Notifications written successfully.")
|
||||
except Exception as e:
|
||||
logger.error(f"[Notification] Failed to write notifications: {e}")
|
||||
|
||||
def get_deserialized(self) -> List[Notification]:
|
||||
"""Return the notifications."""
|
||||
|
||||
def deserialize_with_id(notif):
|
||||
"""Helper to deserialize and return result with ID."""
|
||||
try:
|
||||
return (self._deserialize_notification(notif), None)
|
||||
except Exception as e:
|
||||
msg = f"[Notification] Deserialize failed: {str(e)[:50]}"
|
||||
logger.error(msg)
|
||||
return (None, notif.get("id"))
|
||||
|
||||
# Process all notifications at once
|
||||
results = [
|
||||
deserialize_with_id(notification) for notification in self.all_notifications
|
||||
]
|
||||
|
||||
# Split into successful and failed
|
||||
deserialized = []
|
||||
invalid_ids = []
|
||||
for result, error_id in results:
|
||||
if result is not None:
|
||||
deserialized.append(result)
|
||||
elif error_id is not None:
|
||||
invalid_ids.append(error_id)
|
||||
|
||||
# Clean up invalid notifications
|
||||
for invalid_id in invalid_ids:
|
||||
self.remove_notification(invalid_id)
|
||||
|
||||
return deserialized
|
||||
@@ -1,88 +0,0 @@
|
||||
import dbus
|
||||
from dbus.mainloop.glib import DBusGMainLoop
|
||||
from fabric import Service, Signal
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class PowerProfiles(Service):
|
||||
"""Service to interact with the PowerProfiles service."""
|
||||
|
||||
@Signal
|
||||
def profile(self, value: str) -> None:
|
||||
"""Signal emitted when profile changes."""
|
||||
|
||||
instance = None
|
||||
|
||||
@staticmethod
|
||||
def get_default():
|
||||
if PowerProfiles.instance is None:
|
||||
PowerProfiles.instance = PowerProfiles()
|
||||
|
||||
return PowerProfiles.instance
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
self.bus_name = "net.hadess.PowerProfiles"
|
||||
self.object_path = "/net/hadess/PowerProfiles"
|
||||
|
||||
self.power_profiles = {
|
||||
"power-saver": {
|
||||
"name": "Power Saver",
|
||||
"icon_name": "power-profile-power-saver-symbolic",
|
||||
},
|
||||
"balanced": {
|
||||
"name": "Balanced",
|
||||
"icon_name": "power-profile-balanced-symbolic",
|
||||
},
|
||||
"performance": {
|
||||
"name": "Performance",
|
||||
"icon_name": "power-profile-performance-symbolic",
|
||||
},
|
||||
}
|
||||
|
||||
# Set up the dbus main loop
|
||||
DBusGMainLoop(set_as_default=True)
|
||||
|
||||
self.bus = dbus.SystemBus()
|
||||
|
||||
self.power_profiles_obj = self.bus.get_object(self.bus_name, self.object_path)
|
||||
|
||||
self.iface = dbus.Interface(
|
||||
self.power_profiles_obj, "org.freedesktop.DBus.Properties"
|
||||
)
|
||||
|
||||
# Connect the 'g-properties-changed' signal to the handler
|
||||
self.iface.connect_to_signal("PropertiesChanged", self.handle_property_change)
|
||||
|
||||
def get_current_profile(self):
|
||||
try:
|
||||
return self.iface.Get("net.hadess.PowerProfiles", "ActiveProfile").strip()
|
||||
|
||||
except dbus.DBusException as e:
|
||||
logger.error(f"[PowerProfile] Error retrieving current power profile: {e}")
|
||||
return "balanced"
|
||||
|
||||
def set_power_profile(self, profile):
|
||||
try:
|
||||
self.iface.Set(self.bus_name, "ActiveProfile", dbus.String(profile))
|
||||
logger.info(f"[PowerProfile] Power profile set to {profile}")
|
||||
except dbus.DBusException as e:
|
||||
logger.error(
|
||||
f"[PowerProfile] Could not change power level to {profile}: {e}"
|
||||
)
|
||||
|
||||
# Function to handle properties change signals
|
||||
def handle_property_change(self, proxy, changed, invalidated):
|
||||
# Print the changed ActiveProfile
|
||||
if "ActiveProfile" in changed:
|
||||
logger.info(f"Profile changed: {changed['ActiveProfile']}")
|
||||
self.emit("profile", changed["ActiveProfile"])
|
||||
|
||||
def get_profile_icon(self, profile):
|
||||
return self.power_profiles.get(profile, "balanced").get("icon_name")
|
||||
@@ -1,142 +0,0 @@
|
||||
import datetime
|
||||
import subprocess
|
||||
|
||||
from fabric.core.service import Property, Service, Signal
|
||||
from fabric.utils import exec_shell_command, exec_shell_command_async
|
||||
from gi.repository import Gio, GLib
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class ScreenRecorder(Service):
|
||||
@Signal
|
||||
def recording(self, value: bool) -> None: ...
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self.screenshot_path = GLib.get_home_dir() + "/Pictures/Screenshots/"
|
||||
self.screenrecord_path = GLib.get_home_dir() + "/Videos/Screencasting/"
|
||||
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def screenshot(self, fullscreen=False, save_copy=True):
|
||||
time = datetime.datetime.today().strftime("%Y-%m-%d_%H-%M-%S")
|
||||
file_path = self.screenshot_path + str(time) + ".png"
|
||||
command = (
|
||||
["grimblast", "copysave", "screen", file_path]
|
||||
if save_copy
|
||||
else ["grimblast", "copyscreen"]
|
||||
)
|
||||
if not fullscreen:
|
||||
command[2] = "area"
|
||||
try:
|
||||
subprocess.run(command, check=True)
|
||||
self.send_screenshot_notification(
|
||||
file_path=file_path if file_path else None,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"[SCREENSHOT] Failed to run command: {command}")
|
||||
|
||||
def screencast_start(self, fullscreen=False):
|
||||
if self.is_recording:
|
||||
logger.error(
|
||||
"[SCREENRECORD] Another instance of wf-recorder is already running"
|
||||
)
|
||||
return
|
||||
time = datetime.datetime.today().strftime("%Y-%m-%d_%H-%M-%S")
|
||||
file_path = self.screenrecord_path + str(time) + ".mp4"
|
||||
self._current_screencast_path = file_path
|
||||
area = "" if fullscreen else f"-g '{exec_shell_command('slurp')}'"
|
||||
command = f"wf-recorder --file={file_path} --pixel-format yuv420p {area}"
|
||||
exec_shell_command_async(command)
|
||||
self.emit("recording", True)
|
||||
|
||||
def screencast_stop(self):
|
||||
exec_shell_command_async("killall -INT wf-recorder")
|
||||
self.emit("recording", False)
|
||||
self.send_screencast_notification(self._current_screencast_path)
|
||||
|
||||
def send_screencast_notification(self, file_path):
|
||||
# TODO: Generate a thumbnail
|
||||
# exec_shell_command_async(f"totem-video-thumbnailer {file_path} {FABRIC_CACHE}")
|
||||
|
||||
cmd = ["notify-send"]
|
||||
cmd.extend(
|
||||
[
|
||||
"-A",
|
||||
"files=Show in Files",
|
||||
"-A",
|
||||
"view=View",
|
||||
"-i",
|
||||
"camera-video-symbolic",
|
||||
"-a",
|
||||
"Fabric Screenshot Utility",
|
||||
"Screencast Saved",
|
||||
f"Saved Screencast at {file_path}",
|
||||
]
|
||||
)
|
||||
|
||||
proc: Gio.Subprocess = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.STDOUT_PIPE)
|
||||
|
||||
def do_callback(process: Gio.Subprocess, task: Gio.Task):
|
||||
try:
|
||||
_, stdout, stderr = process.communicate_utf8_finish(task)
|
||||
except Exception:
|
||||
logger.error(
|
||||
f"[SCREENCAST] Failed read notification action with error {stderr}"
|
||||
)
|
||||
return
|
||||
|
||||
match stdout.strip("\n"):
|
||||
case "files":
|
||||
exec_shell_command_async(f"xdg-open {self.screenrecord_path}")
|
||||
case "view":
|
||||
exec_shell_command_async(f"xdg-open {file_path}")
|
||||
|
||||
proc.communicate_utf8_async(None, None, do_callback)
|
||||
|
||||
def send_screenshot_notification(self, file_path=None):
|
||||
cmd = ["notify-send"]
|
||||
cmd.extend(
|
||||
[
|
||||
"-A",
|
||||
"files=Show in Files",
|
||||
"-A",
|
||||
"view=View",
|
||||
"-A",
|
||||
"edit=Edit",
|
||||
"-i",
|
||||
"camera-photo-symbolic",
|
||||
"-a",
|
||||
"Fabric Screenshot Utility",
|
||||
"-h",
|
||||
f"STRING:image-path:{file_path}",
|
||||
"Screenshot Saved",
|
||||
f"Saved Screenshot at {file_path}",
|
||||
]
|
||||
if file_path
|
||||
else ["Screenshot Sent to Clipboard"]
|
||||
)
|
||||
|
||||
proc: Gio.Subprocess = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.STDOUT_PIPE)
|
||||
|
||||
def do_callback(process: Gio.Subprocess, task: Gio.Task):
|
||||
try:
|
||||
_, stdout, stderr = process.communicate_utf8_finish(task)
|
||||
except Exception:
|
||||
logger.error(
|
||||
f"[SCREENSHOT] Failed read notification action with error {stderr}"
|
||||
)
|
||||
return
|
||||
|
||||
match stdout.strip("\n"):
|
||||
case "files":
|
||||
exec_shell_command_async(f"xdg-open {self.screenshot_path}")
|
||||
case "view":
|
||||
exec_shell_command_async(f"xdg-open {file_path}")
|
||||
case "edit":
|
||||
exec_shell_command_async(f"swappy -f {file_path}")
|
||||
|
||||
proc.communicate_utf8_async(None, None, do_callback)
|
||||
|
||||
@Property(bool, "readable", default_value=False)
|
||||
def is_recording(self):
|
||||
return False if len(exec_shell_command("pidof wf-recorder")) == 0 else True
|
||||
Reference in new issue
Block a user