This commit is contained in:
sankalp committed 2025-06-12 03:07:29 +05:45
1 parent 924b46fb89
commit 54732242ed
19 files changed
+16034 -149

No files matched your search

+530
View File
@@ -0,0 +1,530 @@
import json
import logging
import cairo
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.image import Image
from fabric.utils import exec_shell_command, exec_shell_command_async, get_relative_path
from fabric.utils.helpers import get_desktop_applications
from fabric.hyprland.widgets import get_hyprland_connection
from gi.repository import Gdk, GLib, Gtk
import config.data as data
from utils.icon_resolver import IconResolver
def read_config():
"""Read and return the full configuration from the JSON file, handling missing file."""
config_path = get_relative_path("../../../config/dock.json")
try:
with open(config_path, "r") as file:
config_data = json.load(file)
if (
"pinned_apps" in config_data
and config_data["pinned_apps"]
and isinstance(config_data["pinned_apps"][0], str)
):
all_apps = get_desktop_applications()
app_map = {app.name: app for app in all_apps if app.name}
old_pinned = config_data["pinned_apps"]
config_data["pinned_apps"] = []
for app_id in old_pinned:
app = app_map.get(app_id)
if app:
app_data_obj = {
"name": app.name,
"display_name": app.display_name,
"window_class": app.window_class,
"executable": app.executable,
"command_line": app.command_line,
}
config_data["pinned_apps"].append(app_data_obj)
else:
config_data["pinned_apps"].append({"name": app_id})
except (FileNotFoundError, json.JSONDecodeError):
config_data = {"pinned_apps": []}
return config_data
def createSurfaceFromWidget(widget: Gtk.Widget) -> cairo.ImageSurface:
alloc = widget.get_allocation()
surface = cairo.ImageSurface(
cairo.Format.ARGB32,
alloc.width,
alloc.height,
)
cr = cairo.Context(surface)
cr.set_source_rgba(255, 255, 255, 0)
cr.rectangle(0, 0, alloc.width, alloc.height)
cr.fill()
widget.draw(cr)
return surface
class Applications(Box):
def __init__(self, orientation_val="h", dock_instance=None, **kwargs):
super().__init__(
name="applications-dock", orientation=orientation_val, spacing=4, **kwargs
)
# Store reference to the dock instance
self.dock_instance = dock_instance
self.config = read_config()
self.conn = get_hyprland_connection()
self.icon_resolver = IconResolver()
self.pinned = self.config.get("pinned_apps", [])
self.config_path = get_relative_path("../../../config/dock.json")
self.app_map = {}
self._all_apps = get_desktop_applications()
self.app_identifiers = self._build_app_identifiers_map()
self._drag_in_progress = False
# Initialize with empty content
self.update_applications()
# Connect to Hyprland events for live updates
if self.conn.ready:
self.update_applications()
else:
self.conn.connect("event::ready", lambda *args: self.update_applications())
# Connect to window events to update in real-time
for ev in ("activewindow", "openwindow", "closewindow", "changefloatingmode"):
self.conn.connect(f"event::{ev}", lambda *args: self.update_applications())
# Set up periodic refresh for config changes
GLib.timeout_add_seconds(1, self.check_config_change)
def _build_app_identifiers_map(self):
app_map = {}
for app in self._all_apps:
if app.window_class:
app_map[app.window_class.lower()] = app
if app.executable:
app_map[app.executable.lower()] = app
if app.name:
app_map[app.name.lower()] = app
return app_map
def find_app_by_key(self, key):
if not key:
return None
key_lower = key.lower() if isinstance(key, str) else ""
return self.app_identifiers.get(key_lower)
def find_app(self, app_identifier):
if not app_identifier:
return None
if isinstance(app_identifier, dict):
for key in [
"window_class",
"executable",
"command_line",
"name",
"display_name",
]:
if key in app_identifier and app_identifier[key]:
app = self.find_app_by_key(app_identifier[key])
if app:
return app
return None
return self.find_app_by_key(app_identifier)
def create_button(self, app_identifier, instances):
desktop_app = self.find_app(app_identifier)
icon_img = None
display_name = None
icon_size = data.DOCK_ICON_SIZE
if desktop_app:
icon_img = desktop_app.get_icon_pixbuf(size=icon_size)
display_name = desktop_app.display_name or desktop_app.name
id_value = (
app_identifier["name"]
if isinstance(app_identifier, dict)
else app_identifier
)
if not icon_img:
icon_img = self.icon_resolver.get_icon_pixbuf(id_value, icon_size)
if not icon_img:
icon_img = self.icon_resolver.get_icon_pixbuf(
"application-x-executable-symbolic", icon_size
)
if not icon_img:
icon_img = self.icon_resolver.get_icon_pixbuf(
"image-missing", icon_size
)
items = [Image(pixbuf=icon_img)]
tooltip = display_name or (id_value if isinstance(id_value, str) else "Unknown")
if not display_name and instances and instances[0].get("title"):
tooltip = instances[0]["title"]
button = Button(
child=Box(
name="dock-icon", orientation="v", h_align="center", children=items
),
on_clicked=lambda *a: self.handle_app(
app_identifier, instances, desktop_app
),
tooltip_text=tooltip,
name="dock-app-button",
)
button.app_identifier = app_identifier
button.desktop_app = desktop_app
button.instances = instances
if instances:
button.add_style_class("instance")
button.drag_source_set(
Gdk.ModifierType.BUTTON1_MASK,
[Gtk.TargetEntry.new("text/plain", Gtk.TargetFlags.SAME_APP, 0)],
Gdk.DragAction.MOVE,
)
# Add drag destination capabilities
button.drag_dest_set(
Gtk.DestDefaults.ALL,
[Gtk.TargetEntry.new("text/plain", Gtk.TargetFlags.SAME_APP, 0)],
Gdk.DragAction.MOVE,
)
# Connect drag signals
button.connect("drag-begin", self.on_drag_begin)
button.connect("drag-data-get", self.on_drag_data_get)
button.connect("drag-data-received", self.on_drag_data_received)
button.connect("drag-end", self.on_drag_end)
button.connect("button-press-event", self.on_button_press)
return button
def handle_app(self, app_identifier, instances, desktop_app=None):
if not instances:
if not desktop_app:
desktop_app = self.find_app(app_identifier)
if desktop_app:
launch_success = desktop_app.launch()
if not launch_success:
if desktop_app.command_line:
exec_shell_command_async(f"nohup {desktop_app.command_line} &")
elif desktop_app.executable:
exec_shell_command_async(f"nohup {desktop_app.executable} &")
else:
cmd_to_run = None
if isinstance(app_identifier, dict):
if (
"command_line" in app_identifier
and app_identifier["command_line"]
):
cmd_to_run = app_identifier["command_line"]
elif (
"executable" in app_identifier and app_identifier["executable"]
):
cmd_to_run = app_identifier["executable"]
elif "name" in app_identifier and app_identifier["name"]:
cmd_to_run = app_identifier["name"]
else:
cmd_to_run = app_identifier
if cmd_to_run:
exec_shell_command_async(f"nohup {cmd_to_run} &")
else:
address = instances[0].get("address")
if address:
exec_shell_command(f"hyprctl dispatch focuswindow address:{address}")
def update_pinned_apps_file(self):
config_path = get_relative_path("../../../config/dock.json")
try:
with open(config_path, "w") as file:
json.dump(self.config, file, indent=4)
return True
except Exception as e:
logging.error(f"Failed to write dock config: {e}")
return False
def check_config_change(self):
new_config = read_config()
if new_config.get("pinned_apps", []) != self.config.get("pinned_apps", []):
self.config = new_config
self.pinned = self.config.get("pinned_apps", [])
self.update_app_map()
self.update_applications()
return True
def update_app_map(self):
self._all_apps = get_desktop_applications()
self.app_identifiers = self._build_app_identifiers_map()
def _normalize_window_class(self, class_name):
# Implementation from dock.py
return class_name.lower()
def update_applications(self):
# Clear existing children
for child in self.get_children():
self.remove(child)
# Get running windows from Hyprland
clients = []
try:
clients_data = self.conn.send_command("j/clients").reply
if clients_data:
clients = json.loads(clients_data.decode("utf-8"))
except Exception as e:
logging.error(f"Failed to get clients: {e}")
# Group clients by window class
running_windows = {}
for client in clients:
class_name = client.get("class", "").lower()
if class_name:
if class_name not in running_windows:
running_windows[class_name] = []
running_windows[class_name].append(client)
# Create pinned app buttons
pinned_buttons = []
used_window_classes = set()
for app_data_item in self.pinned:
app = self.find_app(app_data_item)
instances = []
matched_class = None
possible_identifiers = []
if isinstance(app_data_item, dict):
for key in [
"window_class",
"executable",
"command_line",
"name",
"display_name",
]:
if key in app_data_item and app_data_item[key]:
possible_identifiers.append(app_data_item[key].lower())
elif isinstance(app_data_item, str):
possible_identifiers.append(app_data_item.lower())
# Try to match with running windows
for class_name, window_instances in running_windows.items():
for identifier in possible_identifiers:
if identifier in class_name or class_name in identifier:
instances = window_instances
matched_class = class_name
break
if matched_class:
break
if matched_class:
used_window_classes.add(matched_class)
pinned_buttons.append(self.create_button(app_data_item, instances))
# Create buttons for non-pinned running apps
open_buttons = []
for class_name, instances in running_windows.items():
if class_name not in used_window_classes:
app = self.find_app_by_key(class_name)
if app:
app_data_obj = {
"name": app.name,
"display_name": app.display_name,
"window_class": app.window_class,
"executable": app.executable,
"command_line": app.command_line,
}
identifier = app_data_obj
else:
identifier = class_name
open_buttons.append(self.create_button(identifier, instances))
# Add all buttons to the dock
children = pinned_buttons
if pinned_buttons and open_buttons:
separator_orientation = (
Gtk.Orientation.VERTICAL
if self.get_orientation() == Gtk.Orientation.HORIZONTAL
else Gtk.Orientation.HORIZONTAL
)
children.append(
Box(
orientation=separator_orientation,
v_expand=False,
h_expand=False,
h_align="center",
v_align="center",
name="dock-separator",
)
)
children.extend(open_buttons)
# Add all children to the dock
for child in children:
self.add(child)
# Show or hide based on whether there are any applications
if len(children) > 0:
self.show_all()
else:
self.hide()
return True
def on_button_press(self, widget, event):
# Handle right-click to pin/unpin applications
if event.button == 3: # Right mouse button
app_identifier = widget.app_identifier
instances = widget.instances
# Check if app is already pinned
is_pinned = False
for pinned_app in self.pinned:
if (
isinstance(app_identifier, dict)
and isinstance(pinned_app, dict)
and app_identifier.get("name") == pinned_app.get("name")
):
is_pinned = True
break
elif app_identifier == pinned_app:
is_pinned = True
break
# If not pinned, add to pinned apps
if not is_pinned:
if hasattr(widget, "desktop_app") and widget.desktop_app:
app = widget.desktop_app
app_data_obj = {
"name": app.name,
"display_name": app.display_name,
"window_class": app.window_class,
"executable": app.executable,
"command_line": app.command_line,
}
self.pinned.append(app_data_obj)
else:
self.pinned.append(app_identifier)
self.config["pinned_apps"] = self.pinned
self.update_pinned_apps_file()
self.update_applications()
return True
return False
def on_drag_begin(self, widget, drag_context):
self._drag_in_progress = True
Gtk.drag_set_icon_surface(drag_context, createSurfaceFromWidget(widget))
# Prevent dock from hiding during drag
if self.dock_instance:
self.dock_instance.prevent_hiding(True)
def on_drag_data_get(self, widget, drag_context, data_obj, info, time):
# Find the index of the button being dragged
children = self.get_children()
index = children.index(widget)
data_obj.set_text(str(index), -1)
def on_drag_data_received(self, widget, drag_context, x, y, data_obj, info, time):
# Get source and target indices
try:
source_index = int(data_obj.get_text())
except (TypeError, ValueError):
return
children = self.get_children()
target_index = children.index(widget)
if source_index != target_index:
# Find separator index to determine pinned vs unpinned sections
separator_index = -1
for i, child in enumerate(children):
if child.get_name() == "dock-separator":
separator_index = i
break
# Check if we're dragging from unpinned to pinned section
cross_section_drag_to_pin = separator_index != -1 and (
source_index > separator_index and target_index < separator_index
)
# Check if we're dragging from pinned to unpinned section
cross_section_drag_to_unpin = separator_index != -1 and (
source_index < separator_index and target_index > separator_index
)
# If dragging from unpinned to pinned, add to pinned apps
if cross_section_drag_to_pin:
source_widget = children[source_index]
app_identifier = source_widget.app_identifier
# Add to pinned apps if not already pinned
if hasattr(source_widget, "desktop_app") and source_widget.desktop_app:
app = source_widget.desktop_app
app_data_obj = {
"name": app.name,
"display_name": app.display_name,
"window_class": app.window_class,
"executable": app.executable,
"command_line": app.command_line,
}
self.pinned.insert(target_index, app_data_obj)
else:
self.pinned.insert(target_index, app_identifier)
self.config["pinned_apps"] = self.pinned
self.update_pinned_apps_file()
self.update_applications()
# If dragging from pinned to unpinned, remove from pinned apps
elif cross_section_drag_to_unpin:
# Remove from pinned apps
if source_index < len(self.pinned):
self.pinned.pop(source_index)
self.config["pinned_apps"] = self.pinned
self.update_pinned_apps_file()
self.update_applications()
else:
# Regular reordering within the same section
if source_index < separator_index and target_index < separator_index:
# Reorder within pinned section
item = self.pinned.pop(source_index)
self.pinned.insert(target_index, item)
self.config["pinned_apps"] = self.pinned
self.update_pinned_apps_file()
self.update_applications()
elif source_index > separator_index and target_index > separator_index:
# Reorder within unpinned section - just reorder the widgets
# since unpinned apps aren't stored in config
child_to_move = children.pop(source_index)
children.insert(target_index, child_to_move)
# Update the UI
for child in self.get_children():
self.remove(child)
for child in children:
self.add(child)
self.show_all()
def on_drag_end(self, widget, drag_context):
if not self._drag_in_progress:
return
# Process drag end...
self._drag_in_progress = False
# Allow dock to hide again
if self.dock_instance:
self.dock_instance.prevent_hiding(False)
+252
View File
@@ -0,0 +1,252 @@
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)
# Only show percentage label if not at 100%
if percentage < 100:
self.level.set_label(f"{int(percentage)}%")
self.level.set_visible(True)
else:
self.level.set_visible(False)
# 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
+138
View File
@@ -0,0 +1,138 @@
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 fabric.system_tray.widgets import SystemTray
from .battery import Battery
from .metrics import Metrics
from .controls import Controls
from .workspaces import workspace
from .indicators import Indicators
from .applications import Applications
class DockComponents(Box):
def __init__(self, orientation_val="h", dock_instance=None, **kwargs):
super().__init__(
name="dock-components", orientation=orientation_val, spacing=8, **kwargs
)
self.dock_instance = dock_instance
# Create applications with dock reference
self.applications = Applications(
orientation_val=orientation_val, dock_instance=dock_instance
)
# Initialize component visibility from data
self.component_visibility = data.DOCK_COMPONENTS_VISIBILITY
# Create components
self.workspaces = workspace # Use the workspace instance directly
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()
self.systray = SystemTray(
icon_size=18,
spacing=4,
name="tray",
orientation="h" if not data.VERTICAL else "v",
)
# Create list of components to add
self.themed_children = [
self.workspaces,
self.metrics,
self.controls,
self.applications,
self.indicators,
self.battery,
self.date_time,
self.systray,
]
# Add each child individually
for child in self.themed_children:
self.add(child)
# Apply initial visibility
self.apply_component_props()
# Apply invert style for specific themes and positions
should_invert = (
data.DOCK_THEME in ["Dense", "Edge"] or
(data.DOCK_THEME == "Pills" and data.DOCK_POSITION in ["Left", "Right"])
)
if should_invert:
for child in self.themed_children:
if hasattr(child, "add_style_class"):
child.add_style_class("invert")
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,
"systray": self.systray,
"applications": self.applications,
}
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,
"systray": self.systray,
"applications": self.applications,
}
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()
+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
+170
View File
@@ -0,0 +1,170 @@
import config.data as data
from gi.repository import Gdk, GdkPixbuf, GLib, Gray, Gtk
from fabric.widgets.box import Box
import logging
import gi
gi.require_version("Gray", "0.1")
logger = logging.getLogger(__name__)
class SystemTray(Box):
def __init__(
self, pixel_size: int = 20, refresh_interval: int = 1, **kwargs
) -> None:
orientation = (
Gtk.Orientation.HORIZONTAL
if not data.VERTICAL
else Gtk.Orientation.VERTICAL
)
super().__init__(name="tray", orientation=orientation, spacing=8, **kwargs)
self.enabled = True
super().set_visible(False)
self.pixel_size = pixel_size
self.refresh_interval = refresh_interval
self.buttons_by_id = {}
self.items_by_id = {}
self.watcher = Gray.Watcher()
self.watcher.connect("item-added", self.on_watcher_item_added)
GLib.timeout_add_seconds(self.refresh_interval, self._refresh_all_items)
def set_visible(self, visible: bool):
self.enabled = visible
self._update_visibility()
def _update_visibility(self):
has = len(self.get_children()) > 0
super().set_visible(self.enabled and has)
def _get_item_pixbuf(self, item: Gray.Item) -> GdkPixbuf.Pixbuf:
try:
pm = Gray.get_pixmap_for_pixmaps(item.get_icon_pixmaps(), self.pixel_size)
if pm:
return pm.as_pixbuf(self.pixel_size, GdkPixbuf.InterpType.HYPER)
name = item.get_icon_name()
theme = Gtk.IconTheme.new()
path = item.get_icon_theme_path()
if path:
theme.prepend_search_path(path)
return theme.load_icon(
name, self.pixel_size, Gtk.IconLookupFlags.FORCE_SIZE
)
except GLib.Error as e:
logger.debug(f"Icon load error {e}")
return Gtk.IconTheme.get_default().load_icon(
"image-missing", self.pixel_size, Gtk.IconLookupFlags.FORCE_SIZE
)
def _refresh_item_ui(self, identifier: str, item: Gray.Item, button: Gtk.Button):
pixbuf = self._get_item_pixbuf(item)
img = button.get_image()
if isinstance(img, Gtk.Image):
img.set_from_pixbuf(pixbuf)
else:
new = Gtk.Image.new_from_pixbuf(pixbuf)
button.set_image(new)
new.show()
tip = None
if hasattr(item, "get_tooltip_text"):
tip = item.get_tooltip_text()
elif hasattr(item, "get_title"):
tip = item.get_title()
if tip:
button.set_tooltip_text(tip)
else:
button.set_has_tooltip(False)
def _refresh_all_items(self) -> bool:
for ident, item in self.items_by_id.items():
btn = self.buttons_by_id.get(ident)
if btn:
self._refresh_item_ui(ident, item, btn)
return True
def on_watcher_item_added(self, _, identifier: str):
item = self.watcher.get_item_for_identifier(identifier)
if not item:
return
if identifier in self.buttons_by_id:
self.buttons_by_id[identifier].destroy()
del self.buttons_by_id[identifier]
del self.items_by_id[identifier]
btn = self.do_bake_item_button(item)
self.buttons_by_id[identifier] = btn
self.items_by_id[identifier] = item
item.connect(
"notify::icon-pixmaps",
lambda itm, pspec: self._refresh_item_ui(identifier, itm, btn),
)
item.connect(
"notify::icon-name",
lambda itm, pspec: self._refresh_item_ui(identifier, itm, btn),
)
try:
item.connect(
"updated", lambda itm: self._refresh_item_ui(identifier, itm, btn)
)
except TypeError:
pass
item.connect(
"removed", lambda itm: self.on_item_instance_removed(identifier, itm)
)
self.add(btn)
btn.show_all()
self._update_visibility()
def do_bake_item_button(self, item: Gray.Item) -> Gtk.Button:
btn = Gtk.Button()
btn.connect("button-press-event", lambda b, e: self.on_button_click(b, item, e))
img = Gtk.Image.new_from_pixbuf(self._get_item_pixbuf(item))
btn.set_image(img)
tip = (
item.get_tooltip_text()
if hasattr(item, "get_tooltip_text")
else getattr(item, "get_title", lambda: None)()
)
if tip:
btn.set_tooltip_text(tip)
return btn
def on_item_instance_removed(self, identifier: str, removed_item: Gray.Item):
if self.items_by_id.get(identifier) is removed_item:
btn = self.buttons_by_id.pop(identifier, None)
self.items_by_id.pop(identifier, None)
if btn:
btn.destroy()
self._update_visibility()
def on_button_click(
self, button: Gtk.Button, item: Gray.Item, event: Gdk.EventButton
):
if event.button == Gdk.BUTTON_PRIMARY:
try:
item.activate(int(event.x_root), int(event.y_root))
except Exception as e:
logger.error(f"Activate error: {e}")
elif event.button == Gdk.BUTTON_SECONDARY:
menu = getattr(item, "get_menu", lambda: None)()
if isinstance(menu, Gtk.Menu):
menu.popup_at_widget(
button, Gdk.Gravity.SOUTH_WEST, Gdk.Gravity.NORTH_WEST, event
)
else:
cm = getattr(item, "context_menu", None)
if cm:
try:
cm(int(event.x_root), int(event.y_root))
except Exception as e:
logger.error(f"ContextMenu error: {e}")
+148
View File
@@ -0,0 +1,148 @@
import json
import subprocess
from fabric.hyprland.service import Hyprland
from fabric.hyprland.widgets import Workspaces, WorkspaceButton
from fabric.widgets.button import Button
import config.data as data
from fabric.widgets.box import Box
CHINESE_NUMERALS = ["一", "二", "三", "四", "五", "六", "七", "八", "九", "〇"]
class Workspace(Button):
def __init__(self):
# Determine if invert style should be applied
should_invert = (
data.DOCK_THEME in ["Dense", "Edge"] or
(data.DOCK_THEME == "Pills" and data.DOCK_POSITION in ["Left", "Right"])
)
super().__init__(
name="workspace-single",
can_focus=True,
receives_default=True,
relief="none",
style_classes="invert" if should_invert else None,
)
self.connection = Hyprland()
self.activeWorkspace = str(self._get_active_workspace())
self.set_label(str(self.activeWorkspace))
self.connect("clicked", self._on_click)
self.connection.connect("event::workspace", self._on_workspace_changed)
self.set_sensitive(True)
self.show_all()
def _get_active_workspace(self):
active = self.connection.send_command("j/activeworkspace").reply
return json.loads(active.decode("utf-8"))["name"]
def _get_active_workspaces(self):
"""Get list of workspaces that have windows"""
clients_data = self.connection.send_command("j/clients").reply
clients = json.loads(clients_data.decode("utf-8"))
active_workspaces = set()
for client in clients:
if "workspace" in client and "name" in client["workspace"]:
workspace_name = str(client["workspace"]["name"])
active_workspaces.add(workspace_name)
return sorted(
list(active_workspaces), key=lambda x: int(x) if x.isdigit() else x
)
def _on_click(self, _widget):
"""Handle button click to cycle through workspaces"""
self._cycle_workspace()
def _cycle_workspace(self):
"""Core logic to cycle through workspaces"""
active_workspaces = self._get_active_workspaces()
if not active_workspaces:
return
if str(self.activeWorkspace) not in active_workspaces:
active_workspaces.append(str(self.activeWorkspace))
active_workspaces.sort(key=lambda x: int(x) if x.isdigit() else x)
current_index = active_workspaces.index(self.activeWorkspace)
next_index = (current_index + 1) % len(active_workspaces)
next_workspace = active_workspaces[next_index]
cmd = ["hyprctl", "dispatch", "workspace", str(next_workspace)]
subprocess.run(cmd, capture_output=True, text=True, timeout=2)
self.activeWorkspace = str(next_workspace)
self.set_label(str(self.activeWorkspace))
def _on_workspace_changed(self, _obj, signal):
"""Handle workspace change events from Hyprland"""
workspace_data = json.loads(signal.data[0])
self.activeWorkspace = str(workspace_data)
self.set_label(str(self.activeWorkspace))
# Create workspace components using Workspaces widget
workspaces = Workspaces(
name="workspaces",
invert_scroll=True,
empty_scroll=True,
v_align="fill",
orientation="h" if not data.VERTICAL else "v",
spacing=8,
buttons=[
WorkspaceButton(
h_expand=False,
v_expand=False,
h_align="center",
v_align="center",
id=i,
label=None,
style_classes=["vertical"] if data.VERTICAL else None,
)
for i in range(1, 11)
],
)
workspaces_num = Workspaces(
name="workspaces-num",
invert_scroll=True,
empty_scroll=True,
v_align="fill",
orientation="h" if not data.VERTICAL else "v",
spacing=0 if not data.WORKSPACE_USE_CHINESE_NUMERAL else 4,
buttons=[
WorkspaceButton(
h_expand=False,
v_expand=False,
h_align="center",
v_align="center",
id=i,
label=CHINESE_NUMERALS[i - 1]
if data.WORKSPACE_USE_CHINESE_NUMERAL and 1 <= i <= len(CHINESE_NUMERALS)
else str(i),
)
for i in range(1, 11)
],
)
# Create the workspace button instance
workspaceactive = Workspace()
# Export the workspace widget based on configuration
workspace = Box(
name="workspaces-container",
children=[
workspaces
if data.WORKSPACE_DOTS
else workspaces_num
if data.WORKSPACE_NUMS
else workspaceactive
],
)