import re
import subprocess
import traceback
from time import sleep
import notify2
import six
from dbus import DBusException
from libqtile import widget
from libqtile.backend.x11.window import Internal
from libqtile.bar import Bar
from notify2 import Notification, URGENCY_NORMAL
from libqtile.log_utils import logger
try:
notify2.init("QTileWM")
except DBusException as e:
logger.error("Could not initialize notify2: {}".format(e))
BUTTON_LEFT = 1
BUTTON_MIDDLE = 2
BUTTON_RIGHT = 3
BUTTON_UP = 4
BUTTON_DOWN = 5
BUTTON_MUTE = 1
BUTTON_SCROLL_UP = 4
BUTTON_SCROLL_DOWN = 5
def is_running(process):
s = subprocess.Popen(["ps", "axuw"], stdout=subprocess.PIPE)
if isinstance(process, list):
process = "".join(process)
for x in s.stdout:
if re.search(process, x.decode('utf-8')):
return True
return False
def execute(process):
if isinstance(process, list):
return subprocess.Popen(process)
elif isinstance(process, str):
return subprocess.Popen(process.split())
else:
pass
def execute_once(process):
if not is_running(process):
if isinstance(process, list):
return subprocess.Popen(process)
elif isinstance(process, str):
return subprocess.Popen(process.split())
else:
pass
def call_process(command, **kwargs):
"""
This method uses `subprocess.check_output` to run the given command
and return the string from stdout, which is decoded when using
Python 3.
"""
output = subprocess.check_output(command, **kwargs)
if six.PY3:
output = output.decode()
return output
def get_screen_count():
try:
output = subprocess.check_output("xrandr -q".split()).decode('utf-8')
output = [x for x in output.split("\n") if " connected" in x]
except subprocess.CalledProcessError:
return 1
if output:
return len(output)
else:
return 1
def bar_separator(config):
return widget.Sep(foreground=config.get('colour_spacer_background', '#777777'),
linewidth=config.get('width_spacer', 1),
padding=config.get('padding_spacer', 4),
)
def notify(title, content, urgency=URGENCY_NORMAL, timeout=5000, image=None):
if image is not None:
notification = Notification(
summary=title, message=content,
icon=image
)
else:
notification = Notification(
summary=title, message=content
)
notification.set_timeout(timeout)
notification.set_urgency(urgency)
try:
return notification.show()
except notify2.UninittedError:
logger.warning("Notify2 was uninitialized, initializing...")
notify2.init("qtile")
return notification.show()
except DBusException as e:
logger.warning("Showing notification failed: {}".format(e))
logger.warning(traceback.format_exc())
def spawn_popup(qtile, x, y, text):
"""
:param qtile: The main qtile instance
:type qtile: Qtile
:param x: x-coordinate
:type x: int
:param y: y-coordinate
:type y: int
:param text: String to display
:type text: str
:return: The popup instance
:rtype: Internal
"""
popup = Internal.create(
qtile, x, y, 100, 100, opacity=1
)
# Create textwidget for in window
popup.bordercolor = "#000000"
popup.borderwidth = 1
popup.focus(False)
#popup.
return popup
def despawn_popup(popup):
"""
:type popup: Internal
:param popup: The popup to despawn
"""
popup.kill()
def test_popups(qtile):
popup = spawn_popup(qtile, 10, 10, "Hello World!")
sleep(3)
despawn_popup(popup)
def display_wm_class(qtile):
window = qtile.currentWindow if qtile else None
if window:
wm_class = window.window.get_wm_class() or None
name = window.name
if wm_class:
notify(title="WM_Class of {}".format(name),
content="{}".format(wm_class),
urgency=notify2.URGENCY_CRITICAL)
def bluetooth_audio_sink():
try:
output = subprocess.check_output("pamixer --list-sinks".split()).decode("utf-8")
output = [x for x in output.split('\n') if "blue" in x.lower()]
except (subprocess.CalledProcessError, FileNotFoundError):
return -1
sink = -1
try:
sink = int(output[0].split()[0])
except IndexError:
pass
except AttributeError:
pass
except ValueError:
pass
return sink
def bluetooth_audio_connected():
return bluetooth_audio_sink() != -1
class KuroTopBar(Bar):
def __init__(self, theme, widgets, size, **config):
self.theme = theme
super(KuroTopBar, self).__init__(widgets, size, **config)
def _configure(self, qtile, screen, *args, **kwargs):
super(KuroTopBar, self)._configure(qtile, screen)
self.window.handle_EnterNotify = self.handle_enter_notify
self.window.handle_LeaveNotify = self.handle_leave_notify
self.window.window.set_property("_NET_WM_NAME", "KuroTopBar")
self.window.update_name()
def draw(self):
if not self.widgets:
return
if not self._draw_queued:
self.future = self.qtile.call_soon(self._actual_draw)
self._draw_queued = True
def _actual_draw(self):
self._draw_queued = False
self._resize(self._length, self.widgets)
for i in self.widgets:
i.draw()
if self.widgets:
end = i.offset + i.length
if end < self._length:
if self.horizontal:
self.drawer.draw(offsetx=end, width=self._length - end)
else:
self.drawer.draw(offsety=end, height=self._length - end)
self.theme.update_visualizers()
def handle_enter_notify(self, e):
# self.theme.log_debug("Bar HandleEnterNotify")
#
# self.window.opacity = Config.get('bar_hover_opacity', 1.0)
# print("Bar Hover Enter")
#
# try:
# hovered_widget = [x for x in self.widgets if (x.offsetx + x.width) >= e.event_x][0]
# except IndexError:
# hovered_widget = None
#
# self.theme.log_debug("Hovered over {}".format(hovered_widget))
#
# if hasattr(hovered_widget, "handle_hover_enter"):
# hovered_widget.handle_hover_enter(e)
self.draw()
def handle_leave_notify(self, e):
# self.theme.log_debug("Bar HandleLeaveNotify")
#
# self.window.opacity = Config.get('bar_opacity', 1.0)
# print("Bar Hover Leave")
#
# try:
# hovered_widget = [x for x in self.widgets if (x.offsetx + x.width) >= e.event_x][0]
# except IndexError:
# hovered_widget = None
#
# self.theme.log_debug("Hovered over {}".format(hovered_widget))
#
# if hasattr(hovered_widget, "handle_hover_leave"):
# hovered_widget.handle_hover_leave(e)
self.draw()