This commit is contained in:
sankalp committed 2024-11-22 20:36:57 +05:45
1 parent 721471cdb0
commit 7d9300a5e2
6 files changed
+299 -12

No files matched your search

+4
View File
@@ -0,0 +1,4 @@
{
"favorite_apps": ["Firefox", "materialgram", "Vesktop", "Spotify", "kitty"],
"city": "Kohalpur"
}
+2 -3
View File
@@ -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,
+10 -8
View File
@@ -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
+8 -1
View File
@@ -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",
]
+249
View File
@@ -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()
+26
View File
@@ -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)