colin/nwg-panel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add system tray module

Browse Source
This commit is contained in:
Christian Schulze
2022-01-29 22:59:35 +11:00
parent 1a436269d1
commit f6c2010216
9 changed files with 631 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@@ -1,5 +1,7 @@
/.idea
/venv
/nwg-panel.egg-info/
/nwg_panel.egg-info/
/build/
/dist/
/dist/
__pycache__

1
nwg_panel/common.py
View File

@@ -13,6 +13,7 @@ taskbars_list = []
scratchpads_list = []
workspaces_list = []
controls_list = []
tray_list = []
config_dir = ""
dwl_data_file = None
dwl_instances = []

13
nwg_panel/main.py
View File

@@ -50,6 +50,7 @@ from nwg_panel.modules.menu_start import MenuStart
dir_name = os.path.dirname(__file__)
from nwg_panel import common
from nwg_panel.modules import sni_system_tray
sway = os.getenv('SWAYSOCK') is not None
if sway:
@@ -72,6 +73,7 @@ def signal_handler(sig, frame):
desc = {2: "SIGINT", 15: "SIGTERM", 10: "SIGUSR1"}
if sig == 2 or sig == 15:
print("Terminated with {}".format(desc[sig]))
sni_system_tray.deinit_tray()
Gtk.main_quit()
elif sig == sig_dwl:
refresh_dwl()
@@ -218,6 +220,14 @@ def instantiate_content(panel, container, content_list, icons_path=""):
else:
print("{} data file not found".format(common.dwl_data_file))
if item == "tray":
tray_settings = {}
if "tray-settings" in panel:
tray_settings = panel["tray-settings"]
tray = sni_system_tray.Tray(tray_settings, icons_path)
common.tray_list.append(tray)
container.pack_start(tray, False, False, panel["items-padding"])
def main():
common.config_dir = get_config_dir()
@@ -540,6 +550,9 @@ def main():
common.outputs_num = len(common.outputs)
Gdk.threads_add_timeout(GLib.PRIORITY_DEFAULT_IDLE, 200, check_tree)
if len(common.tray_list) > 0:
sni_system_tray.init_tray(common.tray_list)
Gtk.main()

20
nwg_panel/modules/sni_system_tray/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
import typing
from threading import Thread
from . import host, watcher
from .tray import Tray
def init_tray(trays: typing.List[Tray]):
host_thread = Thread(target=host.init, args=[0, trays])
host_thread.daemon = True
host_thread.start()
watcher_thread = Thread(target=watcher.init)
watcher_thread.daemon = True
watcher_thread.start()
def deinit_tray():
host.deinit()
watcher.deinit()

128
nwg_panel/modules/sni_system_tray/host.py Normal file
View File

