feat(update): huge changes

services/__init__.py

@@ -1,9 +1,10 @@
 from fabric.audio import Audio
 from .brightness import Brightness
-from .screen_record import ScreenRecorder
-from .network import *
-
+from .network import NetworkClient
+from .notification import *
+from .powerprofile import *
 
+network_client = NetworkClient()
 brightness = Brightness()
 audio = Audio()
-sc = ScreenRecorder()
+notification_service = CustomNotifications()

services/notification.py

@@ -0,0 +1,226 @@
+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

services/powerprofile.py

@@ -0,0 +1,88 @@
+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")