1316 lines
49 KiB
Python
1316 lines
49 KiB
Python
import math
|
|
import os
|
|
import re
|
|
import subprocess
|
|
|
|
import cairocffi
|
|
import iwlib
|
|
import netifaces
|
|
import psutil
|
|
import six
|
|
from libqtile import bar, pangocffi
|
|
from libqtile.log_utils import logger
|
|
from libqtile.widget import base
|
|
from libqtile.widget.base import ORIENTATION_HORIZONTAL
|
|
from libqtile.widget.battery import default_icon_path, load_battery, BatteryState
|
|
from libqtile.widget.check_updates import CheckUpdates
|
|
from libqtile.widget.currentlayout import CurrentLayoutIcon
|
|
from libqtile.widget.graph import _Graph
|
|
from libqtile.widget.tasklist import TaskList
|
|
from libqtile.widget.wlan import get_status
|
|
from libqtile.backend.x11.window import Window
|
|
|
|
from kuro.utils.general import notify, BUTTON_LEFT, BUTTON_MIDDLE, BUTTON_RIGHT, BUTTON_DOWN, BUTTON_UP, BUTTON_MUTE, \
|
|
call_process
|
|
|
|
|
|
class CheckUpdatesYay(CheckUpdates):
|
|
def __init__(self, **config):
|
|
super(CheckUpdatesYay, self).__init__(**config)
|
|
# Override command and output with yay command
|
|
self.cmd = "yay -Qu".split()
|
|
self.status_cmd = "yay -Qu --color never".split()
|
|
self.update_cmd = "sudo yay".split()
|
|
self.subtr = 0
|
|
|
|
def _check_updates(self):
|
|
#subprocess.check_output(self.update_cmd)
|
|
res = super(CheckUpdatesYay, self)._check_updates()
|
|
return res
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
output = subprocess.check_output(self.status_cmd).decode('utf-8').split('\n')
|
|
|
|
num_updates = len(output)-1
|
|
msg = "{} updates available.".format(num_updates)
|
|
|
|
if num_updates > 0:
|
|
msg += "\n\n"
|
|
for x in range(min(num_updates, 9)):
|
|
msg += output[x] + "\n"
|
|
if num_updates > 9:
|
|
msg += "and {} more...".format(num_updates-9)
|
|
|
|
notify(
|
|
"System updates",
|
|
msg
|
|
)
|
|
|
|
elif button == BUTTON_MIDDLE and self.execute is not None:
|
|
subprocess.Popen(self.execute, shell=True)
|
|
|
|
|
|
class DualPaneTextboxBase(base._Widget):
|
|
"""
|
|
Base class for widgets that are two boxes next to each other both containing text.
|
|
"""
|
|
orientations = ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
("font", "sans", "Default font"),
|
|
("fontsize_left", None, "Font size of left text field. Calculated if None."),
|
|
("fontsize_right", None, "Font size of right text field. Calculated if None."),
|
|
("padding", None, "Padding. Calculated if None."),
|
|
("padding_between", None, "Padding between left and right. Calculated if None."),
|
|
("foreground", "ffffff", "Foreground colour"),
|
|
("fontshadow", None, "font shadow color, default is None(no shadow)"),
|
|
("markup", False, "Whether or not to use pango markup"),
|
|
]
|
|
changed = False
|
|
|
|
def __init__(self, text_left=" ", text_right=" ", width=bar.CALCULATED, **config):
|
|
self.layout_left = None
|
|
self.layout_right = None
|
|
base._Widget.__init__(self, width, **config)
|
|
self._text_left = None
|
|
self._text_right = None
|
|
self.text_left = text_left
|
|
self.text_right = text_right
|
|
self.changed = False
|
|
self.add_defaults(DualPaneTextboxBase.defaults)
|
|
|
|
@property
|
|
def text_left(self):
|
|
return self._text_left
|
|
|
|
@text_left.setter
|
|
def text_left(self, value):
|
|
assert value is None or isinstance(value, str)
|
|
if value != self._text_left:
|
|
self.changed = True
|
|
self._text_left = value
|
|
if self.layout_left:
|
|
self.layout_left.text = value
|
|
@property
|
|
def text_right(self):
|
|
return self._text_right
|
|
|
|
@text_right.setter
|
|
def text_right(self, value):
|
|
assert value is None or isinstance(value, str)
|
|
if value != self._text_right:
|
|
self.changed = True
|
|
self._text_right = value
|
|
if self.layout_right:
|
|
self.layout_right.text = value
|
|
|
|
@property
|
|
def foreground(self):
|
|
return self._foreground
|
|
|
|
@foreground.setter
|
|
def foreground(self, fg):
|
|
self._foreground = fg
|
|
if self.layout_left:
|
|
self.layout_left.colour = fg
|
|
if self.layout_right:
|
|
self.layout_right.colour = fg
|
|
|
|
@property
|
|
def font(self):
|
|
return self._font
|
|
|
|
@font.setter
|
|
def font(self, value):
|
|
self._font = value
|
|
if self.layout_left:
|
|
self.layout_left.font = value
|
|
if self.layout_right:
|
|
self.layout_right.font = value
|
|
|
|
@property
|
|
def fontshadow(self):
|
|
return self._fontshadow
|
|
|
|
@fontshadow.setter
|
|
def fontshadow(self, value):
|
|
self._fontshadow = value
|
|
if self.layout_left:
|
|
self.layout_left.font_shadow = value
|
|
if self.layout_right:
|
|
self.layout_right.font_shadow = value
|
|
|
|
@property
|
|
def actual_padding(self):
|
|
if self.padding is None:
|
|
return min(self.fontsize_left, self.fontsize_right) / 2
|
|
else:
|
|
return self.padding
|
|
|
|
@property
|
|
def actual_padding_between(self):
|
|
if self.padding_between is None:
|
|
return max(self.fontsize_left, self.fontsize_right) / 4
|
|
else:
|
|
return self.padding_between
|
|
|
|
def _configure(self, qtile, bar):
|
|
base._Widget._configure(self, qtile, bar)
|
|
if self.fontsize_left is None:
|
|
self.fontsize_left = self.bar.height - self.bar.height / 5
|
|
if self.fontsize_right is None:
|
|
self.fontsize_right = self.bar.height - self.bar.height / 5
|
|
self.layout_left = self.drawer.textlayout(
|
|
self.text_left,
|
|
self.foreground,
|
|
self.font,
|
|
self.fontsize_left,
|
|
self.fontshadow,
|
|
markup=self.markup,
|
|
)
|
|
self.layout_right = self.drawer.textlayout(
|
|
self.text_right,
|
|
self.foreground,
|
|
self.font,
|
|
self.fontsize_right,
|
|
self.fontshadow,
|
|
markup=self.markup,
|
|
)
|
|
|
|
def calculate_length(self):
|
|
length = 0
|
|
if self.text_left:
|
|
length += min(
|
|
self.layout_left.width,
|
|
self.bar.width
|
|
) + self.actual_padding * 2
|
|
if self.text_right:
|
|
length += min(
|
|
self.layout_right.width,
|
|
self.bar.width
|
|
) + self.actual_padding * 2
|
|
return length
|
|
|
|
def draw(self):
|
|
# if the bar hasn't placed us yet
|
|
if self.offsetx is None:
|
|
return
|
|
self.drawer.clear(self.background or self.bar.background)
|
|
self.layout_left.draw(
|
|
self.actual_padding or 0,
|
|
int(self.bar.height / 2.0 - self.layout_left.height / 2.0) + 1
|
|
)
|
|
left_width = self.layout_left.width
|
|
self.layout_right.draw(
|
|
(self.actual_padding or 0) + left_width + (self.actual_padding_between or 0),
|
|
int(self.bar.height / 2.0 - self.layout_right.height / 2.0) + 1
|
|
)
|
|
self.drawer.draw(offsetx=self.offsetx, width=self.width)
|
|
if self.changed:
|
|
# Text changed, update the bar to propagate any changes in width
|
|
self.bar.draw()
|
|
self.changed = False
|
|
|
|
def cmd_set_font(self, font=base.UNSPECIFIED, fontsize_left=base.UNSPECIFIED, fontsize_right=base.UNSPECIFIED, fontshadow=base.UNSPECIFIED):
|
|
"""
|
|
Change the font used by this widget. If font is None, the current
|
|
font is used.
|
|
"""
|
|
if font is not base.UNSPECIFIED:
|
|
self.font = font
|
|
if fontsize_left is not base.UNSPECIFIED:
|
|
self.fontsize_left = fontsize_left
|
|
if fontsize_right is not base.UNSPECIFIED:
|
|
self.fontsize_right = fontsize_right
|
|
if fontshadow is not base.UNSPECIFIED:
|
|
self.fontshadow = fontshadow
|
|
self.bar.draw()
|
|
|
|
def info(self):
|
|
d = base._Widget.info(self)
|
|
d['foreground'] = self.foreground
|
|
d['text_left'] = self.text_left
|
|
d['text_right'] = self.text_right
|
|
return d
|
|
|
|
|
|
class MediaWidget(base.InLoopPollText):
|
|
"""Media Status Widget"""
|
|
|
|
class Status:
|
|
OFFLINE = 0
|
|
PLAYING = 1
|
|
PAUSED = 2
|
|
STOPPED = 3
|
|
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
('off_text', '', 'The pattern for the text if no players are found.'),
|
|
('on_text_play', ' {}', 'The pattern for the text if music is playing.'),
|
|
('on_text_pause', ' {}', 'The pattern for the text if music is paused.'),
|
|
('on_text_stop', ' {}', 'The pattern for the text if music is stopped.'),
|
|
('update_interval', 1, 'The update interval.'),
|
|
('max_chars_per_player', 50, 'Maximum characters of text per player.'),
|
|
]
|
|
|
|
player_icons = {
|
|
'spotify': '',
|
|
'vlc': '',
|
|
'firefox': '',
|
|
'mpv': '',
|
|
}
|
|
|
|
custom_player_data = {
|
|
'firefox': {
|
|
'showing': False,
|
|
'title': '',
|
|
'state': Status.STOPPED,
|
|
}
|
|
}
|
|
|
|
image_urls = {}
|
|
current_image_url = None
|
|
player_to_control = None
|
|
|
|
def __init__(self, **config):
|
|
super(MediaWidget, self).__init__(**config)
|
|
self.add_defaults(MediaWidget.defaults)
|
|
self.surfaces = {}
|
|
self.player_to_control = None
|
|
|
|
def _player_to_control(self):
|
|
info = self._get_info()
|
|
players = {}
|
|
for player in info.keys():
|
|
if player not in self.custom_player_data.keys():
|
|
if info[player][0] in [MediaWidget.Status.PLAYING, MediaWidget.Status.PAUSED]:
|
|
players[player] = info[player]
|
|
|
|
if self.player_to_control is not None and self.player_to_control not in players.keys():
|
|
self.player_to_control = None
|
|
|
|
if self.player_to_control is not None:
|
|
players = {self.player_to_control: players[self.player_to_control]}
|
|
|
|
if len(players.keys()) == 1:
|
|
player = list(players.keys())[0]
|
|
self.player_to_control = player
|
|
return player
|
|
|
|
elif len(players) == 0:
|
|
notify("MediaWidget", "Nothing to control!")
|
|
else:
|
|
notify("MediaWidget", "Multiple players to control, I don't know what you want to do!")
|
|
|
|
return None
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
player = self._player_to_control()
|
|
if player is not None:
|
|
command = ["playerctl", "-p", player, "play-pause"]
|
|
_ = self.call_process(command)
|
|
notify("MediaWidget", "Toggled {}".format(player))
|
|
if button == BUTTON_RIGHT:
|
|
player = self._player_to_control()
|
|
if player is not None:
|
|
command = ["playerctl", "-p", player, "next"]
|
|
_ = self.call_process(command)
|
|
if button == BUTTON_MIDDLE:
|
|
# Jump to the screen that the player is on
|
|
# clients = list(self.bar.qtile.windows_map.values())
|
|
# logger.warning("{}")
|
|
pass
|
|
|
|
def cmd_update_custom_player(self, player_name, data):
|
|
# Update firefox player
|
|
if player_name.startswith("firefox"):
|
|
if data['playing'] and data['muted']:
|
|
self.custom_player_data['firefox']['showing'] = True
|
|
self.custom_player_data['firefox']['state'] = MediaWidget.Status.PAUSED
|
|
self.custom_player_data['firefox']['title'] = data['title'][:50]
|
|
elif data['playing'] and not data['muted']:
|
|
self.custom_player_data['firefox']['showing'] = True
|
|
self.custom_player_data['firefox']['state'] = MediaWidget.Status.PLAYING
|
|
self.custom_player_data['firefox']['title'] = data['title'][:50]
|
|
elif not data['playing'] and data['muted']:
|
|
self.custom_player_data['firefox']['showing'] = True
|
|
self.custom_player_data['firefox']['state'] = MediaWidget.Status.STOPPED
|
|
self.custom_player_data['firefox']['title'] = data['title'][:50]
|
|
elif not data['playing'] and not data['muted']:
|
|
self.custom_player_data['firefox']['showing'] = False
|
|
self.custom_player_data['firefox']['state'] = MediaWidget.Status.OFFLINE
|
|
self.custom_player_data['firefox']['title'] = data['title'][:50]
|
|
|
|
def _get_players(self):
|
|
players = []
|
|
|
|
# Playerctl players
|
|
try:
|
|
result = self.call_process(["playerctl", "-l"])
|
|
except subprocess.CalledProcessError:
|
|
result = None
|
|
|
|
if result:
|
|
players.extend([x for x in result.split("\n") if x])
|
|
|
|
# Custom players - Firefox
|
|
if self.custom_player_data['firefox']['showing']:
|
|
players.append('firefox')
|
|
|
|
if players:
|
|
return players
|
|
else:
|
|
return None
|
|
|
|
def _get_info(self):
|
|
players = self._get_players()
|
|
|
|
if not players:
|
|
return {}
|
|
else:
|
|
result = {}
|
|
|
|
for player in players:
|
|
if player in self.custom_player_data.keys():
|
|
# Custom player -- Firefox
|
|
if player == "firefox":
|
|
result[player] = [self.custom_player_data['firefox']['state'], self.custom_player_data['firefox']['title']]
|
|
|
|
# Other custom players -- generic attempt with error catching
|
|
else:
|
|
try:
|
|
result[player] = [self.custom_player_data[player]['state'],
|
|
self.custom_player_data[player]['title']]
|
|
except KeyError:
|
|
pass
|
|
|
|
else:
|
|
# PlayerCtl player
|
|
command = ["playerctl", "-p", player, "status"]
|
|
cmd_result = self.call_process(command).strip()
|
|
|
|
text = "Unknown"
|
|
if cmd_result in ["Playing", "Paused"]:
|
|
try:
|
|
artist = self.call_process(['playerctl', '-p', player, 'metadata', 'artist']).strip()
|
|
except subprocess.CalledProcessError:
|
|
artist = None
|
|
try:
|
|
title = self.call_process(['playerctl', '-p', player, 'metadata', 'title']).strip()
|
|
except subprocess.CalledProcessError:
|
|
title = None
|
|
|
|
if artist and title:
|
|
text = "{} - {}".format(artist, title)
|
|
elif artist:
|
|
text = artist
|
|
elif title:
|
|
text = title
|
|
|
|
if cmd_result == "Playing":
|
|
result[player] = [MediaWidget.Status.PLAYING, text]
|
|
elif cmd_result == "Paused":
|
|
result[player] = [MediaWidget.Status.PAUSED, text]
|
|
elif cmd_result == "Stopped":
|
|
result[player] = [MediaWidget.Status.STOPPED, ""]
|
|
|
|
return result
|
|
|
|
def _get_formatted_text(self, status):
|
|
if status[0] == MediaWidget.Status.PLAYING:
|
|
res = self.on_text_play.format(status[1])
|
|
elif status[0] == MediaWidget.Status.PAUSED:
|
|
res = self.on_text_pause.format(status[1])
|
|
elif status[0] == MediaWidget.Status.STOPPED:
|
|
res = self.on_text_stop.format(status[1])
|
|
else:
|
|
res = "Unknown"
|
|
res = pangocffi.markup_escape_text(res)
|
|
if len(res) > self.max_chars_per_player:
|
|
res = res[:self.max_chars_per_player] + "..."
|
|
return res
|
|
|
|
def draw(self):
|
|
super(MediaWidget, self).draw()
|
|
|
|
def poll(self):
|
|
text = []
|
|
status = self._get_info()
|
|
if not status:
|
|
return self.off_text
|
|
else:
|
|
for player in status.keys():
|
|
# Shorten firefox.instance[0-9]+ to just firefox for icon finding
|
|
if player.startswith("firefox"):
|
|
player_icon = "firefox"
|
|
else:
|
|
player_icon = player
|
|
icon = self.player_icons.get(player_icon, player_icon)
|
|
text.append("{} {}".format(icon, self._get_formatted_text(status[player])))
|
|
|
|
return " | ".join(text) if text else self.off_text
|
|
|
|
|
|
class AudioVisualizerWidget(_Graph):
|
|
"""Display Audio Visualization graph"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
fixed_upper_bound = True
|
|
defaults = [
|
|
("graph_color", "FFFFFF.0", "Graph color"),
|
|
("fill_color", "FFFFFF.0", "Fill color for linefill graph"),
|
|
("border_color", "FFFFFF.0", "Widget border color"),
|
|
("border_width", 0, "Widget border width"),
|
|
("line_width", 0, "Line width"),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
_Graph.__init__(self, **config)
|
|
self.add_defaults(AudioVisualizerWidget.defaults)
|
|
|
|
self.client = None
|
|
self.screen = None
|
|
|
|
self.old_position = None
|
|
|
|
def set_client(self, c, s):
|
|
self.client = c
|
|
self.screen = s
|
|
|
|
def update_graph(self):
|
|
if self.client is not None:
|
|
viz_info = self.info()
|
|
pos_x = viz_info['offset'] + self.margin_x + self.screen.x
|
|
pos_y = 0 + self.margin_y + self.screen.y
|
|
if self.old_position != (pos_x, pos_y):
|
|
self.old_position = (pos_x, pos_y)
|
|
|
|
# Check if a window on this screen is full-screen
|
|
fullscreen = False
|
|
for window in self.screen.group.windows:
|
|
if isinstance(window, Window):
|
|
if window.fullscreen:
|
|
fullscreen = True
|
|
break
|
|
|
|
logger.debug("Repositioning {} {} to {}x{}".format(self.client, self.client.window.wid, pos_x, pos_y))
|
|
self.client.reposition(pos_x, pos_y, above=not fullscreen)
|
|
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
self.drawer.clear(self.background or self.bar.background)
|
|
self.drawer.draw(offsetx=self.offset, width=self.width)
|
|
|
|
|
|
class KuroCurrentLayoutIcon(CurrentLayoutIcon):
|
|
def _get_layout_names(self):
|
|
names = super(KuroCurrentLayoutIcon, self)._get_layout_names()
|
|
|
|
from kuro.utils import layouts as kuro_layouts
|
|
from libqtile.layout.base import Layout
|
|
klayouts = [
|
|
layout_class_name.lower()
|
|
for layout_class, layout_class_name
|
|
in map(lambda x: (getattr(kuro_layouts, x), x), dir(kuro_layouts))
|
|
if isinstance(layout_class, six.class_types) and issubclass(layout_class, Layout)
|
|
]
|
|
names.extend(klayouts)
|
|
|
|
return list(set(names))
|
|
|
|
|
|
class KuroTaskList(TaskList):
|
|
defaults = [
|
|
(
|
|
'txt_pinned',
|
|
'P ',
|
|
'Text representation of the pinned window state. '
|
|
'e.g., "P " or "\U0001F5D7 "'
|
|
),
|
|
(
|
|
'markup_pinned',
|
|
None,
|
|
'Text markup of the pinned window state. Supports pangomarkup with markup=True.'
|
|
'e.g., "{}" or "<span underline="low">{}</span>"'
|
|
),
|
|
]
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(KuroTaskList, self).__init__(*args, **kwargs)
|
|
self.add_defaults(KuroTaskList.defaults)
|
|
|
|
def get_taskname(self, window):
|
|
"""
|
|
Get display name for given window.
|
|
Depending on its state minimized, maximized and floating
|
|
appropriate characters are prepended.
|
|
"""
|
|
state = ''
|
|
markup_str = self.markup_normal
|
|
|
|
# Enforce markup and new string format behaviour when
|
|
# at least one markup_* option is used.
|
|
# Mixing non markup and markup may cause problems.
|
|
if self.markup_minimized or self.markup_maximized\
|
|
or self.markup_floating or self.markup_focused or self.markup_pinned:
|
|
enforce_markup = True
|
|
else:
|
|
enforce_markup = False
|
|
|
|
if window is None:
|
|
pass
|
|
elif hasattr(window, "is_static_window") and window.is_static_window:
|
|
state = self.txt_pinned
|
|
markup_str = self.markup_pinned
|
|
elif window.minimized:
|
|
state = self.txt_minimized
|
|
markup_str = self.markup_minimized
|
|
elif window.maximized:
|
|
state = self.txt_maximized
|
|
markup_str = self.markup_maximized
|
|
elif window.floating:
|
|
state = self.txt_floating
|
|
markup_str = self.markup_floating
|
|
elif window is window.group.current_window:
|
|
markup_str = self.markup_focused
|
|
|
|
window_name = window.name if window and window.name else "?"
|
|
|
|
# Emulate default widget behavior if markup_str is None
|
|
if enforce_markup and markup_str is None:
|
|
markup_str = "%s{}" % (state)
|
|
|
|
if markup_str is not None:
|
|
self.markup = True
|
|
window_name = pangocffi.markup_escape_text(window_name)
|
|
return markup_str.format(window_name)
|
|
|
|
return "%s%s" % (state, window_name)
|
|
|
|
|
|
class GPUStatusWidget(base._TextBox):
|
|
"""Displays the currently used GPU."""
|
|
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
('check_command', 'optimus-manager --print-mode', 'The command that shows the current mode.'),
|
|
('next_command', 'optimus-manager --print-next-mode', 'The command that shows the mode after reboot.'),
|
|
('update_interval', 60, 'The update interval in seconds.'),
|
|
('theme_path', default_icon_path(), 'Path of the icons'),
|
|
('custom_icons', {}, 'dict containing key->filename icon map'),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(GPUStatusWidget, self).__init__("GPU", bar.CALCULATED, **config)
|
|
self.add_defaults(GPUStatusWidget.defaults)
|
|
|
|
if self.theme_path:
|
|
self.length_type = bar.STATIC
|
|
self.length = 0
|
|
self.surfaces = {}
|
|
self.current_icon = 'gpu-unknown'
|
|
self.icons = dict([(x, '{0}.png'.format(x)) for x in (
|
|
'gpu-intel',
|
|
'gpu-nvidia',
|
|
'gpu-unknown',
|
|
)])
|
|
self.current_status = "Unknown"
|
|
self.icons.update(self.custom_icons)
|
|
|
|
def _get_info(self):
|
|
output = self.call_process(self.check_command, shell=True)
|
|
mode = "nvidia" if "nvidia" in output else "intel" if "intel" in output else "unknown"
|
|
return {'error': False, 'mode': mode}
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _configure(self, qtile, bar):
|
|
super(GPUStatusWidget, self)._configure(qtile, bar)
|
|
self.setup_images()
|
|
|
|
def _get_icon_key(self):
|
|
key = 'gpu'
|
|
info = self._get_info()
|
|
if info.get('mode') == "intel":
|
|
key += '-intel'
|
|
self.current_status = "Intel"
|
|
elif info.get('mode') == "nvidia":
|
|
key += '-nvidia'
|
|
self.current_status = "NVidia"
|
|
else:
|
|
key += '-unknown'
|
|
self.current_status = "Unknown"
|
|
return key
|
|
|
|
def update(self):
|
|
icon = self._get_icon_key()
|
|
if icon != self.current_icon:
|
|
self.current_icon = icon
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
if self.theme_path:
|
|
self.drawer.clear(self.background or self.bar.background)
|
|
self.drawer.ctx.set_source(self.surfaces[self.current_icon])
|
|
self.drawer.ctx.paint()
|
|
self.drawer.draw(offsetx=self.offset, width=self.length)
|
|
else:
|
|
self.text = self.current_icon[8:]
|
|
base._TextBox.draw(self)
|
|
|
|
def setup_images(self):
|
|
for key, name in self.icons.items():
|
|
try:
|
|
path = os.path.join(self.theme_path, name)
|
|
img = cairocffi.ImageSurface.create_from_png(path)
|
|
except cairocffi.Error:
|
|
self.theme_path = None
|
|
logger.warning('GPU Status Icon switching to text mode')
|
|
return
|
|
input_width = img.get_width()
|
|
input_height = img.get_height()
|
|
|
|
sp = input_height / (self.bar.height - 1)
|
|
|
|
width = input_width / sp
|
|
if width > self.length:
|
|
# cast to `int` only after handling all potentially-float values
|
|
self.length = int(width + self.actual_padding * 2)
|
|
|
|
imgpat = cairocffi.SurfacePattern(img)
|
|
|
|
scaler = cairocffi.Matrix()
|
|
|
|
scaler.scale(sp, sp)
|
|
scaler.translate(self.actual_padding * -1, 0)
|
|
imgpat.set_matrix(scaler)
|
|
|
|
imgpat.set_filter(cairocffi.FILTER_BEST)
|
|
self.surfaces[key] = imgpat
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
try:
|
|
next_gpu = self.call_process(self.next_command, shell=True).split(":")[1].strip()
|
|
except IndexError:
|
|
next_gpu = "Unknown"
|
|
|
|
if self.current_status == "Unknown":
|
|
notify("GPU Status", "The currently used GPU is <b>unknown</b>.\n\nAfter the next login it will be the <b>{}</b> GPU.".format(next_gpu),
|
|
image=os.path.join(self.theme_path, "gpu-unknown.png"))
|
|
else:
|
|
notify("GPU Status", "The system is currently running on the <b>{}</b> GPU. Press the middle mouse "
|
|
"button on this icon to switch GPUs.\n\nAfter the next login it will be the <b>{}</b> GPU.".format(
|
|
self.current_status, next_gpu
|
|
),
|
|
image=os.path.join(self.theme_path, "gpu-{}.png".format(self.current_status.lower()))
|
|
)
|
|
|
|
if button == BUTTON_MIDDLE:
|
|
command = ["optimus-manager", "--no-confirm", "--switch", "auto"]
|
|
output = self.call_process(command)
|
|
if "nvidia" in output:
|
|
notify("GPU Switched", "The GPU has been switched from Intel to NVidia.\n"
|
|
"Please log out and log back in to apply the changes to the session.",
|
|
image=os.path.join(self.theme_path, "gpu-nvidia.png"))
|
|
elif "intel" in output:
|
|
notify("GPU Switched", "The GPU has been switched from NVidia to Intel.\n"
|
|
"Please log out and log back in to apply the changes to the session.",
|
|
image=os.path.join(self.theme_path, "gpu-intel.png"))
|
|
else:
|
|
notify("GPU Switch Error", "I could not determine if the GPU was switched successfully.\n"
|
|
"Please log out and log back in to clear up the inconsistency.",
|
|
image=os.path.join(self.theme_path, "gpu-unknown.png"))
|
|
|
|
|
|
class TextSpacerWidget(base._TextBox):
|
|
"""Displays a text separator"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
('spacer', None, 'The character/text to use as separator. Default "|" if None.'),
|
|
('color', "#ffffff", "Color of the text."),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(TextSpacerWidget, self).__init__("Separator", bar.CALCULATED, **config)
|
|
self.add_defaults(TextSpacerWidget.defaults)
|
|
self.text = self.spacer or "|"
|
|
|
|
def draw(self):
|
|
base._TextBox.draw(self)
|
|
|
|
|
|
class ThermalSensorWidget(DualPaneTextboxBase):
|
|
defaults = [
|
|
('show_tag', False, 'Show tag sensor'),
|
|
('update_interval', 2, 'Update interval in seconds'),
|
|
('tag_sensor', None, 'Tag of the temperature sensor. For example: "temp1" or "Core 0"'),
|
|
('chip', None, 'Chip argument for sensors command'),
|
|
('threshold', 70, 'If the current temperature value is above, then change to foreground_alert colour'),
|
|
('foreground_alert', 'ff0000', 'Foreground colour alert'),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(ThermalSensorWidget, self).__init__("TEMP", "", bar.CALCULATED, **config)
|
|
self.add_defaults(ThermalSensorWidget.defaults)
|
|
self.sensors_temp = re.compile(
|
|
(r"\n([\w ]+):" # Sensor tag name
|
|
r"\s+[+|-]" # temp signed
|
|
r"(\d+\.\d+)" # temp value
|
|
"({degrees}" # degree symbol match
|
|
"[C|F])" # Celsius or Fahrenheit
|
|
).format(degrees=u"\xb0"),
|
|
re.UNICODE | re.VERBOSE
|
|
)
|
|
self.value_temp = re.compile(r"\d+\.\d+")
|
|
self.foreground_normal = self.foreground
|
|
self.values = None
|
|
|
|
def get_command(self):
|
|
command = ["sensors"]
|
|
if self.chip:
|
|
command.append(self.chip)
|
|
return command
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _update_values(self):
|
|
sensors_out = self.call_process(self.get_command())
|
|
temperature_values = {}
|
|
for name, temp, symbol in self.sensors_temp.findall(sensors_out):
|
|
name = name.strip()
|
|
temperature_values[name] = temp, symbol
|
|
self.values = temperature_values
|
|
|
|
def update(self):
|
|
self._update_values()
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
self.text_left = ""
|
|
if self.values is None:
|
|
self.text_right = None
|
|
else:
|
|
text = ""
|
|
if self.show_tag and self.tag_sensor is not None:
|
|
text = self.tag_sensor + ": "
|
|
text += "".join(self.values.get(self.tag_sensor, ['N/A']))
|
|
temp_value = float(self.values.get(self.tag_sensor, [0])[0])
|
|
if temp_value > self.threshold:
|
|
self.layout_right.colour = self.foreground_alert
|
|
else:
|
|
self.layout_right.colour = self.foreground_normal
|
|
self.text_right = text
|
|
super(ThermalSensorWidget, self).draw()
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
notify("Temperature Information", "\n".join(
|
|
"{}: {}{}".format(name, *values) for name, values in self.values.items()
|
|
))
|
|
|
|
|
|
class CPUInfoWidget(DualPaneTextboxBase):
|
|
"""Displays information about the CPU usage"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
('update_interval', 3, 'The update interval in seconds.'),
|
|
('text_pattern', "{cpu}%", 'The pattern for the text that is displayed.'),
|
|
('normal_color', "#ffffff", "Color when value is normal"),
|
|
('warning_color', "#ffffff", "Color when value is warning"),
|
|
('critical_color', "#ffffff", "Color when value is critical"),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(CPUInfoWidget, self).__init__("CPU", "", bar.CALCULATED, **config)
|
|
self.add_defaults(CPUInfoWidget.defaults)
|
|
self.text = "..."
|
|
self.cpu = 0
|
|
self.cpu_old = [0, 0, 0, 0]
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _update_values(self):
|
|
cpu = psutil.cpu_times()
|
|
user = cpu.user * 100
|
|
nice = cpu.nice * 100
|
|
sys = cpu.system * 100
|
|
idle = cpu.idle * 100
|
|
nval = (int(user), int(nice), int(sys), int(idle))
|
|
oval = self.cpu_old
|
|
busy = nval[0] + nval[1] + nval[2] - oval[0] - oval[1] - oval[2]
|
|
total = busy + nval[3] - oval[3]
|
|
if total:
|
|
self.cpu = math.ceil(busy * 100.0 / total)
|
|
self.cpu_old = nval
|
|
|
|
def update(self):
|
|
self._update_values()
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
self.text_left = ""
|
|
self.text_right = self.text_pattern.format(cpu=self.cpu)
|
|
super(CPUInfoWidget, self).draw()
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
total = sum([self.cpu_old[0], self.cpu_old[1], self.cpu_old[2], self.cpu_old[3]])
|
|
notify("CPU Information", "user: {} %\nnice: {} %\nsys: {} %\nidle: {} %\ntotal: {} %".format(
|
|
math.ceil((self.cpu_old[0] / total) * 100),
|
|
math.ceil((self.cpu_old[1] / total) * 100),
|
|
math.ceil((self.cpu_old[2] / total) * 100),
|
|
math.ceil((self.cpu_old[3] / total) * 100),
|
|
self.cpu
|
|
))
|
|
|
|
|
|
class MemoryInfoWidget(DualPaneTextboxBase):
|
|
"""Displays information about the Memory usage"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
('update_interval', 3, 'The update interval in seconds.'),
|
|
('text_pattern', "{mem}%", 'The pattern for the text that is displayed.'),
|
|
('normal_color', "#ffffff", "Color when value is normal"),
|
|
('warning_color', "#ffffff", "Color when value is warning"),
|
|
('critical_color', "#ffffff", "Color when value is critical"),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(MemoryInfoWidget, self).__init__("MEM", "", bar.CALCULATED, **config)
|
|
self.add_defaults(MemoryInfoWidget.defaults)
|
|
self.text = "..."
|
|
self.memory = 0
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _update_values(self):
|
|
mem = psutil.virtual_memory()
|
|
self.memory = math.ceil((mem.used / mem.total) * 100)
|
|
|
|
def update(self):
|
|
self._update_values()
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
self.text_left = ""
|
|
self.text_right = self.text_pattern.format(mem=self.memory)
|
|
super(MemoryInfoWidget, self).draw()
|
|
|
|
def button_press(self, x, y, button):
|
|
mem = psutil.virtual_memory()
|
|
swap = psutil.swap_memory()
|
|
val = {}
|
|
val['MemUsed'] = mem.used // 1024 // 1024
|
|
val['MemTotal'] = mem.total // 1024 // 1024
|
|
val['SwapTotal'] = swap.total // 1024 // 1024
|
|
val['SwapUsed'] = swap.used // 1024 // 1024
|
|
|
|
if button == BUTTON_LEFT:
|
|
notify("Memory Information", "Memory: {}MB / {}MB\n {}%\nSwap: {}MB / {}MB\n {}%".format(
|
|
val['MemUsed'], val['MemTotal'],
|
|
math.ceil((mem.used / mem.total) * 100),
|
|
val['SwapUsed'], val['SwapTotal'],
|
|
math.ceil((swap.used / swap.total) * 100),
|
|
))
|
|
|
|
|
|
class DiskIOInfoWidget(DualPaneTextboxBase):
|
|
"""Displays information about the DiskIO usage"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
('update_interval', 3, 'The update interval in seconds.'),
|
|
('text_pattern', "{io}ms", 'The pattern for the text that is displayed.'),
|
|
('hdd_device', 'nvme0n1', 'The device name of the disk to use for IO stats'),
|
|
('normal_color', "#ffffff", "Color when value is normal"),
|
|
('warning_color', "#ffffff", "Color when value is warning"),
|
|
('critical_color', "#ffffff", "Color when value is critical"),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(DiskIOInfoWidget, self).__init__("DiskIO", "", bar.CALCULATED, **config)
|
|
self.add_defaults(DiskIOInfoWidget.defaults)
|
|
self.text = "..."
|
|
self.io = 0
|
|
self.io_old = 0
|
|
self.hdd_path = '/sys/block/{dev}/stat'.format(
|
|
dev=self.hdd_device
|
|
)
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _update_values(self):
|
|
try:
|
|
# io_ticks is field number 9
|
|
with open(self.hdd_path) as f:
|
|
io_ticks = int(f.read().split()[9])
|
|
except IOError:
|
|
return 0
|
|
activity = io_ticks - self.io_old
|
|
self.io_old = io_ticks
|
|
self.io = math.ceil(activity)
|
|
|
|
def update(self):
|
|
self._update_values()
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
self.text_left = ""
|
|
self.text_right = self.text_pattern.format(io=self.io)
|
|
super(DiskIOInfoWidget, self).draw()
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
notify("Disk IO Information",
|
|
"Time that there were IO requests queued for /dev/{}: {} ms".format(self.hdd_device, self.io))
|
|
|
|
|
|
class NetworkInfoWidget(DualPaneTextboxBase):
|
|
"""Displays information about the (wireless) network that we are connected to"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
wired_up_regex = re.compile(" state (DOWN|UP) ")
|
|
defaults = [
|
|
('update_interval', 10, 'The update interval in seconds.'),
|
|
('text_pattern', "{name}", 'The pattern for the text that is displayed.'),
|
|
('text_disconnected', "", "The text to show when not connected using WiFi"),
|
|
('normal_color', "#ffffff", "Color when value is normal"),
|
|
('warning_color', "#ffffff", "Color when value is warning"),
|
|
('critical_color', "#ffffff", "Color when value is critical"),
|
|
('wireless_interface', "wifi0", "Wireless interface device name"),
|
|
('wired_interface', "enp7s0", "Wired interface device name"),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(NetworkInfoWidget, self).__init__("NET", "", bar.CALCULATED, **config)
|
|
self.add_defaults(NetworkInfoWidget.defaults)
|
|
self.wireless_text = "..."
|
|
self.wireless_quality = 0
|
|
self.wireless_signal = 0
|
|
self.wireless_name = 0
|
|
self.wireless_connected = False
|
|
self.wireless_accesspoint = ""
|
|
self.wireless_frequency = ""
|
|
self.wireless_ipv4 = ""
|
|
self.wireless_ipv6 = ""
|
|
self.wireless_mac = ""
|
|
|
|
self.wired_name = 0
|
|
self.wired_connected = False
|
|
self.wired_ipv4 = ""
|
|
self.wired_ipv6 = ""
|
|
self.wired_mac = ""
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _update_values(self):
|
|
# Wifi
|
|
try:
|
|
essid, quality = get_status(self.wireless_interface)
|
|
status = iwlib.get_iwconfig(self.wireless_interface)
|
|
self.wireless_ips = netifaces.ifaddresses(self.wireless_interface)
|
|
disconnected = essid is None
|
|
percent = math.ceil((quality / 70) * 100)
|
|
self.wireless_quality = quality
|
|
self.wireless_signal = percent
|
|
self.wireless_name = essid
|
|
self.wireless_connected = not disconnected
|
|
self.wireless_accesspoint = status.get('Access Point', b'Unknown').decode()
|
|
self.wireless_frequency = status.get('Frequency', b'Unknown').decode()
|
|
self.wireless_ipv4 = self.wireless_ips.get(netifaces.AF_INET, [{'addr': ""}])[0]['addr']
|
|
self.wireless_ipv6 = self.wireless_ips.get(netifaces.AF_INET6, [{'addr': ""}])[0]['addr']
|
|
self.wireless_mac = self.wireless_ips.get(netifaces.AF_LINK, [{'addr': ""}])[0]['addr']
|
|
except EnvironmentError:
|
|
pass
|
|
|
|
# Wired
|
|
try:
|
|
self.wired_ips = netifaces.ifaddresses(self.wired_interface)
|
|
self.wired_ipv4 = self.wired_ips.get(netifaces.AF_INET, [{'addr': ""}])[0]['addr']
|
|
self.wired_ipv6 = self.wired_ips.get(netifaces.AF_INET6, [{'addr': ""}])[0]['addr']
|
|
self.wired_mac = self.wired_ips.get(netifaces.AF_LINK, [{'addr': ""}])[0]['addr']
|
|
eth_status = call_process(["ip", "link", "show", "{}".format(self.wired_interface)])
|
|
m = self.wired_up_regex.search(eth_status)
|
|
if m:
|
|
self.wired_connected = "UP" in m.group(1)
|
|
else:
|
|
self.wired_connected = False
|
|
|
|
except (EnvironmentError, ValueError):
|
|
pass
|
|
|
|
def update(self):
|
|
self._update_values()
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
if self.wireless_connected:
|
|
strength = ""
|
|
if self.wireless_signal < 66:
|
|
strength = ""
|
|
if self.wireless_signal < 33:
|
|
strength = ""
|
|
self.text_left = strength
|
|
else:
|
|
self.text_left = ""
|
|
|
|
if self.wired_connected:
|
|
self.text_right = ""
|
|
else:
|
|
self.text_right = ""
|
|
|
|
super(NetworkInfoWidget, self).draw()
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
if self.wireless_connected:
|
|
title = "Wireless {}".format(self.wireless_interface)
|
|
wireless_ipv4 = "\nIPv4: {}".format(self.wireless_ipv4) if self.wireless_ipv4 else ""
|
|
wireless_ipv6 = "\nIPv6: {}".format(self.wireless_ipv6) if self.wireless_ipv6 else ""
|
|
wireless_mac = "\nMAC: {}".format(self.wireless_mac) if self.wireless_mac else ""
|
|
wifi_text = "SSID: {}\n" \
|
|
"Quality: {} / 70\n"\
|
|
"Strength: {}%" \
|
|
"{}{}{}".format(self.wireless_name, self.wireless_quality,
|
|
self.wireless_signal, wireless_ipv4, wireless_ipv6, wireless_mac)
|
|
else:
|
|
title = "Wireless: Not connected"
|
|
wifi_text = ""
|
|
|
|
if self.wired_connected:
|
|
wired_ipv4 = "\nIPv4: {}".format(self.wired_ipv4) if self.wired_ipv4 else ""
|
|
wired_ipv6 = "\nIPv6: {}".format(self.wired_ipv6) if self.wired_ipv6 else ""
|
|
wired_mac = "\nMAC: {}".format(self.wired_mac) if self.wired_mac else ""
|
|
wired_text = "<b>Wired {}</b>" \
|
|
"{}{}{}".format(self.wired_interface, wired_ipv4, wired_ipv6, wired_mac)
|
|
else:
|
|
wired_text = "<b>Wired: Not connected</b>"
|
|
|
|
if wifi_text:
|
|
notify(title, "{}\n\n{}".format(wifi_text, wired_text))
|
|
else:
|
|
notify(title, "\n{}".format(wired_text))
|
|
|
|
|
|
class BatteryInfoWidget(DualPaneTextboxBase):
|
|
"""Displays information about the battery"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
status_cmd = "acpi"
|
|
defaults = [
|
|
('update_interval', 10, 'The update interval in seconds.'),
|
|
('text_pattern', "{percentage}%", 'The pattern for the text that is displayed.'),
|
|
('charging_color', "#ffffff", "Color when battery is charging"),
|
|
('normal_color', "#ffffff", "Color when value is normal"),
|
|
('warning_color', "#ffffff", "Color when value is warning"),
|
|
('critical_color', "#ffffff", "Color when value is critical"),
|
|
('battery', 0, 'Which battery should be monitored'),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(BatteryInfoWidget, self).__init__("BAT", "", bar.CALCULATED, **config)
|
|
self.add_defaults(BatteryInfoWidget.defaults)
|
|
self.text = "..."
|
|
self.state = 0
|
|
self.percentage = 0
|
|
self.power = 0
|
|
self.time = 0
|
|
self._battery = self._load_battery(**config)
|
|
|
|
@staticmethod
|
|
def _load_battery(**config):
|
|
return load_battery(**config)
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _update_values(self):
|
|
status = self._battery.update_status()
|
|
self.state = status.state
|
|
self.percentage = math.ceil(status.percent * 100)
|
|
self.power = status.power
|
|
self.time = status.time
|
|
|
|
def update(self):
|
|
self._update_values()
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
if self.state == BatteryState.EMPTY:
|
|
self.text_left = ""
|
|
self.text_right = self.text_pattern.format(percentage=0)
|
|
elif self.state == BatteryState.CHARGING:
|
|
self.text_left = ""
|
|
self.text_right = None
|
|
elif self.state == BatteryState.DISCHARGING:
|
|
capacity = ""
|
|
if self.percentage < 80:
|
|
capacity = ""
|
|
if self.percentage < 60:
|
|
capacity = ""
|
|
if self.percentage < 40:
|
|
capacity = ""
|
|
if self.percentage < 20:
|
|
capacity = ""
|
|
self.text_left = capacity
|
|
self.text_right = self.text_pattern.format(percentage=self.percentage)
|
|
elif self.state == BatteryState.FULL:
|
|
self.text_left = ""
|
|
self.text_right = self.text_pattern.format(percentage=100)
|
|
else:
|
|
self.text_left = ""
|
|
self.text_right = self.text_pattern.format(percentage=self.percentage)
|
|
super(BatteryInfoWidget, self).draw()
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
output = subprocess.check_output(self.status_cmd).decode('utf-8')
|
|
notify("Battery Status", output)
|
|
|
|
|
|
class VolumeInfoWidget(DualPaneTextboxBase):
|
|
"""Displays information about the volume"""
|
|
orientations = base.ORIENTATION_HORIZONTAL
|
|
defaults = [
|
|
('update_interval', 10, 'The update interval in seconds.'),
|
|
('text_pattern', "{percentage}%", 'The pattern for the text that is displayed.'),
|
|
('charging_color', "#ffffff", "Color when battery is charging"),
|
|
('normal_color', "#ffffff", "Color when value is normal"),
|
|
('warning_color', "#ffffff", "Color when value is warning"),
|
|
('critical_color', "#ffffff", "Color when value is critical"),
|
|
("pulse_sink", None, "PulseAudio sink name to control"),
|
|
("status_cmd", "pamixer{sink} --get-volume", "Command to get current volume"),
|
|
("mute_cmd", "pamixer{sink} -t", "Command to mute volume"),
|
|
("up_cmd", "pamixer{sink} -i 2", "Command to turn volume up"),
|
|
("down_cmd", "pamixer{sink} -d 2", "Command to turn volume down"),
|
|
("volume_app", "pavucontrol", "Volume mixer app to open on middle click"),
|
|
]
|
|
|
|
def __init__(self, **config):
|
|
super(VolumeInfoWidget, self).__init__("VOL", "", bar.CALCULATED, **config)
|
|
self.add_defaults(VolumeInfoWidget.defaults)
|
|
self.text = "..."
|
|
self.percentage = 0
|
|
self.volume = 0
|
|
|
|
def get_volume(self):
|
|
try:
|
|
if "{sink}" in self.status_cmd:
|
|
cmd = self.status_cmd.format(sink=" --sink {}".format(self.pulse_sink) if self.pulse_sink else "")
|
|
else:
|
|
cmd = self.status_cmd
|
|
mixer_out = self.call_process(cmd.split(" "))
|
|
except subprocess.CalledProcessError:
|
|
return -1
|
|
try:
|
|
return int(mixer_out)
|
|
except ValueError:
|
|
return -1
|
|
|
|
def timer_setup(self):
|
|
self.update()
|
|
self.timeout_add(self.update_interval, self.timer_setup)
|
|
|
|
def _update_values(self):
|
|
self.volume = self.get_volume()
|
|
|
|
def update(self):
|
|
self._update_values()
|
|
self.draw()
|
|
|
|
def draw(self):
|
|
mute = ""
|
|
volume = ""
|
|
if self.volume < 75:
|
|
volume = ""
|
|
if self.volume < 50:
|
|
volume = ""
|
|
if self.volume < 25:
|
|
volume = ""
|
|
if self.volume < 0:
|
|
self.text_left = mute
|
|
self.text_right = ""
|
|
else:
|
|
self.text_left = volume
|
|
self.text_right = self.text_pattern.format(percentage=self.volume)
|
|
super(VolumeInfoWidget, self).draw()
|
|
|
|
def button_press(self, x, y, button):
|
|
if button == BUTTON_LEFT:
|
|
if "{sink}" in self.status_cmd:
|
|
cmd = self.status_cmd.format(sink=" --sink {}".format(self.pulse_sink) if self.pulse_sink else "")
|
|
else:
|
|
cmd = self.status_cmd
|
|
output = subprocess.check_output(cmd.split(" ")).decode('utf-8')
|
|
|
|
sink = "Sink {}\n".format(self.pulse_sink) if self.pulse_sink else ""
|
|
notify("Volume Status", sink+output)
|
|
|
|
elif button == BUTTON_RIGHT:
|
|
if "{sink}" in self.volume_app:
|
|
cmd = self.volume_app.format(sink=" --sink {}".format(self.pulse_sink) if self.pulse_sink else "")
|
|
else:
|
|
cmd = self.volume_app
|
|
subprocess.Popen(cmd.split(" "))
|
|
|
|
elif button == BUTTON_DOWN:
|
|
if "{sink}" in self.down_cmd:
|
|
cmd = self.down_cmd.format(sink=" --sink {}".format(self.pulse_sink) if self.pulse_sink else "")
|
|
else:
|
|
cmd = self.down_cmd
|
|
subprocess.call(cmd.split(" "))
|
|
|
|
elif button == BUTTON_UP:
|
|
if "{sink}" in self.up_cmd:
|
|
cmd = self.up_cmd.format(sink=" --sink {}".format(self.pulse_sink) if self.pulse_sink else "")
|
|
else:
|
|
cmd = self.up_cmd
|
|
subprocess.call(cmd.split(" "))
|
|
|
|
elif button == BUTTON_MUTE:
|
|
if "{sink}" in self.mute_cmd:
|
|
cmd = self.mute_cmd.format(sink=" --sink {}".format(self.pulse_sink) if self.pulse_sink else "")
|
|
else:
|
|
cmd = self.mute_cmd
|
|
subprocess.call(cmd.split(" "))
|
|
|
|
self.update()
|
|
|
|
def cmd_increase_vol(self):
|
|
# Emulate button press.
|
|
self.button_press(0, 0, BUTTON_UP)
|
|
|
|
def cmd_decrease_vol(self):
|
|
# Emulate button press.
|
|
self.button_press(0, 0, BUTTON_DOWN)
|
|
|
|
def cmd_mute(self):
|
|
# Emulate button press.
|
|
self.button_press(0, 0, BUTTON_MUTE)
|
|
|
|
def cmd_run_app(self):
|
|
# Emulate button press.
|
|
self.button_press(0, 0, BUTTON_RIGHT)
|