@@ -0,0 +1,128 @@
import typing
import os
from dasbus.connection import SessionMessageBus
from dasbus.loop import EventLoop
from dasbus.client.observer import DBusObserver
from dasbus.client.proxy import disconnect_proxy
from .watcher import WATCHER_SERVICE_NAME, WATCHER_OBJECT_PATH
from .tray import Tray
from .item import StatusNotifierItem
HOST_SERVICE_NAME_TEMPLATE = "org.kde.StatusNotifierHost-{}-{}"
HOST_OBJECT_PATH_TEMPLATE = "/StatusNotifierHost/{}"
dasbus_event_loop: typing.Union[EventLoop, None] = None
def get_service_name_and_object_path(service: str) -> (str, str):
index = service.find("/")
if index != len(service):
return service[0:index], service[index:]
return service, "/StatusNotifierItem"
class StatusNotifierHostInterface(object):
def __init__(self, host_id, trays: typing.List[Tray]):
self.host_id = host_id
self.trays = trays
self._statusNotifierItems = []
self.watcher_proxy = None
self.session_bus = SessionMessageBus()
self.host_service_name = HOST_SERVICE_NAME_TEMPLATE.format(os.getpid(), self.host_id)
self.host_object_path = HOST_OBJECT_PATH_TEMPLATE.format(self.host_id)
self.session_bus.register_service(self.host_service_name)
self.watcher_service_observer = DBusObserver(
message_bus=self.session_bus,
service_name=WATCHER_SERVICE_NAME
)
self.watcher_service_observer.service_available.connect(
self.watcher_available_handler
)
self.watcher_service_observer.service_unavailable.connect(
self.watcher_unavailable_handler
)
self.watcher_service_observer.connect_once_available()
def __del__(self):
if self.watcher_proxy is not None:
disconnect_proxy(self.watcher_proxy)
self.watcher_service_observer.disconnect()
self.session_bus.disconnect()
def watcher_available_handler(self, _observer):
print("StatusNotifierHostInterface -> watcher_available_handler")
self.watcher_proxy = self.session_bus.get_proxy(WATCHER_SERVICE_NAME, WATCHER_OBJECT_PATH)
self.watcher_proxy.StatusNotifierItemRegistered.connect(self.item_registered_handler)
self.watcher_proxy.StatusNotifierItemUnregistered.connect(self.item_unregistered_handler)
self.watcher_proxy.RegisterStatusNotifierHost(self.host_object_path, callback=lambda _: None)
def watcher_unavailable_handler(self, _observer):
print("StatusNotifierHostInterface -> watcher_unavailable_handler")
self._statusNotifierItems.clear()
disconnect_proxy(self.watcher_proxy)
self.watcher_proxy = None
def item_registered_handler(self, full_service_service):
print(
"StatusNotifierHostInterface -> item_registered_handler\n full_service_name: {}".format(
full_service_service
)
)
service_name, object_path = get_service_name_and_object_path(full_service_service)
if self.find_item(service_name, object_path) is None:
item = StatusNotifierItem(service_name, object_path)
item.set_on_loaded_callback(self.item_loaded_handler)
item.set_on_updated_callback(self.item_updated_handler)
self._statusNotifierItems.append(item)
def item_unregistered_handler(self, full_service_service):
print(
"StatusNotifierHostInterface -> item_unregistered_handler\n full_service_name: {}".format(
full_service_service
)
)
service_name, object_path = get_service_name_and_object_path(full_service_service)
item = self.find_item(service_name, object_path)
if item is not None:
self._statusNotifierItems.remove(item)
for tray in self.trays:
tray.remove_item(item)
def find_item(self, service_name, object_path) -> typing.Union[StatusNotifierItem, None]:
for item in self._statusNotifierItems:
if item.service_name == service_name and item.object_path == object_path:
return item
else:
return None
def item_loaded_handler(self, item):
for tray in self.trays:
tray.add_item(item)
def item_updated_handler(self, item, changed_properties):
for tray in self.trays:
tray.update_item(item, changed_properties)
def init(host_id, trays: typing.List[Tray]):
_status_notifier_host_interface = StatusNotifierHostInterface(host_id, trays)
global dasbus_event_loop
if dasbus_event_loop is None:
print("host.init(): running dasbus.EventLoop")
dasbus_event_loop = EventLoop()
dasbus_event_loop.run()
def deinit():
global dasbus_event_loop
if dasbus_event_loop is not None:
print("host.deinit(): quitting dasbus.EventLoop")
dasbus_event_loop.quit()
if dasbus_event_loop is not None:
dasbus_event_loop = None

96
nwg_panel/modules/sni_system_tray/item.py Normal file
View File

@@ -0,0 +1,96 @@
from dasbus.connection import SessionMessageBus
from dasbus.client.observer import DBusObserver
from dasbus.client.proxy import disconnect_proxy
PROPERTIES = [
"Id",
"Category",
"Title",
"Status",
"WindowId",
"IconName",
"IconPixmap",
"OverlayIconName",
"OverlayIconPixmap",
"AttentionIconName",
"AttentionIconPixmap",
"AttentionMovieName",
"ToolTip",
"IconThemePath",
"ItemIsMenu",
"Menu"
]
class StatusNotifierItem(object):
def __init__(self, service_name, object_path):
self.service_name = service_name
self.object_path = object_path
self.on_loaded_callback = None
self.on_updated_callback = None
self.session_bus = SessionMessageBus()
self.properties = {
"ItemIsMenu": True
}
self.item_proxy = None
self.item_observer = DBusObserver(
message_bus=self.session_bus,
service_name=self.service_name
)
self.item_observer.service_available.connect(
self.item_available_handler
)
self.item_observer.service_unavailable.connect(
self.item_unavailable_handler
)
self.item_observer.connect_once_available()
def __del__(self):
if self.item_proxy is not None:
disconnect_proxy(self.item_proxy)
self.item_observer.disconnect()
self.session_bus.disconnect()
def item_available_handler(self, _observer):
self.item_proxy = self.session_bus.get_proxy(self.service_name, self.object_path)
self.item_proxy.PropertiesChanged.connect(
lambda _if, changed_properties, _invalid: self.change_handler(list(changed_properties))
)
self.item_proxy.NewTitle.connect(
lambda _title: self.change_handler(["Title"])
)
self.item_proxy.NewIcon.connect(
lambda _icon_name, _icon_pixmap: self.change_handler(["IconName", "IconPixmap"])
)
self.item_proxy.NewAttentionIcon.connect(
lambda _icon_name, _icon_pixmap: self.change_handler(["AttentionIconName", "AttentionIconPixmap"])
)
self.item_proxy.NewIconThemePath.connect(
lambda _icon_theme_path: self.change_handler(["IconThemePath"])
)
self.item_proxy.NewStatus.connect(
lambda _status: self.change_handler(["Status"])
)
for name in PROPERTIES:
if hasattr(self.item_proxy, name):
self.properties[name] = getattr(self.item_proxy, name)
if self.on_loaded_callback is not None:
self.on_loaded_callback(self)
def item_unavailable_handler(self, _observer):
disconnect_proxy(self.item_proxy)
self.item_proxy = None
def change_handler(self, changed_properties: list[str]):
if len(changed_properties) > 0:
for name, value in changed_properties:
self.properties[name] = value
if self.on_updated_callback is not None:
self.on_updated_callback(self, changed_properties)
def set_on_loaded_callback(self, callback):
self.on_loaded_callback = callback
def set_on_updated_callback(self, callback):
self.on_updated_callback = callback

