chore(initial):

This commit is contained in:
sankalp committed 2025-05-11 12:17:00 +05:45
1 parent b0b8c19562
commit 7f956a59b7
42 files changed
+21627

No files matched your search

+128
View File
@@ -0,0 +1,128 @@
from fabric.widgets.box import Box
from fabric.widgets.eventbox import EventBox
from fabric.widgets.datetime import DateTime
from fabric.widgets.button import Button
from fabric.widgets.centerbox import CenterBox
from fabric.system_tray.widgets import SystemTray
from fabric.widgets.wayland import WaylandWindow as Window
from fabric.hyprland.widgets import ActiveWindow
from fabric.utils import truncate, FormattedString
from modules.bar.components.workspaces import workspace
from modules.bar.components.language import LanguageWidget
from modules.bar.components.auto_hide import AutoHideBarController
from modules.corners import MyCorner
from modules.bar.components.power_menu import Power
from modules.bar.components.system_indicators import SystemIndicators
from modules.bar.components.metrics import Metrics
class Bar(Window):
"""Main bar widget for the desktop environment."""
def __init__(self):
super().__init__(
layer="top",
anchor="top",
visible=True,
)
self.active_window = ActiveWindow(
name="active-window",
h_expand=True,
h_align="fill",
formatter=FormattedString(
f"{{'Desktop' if not win_class or win_class == 'unknown' else win_class}}",
truncate=truncate,
),
)
self.workspaces = Button(
name="workspaces",
child=workspace,
)
self.language = LanguageWidget()
self.date_time = DateTime(name="date-time", formatters=["%-I:%M:%p 󰧞 %a %d %b"])
self.tray = SystemTray(name="tray", spacing=4, icon_size=16)
self.power_menu = Power()
self.system_indicators = SystemIndicators()
self.metrics = Metrics()
# Main bar content
self.bar = CenterBox(
name="bar",
start_children=Box(spacing=4, orientation="h", children=self.workspaces),
center_children=Box(
spacing=4, orientation="h", children=self.active_window
),
end_children=Box(
spacing=4,
orientation="h",
children=[
self.date_time,
self.metrics,
self.system_indicators,
self.tray,
self.power_menu,
],
),
)
# Event handling and corners
self.eventbox = EventBox()
self.eventbox.add(self.bar)
self.corner_left = Box(
name="bar-corner-left",
orientation="v",
h_align="start",
children=[
MyCorner("top-right"),
Box(),
],
)
self.corner_left.set_margin_start(56)
self.corner_right = Box(
name="bar-corner-right",
orientation="v",
h_align="end",
children=[
MyCorner("top-left"),
Box(),
],
)
self.corner_right.set_margin_end(56)
self.main_box = Box(
name="bar-container",
orientation="h",
h_expand=True,
children=[
self.corner_left,
self.eventbox,
self.corner_right,
],
)
# Add hover activator
self.hover_activator = EventBox(name="hover-activator", v_align="start")
self.hover_activator.set_size_request(-1, 1)
# Root container
self.root_box = Box(
orientation="v", children=[self.main_box, self.hover_activator]
)
self.add(self.root_box)
# Setup auto-hide controller
self.auto_hide = AutoHideBarController(self)
# Connect event handlers for hover
self.eventbox.connect("enter-notify-event", self.auto_hide.on_bar_enter)
self.eventbox.connect("leave-notify-event", self.auto_hide.on_bar_leave)
self.hover_activator.connect(
"enter-notify-event", self.auto_hide.on_hover_enter
)
self.show_all()
+1
View File
@@ -0,0 +1 @@
# Bar components package
+202
View File
@@ -0,0 +1,202 @@
import json
from gi.repository import GLib, Gdk
from fabric.hyprland.widgets import get_hyprland_connection
from utils.occlusion import check_occlusion
class AutoHideBarController:
"""Controller for bar auto-hide behavior."""
# Constants for configuration
OCCLUSION_CHECK_INTERVAL = 250 # ms
HIDE_DELAY = 300 # ms
def __init__(self, bar):
self.bar = bar
self.is_hidden = False
self.hide_id = None
self.is_hovered = False
self.conn = get_hyprland_connection()
self.last_occlusion_state = False
self._setup_event_handlers()
self._start_occlusion_checking()
def _setup_event_handlers(self):
"""Initialize event handlers for window management"""
# Initial state check
if self.conn.ready:
self.check_hide()
else:
self.conn.connect("event::ready", lambda *_: self.check_hide())
# Monitor window events
window_events = ("activewindow", "openwindow", "closewindow", "changefloatingmode")
for event in window_events:
self.conn.connect(f"event::{event}", lambda *_: self.check_hide())
self.conn.connect("event::workspace", lambda *_: GLib.timeout_add(50, self.check_hide))
def _start_occlusion_checking(self):
"""Start periodic occlusion check"""
GLib.timeout_add(self.OCCLUSION_CHECK_INTERVAL, self.check_occlusion_state)
def _get_bar_height(self):
"""Get the height of the bar with fallback"""
try:
height = self.bar.get_allocated_height()
return height if height > 0 else 30
except Exception:
return 30 # Default fallback
def _get_occlusion_region(self):
"""Get the region to check for occlusion"""
return ("top", self._get_bar_height())
def check_occlusion_state(self):
"""Periodic occlusion check"""
# Skip occlusion check if hovered
if self.is_hovered:
self.bar.main_box.remove_style_class("occluded")
return True
# Check occlusion using the top region
is_occluded = check_occlusion(self._get_occlusion_region())
# Only apply changes if state has changed to avoid unnecessary CSS operations
if is_occluded != self.last_occlusion_state:
self.last_occlusion_state = is_occluded
if is_occluded:
self.bar.main_box.add_style_class("occluded")
else:
self.bar.main_box.remove_style_class("occluded")
if not self.is_hovered and self.is_hidden:
# Debounce showing the bar to prevent flickering
GLib.timeout_add(100, lambda: self.toggle_bar(not self._should_stay_hidden()))
return True
def _should_stay_hidden(self):
"""Determine if the bar should stay hidden based on window state"""
clients = self.get_clients()
current_ws = self.get_workspace()
ws_clients = [w for w in clients if w["workspace"]["id"] == current_ws]
# If we have normal windows and are not being hovered, stay hidden
return self._has_normal_windows(ws_clients) and not self.is_hovered
def on_bar_enter(self, widget, event):
"""Handle hover over bar content."""
self.is_hovered = True
self.bar.main_box.remove_style_class("occluded")
# Cancel any pending hide operations
self._cancel_hide_timer()
self.toggle_bar(show=True)
return True # Stop event propagation
def on_bar_leave(self, widget, event):
"""Handle leave from bar content."""
# Only trigger if mouse actually left the entire bar area
if event.detail == Gdk.NotifyType.INFERIOR:
return False # Ignore child-to-child mouse movements
self.is_hovered = False
# Schedule hide if needed
if self._should_stay_hidden() or check_occlusion(self._get_occlusion_region()):
self.delay_hide()
# Update occlusion state
if check_occlusion(self._get_occlusion_region()):
self.last_occlusion_state = True
self.bar.main_box.add_style_class("occluded")
return True # Stop event propagation
def on_hover_enter(self, *args):
"""Handle hover over top activation area."""
self.bar.main_box.remove_style_class("occluded")
self.toggle_bar(show=True)
return False
def _cancel_hide_timer(self):
"""Cancel any pending hide timer"""
if self.hide_id:
GLib.source_remove(self.hide_id)
self.hide_id = None
def toggle_bar(self, show):
"""Show or hide the bar immediately."""
if show:
if self.is_hidden:
self.is_hidden = False
self.bar.main_box.remove_style_class("hide-bar")
self.bar.main_box.add_style_class("show-bar")
self._cancel_hide_timer()
else:
if not self.is_hidden:
self.is_hidden = True
self.bar.main_box.remove_style_class("show-bar")
self.bar.main_box.add_style_class("hide-bar")
def delay_hide(self):
"""Schedule hiding after short delay."""
self._cancel_hide_timer()
self.hide_id = GLib.timeout_add(self.HIDE_DELAY, self.hide_bar)
def hide_bar(self):
"""Finalize hiding procedure."""
if not self.is_hovered:
self.toggle_bar(show=False)
self.hide_id = None
return False
def get_clients(self):
"""Get current client list."""
try:
return json.loads(self.conn.send_command("j/clients").reply.decode())
except (json.JSONDecodeError, AttributeError):
return []
def get_workspace(self):
"""Get current workspace ID."""
try:
return json.loads(
self.conn.send_command("j/activeworkspace").reply.decode()
).get("id", 0)
except (json.JSONDecodeError, AttributeError):
return 0
def _has_normal_windows(self, clients):
"""Check if any non-floating, non-fullscreen windows exist"""
return any(not w.get("floating") and not w.get("fullscreen") for w in clients)
def check_hide(self, *args):
"""Determine if bar should auto-hide based on window state."""
# Get window data
clients = self.get_clients()
current_ws = self.get_workspace()
ws_clients = [w for w in clients if w["workspace"]["id"] == current_ws]
# Check for occlusion
is_occluded = check_occlusion(self._get_occlusion_region())
self.last_occlusion_state = is_occluded
# Determine visibility state
if not ws_clients:
# No windows in current workspace, show the bar
self.bar.main_box.remove_style_class("occluded")
self.toggle_bar(show=True)
elif self._has_normal_windows(ws_clients):
# Normal windows exist, handle occlusion and hide if not hovered
if is_occluded:
self.bar.main_box.add_style_class("occluded")
if not self.is_hovered:
self.toggle_bar(show=False)
else:
# Only floating/fullscreen windows, show the bar
self.bar.main_box.remove_style_class("occluded")
self.toggle_bar(show=True)
return False
+24
View File
@@ -0,0 +1,24 @@
from fabric.widgets.button import Button
from fabric.widgets.label import Label
from fabric.hyprland.widgets import Language, HyprlandEvent, get_hyprland_connection
class LanguageWidget(Button):
"""Widget that displays the current keyboard layout."""
def __init__(self, **kwargs):
self.lang_label = Label(name="lang-label")
super().__init__(
name="language", h_align="center", v_align="center",
child=self.lang_label, **kwargs
)
self.conn = get_hyprland_connection()
self.update_language()
self.conn.connect("event::activelayout", self.update_language)
def update_language(self, _=None, event: HyprlandEvent = None):
"""Update the language indicator based on Hyprland events."""
lang = event.data[1] if event else Language().get_label()
self.set_tooltip_text(lang)
self.lang_label.set_label(lang[:2].lower())
+327
View File
@@ -0,0 +1,327 @@
import subprocess
import json
import psutil
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.revealer import Revealer
from fabric.core.fabricator import Fabricator
from fabric.widgets.button import Button
import utils.icons as icons
import config.data as data
class MetricsProvider:
def __init__(self):
self.cpu = 0.0
self.mem = 0.0
self.swap = 0.0
self.bat_percent = 0.0
self.bat_charging = None
GLib.timeout_add_seconds(1, self._update)
def _update(self):
# Get non-blocking usage percentages (interval=0)
# The first call may return 0, but subsequent calls will provide consistent values.
self.cpu = psutil.cpu_percent(interval=0)
mem = psutil.virtual_memory()
self.mem = mem.percent
swap = psutil.swap_memory()
self.swap = swap.percent
battery = psutil.sensors_battery()
if battery is None:
self.bat_percent = 0.0
self.bat_charging = None
else:
self.bat_percent = battery.percent
self.bat_charging = battery.power_plugged
return True
def get_metrics(self):
return (self.cpu, self.mem, self.swap)
def get_battery(self):
return (self.bat_percent, self.bat_charging)
shared_provider = MetricsProvider()
class SingularMetric:
def __init__(self, id, name, icon):
self.is_vertical_layout = data.VERTICAL
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=26,
line_width=2,
start_angle=150,
end_angle=390,
style_classes=id,
child=self.icon,
)
self.circle_button = Button(
name=f"metrics-{id}-button",
child=self.circle,
style="padding: 0; margin: 0; background: none;",
)
self.circle_button.add_events(
Gdk.EventMask.ENTER_NOTIFY_MASK | Gdk.EventMask.LEAVE_NOTIFY_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("clicked", self.on_circle_clicked)
self.level = Label(
name="metrics-level",
style_classes=id,
label="0%",
style="margin: 4px 0px;" if self.is_vertical_layout else "margin: 0 4px;",
)
self.revealer = Revealer(
name=f"metrics-{id}-revealer",
transition_duration=250,
transition_type="slide-up" if self.is_vertical_layout else "slide-left",
child=self.level,
child_revealed=False,
)
self.box = Box(
name=f"metrics-{id}-box",
orientation="v" if self.is_vertical_layout else "h",
spacing=0,
children=[self.circle_button, self.revealer],
)
# Connect events for cursor changes to the box as well
self.box.connect("enter-notify-event", self.on_button_enter)
self.box.connect("leave-notify-event", self.on_button_leave)
# Set initial tooltip - moved after box is created
self.update_tooltip("0%")
def update_tooltip(self, value_text):
tooltip = f"{self.icon_markup} {self.name_markup}: {value_text}"
# self.box.set_tooltip_markup(tooltip)
self.circle_button.set_tooltip_markup(tooltip)
self.circle.set_tooltip_markup(tooltip)
self.icon.set_tooltip_markup(tooltip)
def markup(self):
"""Get formatted text for tooltip display"""
return f"{self.icon_markup} {self.name_markup}: {self.level.get_label()}"
def on_button_enter(self, widget, event):
window = widget.get_window()
if window:
window.set_cursor(Gdk.Cursor(Gdk.CursorType.HAND2))
def on_button_leave(self, widget, event):
window = widget.get_window()
if window:
window.set_cursor(None)
def on_circle_clicked(self, button):
# Toggle the revealed state of the label
current_state = self.revealer.get_child_revealed()
self.revealer.set_reveal_child(not current_state)
class Metrics(Box):
def __init__(self, **kwargs):
# Determine if we should use vertical layout for components
self.is_vertical_layout = data.VERTICAL
super().__init__(
name="metrics",
spacing=0,
orientation="h" if not self.is_vertical_layout else "v",
visible=True,
all_visible=True,
style="padding:4px 0px" if self.is_vertical_layout else "padding: 0px 4px;",
**kwargs,
)
self.batt_fabricator = Fabricator(
poll_from=lambda v: shared_provider.get_battery(),
on_changed=lambda f, v: self.update_battery,
interval=1000,
stream=False,
default_value=0,
)
self.batt_fabricator.changed.connect(self.update_battery)
GLib.idle_add(self.update_battery, None, shared_provider.get_battery())
self.cpu = (
SingularMetric("cpu", "CPU", icons.cpu)
)
self.ram = (
SingularMetric("ram", "RAM", icons.memory)
)
self.swap = (
SingularMetric("swap", "SWAP", icons.swap)
)
self.bat = (
SingularMetric("bat", "BATTERY", icons.battery)
)
# Add only enabled metrics
if self.ram:
self.add(self.ram.box)
self.add(Box(name="metrics-sep"))
if self.cpu:
self.add(self.cpu.box)
if self.swap:
self.add(Box(name="metrics-sep"))
self.add(self.swap.box)
if self.bat:
self.add(Box(name="metrics-sep"))
self.add(self.bat.box)
# Connect events directly to the button
self.connect("enter-notify-event", self.on_mouse_enter)
self.connect("leave-notify-event", self.on_mouse_leave)
GLib.timeout_add_seconds(1, self.update_metrics)
self.hide_timer = None
self.hover_counter = 0
def _format_percentage(self, value: int) -> str:
return f"{value}%"
def on_mouse_enter(self, widget, event):
window = widget.get_window()
if window:
window.set_cursor(Gdk.Cursor(Gdk.CursorType.HAND2))
if not data.VERTICAL:
self.hover_counter += 1
if self.hide_timer is not None:
GLib.source_remove(self.hide_timer)
self.hide_timer = None
if self.cpu:
self.cpu.revealer.set_reveal_child(True)
if self.ram:
self.ram.revealer.set_reveal_child(True)
if self.swap:
self.swap.revealer.set_reveal_child(True)
if self.bat:
self.bat.revealer.set_reveal_child(True)
return False
def on_mouse_leave(self, widget, event):
window = widget.get_window()
if window:
window.set_cursor(None)
if not data.VERTICAL:
if self.hover_counter > 0:
self.hover_counter -= 1
if self.hover_counter == 0:
if self.hide_timer is not None:
GLib.source_remove(self.hide_timer)
self.hide_timer = GLib.timeout_add(500, self.hide_revealer)
return False
def hide_revealer(self):
if not data.VERTICAL:
if self.cpu:
self.cpu.revealer.set_reveal_child(False)
if self.ram:
self.ram.revealer.set_reveal_child(False)
self.hide_timer = None
return False
def update_metrics(self):
cpu, mem, swap = shared_provider.get_metrics()
if self.cpu:
self.cpu.circle.set_value(cpu / 100.0)
value_text = self._format_percentage(int(cpu))
self.cpu.level.set_label(value_text)
self.cpu.update_tooltip(value_text)
if self.ram:
self.ram.circle.set_value(mem / 100.0)
value_text = self._format_percentage(int(mem))
self.ram.level.set_label(value_text)
self.ram.update_tooltip(value_text)
if self.swap:
self.swap.circle.set_value(swap / 100.0)
value_text = self._format_percentage(int(swap))
self.swap.level.set_label(value_text)
self.swap.update_tooltip(value_text)
if self.bat:
bat_percent, _ = shared_provider.get_battery()
self.bat.circle.set_value(bat_percent / 100.0)
value_text = self._format_percentage(int(bat_percent))
self.bat.level.set_label(value_text)
self.bat.update_tooltip(value_text)
# Tooltip: only show enabled metrics
tooltip_metrics = []
if self.swap: tooltip_metrics.append(self.swap)
if self.bat: tooltip_metrics.append(self.bat)
if self.ram: tooltip_metrics.append(self.ram)
if self.cpu: tooltip_metrics.append(self.cpu)
return True
def update_battery(self, sender, battery_data):
if self.bat is None:
return
value, charging = battery_data
if value == 0:
if self.bat:
self.bat.box.set_visible(False)
else:
if self.bat:
self.bat.box.set_visible(True)
self.bat.circle.set_value(value / 100)
percentage = int(value)
self.bat.level.set_label(self._format_percentage(percentage))
# Apply alert styling if battery is low AND not charging
if percentage <= 15 and charging == False:
self.bat.icon.add_style_class("alert")
self.bat.circle.add_style_class("alert")
else:
self.bat.icon.remove_style_class("alert")
self.bat.circle.remove_style_class("alert")
# Choose the icon based on charging state first, then battery level
if percentage == 100:
self.bat.icon.set_markup(icons.battery)
charging_status = f"{icons.bat_full} Fully Charged"
elif charging == True:
self.bat.icon.set_markup(icons.charging)
charging_status = f"{icons.bat_charging} Charging"
elif percentage <= 15 and charging == False:
self.bat.icon.set_markup(icons.alert)
charging_status = f"{icons.bat_low} Low Battery"
elif charging == False:
self.bat.icon.set_markup(icons.discharging)
charging_status = f"{icons.bat_discharging} Discharging"
else:
self.bat.icon.set_markup(icons.battery)
charging_status = "Battery"
# Update individual tooltip for battery
self.bat.update_tooltip(f"{charging_status}: {self._format_percentage(percentage)}")
# Update the combined tooltip
self.update_metrics()
+180
View File
@@ -0,0 +1,180 @@
from fabric.widgets.button import Button
from fabric.widgets.box import Box
from fabric.widgets.label import Label
from fabric.widgets.stack import Stack
from fabric.widgets.revealer import Revealer
from fabric.utils import exec_shell_command_async
from gi.repository import GLib, Gdk, Gtk
import utils.icons as icons
class Power(Button):
TRANSITION_TYPE = "slide-up-down"
TRANSITION_DURATION = 300 # milliseconds
LABEL_TRANSITION_TYPE = "slide-left"
LONG_PRESS_DURATION = 500 # milliseconds
SCROLL_COOLDOWN = 100 # milliseconds
def __init__(self):
self.current_action = "shutdown"
self.is_locked = True
self.long_press_timeout = None
self.scroll_timeout = None
self.just_unlocked = False
# Initialize stacks for icons and labels
for name in ["icon_stack", "label_stack"]:
setattr(
self,
name,
Stack(
transition_type=self.TRANSITION_TYPE,
transition_duration=self.TRANSITION_DURATION,
),
)
# Define power actions
self.power_actions = {
"shutdown": {
"icon": icons.shutdown,
"label": "Shutdown",
"command": "systemctl poweroff"
},
"reboot": {
"icon": icons.reboot,
"label": "Reboot",
"command": "systemctl reboot"
},
"suspend": {
"icon": icons.suspend,
"label": "Suspend",
"command": "systemctl suspend"
},
"logout": {
"icon": icons.logout,
"label": "Logout",
"command": "hyprctl dispatch exit"
},
"lock": {
"icon": icons.lock,
"label": "Lock",
"command": "hyprlock --immediate"
}
}
# Add all actions to stacks
for action, data in self.power_actions.items():
self.icon_stack.add_named(
Label(
markup=data["icon"],
style_classes=["power-icon"],
),
name=action,
)
self.label_stack.add_named(
Label(
label=data["label"],
style_classes=["power-label"]
),
name=action,
)
self.revealer = Revealer(
transition_type=self.LABEL_TRANSITION_TYPE,
transition_duration=self.TRANSITION_DURATION,
child=self.label_stack,
)
super().__init__(
name="power-menu",
on_enter_notify_event=lambda *args: self.revealer.set_reveal_child(True),
on_leave_notify_event=self.on_leave,
on_pressed=self.on_pressed,
on_released=self.on_released,
on_scroll_event=self.on_scroll,
child=Box(
children=[
self.icon_stack,
self.revealer,
]
),
)
self.add_events(
Gdk.EventMask.SCROLL_MASK |
Gdk.EventMask.BUTTON_PRESS_MASK |
Gdk.EventMask.SMOOTH_SCROLL_MASK
)
self.add_style_class("locked")
# Add cursor change on hover
self.connect(
"state-flags-changed",
lambda btn, *_: (
btn.set_cursor("pointer")
if btn.get_state_flags() & Gtk.StateFlags.PRELIGHT # type: ignore
else btn.set_cursor("default"),
),
)
def on_pressed(self, widget, event=None):
if self.is_locked:
self.long_press_timeout = GLib.timeout_add(
self.LONG_PRESS_DURATION,
self.on_long_press
)
def on_released(self, widget, event=None):
timeout = self.long_press_timeout
self.long_press_timeout = None
if timeout:
GLib.source_remove(timeout)
if not self.is_locked and not self.just_unlocked:
exec_shell_command_async(self.power_actions[self.current_action]["command"])
self.is_locked = True
self.add_style_class("locked")
self.just_unlocked = False
def on_long_press(self):
if self.is_locked:
self.is_locked = False
self.just_unlocked = True
self.remove_style_class("locked")
self.long_press_timeout = None
return False
def _change_action(self, direction):
if self.is_locked:
return
actions = list(self.power_actions.keys())
current_index = actions.index(self.current_action)
new_index = (current_index + direction) % len(actions)
self.current_action = actions[new_index]
self.icon_stack.set_visible_child_name(self.current_action)
self.label_stack.set_visible_child_name(self.current_action)
def on_scroll(self, widget, event):
if self.is_locked:
return True
if event.direction == Gdk.ScrollDirection.SMOOTH:
if event.delta_y < -0.1:
self._change_action(-1)
elif event.delta_y > 0.1:
self._change_action(1)
elif event.direction == Gdk.ScrollDirection.UP:
self._change_action(-1)
elif event.direction == Gdk.ScrollDirection.DOWN:
self._change_action(1)
self.scroll_timeout = GLib.timeout_add(self.SCROLL_COOLDOWN, lambda: None)
return True
def on_leave(self, widget, event=None):
self.revealer.set_reveal_child(False)
if not self.is_locked:
self.is_locked = True
self.add_style_class("locked")
+214
View File
@@ -0,0 +1,214 @@
import subprocess
from fabric import Fabricator
from fabric.widgets.label import Label
from fabric.widgets.box import Box
from fabric.bluetooth import BluetoothClient
from services import network_client, audio
import utils.icons as icons
from fabric.hyprland.widgets import Language, HyprlandEvent, get_hyprland_connection
class SystemIndicators(Box):
def __init__(self, **kwargs):
super().__init__(
orientation="h", spacing=2, name="system-indicator", **kwargs
)
self.bluetooth_icon = Label(name="system-indicator-icon")
self.wifi_icon = Label(name="system-indicator-icon")
self.volume_icon_button = Label(name="system-indicator-icon")
self.language_icon = Label(name="lang-label")
self.idle_label = Label(name="system-indicator-icon")
self.night_label = Label(name="system-indicator-icon")
self.power_label = Label(name="system-indicator-icon")
for widget in [
self.language_icon,
self.bluetooth_icon,
self.wifi_icon,
self.volume_icon_button,
self.idle_label,
self.night_label,
self.power_label,
]:
self.add(widget)
self.audio_service = audio
self.bluetooth_client = BluetoothClient()
self.network_client = network_client
self.hyprland = get_hyprland_connection()
# Connect signals
self.bluetooth_client.connect("changed", self.update_bluetooth_status)
self.audio_service.connect("changed", self.update_volume_status)
self.network_client.connect("device_ready", self.update_network_status)
self.hyprland.connect("event::activelayout", self.update_language)
if self.network_client.wifi_device:
self.network_client.wifi_device.connect(
"changed", self.update_network_status
)
if self.network_client.ethernet_device:
self.network_client.ethernet_device.connect(
"changed", self.update_network_status
)
Fabricator(interval=1000, poll_from=self.update_all_statuses)
# Initial updates
self.update_bluetooth_status()
self.update_language()
self.update_idle_night_status()
def update_all_statuses(self, *_args):
self.update_idle_night_status()
self.update_power_profile()
self.update_network_status()
def update_language(self, _=None, event: HyprlandEvent = None):
"""Update the language indicator based on Hyprland events."""
lang = event.data[1] if event else Language().get_label()
self.language_icon.set_tooltip_text(lang)
self.language_icon.set_label(lang[:2].lower())
def update_idle_night_status(self, *_):
# Update idle status
has_idle = (
subprocess.run(["pgrep", "-x", "wlinhibit"], capture_output=True).returncode
== 0
)
self.idle_label.set_visible(has_idle)
self.idle_label.set_markup(icons.coffee if has_idle else "")
# Update night mode status
has_night = (
subprocess.run(
["pgrep", "-x", "hyprsunset"], capture_output=True
).returncode
== 0
)
self.night_label.set_visible(has_night)
self.night_label.set_markup(icons.night if has_night else "")
def update_volume_status(self, *_):
stream = self.audio_service.speaker
if not stream:
self.volume_icon_button.set_visible(False)
return
self.volume_icon_button.set_visible(True)
# Normalize volume to integer percentage (0-100)
volume_level = stream.volume
if volume_level > 1:
volume_level = min(int(volume_level), 100)
else:
volume_level = int(volume_level * 100)
is_muted = stream.muted
if is_muted:
icon = icons.vol_off
elif volume_level > 74:
icon = icons.vol_high
elif volume_level > 0:
icon = icons.vol_medium
else:
icon = icons.vol_mute
self.volume_icon_button.set_markup(icon)
tooltip_text = "Muted" if is_muted else f"Volume: {volume_level}%"
self.volume_icon_button.set_tooltip_text(tooltip_text)
def update_bluetooth_status(self, *_):
if self.bluetooth_client.enabled:
self.bluetooth_icon.set_visible(True)
# Get a list of connected devices
connected_devices = [
device.name
for device in self.bluetooth_client.devices
if device.connected
]
if connected_devices:
self.bluetooth_icon.set_markup(icons.bluetooth_connected)
self.bluetooth_icon.set_tooltip_text(f"{', '.join(connected_devices)}")
else:
self.bluetooth_icon.set_markup(icons.bluetooth)
self.bluetooth_icon.set_tooltip_text(
"Bluetooth is enabled, no devices connected"
)
else:
self.bluetooth_icon.set_visible(True) # Keep it visible
self.bluetooth_icon.set_markup(icons.bluetooth_off)
self.bluetooth_icon.set_tooltip_text("Bluetooth is disabled")
def update_network_status(self, *_):
primary_device = self.network_client.primary_device
self.wifi_icon.set_visible(True)
if primary_device == "wifi" and self.network_client.wifi_device:
wifi_device = self.network_client.wifi_device
self.wifi_icon.set_tooltip_text(
wifi_device.ssid if wifi_device.ssid else "No WiFi Connection"
)
if wifi_device.enabled:
strength = wifi_device.strength
if strength > 0:
if strength < 25:
icon_label = icons.wifi_0
elif strength < 50:
icon_label = icons.wifi_1
elif strength < 75:
icon_label = icons.wifi_2
else:
icon_label = icons.wifi_3
else:
icon_label = icons.wifi_off
else:
icon_label = icons.wifi_off
elif primary_device == "wired":
icon_label = icons.lan
self.wifi_icon.set_tooltip_text("Wired Connection")
else:
icon_label = icons.wifi_off # No network
self.wifi_icon.set_markup(icon_label)
def update_power_profile(self, *_):
try:
result = subprocess.run(
["powerprofilesctl", "get"], capture_output=True, text=True
)
current_profile = result.stdout.strip()
profile_icons = {
"performance": icons.power_performance,
"balanced": "",
"power-saver": icons.power_saving,
}
icon = profile_icons.get(current_profile)
tooltip = (
f"Power Profile: {current_profile.capitalize()}"
if current_profile
else "Unknown Profile"
)
self.power_label.set_visible(bool(icon)) # Only show if there's an icon
self.power_label.set_markup(icon)
self.power_label.set_tooltip_text(tooltip)
except Exception as e:
self.power_label.set_visible(False)
self.power_label.set_tooltip_text(f"Error fetching power profile: {e}")
+19
View File
@@ -0,0 +1,19 @@
import json
from fabric.hyprland.service import Hyprland
from fabric.widgets.label import Label
connection = Hyprland()
workspaceData = connection.send_command("j/activeworkspace").reply
activeWorkspace = json.loads(workspaceData.decode("utf-8"))["name"]
workspace = Label(label=f"{activeWorkspace}")
def on_workspace(obj, signal):
global activeWorkspace
activeWorkspace = json.loads(signal.data[0])
workspace.set_label(f"{activeWorkspace}")
connection.connect("event::workspace", on_workspace)
+62
View File
@@ -0,0 +1,62 @@
from fabric.widgets.box import Box
from fabric.widgets.wayland import WaylandWindow as Window
from fabric.widgets.shapes import Corner
class MyCorner(Box):
def __init__(self, corner):
super().__init__(
name="corner-container",
children=Corner(
name="corner",
orientation=corner,
size=20,
),
)
class Corners(Window):
def __init__(self):
super().__init__(
name="corners",
layer="top",
anchor="top bottom left right",
exclusivity="normal",
pass_through=True,
visible=False,
all_visible=False,
)
self.all_corners = Box(
name="all-corners",
orientation="v",
h_expand=True,
v_expand=True,
h_align="fill",
v_align="fill",
children=[
Box(
name="top-corners",
orientation="h",
h_align="fill",
children=[
MyCorner("top-left"),
Box(h_expand=True),
MyCorner("top-right"),
],
),
Box(v_expand=True),
Box(
name="bottom-corners",
orientation="h",
h_align="fill",
children=[
MyCorner("bottom-left"),
Box(h_expand=True),
MyCorner("bottom-right"),
],
),
],
)
self.add(self.all_corners)
self.show_all()
+948
View File
@@ -0,0 +1,948 @@
import json
from gi.repository import GLib, Gtk, Gdk
import cairo
from utils.icon_resolver import IconResolver
from utils.occlusion import check_occlusion
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.image import Image
from fabric.widgets.eventbox import EventBox
from fabric.widgets.wayland import WaylandWindow as Window
from fabric.hyprland.widgets import get_hyprland_connection
from fabric.utils import exec_shell_command, exec_shell_command_async, idle_add, remove_handler, get_relative_path
from fabric.utils.helpers import get_desktop_applications
from modules.corners import MyCorner # Add this import for corners
import config.data as data
import logging
OCCLUSION = 36 + data.DOCK_ICON_SIZE
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:
data = json.load(file)
# Migration: Convert old format (simple string array) to new format (array of objects with full app data)
if "pinned_apps" in data and data["pinned_apps"] and isinstance(data["pinned_apps"][0], str):
# Get all desktop apps for lookup during migration
all_apps = get_desktop_applications()
app_map = {app.name: app for app in all_apps if app.name}
# This is the old format - convert it with full app data
old_pinned = data["pinned_apps"]
data["pinned_apps"] = []
for app_id in old_pinned:
# Try to find the app by name
app = app_map.get(app_id)
if app:
# Create comprehensive app data
app_data = {
"name": app.name,
"display_name": app.display_name,
"window_class": app.window_class,
"executable": app.executable,
"command_line": app.command_line
}
data["pinned_apps"].append(app_data)
else:
# Fallback to just the name if app not found
data["pinned_apps"].append({"name": app_id})
except (FileNotFoundError, json.JSONDecodeError):
data = {"pinned_apps": []} # Default to empty pinned apps
return data
# Credit to Aylur for the createSurfaceFromWidget code
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 Dock(Window):
# Static registry to track all active dock instances
_instances = []
def __init__(self, **kwargs):
super().__init__(
name="dock-window",
layer="top",
anchor="bottom center",
margin="-8px 0 -8px 0",
exclusivity="none",
**kwargs,
)
# Add this instance to the registry
Dock._instances.append(self)
# Initialize configuration
self.config = read_config()
self.conn = get_hyprland_connection()
self.icon = IconResolver()
self.pinned = self.config.get("pinned_apps", [])
self.config_path = get_relative_path("../config/dock.json")
self.app_map = {} # Initialize the app map
self._all_apps = get_desktop_applications() # Get all apps for lookup
# Create app identifiers mapping
self.app_identifiers = self._build_app_identifiers_map()
self.is_hidden = False
self.hide_id = None
self._arranger_handler = None
self._drag_in_progress = False # Drag lock flag
self.is_hovered = False
self.always_occluded = data.DOCK_ALWAYS_OCCLUDED # Track the always occluded setting
# Set up UI containers
self.view = Box(name="viewport", orientation="h", spacing=4)
self.wrapper = Box(name="dock", orientation="v", children=[self.view])
# Main dock container with hover handling
self.dock_eventbox = EventBox()
self.dock_eventbox.add(self.wrapper)
self.dock_eventbox.connect("enter-notify-event", self._on_dock_enter)
self.dock_eventbox.connect("leave-notify-event", self._on_dock_leave)
# Create corners in the orientation-appropriate direction
self.corner_left = Box(
name="dock-corner-left",
orientation="v",
h_align="start",
children=[
Box(v_expand=True, v_align="fill"),
MyCorner("bottom-right"),
]
)
self.corner_right = Box(
name="dock-corner-right",
orientation="v",
h_align="end",
children=[
Box(v_expand=True, v_align="fill"),
MyCorner("bottom-left"),
]
)
# Create a horizontal box that contains the corners and the dock
self.dock_full = Box(
name="dock-full",
orientation="h",
h_expand=True,
h_align="fill",
children=[
self.corner_left,
self.dock_eventbox,
self.corner_right,
]
)
# Bottom/side hover activation area
self.hover_activator = EventBox()
self.hover_activator.set_size_request(-1, 1)
self.hover_activator.connect("enter-notify-event", self._on_hover_enter)
self.hover_activator.connect("leave-notify-event", self._on_hover_leave)
# Update main_box to include the dock_full and activator
self.main_box = Box(orientation="v",
children=[self.dock_full, self.hover_activator])
self.add(self.main_box)
# Drag-and-drop setup for the viewport
self.view.drag_source_set(
Gdk.ModifierType.BUTTON1_MASK,
[Gtk.TargetEntry.new("text/plain", Gtk.TargetFlags.SAME_APP, 0)],
Gdk.DragAction.MOVE
)
self.view.drag_dest_set(
Gtk.DestDefaults.ALL,
[Gtk.TargetEntry.new("text/plain", Gtk.TargetFlags.SAME_APP, 0)],
Gdk.DragAction.MOVE
)
self.view.connect("drag-data-get", self.on_drag_data_get)
self.view.connect("drag-data-received", self.on_drag_data_received)
self.view.connect("drag-begin", self.on_drag_begin)
self.view.connect("drag-end", self.on_drag_end)
# Initialization
if self.conn.ready:
self.update_dock()
GLib.timeout_add(500, self.check_hide)
else:
self.conn.connect("event::ready", self.update_dock)
self.conn.connect("event::ready", self.check_hide)
for ev in ("activewindow", "openwindow", "closewindow", "changefloatingmode"):
self.conn.connect(f"event::{ev}", self.update_dock)
self.conn.connect("event::workspace", self.check_hide)
GLib.timeout_add(250, self.check_occlusion_state)
# Monitor dock.json for changes
GLib.timeout_add_seconds(1, self.check_config_change)
# Check if dock should be visible based on configuration
if not data.DOCK_ENABLED:
self.set_visible(False)
# Apply always occluded state if enabled in settings
if self.always_occluded:
self.dock_full.add_style_class("occluded")
def _build_app_identifiers_map(self):
"""Build a mapping of app identifiers (class names, executables, names) to DesktopApp objects"""
identifiers = {}
for app in self._all_apps:
# Map by name (lowercase)
if app.name:
identifiers[app.name.lower()] = app
# Map by display name
if app.display_name:
identifiers[app.display_name.lower()] = app
# Map by window class if available
if app.window_class:
identifiers[app.window_class.lower()] = app
# Map by executable name if available
if app.executable:
exe_basename = app.executable.split('/')[-1].lower()
identifiers[exe_basename] = app
# Map by command line if available (without parameters)
if app.command_line:
cmd_base = app.command_line.split()[0].split('/')[-1].lower()
identifiers[cmd_base] = app
return identifiers
def _normalize_window_class(self, class_name):
"""Normalize window class by removing common suffixes and lowercase."""
if not class_name:
return ""
normalized = class_name.lower()
# Remove common suffixes
suffixes = [".bin", ".exe", ".so", "-bin", "-gtk"]
for suffix in suffixes:
if normalized.endswith(suffix):
normalized = normalized[:-len(suffix)]
return normalized
def _classes_match(self, class1, class2):
"""Check if two window class names match with stricter comparison."""
if not class1 or not class2:
return False
# Normalize both classes
norm1 = self._normalize_window_class(class1)
norm2 = self._normalize_window_class(class2)
# Direct match after normalization
if norm1 == norm2:
return True
# Don't do substring matching as it's too error-prone
# This avoids incorrectly matching flatpak apps and others
return False
def on_drag_begin(self, widget, drag_context):
"""Handle drag begin event by setting the drag lock flag."""
self._drag_in_progress = True
# Set custom drag icon using the widget surface
Gtk.drag_set_icon_surface(drag_context, createSurfaceFromWidget(widget))
def _on_hover_enter(self, *args):
"""Handle hover over bottom activation area"""
self.toggle_dock(show=True)
def _on_hover_leave(self, *args):
"""Handle leave from bottom activation area"""
self.delay_hide()
def _on_dock_enter(self, widget, event):
"""Handle hover over dock content"""
self.is_hovered = True
self.dock_full.remove_style_class("occluded")
# Cancel any pending hide operations
if self.hide_id:
GLib.source_remove(self.hide_id)
self.hide_id = None
self.toggle_dock(show=True)
return True # Important: Stop event propagation
def _on_dock_leave(self, widget, event):
"""Handle leave from dock content"""
# Only trigger if mouse actually left the entire dock area
if event.detail == Gdk.NotifyType.INFERIOR:
return False # Ignore child-to-child mouse movements
self.is_hovered = False
self.delay_hide()
# Immediate occlusion check on true leave using simplified format
occlusion_region = ("bottom", OCCLUSION)
# Add occlusion style if not dragging an icon or if always_occluded is enabled
if (self.always_occluded or (not self._drag_in_progress and (check_occlusion(occlusion_region) or not self.view.get_children()))):
self.dock_full.add_style_class("occluded")
return True
# Enhanced app lookup methods
def find_app(self, app_identifier):
"""Return the DesktopApp object by matching any app identifier.
If app_identifier is a dict, it will use all available keys for matching.
"""
if not app_identifier:
return None
# If we got a dict object with app data (new format)
if isinstance(app_identifier, dict):
# Try to find by all available identifiers in priority order
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
# Simple string identifier (backward compatibility)
return self.find_app_by_key(app_identifier)
def find_app_by_key(self, key_value):
"""Find app by a single identifier value"""
if not key_value:
return None
# Try direct lookup in our identifiers map
normalized_id = str(key_value).lower()
if normalized_id in self.app_identifiers:
return self.app_identifiers[normalized_id]
# Try fuzzy matching - find apps where the identifier is part of their names
for app in self._all_apps:
if app.name and normalized_id in app.name.lower():
return app
if app.display_name and normalized_id in app.display_name.lower():
return app
if app.window_class and normalized_id in app.window_class.lower():
return app
if app.executable and normalized_id in app.executable.lower():
return app
if app.command_line and normalized_id in app.command_line.lower():
return app
return None
# Update the dock's app map using DesktopApp objects from the system.
def update_app_map(self):
"""Updates the mapping of commands to DesktopApp objects."""
self._all_apps = get_desktop_applications() # Refresh all apps
self.app_map = {app.name: app for app in self._all_apps if app.name} # Map app names to DesktopApp objects
self.app_identifiers = self._build_app_identifiers_map() # Rebuild identifiers map
def create_button(self, app_identifier, instances):
"""Create dock application button"""
desktop_app = self.find_app(app_identifier) # Find app by identifier
icon_img = None
display_name = None
if desktop_app:
icon_img = desktop_app.get_icon_pixbuf(size=data.DOCK_ICON_SIZE)
display_name = desktop_app.display_name or desktop_app.name
# Extract identifier for fallback
id_value = app_identifier["name"] if isinstance(app_identifier, dict) else app_identifier
if not icon_img:
# Fallback to IconResolver with the app command
icon_img = self.icon.get_icon_pixbuf(id_value, data.DOCK_ICON_SIZE) # Use identifier for fallback
if not icon_img: # Double check after exec path try
# Fallback icon if no DesktopApp is found
icon_img = self.icon.get_icon_pixbuf("application-x-executable-symbolic", data.DOCK_ICON_SIZE)
# Final fallback
if not icon_img:
icon_img = self.icon.get_icon_pixbuf("image-missing", data.DOCK_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), # Pass desktop_app as well
tooltip_text=tooltip,
name="dock-app-button",
)
# Store app data with the button for future reference
button.app_identifier = app_identifier
button.desktop_app = desktop_app
button.instances = instances
if instances:
button.add_style_class("instance") # Style running apps
# Enable DnD for ALL apps
button.drag_source_set(
Gdk.ModifierType.BUTTON1_MASK,
[Gtk.TargetEntry.new("text/plain", Gtk.TargetFlags.SAME_APP, 0)],
Gdk.DragAction.MOVE
)
button.connect("drag-begin", self.on_drag_begin)
button.connect("drag-end", self.on_drag_end) # Connect drag-end to ALL buttons
# Make all buttons drop targets for better reordering
button.drag_dest_set(
Gtk.DestDefaults.ALL,
[Gtk.TargetEntry.new("text/plain", Gtk.TargetFlags.SAME_APP, 0)],
Gdk.DragAction.MOVE
)
button.connect("drag-data-get", self.on_drag_data_get)
button.connect("drag-data-received", self.on_drag_data_received)
button.connect("enter-notify-event", self._on_child_enter)
return button
# Enhanced app launching with multiple fallbacks
def handle_app(self, app_identifier, instances, desktop_app=None):
"""Handle application button clicks with improved fallbacks"""
if not instances:
# Try to launch the app with multiple fallback methods
if not desktop_app:
desktop_app = self.find_app(app_identifier)
if desktop_app:
# Try the standard launch method first
launch_success = desktop_app.launch()
# If that fails, try command line or executable
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:
# No desktop entry found, try direct execution
if isinstance(app_identifier, dict):
# Try command_line first, then executable, then name as last resort
if "command_line" in app_identifier and app_identifier["command_line"]:
exec_shell_command_async(f"nohup {app_identifier['command_line']}")
elif "executable" in app_identifier and app_identifier["executable"]:
exec_shell_command_async(f"nohup {app_identifier['executable']}")
elif "name" in app_identifier and app_identifier["name"]:
exec_shell_command_async(f"nohup {app_identifier['name']}")
elif isinstance(app_identifier, str):
# Try direct execution with the identifier as fallback
exec_shell_command_async(f"nohup {app_identifier}")
else:
# Handle window switching for running instances
focused = self.get_focused()
idx = next(
(i for i, inst in enumerate(instances) if inst["address"] == focused),
-1,
)
next_inst = instances[(idx + 1) % len(instances)]
exec_shell_command(
f"hyprctl dispatch focuswindow address:{next_inst['address']}"
)
def _on_child_enter(self, widget, event):
"""Maintain hover state when entering child widgets"""
self.is_hovered = True
if self.hide_id:
GLib.source_remove(self.hide_id)
self.hide_id = None
return False # Continue event propagation
def toggle_dock(self, show):
"""Show or hide the dock immediately"""
if show:
if self.is_hidden:
self.is_hidden = False
self.dock_full.add_style_class("show-dock")
self.dock_full.remove_style_class("hide-dock")
if self.hide_id:
GLib.source_remove(self.hide_id)
self.hide_id = None
else:
if not self.is_hidden:
self.is_hidden = True
self.dock_full.add_style_class("hide-dock")
self.dock_full.remove_style_class("show-dock")
def delay_hide(self):
"""Schedule hiding after short delay"""
if self.hide_id:
GLib.source_remove(self.hide_id)
self.hide_id = GLib.timeout_add(1000, self.hide_dock)
def hide_dock(self):
"""Finalize hiding procedure"""
self.toggle_dock(show=False)
self.hide_id = None
return False
def check_hide(self, *args):
"""Determine if dock should auto-hide"""
clients = self.get_clients()
current_ws = self.get_workspace()
ws_clients = [w for w in clients if w["workspace"]["id"] == current_ws]
if not ws_clients:
self.toggle_dock(show=True)
elif any(not w.get("floating") and not w.get("fullscreen") for w in ws_clients):
self.delay_hide()
else:
self.toggle_dock(show=True)
def update_dock(self, *args):
"""Refresh dock contents and clear drag lock."""
self.update_app_map() # Update app map before creating buttons
arranger_handler = getattr(self, "_arranger_handler", None)
if arranger_handler:
remove_handler(arranger_handler)
clients = self.get_clients()
# Create a mapping of window class to instances
running_windows = {}
for c in clients:
# Try multiple identification methods in order of reliability
window_id = None
# Try initialClass first (most reliable)
if class_name := c.get("initialClass", "").lower():
window_id = class_name
# Try class second (fallback)
elif class_name := c.get("class", "").lower():
window_id = class_name
# Use title as last resort if both class identifiers are missing
elif title := c.get("title", "").lower():
# Extract app name from title (common format: "App Name - Document")
possible_name = title.split(" - ")[0].strip()
if possible_name and len(potential_name) > 1: # Avoid single letter app names
window_id = possible_name
else:
window_id = title # Use full title if we can't extract a good name
# Use generic identifier as absolute fallback
if not window_id:
window_id = "unknown-app"
# Log window for debugging purposes
logging.debug(f"Window detected: {window_id} (from {c.get('initialClass', '')}/{c.get('class', '')}/{c.get('title', '')})")
# Add to running windows - store with both original and normalized keys
running_windows.setdefault(window_id, []).append(c)
# Also store with normalized key for more flexible matching
normalized_id = self._normalize_window_class(window_id)
if normalized_id != window_id:
running_windows.setdefault(normalized_id, []).extend(running_windows[window_id])
# Map pinned apps to their running instances
pinned_buttons = []
used_window_classes = set() # Track which window classes we've already assigned
for app_data in self.pinned:
app = self.find_app(app_data)
# Try to find running instances for this pinned app
instances = []
matched_class = None
# Extract all possible identifiers from app_data for matching
possible_identifiers = []
if isinstance(app_data, dict):
for key in ["window_class", "executable", "command_line", "name", "display_name"]:
if key in app_data and app_data[key]:
possible_identifiers.append(app_data[key].lower())
elif isinstance(app_data, str):
possible_identifiers.append(app_data.lower())
# Add identifiers from DesktopApp if available
if app:
if app.window_class:
possible_identifiers.append(app.window_class.lower())
if app.executable:
possible_identifiers.append(app.executable.split('/')[-1].lower())
if app.command_line:
cmd_parts = app.command_line.split()
if cmd_parts:
possible_identifiers.append(cmd_parts[0].split('/')[-1].lower())
if app.name:
possible_identifiers.append(app.name.lower())
if app.display_name:
possible_identifiers.append(app.display_name.lower())
# Remove duplicates
possible_identifiers = list(set(possible_identifiers))
# Try each identifier for matching
for identifier in possible_identifiers:
# Try direct match
if identifier in running_windows:
instances = running_windows[identifier]
matched_class = identifier
break
# Try normalized version
normalized = self._normalize_window_class(identifier)
if normalized in running_windows:
instances = running_windows[normalized]
matched_class = normalized
break
# Try substring matching for window classes (less reliable but helpful)
for window_class in running_windows:
# For substring matching, ensure the identifier is at least 3 chars
# to avoid too broad matches
if len(identifier) >= 3 and identifier in window_class:
instances = running_windows[window_class]
matched_class = window_class
logging.debug(f"Substring match: {identifier} in {window_class}")
break
if matched_class:
break
# Mark the matched class as used
if matched_class:
used_window_classes.add(matched_class)
# Also mark the normalized version as used
used_window_classes.add(self._normalize_window_class(matched_class))
logging.debug(f"Matched pinned app {app_data} to running instances via {matched_class}")
# Create button for this pinned app with any found instances
pinned_buttons.append(self.create_button(app_data, instances))
# For any remaining window classes that aren't assigned to pinned apps
open_buttons = []
for class_name, instances in running_windows.items():
if class_name not in used_window_classes:
# Enhanced app identification for running windows
app = None
# Try multiple methods to find the correct app
# 1. Direct lookup by class name
app = self.app_identifiers.get(class_name)
# 2. Try with normalized class name
if not app:
norm_class = self._normalize_window_class(class_name)
app = self.app_identifiers.get(norm_class)
# 3. Try with our more robust find_app method
if not app:
app = self.find_app_by_key(class_name)
# 4. Try using window title which often contains app name
if not app and instances and instances[0].get("title"):
title = instances[0].get("title", "")
# Extract potential app name from title (common format: "App Name - Document")
potential_name = title.split(" - ")[0].strip()
if len(potential_name) > 2: # Avoid very short names
app = self.find_app_by_key(potential_name)
# Create comprehensive app data if app was found
if app:
app_data = {
"name": app.name,
"display_name": app.display_name,
"window_class": app.window_class,
"executable": app.executable,
"command_line": app.command_line
}
identifier = app_data
else:
# Fallback to just class name
identifier = class_name
open_buttons.append(self.create_button(identifier, instances))
# Assemble dock layout
children = pinned_buttons
# Only add separator if both pinned and open buttons exist
if pinned_buttons and open_buttons:
children += [Box(orientation="v", v_expand=True, name="dock-separator")]
children += open_buttons
self.view.children = children
idle_add(self._update_size)
self._drag_in_progress = False # Clear the drag lock
def _update_size(self):
"""Update window size based on content"""
width, _ = self.view.get_preferred_width()
self.set_size_request(width, -1)
return False
def get_clients(self):
"""Get current client list"""
try:
return json.loads(self.conn.send_command("j/clients").reply.decode())
except json.JSONDecodeError:
return []
def get_focused(self):
"""Get focused window address"""
try:
return json.loads(
self.conn.send_command("j/activewindow").reply.decode()
).get("address", "")
except json.JSONDecodeError:
return ""
def get_workspace(self):
"""Get current workspace ID"""
try:
return json.loads(
self.conn.send_command("j/activeworkspace").reply.decode()
).get("id", 0)
except json.JSONDecodeError:
return 0
def check_occlusion_state(self):
"""Periodic occlusion check"""
# Skip occlusion check if hovered or dragging an icon
if self.is_hovered or self._drag_in_progress:
self.dock_full.remove_style_class("occluded")
return True
# If always_occluded is enabled, keep the dock occluded
if self.always_occluded:
self.dock_full.add_style_class("occluded")
return True
occlusion_region = ("bottom", OCCLUSION)
if check_occlusion(occlusion_region) or not self.view.get_children():
self.dock_full.add_style_class("occluded")
else:
self.dock_full.remove_style_class("occluded")
return True
def _find_drag_target(self, widget):
"""Find valid drag target in viewport"""
children = self.view.get_children()
while widget is not None and widget not in children:
widget = widget.get_parent() if hasattr(widget, "get_parent") else None
return widget
def on_drag_data_get(self, widget, drag_context, data, info, time):
"""Handle drag start"""
target = self._find_drag_target(widget.get_parent() if isinstance(widget, Box) else widget)
if target is not None:
index = self.view.get_children().index(target)
data.set_text(str(index), -1)
def on_drag_data_received(self, widget, drag_context, x, y, data, info, time):
"""Handle drop event"""
target = self._find_drag_target(widget.get_parent() if isinstance(widget, Box) else widget)
if target is None:
return
try:
source_index = int(data.get_text())
except (TypeError, ValueError):
return
children = self.view.get_children()
try:
target_index = children.index(target)
except ValueError:
return
if source_index != target_index:
# Check for separator to detect cross-section drag
separator_index = -1
for i, child in enumerate(children):
if child.get_name() == "dock-separator":
separator_index = i
break
# Detect if we're dragging across the separator (between pinned/unpinned)
cross_section_drag = (separator_index != -1 and
((source_index < separator_index and target_index > separator_index) or
(source_index > separator_index and target_index < separator_index)))
# Move the item in the children list
child = children.pop(source_index)
children.insert(target_index, child)
self.view.children = children # Update view immediately
# Update pinned apps configuration
self.update_pinned_apps(skip_update=not cross_section_drag)
# Force immediate update if we dragged across sections to update separator visibility
if cross_section_drag:
GLib.idle_add(self.update_dock)
def on_drag_end(self, widget, drag_context):
"""Handles drag end, for unpinning and closing apps."""
# If a drag is already in progress, ignore duplicate calls.
if not self._drag_in_progress:
return
def process_drag_end():
"""Inner function to handle drag end, safe to call with idle_add."""
if self.get_mapped(): # Check if the window is mapped
# Get window geometry *before* any modifications
display = Gdk.Display.get_default()
_, x, y, _ = display.get_pointer() # Get pointer relative to the screen
window = self.get_window()
if window: # Check if we got a window
win_x, win_y, width, height = window.get_geometry()
if not (win_x <= x <= win_x + width and win_y <= y <= win_y + height):
# Drag ended outside the dock - get app data from the button
app_id = widget.app_identifier
instances = widget.instances
# Find this app in pinned list
app_index = -1
for i, pinned_app in enumerate(self.pinned):
# Match by identifier
if pinned_app == app_id:
app_index = i
break
# For dict format, match by name
elif isinstance(pinned_app, dict) and isinstance(app_id, dict):
if pinned_app.get("name") == app_id.get("name"):
app_index = i
break
# If it's a pinned app, remove from pinned
if app_index >= 0:
self.pinned.pop(app_index)
self.config["pinned_apps"] = self.pinned
self.update_pinned_apps_file()
self.update_dock() # Update dock after file save
elif instances: # Close if running and unpinned
# Close running app (if not pinned)
address = instances[0].get("address")
if address:
exec_shell_command(f"hyprctl dispatch closewindow address:{address}")
self.update_dock() # Update dock after closing window
self._drag_in_progress = False # Clear the drag lock
GLib.idle_add(process_drag_end) # Deferred execution with idle_add
def check_config_change(self):
"""Check if the config file has been modified."""
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() # Update app_map when config changes
self.update_dock()
return True # Continue the timeout
def update_pinned_apps_file(self):
"""Writes the updated pinned apps list to dock.json"""
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 update_pinned_apps(self, skip_update=False):
"""Update pinned apps configuration by collecting all items before the separator"""
pinned_children = []
for child in self.view.get_children():
if child.get_name() == "dock-separator":
break # Stop at separator
# Use stored app identifier from the button
if hasattr(child, "app_identifier"):
# If the button has a desktop_app, use it to create comprehensive data
if hasattr(child, "desktop_app") and child.desktop_app:
app = child.desktop_app
app_data = {
"name": app.name,
"display_name": app.display_name,
"window_class": app.window_class,
"executable": app.executable,
"command_line": app.command_line
}
pinned_children.append(app_data)
else:
# Keep existing app_identifier (might be dict or string)
pinned_children.append(child.app_identifier)
# Directly update both config and local pinned list
self.config["pinned_apps"] = pinned_children
self.pinned = pinned_children # Update local state immediately
# Write to file first to ensure persistence
file_updated = self.update_pinned_apps_file()
# Only update the dock if needed and file was successfully written
if file_updated and not skip_update:
self.update_dock() # Refresh dock only after file is saved
# Static method to update all dock instances
@staticmethod
def notify_config_change():
"""Notify all dock instances to reload their configuration immediately."""
for dock in Dock._instances:
# Use idle_add to avoid potential issues with direct calls
GLib.idle_add(dock.check_config_change_immediate)
def check_config_change_immediate(self):
"""Immediately check for configuration changes and update if needed."""
new_config = read_config()
# Check if always_occluded setting has changed
previous_always_occluded = self.always_occluded
self.always_occluded = data.DOCK_ALWAYS_OCCLUDED
# If always_occluded changed, update the dock appearance
if previous_always_occluded != self.always_occluded:
if self.always_occluded and not self.is_hovered:
self.dock_full.add_style_class("occluded")
elif not self.always_occluded and not self.is_hovered:
# Perform a normal occlusion check
occlusion_region = ("bottom", OCCLUSION)
if check_occlusion(occlusion_region) or not self.view.get_children():
self.dock_full.add_style_class("occluded")
else:
self.dock_full.remove_style_class("occluded")
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_dock()
return False # Don't repeat
# Add a static method to handle dock visibility changes
@staticmethod
def update_visibility(visible):
"""Update visibility for all dock instances."""
for dock in Dock._instances:
dock.set_visible(visible)