diff --git a/config/fabric/assets/config.json b/config/fabric/assets/config.json new file mode 100644 index 00000000..b3c6e031 --- /dev/null +++ b/config/fabric/assets/config.json @@ -0,0 +1,4 @@ +{ + "favorite_apps": ["Firefox", "materialgram", "Vesktop", "Spotify", "kitty"], + "city": "Kohalpur" +} diff --git a/config/fabric/modules/bar/bar.py b/config/fabric/modules/bar/bar.py index 5185b721..e0fe60c7 100644 --- a/config/fabric/modules/bar/bar.py +++ b/config/fabric/modules/bar/bar.py @@ -8,7 +8,7 @@ from fabric.widgets.datetime import DateTime from fabric.widgets.wayland import WaylandWindow as Window from modules.bar.widgets import ( BatteryLabel, - Bluetooth, + Bluetooth, # Weather, HardwareUsage, IdleIndicator, MicrophoneIndicator, @@ -16,7 +16,6 @@ from modules.bar.widgets import ( PowerProfile, TaskBar, VolumeIndicator, - Weather, workspace, ) @@ -56,7 +55,7 @@ class Bar(Window): self.workspaces = Button(child=workspace, name="workspaces") self.info = HardwareUsage() self.idle = IdleIndicator() - self.weather = Weather("Kohalpur") + # self.weather = Weather() self.applets = Box( name="applets", spacing=4, diff --git a/config/fabric/modules/bar/widgets/weather.py b/config/fabric/modules/bar/widgets/weather.py index 8720c6c0..24afdc16 100644 --- a/config/fabric/modules/bar/widgets/weather.py +++ b/config/fabric/modules/bar/widgets/weather.py @@ -1,32 +1,34 @@ -from services import WeatherInfo - from fabric.utils import invoke_repeater from fabric.widgets.box import Box from fabric.widgets.label import Label +from services import WeatherInfo +from snippets import read_config class Weather(Box): def __init__( self, - city: str, interval: int = 60000, enable_tooltip=True, ): super().__init__(name="tray") self.enable_tooltip = enable_tooltip - self.city = city self.weather_service = WeatherInfo() - self.weather_label = Label(label="weather") self.children = self.weather_label invoke_repeater(interval, self.update_label, initial_call=True) def update_label(self): - res = self.weather_service.simple_weather_info(city=self.city) - self.weather_label.set_label(f"{res["icon"]} {res["temperature"]}") + config = read_config() + city = config.get("city", "London") + + res = self.weather_service.simple_weather_info(city) + + self.weather_label.set_label(f"{res['icon']} {res['temperature']}") if self.enable_tooltip: - self.set_tooltip_text(f"{res["city"]}, {res["condition"]}".strip("'")) + self.set_tooltip_text(f"{res['city']}, {res['condition']}".strip("'")) + return True diff --git a/config/fabric/services/__init__.py b/config/fabric/services/__init__.py index 4b5927ad..1df3d07e 100644 --- a/config/fabric/services/__init__.py +++ b/config/fabric/services/__init__.py @@ -1,5 +1,12 @@ from .brightness import Brightness from .screen_record import ScreenRecorder from .weather import WeatherInfo +from .wifi import NetworkClient, Wifi -__all__ = ["Brightness", "ScreenRecorder", "WeatherInfo"] +__all__ = [ + "Brightness", + "ScreenRecorder", + "WeatherInfo", + "NetworkClient", + "Wifi", +] diff --git a/config/fabric/services/wifi.py b/config/fabric/services/wifi.py new file mode 100644 index 00000000..b7ec7b27 --- /dev/null +++ b/config/fabric/services/wifi.py @@ -0,0 +1,249 @@ +from typing import 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): + @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 NetworkClient(Service): + @Signal + def device_ready(self) -> None: ... + + def __init__(self, **kwargs): + self._client: NM.Client | None = None + self.wifi_device: Wifi | 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 = self._get_device(NM.DeviceType.WIFI) # type: ignore + # eth_device: NM.DeviceEthernet = self._get_device(NM.DeviceType.ETHERNET) # type: ignore + + if wifi_device: + self.wifi_device = Wifi(self._client, wifi_device) + self.emit("device-ready") + + # if eth_device: + # self.eth_device = Ethernet(self._client, eth_device) + + def _get_device(self, device_type) -> NM.Device | None: + devices: List[NM.Device] = self._client.get_devices() # type: ignore + return next((x for x in devices if x.get_device_type() == device_type), 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() diff --git a/config/fabric/snippets/util.py b/config/fabric/snippets/util.py new file mode 100644 index 00000000..aeb1723b --- /dev/null +++ b/config/fabric/snippets/util.py @@ -0,0 +1,26 @@ +import json +from typing import Literal + + +def read_config(): + with open("config.json", "r") as file: + # Load JSON data into a Python dictionary + data = json.load(file) + return data + + +def format_time(secs): + mm, _ = divmod(secs, 60) + hh, mm = divmod(mm, 60) + return "%d h %02d min" % (hh, mm) + + +def convert_bytes(bytes: int, to: Literal["kb", "mb", "gb"]): + multiplier = 1 + + if to == "mb": + multiplier = 2 + elif to == "gb": + multiplier = 3 + + return bytes / (1024**multiplier)