135
nwg_panel/modules/sni_system_tray/tray.py Normal file
View File

@@ -0,0 +1,135 @@
import os
import sys
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, GLib, GdkPixbuf
from nwg_panel.tools import check_key, get_config_dir
from .item import StatusNotifierItem
def load_icon(image, icon_name, icon_size, icons_path=""):
icon_theme = Gtk.IconTheme.get_default()
search_path = icon_theme.get_search_path()
if icons_path:
search_path.append(icons_path)
icon_theme.set_search_path(search_path)
if icon_theme.has_icon(icon_name):
pixbuf = icon_theme.load_icon(icon_name, icon_size, Gtk.IconLookupFlags.FORCE_SIZE)
elif icon_theme.has_icon(icon_name.lower()):
pixbuf = icon_theme.load_icon(icon_name.lower(), icon_size, Gtk.IconLookupFlags.FORCE_SIZE)
else:
try:
pixbuf = icon_theme.load_icon(icon_name, icon_size, Gtk.IconLookupFlags.FORCE_SIZE)
except GLib.GError:
print(
"tray.update_icon -> icon not found\n icon_name: {}\n search_path: {}".format(
icon_name,
search_path
),
file=sys.stderr
)
path = os.path.join(get_config_dir(), "icons_light/icon-missing.svg")
pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_size(path, icon_size, icon_size)
# TODO: if image height is different to icon_size, resize to match, while maintaining
# aspect ratio. Width can be ignored.
image.set_from_pixbuf(pixbuf)
def update_icon(image, item, icon_size, icon_path):
if "IconThemePath" in item.properties:
icon_path = item.properties["IconThemePath"]
load_icon(image, item.properties["IconName"], icon_size, icon_path)
def update_status(event_box, item):
if "Status" in item.properties:
status = item.properties["Status"].lower()
event_box.set_visible(status != "passive")
event_box_style = event_box.get_style_context()
for class_name in event_box_style.list_classes():
event_box_style.remove_class(class_name)
if status == "needsattention":
event_box_style.add_class("needs-attention")
event_box_style.add_class(status)
class Tray(Gtk.EventBox):
def __init__(self, settings, icons_path=""):
self.settings = settings
self.icons_path = icons_path
Gtk.EventBox.__init__(self)
check_key(settings, "icon-size", 16)
check_key(settings, "root-css-name", "tray")
self.set_property("name", settings["root-css-name"])
self.icon_size = settings["icon-size"]
self.box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0)
self.add(self.box)
self.items = {}
def add_item(self, item: StatusNotifierItem):
print("Tray -> add_item: {}".format(item.properties))
full_service_name = "{}{}".format(item.service_name, item.object_path)
if full_service_name not in self.items:
event_box = Gtk.EventBox()
image = Gtk.Image()
if "IconPixmap" in item.properties:
# TODO: handle loading pixbuf from dbus
pass
else:
update_icon(image, item, self.icon_size, self.icons_path)
if "Tooltip" in item.properties:
# TODO: handle tooltip variant type
pass
if "Title" in item.properties:
image.set_tooltip_markup(item.properties["Title"])
update_status(event_box, item)
event_box.add(image)
self.box.pack_start(event_box, False, False, 6)
self.box.show_all()
self.items[full_service_name] = {
"event_box": event_box,
"image": image,
"item": item
}
def update_item(self, item: StatusNotifierItem, changed_properties: list[str]):
full_service_name = "{}{}".format(item.service_name, item.object_path)
event_box = self.items[full_service_name]["event_box"]
image = self.items[full_service_name]["image"]
if "IconPixmap" in changed_properties:
# TODO: handle loading pixbuf from dbus
pass
elif "IconThemePath" in changed_properties or "IconName" in changed_properties:
update_icon(image, item, self.icon_size, self.icons_path)
if "Tooltip" in changed_properties:
# handle tooltip variant type
pass
if "Title" in changed_properties:
image.set_tooltip_markup(item.properties["Title"])
update_status(event_box, item)
event_box.show_all()
def remove_item(self, item: StatusNotifierItem):
full_service_name = "{}{}".format(item.service_name, item.object_path)
self.box.remove(self.items[full_service_name]["event_box"])
self.items.pop(full_service_name)
self.box.show_all()

