dotfiles/config/lemonbar/updaters.py

271 lines
7.4 KiB
Python
Raw Normal View History

2018-09-05 09:07:37 +02:00
#!/usr/bin/env python3
import math
import functools
import threading
import pyinotify
import os
import time
2018-09-06 07:38:22 +02:00
import logging
import coloredlogs
import i3ipc
from display import Text
2018-09-05 09:07:37 +02:00
2021-06-13 11:49:21 +02:00
coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s")
2018-09-06 07:38:22 +02:00
log = logging.getLogger()
# TODO Sync bar update with PeriodicUpdater updates
2018-09-05 09:07:37 +02:00
2018-10-14 16:59:49 +02:00
notBusy = threading.Event()
2018-09-05 09:07:37 +02:00
2021-06-13 11:49:21 +02:00
2018-09-05 09:07:37 +02:00
class Updater:
@staticmethod
def init():
PeriodicUpdater.init()
InotifyUpdater.init()
2018-10-14 16:59:49 +02:00
notBusy.set()
2018-09-05 09:07:37 +02:00
def updateText(self, text):
print(text)
def fetcher(self):
return "{} refreshed".format(self)
2018-09-06 07:38:22 +02:00
def __init__(self):
self.lock = threading.Lock()
2018-09-05 09:07:37 +02:00
def refreshData(self):
2018-09-06 07:38:22 +02:00
# 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 = ""
2018-09-05 09:07:37 +02:00
self.updateText(data)
2018-09-06 07:38:22 +02:00
self.lock.release()
2018-09-05 09:07:37 +02:00
class PeriodicUpdaterThread(threading.Thread):
def run(self):
2018-09-06 07:38:22 +02:00
# TODO Sync with system clock
2018-09-05 09:07:37 +02:00
counter = 0
while True:
2018-10-14 16:59:49 +02:00
notBusy.set()
2021-06-13 11:49:21 +02:00
if PeriodicUpdater.intervalsChanged.wait(
timeout=PeriodicUpdater.intervalStep
):
2018-09-05 09:07:37 +02:00
# ↑ sleeps here
2018-10-14 16:59:49 +02:00
notBusy.clear()
2018-09-05 09:07:37 +02:00
PeriodicUpdater.intervalsChanged.clear()
counter = 0
for providerList in PeriodicUpdater.intervals.copy().values():
for provider in providerList.copy():
provider.refreshData()
else:
2018-10-14 16:59:49 +02:00
notBusy.clear()
2018-09-05 09:07:37 +02:00
counter += PeriodicUpdater.intervalStep
counter = counter % PeriodicUpdater.intervalLoop
for interval in PeriodicUpdater.intervals.keys():
if counter % interval == 0:
for provider in PeriodicUpdater.intervals[interval]:
provider.refreshData()
class PeriodicUpdater(Updater):
"""
Needs to call :func:`PeriodicUpdater.changeInterval` in `__init__`
"""
intervals = dict()
intervalStep = None
intervalLoop = None
updateThread = PeriodicUpdaterThread(daemon=True)
intervalsChanged = threading.Event()
@staticmethod
def gcds(*args):
return functools.reduce(math.gcd, args)
@staticmethod
def lcm(a, b):
"""Return lowest common multiple."""
return a * b // math.gcd(a, b)
@staticmethod
def lcms(*args):
"""Return lowest common multiple."""
return functools.reduce(PeriodicUpdater.lcm, args)
@staticmethod
def updateIntervals():
intervalsList = list(PeriodicUpdater.intervals.keys())
PeriodicUpdater.intervalStep = PeriodicUpdater.gcds(*intervalsList)
PeriodicUpdater.intervalLoop = PeriodicUpdater.lcms(*intervalsList)
PeriodicUpdater.intervalsChanged.set()
@staticmethod
def init():
PeriodicUpdater.updateThread.start()
def __init__(self):
Updater.__init__(self)
self.interval = None
def changeInterval(self, interval):
assert isinstance(interval, int)
if self.interval is not None:
PeriodicUpdater.intervals[self.interval].remove(self)
self.interval = interval
if interval not in PeriodicUpdater.intervals:
PeriodicUpdater.intervals[interval] = set()
PeriodicUpdater.intervals[interval].add(self)
PeriodicUpdater.updateIntervals()
class InotifyUpdaterEventHandler(pyinotify.ProcessEvent):
def process_default(self, event):
# DEBUG
# from pprint import pprint
# pprint(event.__dict__)
# return
assert event.path in InotifyUpdater.paths
if 0 in InotifyUpdater.paths[event.path]:
for provider in InotifyUpdater.paths[event.path][0]:
provider.refreshData()
if event.name in InotifyUpdater.paths[event.path]:
for provider in InotifyUpdater.paths[event.path][event.name]:
provider.refreshData()
class InotifyUpdater(Updater):
"""
Needs to call :func:`PeriodicUpdater.changeInterval` in `__init__`
"""
wm = pyinotify.WatchManager()
paths = dict()
@staticmethod
def init():
2021-06-13 11:49:21 +02:00
notifier = pyinotify.ThreadedNotifier(
InotifyUpdater.wm, InotifyUpdaterEventHandler()
)
2018-09-05 09:07:37 +02:00
notifier.start()
# TODO Mask for folders
MASK = pyinotify.IN_CREATE | pyinotify.IN_MODIFY | pyinotify.IN_DELETE
def addPath(self, path, refresh=True):
path = os.path.realpath(os.path.expanduser(path))
# Detect if file or folder
if os.path.isdir(path):
self.dirpath = path
# 0: Directory watcher
self.filename = 0
elif os.path.isfile(path):
self.dirpath = os.path.dirname(path)
self.filename = os.path.basename(path)
else:
2021-06-13 11:49:21 +02:00
raise FileNotFoundError("No such file or directory: '{}'".format(path))
2018-09-05 09:07:37 +02:00
# Register watch action
if self.dirpath not in InotifyUpdater.paths:
InotifyUpdater.paths[self.dirpath] = dict()
if self.filename not in InotifyUpdater.paths[self.dirpath]:
InotifyUpdater.paths[self.dirpath][self.filename] = set()
InotifyUpdater.paths[self.dirpath][self.filename].add(self)
# Add watch
InotifyUpdater.wm.add_watch(self.dirpath, InotifyUpdater.MASK)
if refresh:
self.refreshData()
class ThreadedUpdaterThread(threading.Thread):
def __init__(self, updater, *args, **kwargs):
self.updater = updater
threading.Thread.__init__(self, *args, **kwargs)
2018-10-06 10:27:36 +02:00
self.looping = True
2018-09-05 09:07:37 +02:00
def run(self):
2018-10-06 10:27:36 +02:00
try:
while self.looping:
self.updater.loop()
except BaseException as e:
log.error("Error with {}".format(self.updater))
log.error(e, exc_info=True)
self.updater.updateText("")
2018-09-05 09:07:37 +02:00
class ThreadedUpdater(Updater):
"""
Must implement loop(), and call start()
"""
def __init__(self):
2018-09-06 07:38:22 +02:00
Updater.__init__(self)
2018-09-05 09:07:37 +02:00
self.thread = ThreadedUpdaterThread(self, daemon=True)
def loop(self):
self.refreshData()
time.sleep(10)
def start(self):
self.thread.start()
2018-09-06 07:38:22 +02:00
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 = Text()
2018-09-06 07:38:22 +02:00
for updater in self.updaters:
text.append(self.texts[updater])
2018-09-06 22:07:32 +02:00
if not len(text):
return None
return text
def __init__(self, *args):
2018-09-06 07:38:22 +02:00
Updater.__init__(self)
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)
2021-06-13 11:49:21 +02:00
self.texts[updater] = ""