initial commit

services/__init__.py

@@ -0,0 +1,9 @@
+from fabric.audio import Audio
+from .brightness import Brightness
+from .screen_record import ScreenRecorder
+from .wifi import *
+
+
+brightness = Brightness()
+audio = Audio()
+sc = ScreenRecorder()

services/brightness.py

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

services/screen_record.py

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

services/wifi.py

@@ -0,0 +1,321 @@
+from typing import Any, List, Literal
+import gi
+from fabric.core.service import Property, Service, Signal
+from fabric.utils import bulk_connect
+from gi.repository import Gio
+from loguru import logger
+
+from fabric.utils import exec_shell_command_async
+
+from gi.repository.NM import DeviceEthernet, DeviceWifi
+
+try:
+    gi.require_version("NM", "1.0")
+    from gi.repository import NM
+except ValueError:
+    logger.error("Failed to start network manager")
+
+
+class Wifi(Service):
+    @Signal
+    def changed(self) -> None: ...
+
+    @Signal
+    def enabled(self) -> bool: ...
+
+    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):
+        self._device.request_scan_async(
+            None,
+            lambda device, result: [
+                device.request_scan_finish(result),
+                self.emit("changed"),
+            ],
+        )
+
+    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):
+    @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 pn in (
+            "active-connection",
+            "icon-name",
+            "internet",
+            "speed",
+            "state",
+        ):
+            self._device.connect(f"notify::{pn}", lambda *_: self.notifier(pn))
+
+        self._device.connect("notify::speed", lambda *_: print(_))
+
+    def notifier(self, pn):
+        self.notify(pn)
+        self.emit("changed")
+
+
+class NetworkClient(Service):
+    @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() != None
+            ),
+            None,
+        )
+
+    def _get_primary_device(self) -> Literal["wifi", "wired"] | None:
+        if not self._client:
+            return None
+        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()
+