230
nwg_panel/modules/sni_system_tray/watcher.py Normal file
View File

@@ -0,0 +1,230 @@
import typing
from dasbus.connection import SessionMessageBus
from dasbus.loop import EventLoop
from dasbus.signal import Signal
from dasbus.client.observer import DBusObserver
from dasbus.server.interface import accepts_additional_arguments
import dasbus.typing
WATCHER_SERVICE_NAME = "org.kde.StatusNotifierWatcher"
WATCHER_OBJECT_PATH = "/StatusNotifierWatcher"
dasbus_event_loop: typing.Union[EventLoop, None] = None
class StatusNotifierWatcherInterface(object):
__dbus_xml__ = """
<!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN" "http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
<node>
<interface name="org.kde.StatusNotifierWatcher">
<annotation name="org.gtk.GDBus.C.Name" value="Watcher" />
<method name="RegisterStatusNotifierItem">
<annotation name="org.gtk.GDBus.C.Name" value="RegisterItem" />
<arg name="service" type="s" direction="in"/>
</method>
<method name="RegisterStatusNotifierHost">
<annotation name="org.gtk.GDBus.C.Name" value="RegisterHost" />
<arg name="service" type="s" direction="in"/>
</method>
<property name="RegisteredStatusNotifierItems" type="as" access="read">
<annotation name="org.gtk.GDBus.C.Name" value="RegisteredItems" />
<annotation name="org.qtproject.QtDBus.QtTypeName.Out0" value="QStringList"/>
</property>
<property name="IsStatusNotifierHostRegistered" type="b" access="read">
<annotation name="org.gtk.GDBus.C.Name" value="IsHostRegistered" />
</property>
<property name="ProtocolVersion" type="i" access="read"/>
<signal name="StatusNotifierItemRegistered">
<annotation name="org.gtk.GDBus.C.Name" value="ItemRegistered" />
<arg type="s" direction="out" name="service" />
</signal>
<signal name="StatusNotifierItemUnregistered">
<annotation name="org.gtk.GDBus.C.Name" value="ItemUnregistered" />
<arg type="s" direction="out" name="service" />
</signal>
<signal name="StatusNotifierHostRegistered">
<annotation name="org.gtk.GDBus.C.Name" value="HostRegistered" />
</signal>
<signal name="StatusNotifierHostUnregistered">
<annotation name="org.gtk.GDBus.C.Name" value="HostUnregistered" />
</signal>
</interface>
</node>
"""
PropertiesChanged = Signal()
StatusNotifierItemRegistered = Signal()
StatusNotifierItemUnregistered = Signal()
StatusNotifierHostRegistered = Signal()
StatusNotifierHostUnregistered = Signal()
def __init__(self):
self._statusNotifierItems = []
self._statusNotifierHosts = []
self._isStatusNotifierHostRegistered = False
self._protocolVersion = 0
self.session_bus = SessionMessageBus()
def __del__(self):
self.session_bus.disconnect()
@accepts_additional_arguments
def RegisterStatusNotifierItem(self, service, call_info):
print(
"StatusNotifierWatcher -> RegisterStatusNotifierItem\n service: {}\n sender: {}".format(
service,
call_info["sender"]
)
)
# libappindicator sends object path, use sender name and object path
if service[0] == "/":
full_service_name = "{}{}".format(call_info["sender"], service)
# xembedsniproxy sends item name, use the item from the argument
elif service[0] == ":":
full_service_name = "{}{}".format(service, "/StatusNotifierItem")
else:
full_service_name = "{}{}".format(call_info["sender"], "/StatusNotifierItem")
if full_service_name not in self._statusNotifierItems:
item_service_observer = DBusObserver(
message_bus=self.session_bus,
service_name=call_info["sender"]
)
item_service_observer.service_available.connect(
lambda _observer: self.item_available_handler(full_service_name)
)
item_service_observer.service_unavailable.connect(
lambda _observer: self.item_unavailable_handler(full_service_name)
)
item_service_observer.connect_once_available()
else:
print(
(
"StatusNotifierWatcher -> RegisterStatusNotifierItem: item already registered\n"
" full_service_name: {}"
).format(full_service_name, service)
)
@accepts_additional_arguments
def RegisterStatusNotifierHost(self, service, call_info):
print("StatusNotifierWatcher -> RegisterStatusNotifierHost: {}".format(service))
if call_info["sender"] not in self._statusNotifierHosts:
host_service_observer = DBusObserver(
message_bus=self.session_bus,
service_name=call_info["sender"]
)
host_service_observer.service_available.connect(
self.host_available_handler
)
host_service_observer.service_unavailable.connect(
self.host_unavailable_handler
)
host_service_observer.connect_once_available()
else:
print(
"StatusNotifierWatcher -> RegisterStatusNotifierHost: host already registered\n service: {}\n sender: {})".format(
service,
call_info["sender"]
)
)
@property
def RegisteredStatusNotifierItems(self) -> list:
print("StatusNotifierWatcher -> RegisteredStatusNotifierItems")
return self._statusNotifierItems
@property
def IsStatusNotifierHostRegistered(self) -> bool:
print(
"StatusNotifierWatcher -> IsStatusNotifierHostRegistered: {}".format(
str(len(self._statusNotifierHosts) > 0)
)
)
return len(self._statusNotifierHosts) > 0
@property
def ProtocolVersion(self) -> int:
print("StatusNotifierWatcher -> ProtocolVersion: ".format(str(self._protocolVersion)))
return self._protocolVersion
def item_available_handler(self, full_service_name):
print(
"StatusNotifierWatcher -> item_available_handler\n full_service_name: {}".format(
full_service_name
)
)
self._statusNotifierItems.append(full_service_name)
self.StatusNotifierItemRegistered.emit(full_service_name)
self.PropertiesChanged.emit(WATCHER_SERVICE_NAME, {
"RegisteredStatusNotifierItems": dasbus.typing.get_variant(
dasbus.typing.List[dasbus.typing.Str],
self._statusNotifierItems
)
}, [])
def item_unavailable_handler(self, full_service_name):
print(
"StatusNotifierWatcher -> item_unavailable_handler\n full_service_name: {}".format(
full_service_name
)
)
if full_service_name in set(self._statusNotifierItems):
self._statusNotifierItems.remove(full_service_name)
self.StatusNotifierItemUnregistered.emit(full_service_name)
self.PropertiesChanged.emit(WATCHER_SERVICE_NAME, {
"RegisteredStatusNotifierItems": dasbus.typing.get_variant(
dasbus.typing.List[dasbus.typing.Str],
self._statusNotifierItems
)
}, [])
def host_available_handler(self, observer):
self._statusNotifierHosts.append(observer.service_name)
self.StatusNotifierHostRegistered.emit()
self.PropertiesChanged.emit(WATCHER_SERVICE_NAME, {
"IsStatusNotifierHostRegistered": dasbus.typing.get_variant(dasbus.typing.Bool, True)
}, [])
def host_unavailable_handler(self, observer):
self._statusNotifierHosts.remove(observer.service_name)
self.StatusNotifierHostUnregistered.emit()
if len(self._statusNotifierHosts) == 0:
self.PropertiesChanged.emit(WATCHER_SERVICE_NAME, {
"IsStatusNotifierHostRegistered": dasbus.typing.get_variant(dasbus.typing.Bool, False)
}, [])
def init():
session_bus = SessionMessageBus()
session_bus.publish_object(WATCHER_OBJECT_PATH, StatusNotifierWatcherInterface())
session_bus.register_service(WATCHER_SERVICE_NAME)
print("watcher.init(): published {}{} on dbus.".format(WATCHER_SERVICE_NAME, WATCHER_OBJECT_PATH))
global dasbus_event_loop
if dasbus_event_loop is None:
print("watcher.init(): running dasbus.EventLoop")
dasbus_event_loop = EventLoop()
dasbus_event_loop.run()
def deinit():
global dasbus_event_loop
if dasbus_event_loop is not None:
print("watcher.deinit(): quitting dasbus.EventLoop")
dasbus_event_loop.quit()
if dasbus_event_loop is not None:
dasbus_event_loop = None

5
requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
PyGObject~=3.42.0
psutil~=5.9.0
i3ipc~=2.2.1
setuptools~=57.0.0
dasbus~=1.6