This commit is contained in:
sankalp committed 2025-06-07 15:11:26 +05:45
1 parent f53b335793
commit e1a85413c2
105 files changed
+6137 -27999

No files matched your search

+246
View File
@@ -0,0 +1,246 @@
from gi.repository import GLib, Gdk
from fabric.widgets.label import Label
from fabric.widgets.box import Box
from fabric.widgets.circularprogressbar import CircularProgressBar
from fabric.widgets.button import Button
from services.battery import Battery as BatteryService
import utils.icons as icons
import config.data as data
class Battery(Box):
def __init__(self, **kwargs):
orientation = "v" if data.VERTICAL else "h"
super().__init__(
name="battery", spacing=0, orientation=orientation, visible=True, **kwargs
)
# Initialize battery service
self._battery = BatteryService()
self._battery.changed.connect(self.update_battery)
# Create the battery icon and circular progress
self.icon = Label(name="battery-icon", markup=icons.battery)
self.circle = CircularProgressBar(
name="battery-circle",
value=0,
size=26,
line_width=2,
start_angle=150,
end_angle=390,
style_classes="battery",
child=self.icon,
)
# Create button to hold the circle
self.circle_button = Button(
name="battery-button",
child=self.circle,
style="padding: 0; margin: 0; background: none;",
)
# Add event handling for hover effects
self.circle_button.add_events(
Gdk.EventMask.ENTER_NOTIFY_MASK
| Gdk.EventMask.LEAVE_NOTIFY_MASK
| Gdk.EventMask.BUTTON_PRESS_MASK
)
self.circle_button.connect("enter-notify-event", self.on_button_enter)
self.circle_button.connect("leave-notify-event", self.on_button_leave)
self.circle_button.connect("button-press-event", self.on_button_press)
# Create percentage label
self.level = Label(
name="battery-level",
style_classes="battery",
label="0%",
)
# Create the container box
self.battery_box = Box(
name="battery-box",
orientation=orientation,
spacing=0,
children=[self.circle_button],
)
# Only add level label in horizontal mode
if not data.VERTICAL:
self.battery_box.add(self.level)
# Enable tooltips
self.circle_button.set_has_tooltip(True)
self.battery_box.set_has_tooltip(True)
self.level.set_has_tooltip(True)
# Connect tooltip signals
self.circle_button.connect("query-tooltip", self.on_query_tooltip)
self.battery_box.connect("query-tooltip", self.on_query_tooltip)
self.level.connect("query-tooltip", self.on_query_tooltip)
# Add the battery box to self
self.add(self.battery_box)
# Initial update
GLib.idle_add(self.update_battery)
def on_button_enter(self, widget, event):
"""Handle mouse enter event"""
window = widget.get_window()
if window:
window.set_cursor(Gdk.Cursor(Gdk.CursorType.HAND2))
def on_button_leave(self, widget, event):
"""Handle mouse leave event"""
window = widget.get_window()
if window:
window.set_cursor(None)
def on_button_press(self, widget, event):
"""Handle button press event - toggle power profiles"""
if event.button == 1: # Left click
try:
if (
not hasattr(self._battery, "_profile_proxy")
or not self._battery._profile_proxy
):
return False
current_profile = self._battery._profile_proxy.ActiveProfile
# Get available profiles from D-Bus property
profiles = self._battery._profile_proxy.Profiles
profile_names = []
# Extract profile names from the D-Bus structure
for profile in profiles:
if isinstance(profile, dict) and "Profile" in profile:
profile_names.append(profile["Profile"])
elif hasattr(profile, "Profile"):
profile_names.append(profile.Profile)
elif isinstance(profile, str):
profile_names.append(profile)
if profile_names and current_profile in profile_names:
next_index = (profile_names.index(current_profile) + 1) % len(
profile_names
)
next_profile = profile_names[next_index]
# Set profile directly on the D-Bus proxy
self._battery._profile_proxy.ActiveProfile = next_profile
else:
return False
except Exception:
return False
return True
return False
def on_query_tooltip(self, widget, x, y, keyboard_mode, tooltip):
"""Handle tooltip query"""
tooltip.set_markup(self.get_tooltip_text())
return True
def get_tooltip_text(self):
"""Get formatted tooltip text in bullet points"""
if not self._battery.is_present:
return "No battery detected"
percentage = self._battery.percentage
state = self._battery.state
charging = state in ["CHARGING", "FULLY_CHARGED"]
# Get capacity as integer by converting from float
capacity = int(float(self._battery.capacity.rstrip("%")))
tooltip_points = []
# Status and percentage
if state == "FULLY_CHARGED":
status = f"{icons.bat_full} Fully Charged"
elif state == "CHARGING":
status = f"{icons.bat_charging} Charging"
elif percentage <= 15 and state == "DISCHARGING":
status = f"{icons.bat_low} Low Battery"
elif state == "DISCHARGING":
status = f"{icons.bat_discharging} Discharging"
else:
status = "Battery"
tooltip_points.append(f"• Status: {status}")
# Battery level and capacity
tooltip_points.append(f"• Level: {int(percentage)}%")
tooltip_points.append(f"• Battery Health: {capacity}%")
# Add power profile if available
power_profile = self._battery.power_profile
if power_profile:
# Add icon based on profile
if power_profile == "power-saver":
profile_icon = icons.power_saving
elif power_profile == "performance":
profile_icon = icons.power_performance
else: # balanced
profile_icon = icons.power_balanced
tooltip_points.append(f"• Profile: {profile_icon} {power_profile}")
# Add temperature if available
temp = self._battery.temperature
if temp != "N/A":
tooltip_points.append(f"• Temperature: {temp}")
# Add time estimates
if state == "CHARGING":
time_to_full = self._battery.time_to_full
if time_to_full:
tooltip_points.append(f"• Time until full: {time_to_full}")
elif state == "DISCHARGING":
time_to_empty = self._battery.time_to_empty
if time_to_empty:
tooltip_points.append(f"• Time remaining: {time_to_empty}")
return "\n".join(tooltip_points)
def update_battery(self, *args):
"""Update battery status and UI"""
if not self._battery.is_present:
self.set_visible(False)
return True
percentage = self._battery.percentage
state = self._battery.state
charging = state in ["CHARGING", "FULLY_CHARGED"]
power_profile = self._battery.power_profile
# Update circle progress
self.circle.set_value(percentage / 100.0)
self.level.set_label(f"{int(percentage)}%")
# Remove any existing profile classes
self.circle.remove_style_class("power-saver")
self.circle.remove_style_class("performance")
self.circle.remove_style_class("balanced")
# Apply power profile styling
if power_profile:
self.circle.add_style_class(power_profile)
# Apply alert styling if battery is low AND not charging
if percentage <= 15 and not charging:
self.icon.add_style_class("alert")
self.circle.add_style_class("alert")
else:
self.icon.remove_style_class("alert")
self.circle.remove_style_class("alert")
# Update icon based on battery state
if state == "FULLY_CHARGED":
self.icon.set_markup(icons.battery)
elif state == "CHARGING":
self.icon.set_markup(icons.charging)
elif percentage <= 15 and state == "DISCHARGING":
self.icon.set_markup(icons.alert)
elif state == "DISCHARGING":
self.icon.set_markup(icons.discharging)
else:
self.icon.set_markup(icons.battery)
self.set_visible(True)
return True
+107
View File
@@ -0,0 +1,107 @@
import os
import json
import config.data as data
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.datetime import DateTime
from modules.dock.battery import Battery
from modules.dock.metrics import Metrics
from modules.dock.controls import Controls
from modules.dock.workspaces import create_workspace_widget
from modules.dock.indicators import Indicators
class DockSystemTray(Box):
def __init__(self, orientation_val="h", **kwargs):
super().__init__(
name="system-tray",
orientation=orientation_val,
spacing=4,
**kwargs
)
# Initialize component visibility from data
self.component_visibility = data.DOCK_COMPONENTS_VISIBILITY
# Create components
self.workspaces = Button(
child=create_workspace_widget(),
name="workspaces",
)
self.metrics = Metrics()
self.battery = Battery()
self.date_time = DateTime(
name="date-time",
formatters=["%I:%M"] if not data.VERTICAL else ["%I\n%M"],
h_align="center" if not data.VERTICAL else "fill",
v_align="center",
h_expand=True,
v_expand=True,
)
self.controls = Controls()
self.indicators = Indicators()
# Add components based on position
if data.DOCK_POSITION == "Right":
self.add(self.date_time)
self.add(self.controls)
self.add(self.indicators)
self.add(self.battery)
self.add(self.metrics)
self.add(self.workspaces)
else: # Bottom or Left position
self.add(self.workspaces)
self.add(self.metrics)
self.add(self.controls)
self.add(self.indicators)
self.add(self.battery)
self.add(self.date_time)
# Apply initial visibility
self.apply_component_props()
def apply_component_props(self):
components = {
"workspaces": self.workspaces,
"metrics": self.metrics,
"battery": self.battery,
"date_time": self.date_time,
"controls": self.controls,
"indicators": self.indicators,
}
for component_name, widget in components.items():
if component_name in self.component_visibility:
widget.set_visible(self.component_visibility[component_name])
def toggle_component_visibility(self, component_name):
components = {
"workspaces": self.workspaces,
"metrics": self.metrics,
"battery": self.battery,
"date_time": self.date_time,
"controls": self.controls,
"indicators": self.indicators,
}
if component_name in components and component_name in self.component_visibility:
self.component_visibility[component_name] = not self.component_visibility[component_name]
components[component_name].set_visible(self.component_visibility[component_name])
config_file = os.path.expanduser(f"~/.config/{data.APP_NAME}/config/config.json")
if os.path.exists(config_file):
try:
with open(config_file, "r") as f:
config = json.load(f)
config[f"dock_{component_name}_visible"] = self.component_visibility[component_name]
with open(config_file, "w") as f:
json.dump(config, f, indent=4)
except Exception as e:
print(f"Error updating config file: {e}")
return self.component_visibility[component_name]
return None
+272
View File
@@ -0,0 +1,272 @@
from fabric.audio.service import Audio
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.circularprogressbar import CircularProgressBar
from fabric.widgets.eventbox import EventBox
from fabric.widgets.label import Label
from fabric.widgets.overlay import Overlay
from fabric.widgets.scale import Scale
from gi.repository import Gdk, GLib
import config.data as data
import utils.icons as icons
from services.brightness import Brightness
class BrightnessSmall(Box):
def __init__(self, **kwargs):
super().__init__(name="button-bar-brightness", **kwargs)
self.brightness = Brightness.get_initial()
if self.brightness.screen_brightness == -1:
self.destroy()
return
self.progress_bar = CircularProgressBar(
name="button-brightness", size=28, line_width=2,
start_angle=150, end_angle=390,
)
self.brightness_label = Label(name="brightness-label", markup=icons.brightness_high)
self.brightness_button = Button(child=self.brightness_label)
self.event_box = EventBox(
events=["scroll", "smooth-scroll"],
child=Overlay(
child=self.progress_bar,
overlays=self.brightness_button
),
)
self.event_box.connect("scroll-event", self.on_scroll)
self.add(self.event_box)
self.add_events(Gdk.EventMask.SCROLL_MASK | Gdk.EventMask.SMOOTH_SCROLL_MASK)
self._updating_from_brightness = False
self._pending_value = None
self._update_source_id = None
self.progress_bar.connect("notify::value", self.on_progress_value_changed)
self.brightness.connect("screen", self.on_brightness_changed)
self.on_brightness_changed()
def on_scroll(self, widget, event):
if self.brightness.max_screen == -1:
return
step_size = 5
current_norm = self.progress_bar.value
if event.delta_y < 0:
new_norm = min(current_norm + (step_size / self.brightness.max_screen), 1)
elif event.delta_y > 0:
new_norm = max(current_norm - (step_size / self.brightness.max_screen), 0)
else:
return
self.progress_bar.value = new_norm
def on_progress_value_changed(self, widget, pspec):
if self._updating_from_brightness:
return
new_norm = widget.value
new_brightness = int(new_norm * self.brightness.max_screen)
self._pending_value = new_brightness
if self._update_source_id is None:
self._update_source_id = GLib.timeout_add(50, self._update_brightness_callback)
def _update_brightness_callback(self):
if self._pending_value is not None and self._pending_value != self.brightness.screen_brightness:
self.brightness.screen_brightness = self._pending_value
self._pending_value = None
return True
else:
self._update_source_id = None
return False
def on_brightness_changed(self, *args):
if self.brightness.max_screen == -1:
return
normalized = self.brightness.screen_brightness / self.brightness.max_screen
self._updating_from_brightness = True
self.progress_bar.value = normalized
self._updating_from_brightness = False
brightness_percentage = int(normalized * 100)
if brightness_percentage >= 75:
self.brightness_label.set_markup(icons.brightness_high)
elif brightness_percentage >= 24:
self.brightness_label.set_markup(icons.brightness_medium)
else:
self.brightness_label.set_markup(icons.brightness_low)
self.set_tooltip_text(f"{brightness_percentage}%")
def destroy(self):
if self._update_source_id is not None:
GLib.source_remove(self._update_source_id)
super().destroy()
class VolumeSmall(Box):
def __init__(self, **kwargs):
super().__init__(name="button-bar-vol", **kwargs)
self.audio = Audio()
self.progress_bar = CircularProgressBar(
name="button-volume", size=28, line_width=2,
start_angle=150, end_angle=390,
)
self.vol_label = Label(name="vol-label", markup=icons.vol_high)
self.vol_button = Button(on_clicked=self.toggle_mute, child=self.vol_label)
self.event_box = EventBox(
events=["scroll", "smooth-scroll"],
child=Overlay(child=self.progress_bar, overlays=self.vol_button),
)
self.audio.connect("notify::speaker", self.on_new_speaker)
if self.audio.speaker:
self.audio.speaker.connect("changed", self.on_speaker_changed)
self.event_box.connect("scroll-event", self.on_scroll)
self.add(self.event_box)
self.on_speaker_changed()
self.add_events(Gdk.EventMask.SCROLL_MASK | Gdk.EventMask.SMOOTH_SCROLL_MASK)
def on_new_speaker(self, *args):
if self.audio.speaker:
self.audio.speaker.connect("changed", self.on_speaker_changed)
self.on_speaker_changed()
def toggle_mute(self, event):
current_stream = self.audio.speaker
if current_stream:
current_stream.muted = not current_stream.muted
if current_stream.muted:
self.on_speaker_changed()
self.progress_bar.add_style_class("muted")
self.vol_label.add_style_class("muted")
else:
self.on_speaker_changed()
self.progress_bar.remove_style_class("muted")
self.vol_label.remove_style_class("muted")
def on_scroll(self, _, event):
if not self.audio.speaker:
return
if event.direction == Gdk.ScrollDirection.SMOOTH:
if abs(event.delta_y) > 0:
self.audio.speaker.volume -= event.delta_y
if abs(event.delta_x) > 0:
self.audio.speaker.volume += event.delta_x
def on_speaker_changed(self, *_):
if not self.audio.speaker:
return
vol_high_icon = icons.vol_high
vol_medium_icon = icons.vol_medium
vol_mute_icon = icons.vol_off
vol_off_icon = icons.vol_mute
if "bluetooth" in self.audio.speaker.icon_name:
vol_high_icon = icons.bluetooth_connected
vol_medium_icon = icons.bluetooth
vol_mute_icon = icons.bluetooth_off
vol_off_icon = icons.bluetooth_disconnected
self.progress_bar.value = self.audio.speaker.volume / 100
if self.audio.speaker.muted:
self.vol_label.set_markup(vol_mute_icon)
self.progress_bar.add_style_class("muted")
self.vol_label.add_style_class("muted")
self.set_tooltip_text("Muted")
return
else:
self.progress_bar.remove_style_class("muted")
self.vol_label.remove_style_class("muted")
self.set_tooltip_text(f"{round(self.audio.speaker.volume)}%")
if self.audio.speaker.volume > 74:
self.vol_label.set_markup(vol_high_icon)
elif self.audio.speaker.volume > 0:
self.vol_label.set_markup(vol_medium_icon)
else:
self.vol_label.set_markup(vol_off_icon)
class MicSmall(Box):
def __init__(self, **kwargs):
super().__init__(name="button-bar-mic", **kwargs)
self.audio = Audio()
self.progress_bar = CircularProgressBar(
name="button-mic", size=28, line_width=2,
start_angle=150, end_angle=390,
)
self.mic_label = Label(name="mic-label", markup=icons.mic)
self.mic_button = Button(on_clicked=self.toggle_mute, child=self.mic_label)
self.event_box = EventBox(
events=["scroll", "smooth-scroll"],
child=Overlay(child=self.progress_bar, overlays=self.mic_button),
)
self.audio.connect("notify::microphone", self.on_new_microphone)
if self.audio.microphone:
self.audio.microphone.connect("changed", self.on_microphone_changed)
self.event_box.connect("scroll-event", self.on_scroll)
self.add_events(Gdk.EventMask.SCROLL_MASK | Gdk.EventMask.SMOOTH_SCROLL_MASK)
self.add(self.event_box)
self.on_microphone_changed()
def on_new_microphone(self, *args):
if self.audio.microphone:
self.audio.microphone.connect("changed", self.on_microphone_changed)
self.on_microphone_changed()
def toggle_mute(self, event):
current_stream = self.audio.microphone
if current_stream:
current_stream.muted = not current_stream.muted
if current_stream.muted:
self.mic_label.set_markup(icons.mic_mute)
self.progress_bar.add_style_class("muted")
self.mic_label.add_style_class("muted")
else:
self.on_microphone_changed()
self.progress_bar.remove_style_class("muted")
self.mic_label.remove_style_class("muted")
def on_scroll(self, _, event):
if not self.audio.microphone:
return
if event.direction == Gdk.ScrollDirection.SMOOTH:
if abs(event.delta_y) > 0:
self.audio.microphone.volume -= event.delta_y
if abs(event.delta_x) > 0:
self.audio.microphone.volume += event.delta_x
def on_microphone_changed(self, *_):
if not self.audio.microphone:
return
self.progress_bar.value = self.audio.microphone.volume / 100
if self.audio.microphone.muted:
self.mic_label.set_markup(icons.mic_mute)
self.progress_bar.add_style_class("muted")
self.mic_label.add_style_class("muted")
self.set_tooltip_text("Muted")
return
else:
self.progress_bar.remove_style_class("muted")
self.mic_label.remove_style_class("muted")
self.set_tooltip_text(f"{round(self.audio.microphone.volume)}%")
if self.audio.microphone.volume > 74:
self.mic_label.set_markup(icons.mic)
elif self.audio.microphone.volume > 0:
self.mic_label.set_markup(icons.mic)
else:
self.mic_label.set_markup(icons.mic_mute)
class Controls(Box):
def __init__(self, **kwargs):
brightness = Brightness.get_initial()
children = []
if (brightness.screen_brightness != -1):
children.append(BrightnessSmall())
children.extend([VolumeSmall(), MicSmall()])
super().__init__(
name="control",
orientation="h" if not data.VERTICAL else "v",
spacing=4,
children=children,
**kwargs,
)
self.show_all()
+196
View File
@@ -0,0 +1,196 @@
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.label import Label
from fabric.widgets.overlay import Overlay
from fabric.widgets.circularprogressbar import CircularProgressBar
from gi.repository import Gdk, GLib
import config.data as data
import utils.icons as icons
from services.network import NetworkClient
from fabric.bluetooth import BluetoothClient
class WifiIndicator(Button):
def __init__(self, **kwargs):
super().__init__(name="button-bar-network", **kwargs)
# Initialize NetworkClient
self.network = NetworkClient()
# Create progress bar and wifi label
self.progress_bar = CircularProgressBar(
name="button-network",
size=28,
line_width=2,
start_angle=150,
end_angle=390,
)
self.wifi_label = Label(name="network-label", markup=icons.loader)
# Create overlay with progress bar and label
self.add(Overlay(
child=self.progress_bar,
overlays=self.wifi_label
))
# Connect signals
self.network.connect("ready", self.on_network_ready)
# If already ready, set up signals immediately
if self.network.is_ready:
self.setup_signals()
def on_network_ready(self, *args):
self.setup_signals()
def setup_signals(self):
# Connect to NetworkClient signals
self.network.connect("changed", self.on_network_changed)
self.network.connect("wifi-device-added", self.on_wifi_device_added)
self.network.connect("wifi-device-removed", self.on_wifi_device_removed)
self.network.connect("ethernet-device-added", self.on_ethernet_device_added)
self.network.connect("ethernet-device-removed", self.on_ethernet_device_removed)
# Connect to WiFi device if available
if self.network.wifi_device:
self.network.wifi_device.connect("changed", self.on_wifi_changed)
self.network.wifi_device.connect("ap-added", self.on_wifi_changed)
self.network.wifi_device.connect("ap-removed", self.on_wifi_changed)
# Connect to Ethernet device if available
if self.network.ethernet_device:
self.network.ethernet_device.connect("changed", self.on_ethernet_changed)
self.update_state()
def update_state(self):
# Check ethernet first
if self.network.ethernet_device and self.network.ethernet_device.state == "activated":
self.wifi_label.set_markup(icons.world)
self.progress_bar.value = 1.0 # Full circle for ethernet
tooltip = f"Connected to Ethernet ({self.network.ethernet_device.speed})"
self.set_tooltip_text(tooltip)
return
# Then check wifi
if not self.network.wifi_device:
self.wifi_label.set_markup(icons.cloud_off)
self.progress_bar.value = 0
tooltip = "No WiFi device found"
elif not self.network.wifi_device.wireless_enabled:
self.wifi_label.set_markup(icons.cloud_off)
self.progress_bar.value = 0
tooltip = "WiFi disabled"
else:
active_ap = self.network.wifi_device.active_access_point
if active_ap:
strength = active_ap.strength
self.progress_bar.value = strength / 100
if strength >= 75:
self.wifi_label.set_markup(icons.wifi_3)
elif strength >= 50:
self.wifi_label.set_markup(icons.wifi_2)
elif strength >= 25:
self.wifi_label.set_markup(icons.wifi_1)
else:
self.wifi_label.set_markup(icons.wifi_0)
tooltip = f"Connected to {active_ap.ssid} ({strength}%)"
else:
self.wifi_label.set_markup(icons.world_off)
self.progress_bar.value = 0
tooltip = "Not connected"
self.set_tooltip_text(tooltip)
def on_wifi_device_added(self, *args):
self.network.wifi_device.connect("changed", self.on_wifi_changed)
self.network.wifi_device.connect("ap-added", self.on_wifi_changed)
self.network.wifi_device.connect("ap-removed", self.on_wifi_changed)
self.update_state()
def on_wifi_device_removed(self, *args):
self.update_state()
def on_network_changed(self, *args):
self.update_state()
def on_wifi_changed(self, *args):
self.update_state()
def on_ethernet_device_added(self, *args):
if self.network.ethernet_device:
self.network.ethernet_device.connect("changed", self.on_ethernet_changed)
self.update_state()
def on_ethernet_device_removed(self, *args):
self.update_state()
def on_ethernet_changed(self, *args):
self.update_state()
class BluetoothIndicator(Button):
def __init__(self, **kwargs):
super().__init__(name="button-bar-bluetooth", **kwargs)
# Initialize BluetoothClient
self.bluetooth = BluetoothClient()
# Create bluetooth label
self.bt_label = Label(name="bluetooth-label", markup=icons.loader)
self.add(self.bt_label)
# Connect signals
self.bluetooth.connect("changed", self.on_bluetooth_changed)
self.bluetooth.connect("device-added", self.on_device_added)
self.bluetooth.connect("device-removed", self.on_device_removed)
# Initial state update
self.update_state()
def update_state(self):
if not self.bluetooth.enabled:
self.bt_label.set_markup(icons.bluetooth_off)
tooltip = "Bluetooth disabled"
else:
connected_devices = self.bluetooth.connected_devices
if connected_devices:
self.bt_label.set_markup(icons.bluetooth_connected)
if len(connected_devices) == 1:
device = connected_devices[0]
tooltip = f"Connected to {device.alias}"
if device.battery_percentage > 0:
tooltip += f" ({device.battery_percentage:.0f}%)"
else:
tooltip = f"Connected to {len(connected_devices)} devices"
else:
if self.bluetooth.scanning:
self.bt_label.set_markup(icons.loader)
tooltip = "Scanning for devices..."
else:
self.bt_label.set_markup(icons.bluetooth)
tooltip = "No devices connected"
self.set_tooltip_text(tooltip)
def on_bluetooth_changed(self, *args):
self.update_state()
def on_device_added(self, _, address):
self.update_state()
def on_device_removed(self, _, address):
self.update_state()
class Indicators(Box):
def __init__(self, **kwargs):
super().__init__(
name="indicator",
orientation="h" if not data.VERTICAL else "v",
spacing=4,
children=[WifiIndicator(), BluetoothIndicator()],
**kwargs,
)
self.show_all()
+1047
View File
File diff suppressed because it is too large. Load diff
+299
View File
@@ -0,0 +1,299 @@
import json
import logging
import subprocess
import psutil
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.circularprogressbar import CircularProgressBar
from fabric.widgets.label import Label
from fabric.widgets.revealer import Revealer
from gi.repository import GLib
import config.data as data
import utils.icons as icons
logger = logging.getLogger(__name__)
class MetricsProvider:
"""
Class responsible for obtaining centralized CPU, memory, disk usage, and battery metrics.
It updates periodically so that all widgets querying it display the same values.
"""
def __init__(self):
self.gpu = []
self.cpu = 0.0
self.mem = 0.0
self.swap = 0.0
self.disk = []
self.bat_percent = 0.0
self.bat_charging = None
self.bat_time = 0
self._gpu_update_running = False
GLib.timeout_add_seconds(1, self._update)
def _update(self):
self.cpu = psutil.cpu_percent(interval=0)
self.mem = psutil.virtual_memory().percent
self.swap = psutil.swap_memory().percent
self.disk = [psutil.disk_usage(path).percent for path in data.METRICS_DISKS]
if not self._gpu_update_running:
self._start_gpu_update_async()
return True
def _start_gpu_update_async(self):
"""Starts a new GLib thread to run nvtop in the background."""
self._gpu_update_running = True
GLib.Thread.new("nvtop-thread", lambda _: self._run_nvtop_in_thread(), None)
def _run_nvtop_in_thread(self):
"""Runs nvtop via subprocess in a separate GLib thread."""
output = None
error_message = None
try:
result = subprocess.check_output(["nvtop", "-s"], text=True, timeout=10)
output = result
except FileNotFoundError:
error_message = "nvtop command not found."
logger.warning(error_message)
except subprocess.CalledProcessError as e:
error_message = f"nvtop failed with exit code {e.returncode}: {e.stderr.strip()}"
logger.error(error_message)
except subprocess.TimeoutExpired:
error_message = "nvtop command timed out."
logger.error(error_message)
except Exception as e:
error_message = f"Unexpected error running nvtop: {e}"
logger.error(error_message)
GLib.idle_add(self._process_gpu_output, output, error_message)
self._gpu_update_running = False
def _process_gpu_output(self, output, error_message):
"""Process nvtop JSON output on the main loop."""
try:
if error_message:
logger.error(f"GPU update failed: {error_message}")
self.gpu = []
elif output:
try:
info = json.loads(output)
if not info or not isinstance(info, list):
logger.warning("Invalid nvtop data format")
self.gpu = []
return False
gpu_utils = []
for gpu in info:
if not isinstance(gpu, dict):
continue
util = gpu.get("gpu_util", "0%")
if not isinstance(util, str):
continue
try:
# Remove % and convert to int, default to 0 if fails
util_value = int(util.rstrip("%") or 0)
gpu_utils.append(util_value)
except (ValueError, TypeError):
continue
self.gpu = gpu_utils if gpu_utils else []
except (KeyError, ValueError, TypeError, AttributeError) as e:
logger.error(f"Failed parsing nvtop JSON: {e}")
self.gpu = []
else:
logger.warning("nvtop returned no output.")
self.gpu = []
except json.JSONDecodeError as e:
logger.error(f"JSON decode error: {e}")
self.gpu = []
except Exception as e:
logger.error(f"Error processing nvtop output: {e}")
self.gpu = []
return False
def get_metrics(self):
return (self.cpu, self.mem, self.swap, self.disk, self.gpu)
def get_battery(self):
return (self.bat_percent, self.bat_charging, self.bat_time)
def get_gpu_info(self):
try:
result = subprocess.check_output(["nvtop", "-s"], text=True, timeout=5)
if not result.strip():
return []
info = json.loads(result)
if not info or not isinstance(info, list):
return []
valid_gpus = []
for gpu in info:
if not isinstance(gpu, dict):
continue
gpu_data = {
'device_name': gpu.get('device_name', 'GPU'),
'gpu_util': gpu.get('gpu_util', '0%'),
'mem_util': gpu.get('mem_util', '0%'),
'temperature': gpu.get('temperature', '0 C'),
'power': gpu.get('power', '0 W')
}
valid_gpus.append(gpu_data)
return valid_gpus
except FileNotFoundError:
logger.warning("nvtop not found; GPU info unavailable.")
return []
except subprocess.CalledProcessError as e:
logger.error(f"nvtop init sync failed: {e}")
return []
except subprocess.TimeoutExpired:
logger.error("nvtop init call timed out.")
return []
except json.JSONDecodeError as e:
logger.error(f"Init JSON parse error: {e}")
return []
except Exception as e:
logger.error(f"Unexpected error during GPU init: {e}")
return []
shared_provider = MetricsProvider()
class SingularMetric:
def __init__(self, id, name, icon):
self.name_markup = name
self.icon_markup = icon
self.icon = Label(name="metrics-icon", markup=icon)
self.circle = CircularProgressBar(
name="metrics-circle",
value=0,
size=28,
line_width=2,
start_angle=150,
end_angle=390,
style_classes=id,
child=self.icon,
)
self.level = Label(name="metrics-level", style_classes=id, label="0%")
self.revealer = Revealer(
name=f"metrics-{id}-revealer",
transition_duration=250,
transition_type="slide-left",
child=self.level,
child_revealed=False,
)
self.box = Box(
name=f"metrics-{id}-box",
orientation="h",
spacing=0,
children=[self.circle, self.revealer],
)
def markup(self):
return f"{self.icon_markup} {self.name_markup}: {self.level.get_label()}"
class Metrics(Button):
def __init__(self, **kwargs):
super().__init__(name="metrics", **kwargs)
main_box = Box(
spacing=0,
orientation="h" if not data.VERTICAL else "v",
visible=True,
all_visible=True,
)
visible = getattr(data, "METRICS_VISIBLE", {'cpu': True, 'ram': True, 'swap': True, 'disk': True, 'gpu': True})
disks = [SingularMetric("disk", f"DISK ({path})" if len(data.METRICS_DISKS) != 1 else "DISK", icons.disk)
for path in data.METRICS_DISKS] if visible.get('disk', True) else []
gpu_info = shared_provider.get_gpu_info()
gpus = [SingularMetric(f"gpu", f"GPU ({v['device_name']})" if len(gpu_info) != 1 else "GPU", icons.gpu)
for v in gpu_info] if visible.get('gpu', True) else []
self.cpu = SingularMetric("cpu", "CPU", icons.cpu) if visible.get('cpu', True) else None
self.ram = SingularMetric("ram", "RAM", icons.memory) if visible.get('ram', True) else None
self.swap = SingularMetric("swap", "SWAP", icons.swap) if visible.get('swap', True) else None
self.disk = disks
self.gpu = gpus
for disk in self.disk:
main_box.add(disk.box)
main_box.add(Box(name="metrics-sep"))
if self.ram:
main_box.add(self.ram.box)
main_box.add(Box(name="metrics-sep"))
if self.swap:
main_box.add(self.swap.box)
main_box.add(Box(name="metrics-sep"))
if self.cpu:
main_box.add(self.cpu.box)
for gpu in self.gpu:
main_box.add(Box(name="metrics-sep"))
main_box.add(gpu.box)
self.add(main_box)
self.connect("clicked", self.on_click)
self.labels_visible = False
GLib.timeout_add_seconds(1, self.update_metrics)
def _format_percentage(self, value: int) -> str:
"""Formato natural del porcentaje sin forzar ancho fijo."""
return f"{value}%"
def on_click(self, widget):
if not data.VERTICAL:
self.labels_visible = not self.labels_visible
if self.cpu: self.cpu.revealer.set_reveal_child(self.labels_visible)
if self.ram: self.ram.revealer.set_reveal_child(self.labels_visible)
if self.swap: self.swap.revealer.set_reveal_child(self.labels_visible)
for disk in self.disk:
disk.revealer.set_reveal_child(self.labels_visible)
for gpu in self.gpu:
gpu.revealer.set_reveal_child(self.labels_visible)
def update_metrics(self):
cpu, mem, swap, disks, gpus = shared_provider.get_metrics()
if self.cpu:
self.cpu.circle.set_value(cpu / 100.0)
self.cpu.level.set_label(self._format_percentage(int(cpu)))
if self.ram:
self.ram.circle.set_value(mem / 100.0)
self.ram.level.set_label(self._format_percentage(int(mem)))
if self.swap:
self.swap.circle.set_value(swap / 100.0)
self.swap.level.set_label(self._format_percentage(int(swap)))
for i, disk in enumerate(self.disk):
if i < len(disks):
disk.circle.set_value(disks[i] / 100.0)
disk.level.set_label(self._format_percentage(int(disks[i])))
for i, gpu in enumerate(self.gpu):
if i < len(gpus):
gpu.circle.set_value(gpus[i] / 100.0)
gpu.level.set_label(self._format_percentage(int(gpus[i])))
tooltip_metrics = []
if self.disk: tooltip_metrics.extend(self.disk)
if self.ram: tooltip_metrics.append(self.ram)
if self.swap: tooltip_metrics.append(self.swap)
if self.cpu: tooltip_metrics.append(self.cpu)
if self.gpu: tooltip_metrics.extend(self.gpu)
self.set_tooltip_markup((" - " if not data.VERTICAL else "\n").join([v.markup() for v in tooltip_metrics]))
return True
+19
View File
@@ -0,0 +1,19 @@
import json
from fabric.hyprland.service import Hyprland
from fabric.widgets.label import Label
connection = Hyprland()
def create_workspace_widget():
workspaceData = connection.send_command("j/activeworkspace").reply
activeWorkspace = json.loads(workspaceData.decode("utf-8"))["name"]
workspace_label = Label(label=f"{activeWorkspace}")
def on_workspace(obj, signal):
nonlocal workspace_label
activeWorkspace = json.loads(signal.data[0])
workspace_label.set_label(f"{activeWorkspace}")
connection.connect("event::workspace", on_workspace)
return workspace_label