This commit is contained in:
Geoffrey Frogeye 2018-09-06 07:38:22 +02:00
parent 7987cdcaae
commit b39ce22885
5 changed files with 513 additions and 142 deletions

View file

@ -2,21 +2,45 @@
from providers import *
# TODO If multiple screen, expand the sections and share them
if __name__ == "__main__":
Bar.init()
Updater.init()
Bar.addSectionAll(I3Provider(), BarGroupType.LEFT)
WORKSPACE_THEME = 0
FOCUS_THEME = 3
URGENT_THEME = 1
# TODO CPU provider
# TODO RAM provider
# TODO Temperature provider
Bar.addSectionAll(I3WorkspacesProvider(theme=WORKSPACE_THEME, themeFocus=FOCUS_THEME, themeUrgent=URGENT_THEME, themeMode=URGENT_THEME), BarGroupType.LEFT)
# TODO Middle
Bar.addSectionAll(MpdProvider(theme=7), BarGroupType.LEFT)
# Bar.addSectionAll(I3WindowTitleProvider(), BarGroupType.LEFT)
SYSTEM_THEME = 2
DANGER_THEME = FOCUS_THEME
CRITICAL_THEME = URGENT_THEME
Bar.addSectionAll(CpuProvider(theme=SYSTEM_THEME, themeDanger=DANGER_THEME, themeCritical=CRITICAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(RamProvider(theme=SYSTEM_THEME, themeDanger=DANGER_THEME, themeCritical=CRITICAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(TemperatureProvider(theme=SYSTEM_THEME, themeDanger=DANGER_THEME, themeCritical=CRITICAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(BatteryProvider(theme=SYSTEM_THEME, themeDanger=DANGER_THEME, themeCritical=CRITICAL_THEME), BarGroupType.RIGHT)
# Peripherals
PERIPHERAL_THEME = 5
NETWORK_THEME = 4
# TODO Disk space provider
# TODO Screen (connected, autorandr configuration, bbswitch) provider
# TODO Unlocked keys provider
# TODO Mail provider
Bar.addSectionAll(TodoProvider(dir='~/.vdirsyncer/currentCalendars/'), BarGroupType.RIGHT)
Bar.addSectionAll(NetworkProvider(), BarGroupType.RIGHT)
Bar.addSectionAll(PulseaudioProvider(), BarGroupType.RIGHT)
Bar.addSectionAll(BatteryProvider(), BarGroupType.RIGHT)
Bar.addSectionAll(TimeProvider(), BarGroupType.RIGHT)
Bar.addSectionAll(PulseaudioProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(NetworkProvider(theme=NETWORK_THEME), BarGroupType.RIGHT)
# Personal
PERSONAL_THEME = 0
Bar.addSectionAll(KeystoreProvider(theme=PERSONAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(NotmuchUnreadProvider(dir='~/.mail/', theme=PERSONAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(TodoProvider(dir='~/.vdirsyncer/currentCalendars/', theme=PERSONAL_THEME), BarGroupType.RIGHT)
TIME_THEME = 6
Bar.addSectionAll(TimeProvider(theme=TIME_THEME), BarGroupType.RIGHT)

View file

@ -13,6 +13,7 @@ import subprocess
# TODO Use bytes rather than strings
# TODO Use default colors of lemonbar sometimes
# TODO Mouse actions
# TODO Adapt bar height with font height
class BarGroupType(enum.Enum):
@ -30,6 +31,7 @@ class Bar:
# Constants
FONTS = ["DejaVu Sans Mono for Powerline", "Font Awesome"]
FONTSIZE = 10
@staticmethod
def init():
@ -37,7 +39,7 @@ class Bar:
cmd = ['lemonbar', '-b']
for font in Bar.FONTS:
cmd += ["-f", font]
cmd += ["-f", "{}:size={}".format(font, Bar.FONTSIZE)]
Bar.process = subprocess.Popen(cmd, stdin=subprocess.PIPE)
# Debug
@ -146,11 +148,15 @@ class BarGroup:
@staticmethod
def fgColor(color):
return "%{F" + color + "}"
return "%{F" + (color or '-') + "}"
@staticmethod
def bgColor(color):
return "%{B" + color + "}"
return "%{B" + (color or '-') + "}"
@staticmethod
def underlineColor(color):
return "%{U" + (color or '-') + "}"
@staticmethod
def color(fg, bg):
@ -190,7 +196,7 @@ class BarGroup:
parts.append(sec)
# TODO Concatenate successive strings
# TODO OPTI Concatenate successive strings
self.parts = parts
if self.childsTextChanged or self.childsThemeChanged:
@ -249,6 +255,7 @@ class Section:
sizeChanging = set()
updateThread = SectionThread(daemon=True)
somethingChanged = threading.Event()
lastChosenTheme = 0
@staticmethod
def init():
@ -257,11 +264,15 @@ class Section:
Section.updateThread.start()
def __init__(self, theme=0):
def __init__(self, theme=None):
#: Displayed section
#: Note: A section can be empty and displayed!
self.visible = False
if theme is None:
theme = Section.lastChosenTheme
Section.lastChosenTheme = (Section.lastChosenTheme + 1) \
% len(Section.THEMES)
self.theme = theme
#: Displayed text
@ -302,17 +313,62 @@ class Section:
for parent in self.parents:
parent.childsTextChanged = True
def parseParts(self, parts, bit='', clo=''):
if isinstance(parts, str):
parts = [parts]
bits = []
clos = []
for part in parts:
if isinstance(part, str):
for char in part:
bit += char
bits.append(bit)
clos.append(clo)
bit = ''
elif isinstance(part, dict):
newBit = ''
newClo = ''
if 'fgColor' in part:
newBit = newBit + BarGroup.fgColor(part['fgColor'])
newClo = BarGroup.fgColor(Section.THEMES[self.theme][0]) + newClo
if 'bgColor' in part:
newBit = newBit + BarGroup.bgColor(part['bgColor'])
newClo = BarGroup.bgColor(Section.THEMES[self.theme][0]) + newClo
if 'underlineColor' in part:
newBit = newBit + BarGroup.underlineColor(part['underlineColor'])
newClo = BarGroup.underlineColor(None) + newClo
if 'underline' in part:
newBit = newBit + '%{+u}'
newClo = '%{-u}' + newClo
if 'overline' in part:
newBit = newBit + '%{+o}'
newClo = '%{-o}' + newClo
newBits, newClos = self.parseParts(part["cont"], bit=newBit, clo=clo+newClo)
bits += newBits
clos += newClos
bit += newClo
else:
raise RuntimeError()
return bits, clos
def updateText(self, text):
# TODO FEAT Actions
# TODO OPTI When srcSize == dstSize, maybe the bit array isn't
# needed
if len(text):
if isinstance(text, str):
# TODO OPTI This common case
text = [text]
text = [' '] + text + [' ']
raw = [(t if isinstance(t, str) else t['text']) for t in text]
self.dstSize = sum([len(t) for t in raw])
# TODO FEAT Handle colors
# TODO OPTI Not like that
self.dstText = ''.join(raw)
self.dstBits, self.dstClos = self.parseParts([' '] + text + [' '])
# TODO BUG Try this and fix some closings that aren't done
# self.dstBits, self.dstClos = self.parseParts(text)
# TODO FEAT Half-spaces
self.dstText = ''.join(self.dstBits)
self.dstSize = len(self.dstBits)
else:
self.dstSize = 0
@ -340,7 +396,7 @@ class Section:
@staticmethod
def fit(text, size):
t = len(text)
return text[:size] if t >= size else text + " " * (size - t)
return text[:size] if t >= size else text + [" "] * (size - t)
def update(self):
# TODO Might profit of a better logic
@ -359,7 +415,9 @@ class Section:
Section.sizeChanging.remove(self)
return
self.curText = Section.fit(self.dstText, self.curSize)
closPos = self.curSize-1
clos = self.dstClos[closPos] if closPos < len(self.dstClos) else ''
self.curText = ''.join(Section.fit(self.dstBits, self.curSize)) + clos
self.informParentsTextChanged()
@staticmethod
@ -377,8 +435,20 @@ class Section:
@staticmethod
def ramp(p, ramp=" ▁▂▃▄▅▆▇█"):
if p > 1:
return ramp[-1]
elif p < 0:
return ramp[0]
else:
return ramp[round(p * (len(ramp)-1))]
class StatefulSection(Section):
# TODO Next thing to do
# TODO Allow to temporary expand the section (e.g. when important change)
pass
if __name__ == '__main__':
Bar.init()
sec = Section(0)

View file

@ -1,29 +0,0 @@
#!/usr/bin/env python3
import psutil
import subprocess
netStats = psutil.net_if_stats()
netAddrs = psutil.net_if_addrs()
netIO = psutil.net_io_counters(pernic=True)
for iface in netStats.keys():
if not netStats[iface].isup or iface.startswith('lo'):
continue
ssid = ''
if iface.startswith('eth') or iface.startswith('enp'):
icon = 'E'
elif iface.startswith('wlan') or iface.startswith('wlp'):
icon = 'W'
cmd = ["iwgetid", iface, "--raw"]
p = subprocess.run(cmd, stdout=subprocess.PIPE)
ssid = p.stdout.strip().decode()
# TODO Real icons
# TODO USB tethering
# TODO tan / tun
else:
icon = '?'
print(icon, iface, ssid)
print(netIO[iface])

View file

@ -11,8 +11,9 @@ import ipaddress
import logging
import coloredlogs
import json
import i3ipc
from pprint import pprint
import notmuch
import mpd
import random
coloredlogs.install(level='DEBUG', fmt='%(levelname)s %(message)s')
log = logging.getLogger()
@ -31,53 +32,125 @@ def humanSize(num):
num /= 1024.0
return "{:d}YiB".format(num)
def randomColor(seed=0):
random.seed(seed)
return '#{:02x}{:02x}{:02x}'.format(*[random.randint(0, 255) for _ in range(3)])
class TimeProvider(Section, PeriodicUpdater):
def fetcher(self):
now = datetime.datetime.now()
return now.strftime('%d/%m/%y %H:%M:%S')
def __init__(self):
def __init__(self, theme=None):
PeriodicUpdater.__init__(self)
Section.__init__(self)
Section.__init__(self, theme)
self.changeInterval(1)
class CpuProvider(Section, PeriodicUpdater):
def fetcher(self):
percents = psutil.cpu_percent(percpu=True)
percent = psutil.cpu_percent(percpu=False)
theme = self.themeCritical if percent >= 90 else \
(self.themeDanger if percent >= 75 else self.themeNormal)
self.updateTheme(theme)
return '' + ''.join([Section.ramp(p/100) for p in percents])
def __init__(self, theme=None, themeDanger=None, themeCritical=None):
PeriodicUpdater.__init__(self)
Section.__init__(self, theme)
self.themeNormal = theme or self.theme
self.themeDanger = themeDanger or self.theme
self.themeCritical = themeCritical or self.theme
self.changeInterval(1)
class RamProvider(Section, PeriodicUpdater):
"""
Shows free RAM
"""
def fetcher(self):
mem = psutil.virtual_memory()
ramp = Section.ramp(1-mem.percent/100)
free = humanSize(mem.available)
theme = self.themeCritical if mem.percent >= 90 else \
(self.themeDanger if mem.percent >= 75 else self.themeNormal)
self.updateTheme(theme)
return '{}{}'.format(ramp, free)
def __init__(self, theme=None, themeDanger=None, themeCritical=None):
PeriodicUpdater.__init__(self)
Section.__init__(self, theme)
self.themeNormal = theme or self.theme
self.themeDanger = themeDanger or self.theme
self.themeCritical = themeCritical or self.theme
self.changeInterval(1)
class TemperatureProvider(Section, PeriodicUpdater):
# TODO FEAT Change color when > high > critical
RAMP = ""
def fetcher(self):
allTemp = psutil.sensors_temperatures()
if 'coretemp' not in allTemp:
# TODO Opti Remove interval
return ''
temp = allTemp['coretemp'][0]
theme = self.themeCritical if temp.current >= temp.critical else \
(self.themeDanger if temp.current >= temp.high
else self.themeNormal)
self.updateTheme(theme)
ramp = Section.ramp(temp.current/temp.high, self.RAMP)
return '{} {:.0f}°C'.format(ramp, temp.current)
def __init__(self, theme=None, themeDanger=None, themeCritical=None):
PeriodicUpdater.__init__(self)
Section.__init__(self, theme=theme)
self.themeNormal = theme or self.theme
self.themeDanger = themeDanger or self.theme
self.themeCritical = themeCritical or self.theme
self.changeInterval(5)
class BatteryProvider(Section, PeriodicUpdater):
# TODO Change color when < thresold%
RAMP = ""
def fetcher(self):
with open(self.batdir + 'status') as f:
status = f.read().strip()
if status == "Full":
return ""
elif status == "Discharging":
icon = ""
elif status == "Charging":
icon = ""
else:
log.warn("Unknwon battery status: {}".format(status))
icon = "?"
with open(self.batdir + 'capacity') as f:
capacity = int(f.read())
icon += self.ramp(capacity/100, self.RAMP)
return '{} {}%'.format(icon, capacity)
bat = psutil.sensors_battery()
text = ''
if not bat:
return text
if bat.power_plugged:
text += ""
text += Section.ramp(bat.percent/100, self.RAMP)
if bat.percent < 100:
text += ' {:.0f}%'.format(bat.percent)
theme = self.themeCritical if bat.percent < 10 else \
(self.themeDanger if bat.percent < 25 else self.themeNormal)
self.updateTheme(theme)
return text
def __init__(self, battery='BAT0'):
def __init__(self, theme=None, themeDanger=None, themeCritical=None):
PeriodicUpdater.__init__(self)
Section.__init__(self)
self.batdir = '/sys/class/power_supply/{}/'.format(battery)
assert os.path.isdir(self.batdir)
Section.__init__(self, theme)
self.themeNormal = theme or self.theme
self.themeDanger = themeDanger or self.theme
self.themeCritical = themeCritical or self.theme
self.changeInterval(5)
class PulseaudioProvider(Section, ThreadedUpdater):
def __init__(self):
def __init__(self, theme=None):
ThreadedUpdater.__init__(self)
Section.__init__(self)
Section.__init__(self, theme)
self.pulseEvents = pulsectl.Pulse('event-handler')
self.pulseEvents.event_mask_set(pulsectl.PulseEventMaskEnum.sink)
@ -111,18 +184,27 @@ class PulseaudioProvider(Section, ThreadedUpdater):
def handleEvent(self, ev):
self.refreshData()
class NetworkProviderSection(Section, Updater):
THEME = 5
def fetcher(self):
text = ''
def getIcon(self):
if self.iface.startswith('eth') or self.iface.startswith('enp'):
if 'u' in self.iface:
return ['']
else:
return ['']
elif self.iface.startswith('wlan') or self.iface.startswith('wlp'):
cmd = ["iwgetid", self.iface, "--raw"]
p = subprocess.run(cmd, stdout=subprocess.PIPE)
if self.iface not in self.parent.stats or \
not self.parent.stats[self.iface].isup or \
self.iface.startswith('lo'):
return text
ssid = p.stdout.strip().decode()
return ['{}'.format(ssid)]
elif self.iface.startswith('tun') or self.iface.startswith('tap'):
return ['']
else:
return ['?']
# Get addresses
def getAddresses(self):
ipv4 = None
ipv6 = None
for address in self.parent.addrs[self.iface]:
@ -130,31 +212,29 @@ class NetworkProviderSection(Section, Updater):
ipv4 = address
elif address.family == socket.AF_INET6:
ipv6 = address
return ipv4, ipv6
def fetcher(self):
text = []
if self.iface not in self.parent.stats or \
not self.parent.stats[self.iface].isup or \
self.iface.startswith('lo'):
return text
# Get addresses
ipv4, ipv6 = self.getAddresses()
if ipv4 is None and ipv6 is None:
return text
# Set icon
if self.iface.startswith('eth') or self.iface.startswith('enp'):
if 'u' in self.iface:
text = ''
else:
text = ''
elif self.iface.startswith('wlan') or self.iface.startswith('wlp'):
text = ''
cmd = ["iwgetid", self.iface, "--raw"]
p = subprocess.run(cmd, stdout=subprocess.PIPE)
text += ' ' + p.stdout.strip().decode()
elif self.iface.startswith('tun') or self.iface.startswith('tap'):
text = ''
else:
text = '?'
text = self.getIcon()
if self.showAddress:
if ipv4:
netStrFull = '{}/{}'.format(ipv4.address, ipv4.netmask)
addr = ipaddress.IPv4Network(netStrFull, strict=False)
text += ' {}/{}'.format(ipv4.address, addr.prefixlen)
addrStr = ' {}/{}'.format(ipv4.address, addr.prefixlen)
text += [addrStr]
# TODO IPV6
# if ipv6:
# text += ' ' + ipv6.address
@ -166,12 +246,12 @@ class NetworkProviderSection(Section, Updater):
- self.parent.prevIO[self.iface].bytes_sent
recvDiff /= self.parent.dt
sentDiff /= self.parent.dt
text += '{}{}'.format(humanSize(recvDiff), humanSize(sentDiff))
text += ['{}{}'.format(humanSize(recvDiff), humanSize(sentDiff))]
if self.showTransfer:
text += '{}{}'.format(
text += ['{}{}'.format(
humanSize(self.parent.IO[self.iface].bytes_recv),
humanSize(self.parent.IO[self.iface].bytes_sent))
humanSize(self.parent.IO[self.iface].bytes_sent))]
return text
@ -188,7 +268,8 @@ class NetworkProviderSection(Section, Updater):
self.showTransfer = state >= 3
def __init__(self, iface, parent):
Section.__init__(self, theme=self.THEME)
Updater.__init__(self)
Section.__init__(self, theme=parent.theme)
self.iface = iface
self.parent = parent
self.changeState(1)
@ -208,9 +289,10 @@ class NetworkProvider(Section, PeriodicUpdater):
self.last = time.perf_counter()
self.dt = self.last - self.prev
def refreshData(self):
def fetcher(self):
self.fetchData()
# Add missing sections
lastSection = self
for iface in sorted(list(self.ifaces)):
if iface not in self.sections.keys():
@ -219,21 +301,117 @@ class NetworkProvider(Section, PeriodicUpdater):
self.sections[iface] = section
else:
section = self.sections[iface]
section.refreshData()
lastSection = section
# Refresh section text
for section in self.sections.values():
section.refreshData()
return ''
def addParent(self, parent):
self.parents.add(parent)
self.refreshData()
def __init__(self):
def __init__(self, theme=None):
PeriodicUpdater.__init__(self)
Section.__init__(self)
Section.__init__(self, theme)
self.sections = dict()
self.last = 0
self.IO = dict()
self.fetchData()
self.changeInterval(5)
class SshAgentProvider(PeriodicUpdater):
def fetcher(self):
cmd = ["ssh-add", "-l"]
proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
if proc.returncode != 0:
return ''
text = []
for line in proc.stdout.split(b'\n'):
if not len(line):
continue
fingerprint = line.split()[1]
text += [{"cont": '',
"fgColor": randomColor(seed=fingerprint)
}]
return text
def __init__(self):
PeriodicUpdater.__init__(self)
self.changeInterval(5)
class GpgAgentProvider(PeriodicUpdater):
def fetcher(self):
cmd = ["gpg-connect-agent", "keyinfo --list", "/bye"]
proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
# proc = subprocess.run(cmd)
if proc.returncode != 0:
return ''
text = []
for line in proc.stdout.split(b'\n'):
if not len(line) or line == b'OK':
continue
spli = line.split()
if spli[6] != b'1':
continue
keygrip = spli[2]
text += [{"cont": '',
"fgColor": randomColor(seed=keygrip)
}]
return text
def __init__(self):
PeriodicUpdater.__init__(self)
self.changeInterval(5)
class KeystoreProvider(Section, MergedUpdater):
def __init__(self, theme=None):
MergedUpdater.__init__(self, SshAgentProvider(), GpgAgentProvider(), prefix=[''])
Section.__init__(self, theme)
class NotmuchUnreadProvider(Section, PeriodicUpdater):
# TODO OPTI Transform InotifyUpdater (watching notmuch folder should be
# enough)
def fetcher(self):
db = notmuch.Database(mode=notmuch.Database.MODE.READ_ONLY, path=self.dir)
text = []
for account in self.accounts:
queryStr = 'folder:/{}/ and tag:unread'.format(account)
query = notmuch.Query(db, queryStr)
nbMsgs = query.count_messages()
if nbMsgs < 1:
continue
text += [' ']
text += [{"cont": str(nbMsgs), "fgColor": self.colors[account]}]
db.close()
if len(text):
return [''] + text
else:
return ''
def __init__(self, dir='~/.mail/', theme=None):
PeriodicUpdater.__init__(self)
Section.__init__(self, theme)
self.dir = os.path.realpath(os.path.expanduser(dir))
assert os.path.isdir(self.dir)
# Fetching account list
self.accounts = sorted([a for a in os.listdir(self.dir)
if not a.startswith('.')])
# Fetching colors
self.colors = dict()
for account in self.accounts:
filename = os.path.join(self.dir, account, 'color')
with open(filename, 'r') as f:
color = f.read().strip()
self.colors[account] = color
self.changeInterval(10)
class TodoProvider(Section, InotifyUpdater):
@ -248,12 +426,12 @@ class TodoProvider(Section, InotifyUpdater):
self.addPath(os.path.join(self.dir, calendar), refresh=False)
self.calendars = calendars
def __init__(self, dir):
def __init__(self, dir, theme=None):
"""
:parm str dir: [main]path value in todoman.conf
"""
InotifyUpdater.__init__(self)
Section.__init__(self)
Section.__init__(self, theme)
self.dir = os.path.realpath(os.path.expanduser(dir))
assert os.path.isdir(self.dir)
@ -286,13 +464,25 @@ class TodoProvider(Section, InotifyUpdater):
if c > 0:
color = self.getColor(calendar)
text += [' ']
text += [{"text": str(c), "fgColor": color}]
text += [{"cont": str(c), "fgColor": color}]
if len(text):
return [''] + text
else:
return ''
class I3Provider(Section, ThreadedUpdater):
class I3WindowTitleProvider(Section, I3Updater):
# TODO FEAT To make this available from start, we need to find the
# `focused=True` element following the `focus` array
# TODO Feat Make this output dependant if wanted
def on_window(self, i3, e):
self.updateText(e.container.name)
def __init__(self, theme=None):
I3Updater.__init__(self)
Section.__init__(self, theme=theme)
self.on("window", self.on_window)
class I3WorkspacesProvider(Section, I3Updater):
# TODO Multi-screen
THEME_NORMAL = 0
@ -319,18 +509,15 @@ class I3Provider(Section, ThreadedUpdater):
for workspace in workspaces:
# if parent.display != workspace["display"]:
# continue
theme = self.THEME_FOCUSED if workspace["focused"] \
else (self.THEME_URGENT if workspace["urgent"]
else self.THEME_NORMAL)
theme = self.themeFocus if workspace["focused"] \
else (self.themeUrgent if workspace["urgent"]
else self.theme)
section = Section(theme=theme)
parent.addSectionAfter(lastSection, section)
self.setName(section, workspace["name"])
self.sections[workspace["num"]] = section
lastSection = section
def on_workspace(self, i3, e):
print(304, e.change)
def on_workspace_init(self, i3, e):
workspace = e.current
i = workspace.num
@ -351,11 +538,11 @@ class I3Provider(Section, ThreadedUpdater):
self.setName(section, None)
def on_workspace_focus(self, i3, e):
self.sections[e.current.num].updateTheme(self.THEME_FOCUSED)
self.sections[e.old.num].updateTheme(self.THEME_NORMAL)
self.sections[e.current.num].updateTheme(self.themeFocus)
self.sections[e.old.num].updateTheme(self.theme)
def on_workspace_urgent(self, i3, e):
self.sections[e.current.num].updateTheme(self.THEME_URGENT)
self.sections[e.current.num].updateTheme(self.themeUrgent)
def on_workspace_rename(self, i3, e):
self.sections[e.current.num].updateText(e.name)
@ -370,28 +557,77 @@ class I3Provider(Section, ThreadedUpdater):
for section in self.sections.values():
section.updateText('')
def __init__(self):
ThreadedUpdater.__init__(self)
def __init__(self, theme=0, themeFocus=3, themeUrgent=1, themeMode=2):
I3Updater.__init__(self)
Section.__init__(self)
self.theme = theme
self.themeFocus = themeFocus
self.themeUrgent = themeUrgent
self.i3 = i3ipc.Connection()
self.sections = dict()
self.i3.on("workspace::init", self.on_workspace_init)
self.i3.on("workspace::focus", self.on_workspace_focus)
self.i3.on("workspace::empty", self.on_workspace_empty)
self.i3.on("workspace::urgent", self.on_workspace_urgent)
self.i3.on("workspace::rename", self.on_workspace_rename)
self.on("workspace::init", self.on_workspace_init)
self.on("workspace::focus", self.on_workspace_focus)
self.on("workspace::empty", self.on_workspace_empty)
self.on("workspace::urgent", self.on_workspace_urgent)
self.on("workspace::rename", self.on_workspace_rename)
# TODO Un-handled/tested: reload, rename, restored, move
self.i3.on("mode", self.on_mode)
self.modeSection = Section(theme=self.THEME_MODE)
self.start()
self.on("mode", self.on_mode)
self.modeSection = Section(theme=themeMode)
def addParent(self, parent):
self.parents.add(parent)
parent.addSection(self.modeSection)
self.initialPopulation(parent)
class MpdProvider(Section, ThreadedUpdater):
MAX_LENGTH = 50
def connect(self):
self.mpd.connect('localhost', 6600)
def __init__(self, theme=None):
ThreadedUpdater.__init__(self)
Section.__init__(self, theme)
self.mpd = mpd.MPDClient()
self.connect()
self.refreshData()
self.start()
def fetcher(self):
cur = self.mpd.currentsong()
if not len(cur):
return ''
infos = []
def tryAdd(field):
if field in cur:
infos.append(cur[field])
tryAdd("title")
tryAdd("album")
tryAdd("artist")
infosStr = " - ".join(infos)
if len(infosStr) > MpdProvider.MAX_LENGTH:
infosStr = infosStr[:MpdProvider.MAX_LENGTH-1] + ''
text = ["{}".format(infosStr)]
return text
def loop(self):
self.i3.main()
try:
self.mpd.idle('player')
self.refreshData()
except mpd.base.ConnectionError as e:
log.warn(e, exc_info=True)
self.connect()
except BaseException as e:
log.error(e, exc_info=True)

View file

@ -6,8 +6,14 @@ import threading
import pyinotify
import os
import time
import logging
import coloredlogs
import i3ipc
# TODO Multiple providers for the same section
coloredlogs.install(level='DEBUG', fmt='%(levelname)s %(message)s')
log = logging.getLogger()
# TODO Sync bar update with PeriodicUpdater updates
class Updater:
@ -22,13 +28,24 @@ class Updater:
def fetcher(self):
return "{} refreshed".format(self)
def __init__(self):
self.lock = threading.Lock()
def refreshData(self):
# TODO OPTI Maybe discard the refresh if there's already another one?
self.lock.acquire()
try:
data = self.fetcher()
except BaseException as e:
log.error(e, exc_info=True)
data = ""
self.updateText(data)
self.lock.release()
class PeriodicUpdaterThread(threading.Thread):
def run(self):
# TODO Sync with system clock
counter = 0
while True:
if PeriodicUpdater.intervalsChanged \
@ -184,6 +201,7 @@ class ThreadedUpdater(Updater):
"""
def __init__(self):
Updater.__init__(self)
self.thread = ThreadedUpdaterThread(self, daemon=True)
def loop(self):
@ -192,3 +210,55 @@ class ThreadedUpdater(Updater):
def start(self):
self.thread.start()
class I3Updater(ThreadedUpdater):
# TODO OPTI One i3 connection for all
def __init__(self):
ThreadedUpdater.__init__(self)
self.i3 = i3ipc.Connection()
self.start()
def on(self, event, function):
self.i3.on(event, function)
def loop(self):
self.i3.main()
class MergedUpdater(Updater):
# TODO OPTI Do not update until end of periodic batch
def fetcher(self):
text = []
for updater in self.updaters:
data = self.texts[updater]
if not len(data):
continue
if isinstance(data, str):
data = [data]
text += data
if not len(text):
return ''
return self.prefix + text + self.suffix
def __init__(self, *args, prefix=[''], suffix=['']):
Updater.__init__(self)
self.prefix = prefix
self.suffix = suffix
self.updaters = []
self.texts = dict()
for updater in args:
assert isinstance(updater, Updater)
def newUpdateText(updater, text):
self.texts[updater] = text
self.refreshData()
updater.updateText = newUpdateText.__get__(updater, Updater)
self.updaters.append(updater)
self.texts[updater] = ''