From e09774c4caac572b58041f391fca578c07885e27 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geoffrey=20=E2=80=9CFrogeye=E2=80=9D=20Preud=27homme?= Date: Tue, 18 Jun 2024 00:31:29 +0200 Subject: [PATCH] frobar: Type hint That was TEDIOUS --- hm/desktop/frobar/frobar/__init__.py | 49 +++--- hm/desktop/frobar/frobar/display.py | 231 ++++++++++++++------------ hm/desktop/frobar/frobar/providers.py | 201 +++++++++++----------- hm/desktop/frobar/frobar/updaters.py | 103 +++++------- 4 files changed, 298 insertions(+), 286 deletions(-) diff --git a/hm/desktop/frobar/frobar/__init__.py b/hm/desktop/frobar/frobar/__init__.py index e2848ba..8a76774 100644 --- a/hm/desktop/frobar/frobar/__init__.py +++ b/hm/desktop/frobar/frobar/__init__.py @@ -1,10 +1,14 @@ #!/usr/bin/env python3 -from frobar.providers import * + +from frobar import providers as fp +from frobar.display import Bar, BarGroupType +from frobar.updaters import Updater # TODO If multiple screen, expand the sections and share them # TODO Graceful exit -def run(): + +def run() -> None: Bar.init() Updater.init() @@ -19,7 +23,7 @@ def run(): full = short + " " + CUSTOM_SUFFIXES[i] customNames[short] = full Bar.addSectionAll( - I3WorkspacesProvider( + fp.I3WorkspacesProvider( theme=WORKSPACE_THEME, themeFocus=FOCUS_THEME, themeUrgent=URGENT_THEME, @@ -30,37 +34,40 @@ def run(): ) # TODO Middle - Bar.addSectionAll(MpdProvider(theme=9), BarGroupType.LEFT) + Bar.addSectionAll(fp.MpdProvider(theme=9), BarGroupType.LEFT) # Bar.addSectionAll(I3WindowTitleProvider(), BarGroupType.LEFT) # TODO Computer modes - SYSTEM_THEME = 3 - DANGER_THEME = 1 - CRITICAL_THEME = 0 - Bar.addSectionAll(CpuProvider(), BarGroupType.RIGHT) - Bar.addSectionAll(LoadProvider(), BarGroupType.RIGHT) - Bar.addSectionAll(RamProvider(), BarGroupType.RIGHT) - Bar.addSectionAll(TemperatureProvider(), BarGroupType.RIGHT) - Bar.addSectionAll(BatteryProvider(), BarGroupType.RIGHT) + Bar.addSectionAll(fp.CpuProvider(), BarGroupType.RIGHT) + Bar.addSectionAll(fp.LoadProvider(), BarGroupType.RIGHT) + Bar.addSectionAll(fp.RamProvider(), BarGroupType.RIGHT) + Bar.addSectionAll(fp.TemperatureProvider(), BarGroupType.RIGHT) + Bar.addSectionAll(fp.BatteryProvider(), BarGroupType.RIGHT) # Peripherals PERIPHERAL_THEME = 6 NETWORK_THEME = 5 # TODO Disk space provider # TODO Screen (connected, autorandr configuration, bbswitch) provider - Bar.addSectionAll(XautolockProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT) - Bar.addSectionAll(PulseaudioProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT) - Bar.addSectionAll(RfkillProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT) - Bar.addSectionAll(NetworkProvider(theme=NETWORK_THEME), BarGroupType.RIGHT) + Bar.addSectionAll(fp.XautolockProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT) + Bar.addSectionAll(fp.PulseaudioProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT) + Bar.addSectionAll(fp.RfkillProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT) + Bar.addSectionAll(fp.NetworkProvider(theme=NETWORK_THEME), BarGroupType.RIGHT) # Personal - PERSONAL_THEME = 7 - # 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) + # PERSONAL_THEME = 7 + # Bar.addSectionAll(fp.KeystoreProvider(theme=PERSONAL_THEME), BarGroupType.RIGHT) + # Bar.addSectionAll( + # fp.NotmuchUnreadProvider(dir="~/.mail/", theme=PERSONAL_THEME), + # BarGroupType.RIGHT, + # ) + # Bar.addSectionAll( + # fp.TodoProvider(dir="~/.vdirsyncer/currentCalendars/", theme=PERSONAL_THEME), + # BarGroupType.RIGHT, + # ) TIME_THEME = 4 - Bar.addSectionAll(TimeProvider(theme=TIME_THEME), BarGroupType.RIGHT) + Bar.addSectionAll(fp.TimeProvider(theme=TIME_THEME), BarGroupType.RIGHT) # Bar.run() diff --git a/hm/desktop/frobar/frobar/display.py b/hm/desktop/frobar/frobar/display.py index 76dd9cf..0896a71 100644 --- a/hm/desktop/frobar/frobar/display.py +++ b/hm/desktop/frobar/frobar/display.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python3 +#!/usr/bin/env python3init import enum import logging @@ -7,6 +7,7 @@ import signal import subprocess import threading import time +import typing import coloredlogs import i3ipc @@ -29,6 +30,12 @@ log = logging.getLogger() # TODO forceSize and changeText are different +Handle = typing.Callable[[], None] +Decorator = Handle | str | None +Element: typing.TypeAlias = typing.Union[str, "Text", None] +Part: typing.TypeAlias = typing.Union[str, "Text", "Section"] + + class BarGroupType(enum.Enum): LEFT = 0 RIGHT = 1 @@ -40,6 +47,7 @@ class BarGroupType(enum.Enum): class BarStdoutThread(threading.Thread): def run(self) -> None: while Bar.running: + assert Bar.process.stdout handle = Bar.process.stdout.readline().strip() if not len(handle): Bar.stop() @@ -70,8 +78,7 @@ class Bar: Bar.process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE ) - Bar.stdoutThread = BarStdoutThread() - Bar.stdoutThread.start() + BarStdoutThread().start() # Debug Bar(0) @@ -90,7 +97,7 @@ class Bar: Bar.forever() i3 = i3ipc.Connection() - def doStop(*args) -> None: + def doStop(*args: list) -> None: Bar.stop() print(88) @@ -102,17 +109,17 @@ class Bar: Bar.stop() # Class globals - everyone = set() + everyone: set["Bar"] = set() string = "" - process = None + process: subprocess.Popen running = False nextHandle = 0 - actionsF2H = dict() - actionsH2F = dict() + actionsF2H: dict[Handle, bytes] = dict() + actionsH2F: dict[bytes, Handle] = dict() @staticmethod - def getFunctionHandle(function): + def getFunctionHandle(function: typing.Callable[[], None]) -> bytes: assert callable(function) if function in Bar.actionsF2H.keys(): return Bar.actionsF2H[function] @@ -126,12 +133,11 @@ class Bar: return handle @staticmethod - def forever(): + def forever() -> None: Bar.process.wait() Bar.stop() - def __init__(self, screen): - assert isinstance(screen, int) + def __init__(self, screen: int) -> None: self.screen = "%{S" + str(screen) + "}" self.groups = dict() @@ -144,23 +150,21 @@ class Bar: self.everyone.add(self) @staticmethod - def addSectionAll(section, group, screens=None): + def addSectionAll( + section: "Section", group: "BarGroupType", screens: None = None + ) -> None: """ .. note:: Add the section before updating it for the first time. """ - assert isinstance(section, Section) - assert isinstance(group, BarGroupType) # TODO screens selection for bar in Bar.everyone: bar.addSection(section, group=group) - def addSection(self, section, group): - assert isinstance(section, Section) - assert isinstance(group, BarGroupType) + def addSection(self, section: "Section", group: "BarGroupType") -> None: self.groups[group].addSection(section) - def update(self): + def update(self) -> None: if self.childsChanged: self.string = self.screen self.string += self.groups[BarGroupType.LEFT].string @@ -169,7 +173,7 @@ class Bar: self.childsChanged = False @staticmethod - def updateAll(): + def updateAll() -> None: if Bar.running: Bar.string = "" for bar in Bar.everyone: @@ -179,6 +183,7 @@ class Bar: Bar.string += BarGroup.color(*Section.EMPTY) # print(Bar.string) + assert Bar.process.stdin Bar.process.stdin.write(bytes(Bar.string + "\n", "utf-8")) Bar.process.stdin.flush() @@ -188,18 +193,16 @@ class BarGroup: One for each group of each bar """ - everyone = set() + everyone: set["BarGroup"] = set() - def __init__(self, groupType, parent): - assert isinstance(groupType, BarGroupType) - assert isinstance(parent, Bar) + def __init__(self, groupType: BarGroupType, parent: Bar): self.groupType = groupType self.parent = parent - self.sections = list() + self.sections: list["Section"] = list() self.string = "" - self.parts = [] + self.parts: list[Part] = [] #: One of the sections that had their theme or visibility changed self.childsThemeChanged = False @@ -209,11 +212,11 @@ class BarGroup: BarGroup.everyone.add(self) - def addSection(self, section): + def addSection(self, section: "Section") -> None: self.sections.append(section) section.addParent(self) - def addSectionAfter(self, sectionRef, section): + def addSectionAfter(self, sectionRef: "Section", section: "Section") -> None: index = self.sections.index(sectionRef) self.sections.insert(index + 1, section) section.addParent(self) @@ -221,20 +224,20 @@ class BarGroup: ALIGNS = {BarGroupType.LEFT: "%{l}", BarGroupType.RIGHT: "%{r}"} @staticmethod - def fgColor(color): + def fgColor(color: str) -> str: return "%{F" + (color or "-") + "}" @staticmethod - def bgColor(color): + def bgColor(color: str) -> str: return "%{B" + (color or "-") + "}" @staticmethod - def color(fg, bg): + def color(fg: str, bg: str) -> str: return BarGroup.fgColor(fg) + BarGroup.bgColor(bg) - def update(self): + def update(self) -> None: if self.childsThemeChanged: - parts = [BarGroup.ALIGNS[self.groupType]] + parts: list[Part] = [BarGroup.ALIGNS[self.groupType]] secs = [sec for sec in self.sections if sec.visible] lenS = len(secs) @@ -283,7 +286,7 @@ class BarGroup: self.childsTextChanged = False @staticmethod - def updateAll(): + def updateAll() -> None: for group in BarGroup.everyone: group.update() Bar.updateAll() @@ -294,7 +297,7 @@ class SectionThread(threading.Thread): ANIMATION_STOP = 0.001 ANIMATION_EVOLUTION = 0.9 - def run(self): + def run(self) -> None: while Section.somethingChanged.wait(): notBusy.wait() Section.updateAll() @@ -311,6 +314,9 @@ class SectionThread(threading.Thread): animTime = self.ANIMATION_STOP +Theme = tuple[str, str] + + class Section: # TODO Update all of that to base16 COLORS = [ @@ -334,20 +340,20 @@ class Section: FGCOLOR = "#fff0f1" BGCOLOR = "#092c0e" - THEMES = list() - EMPTY = (FGCOLOR, BGCOLOR) + THEMES: list[Theme] = list() + EMPTY: Theme = (FGCOLOR, BGCOLOR) - ICON = None + ICON: str | None = None PERSISTENT = False #: Sections that do not have their destination size - sizeChanging = set() - updateThread = SectionThread(daemon=True) + sizeChanging: set["Section"] = set() + updateThread: threading.Thread = SectionThread(daemon=True) somethingChanged = threading.Event() lastChosenTheme = 0 @staticmethod - def init(): + def init() -> None: for t in range(8, 16): Section.THEMES.append((Section.COLORS[0], Section.COLORS[t])) Section.THEMES.append((Section.COLORS[0], Section.COLORS[3])) @@ -355,7 +361,7 @@ class Section: Section.updateThread.start() - def __init__(self, theme=None): + def __init__(self, theme: int | None = None) -> None: #: Displayed section #: Note: A section can be empty and displayed! self.visible = False @@ -378,12 +384,12 @@ class Section: self.dstSize = 0 #: Groups that have this section - self.parents = set() + self.parents: set[BarGroup] = set() self.icon = self.ICON self.persistent = self.PERSISTENT - def __str__(self): + def __str__(self) -> str: try: return "<{}><{}>{:01d}{}{:02d}/{:02d}".format( self.curText, @@ -393,26 +399,26 @@ class Section: self.curSize, self.dstSize, ) - except: + except Exception: return super().__str__() - def addParent(self, parent): + def addParent(self, parent: BarGroup) -> None: self.parents.add(parent) - def appendAfter(self, section): + def appendAfter(self, section: "Section") -> None: assert len(self.parents) for parent in self.parents: parent.addSectionAfter(self, section) - def informParentsThemeChanged(self): + def informParentsThemeChanged(self) -> None: for parent in self.parents: parent.childsThemeChanged = True - def informParentsTextChanged(self): + def informParentsTextChanged(self) -> None: for parent in self.parents: parent.childsTextChanged = True - def updateText(self, text): + def updateText(self, text: Element) -> None: if isinstance(text, str): text = Text(text) elif isinstance(text, Text) and not len(text.elements): @@ -439,14 +445,13 @@ class Section: Section.sizeChanging.add(self) Section.somethingChanged.set() - def setDecorators(self, **kwargs): + def setDecorators(self, **kwargs: Handle) -> None: self.dstText.setDecorators(**kwargs) self.curText = str(self.dstText) self.informParentsTextChanged() Section.somethingChanged.set() - def updateTheme(self, theme): - assert isinstance(theme, int) + def updateTheme(self, theme: int) -> None: assert theme < len(Section.THEMES) if theme == self.theme: return @@ -454,19 +459,18 @@ class Section: self.informParentsThemeChanged() Section.somethingChanged.set() - def updateVisibility(self, visibility): - assert isinstance(visibility, bool) + def updateVisibility(self, visibility: bool) -> None: self.visible = visibility self.informParentsThemeChanged() Section.somethingChanged.set() @staticmethod - def fit(text, size): + def fit(text: str, size: int) -> str: 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): + def update(self) -> None: # TODO Might profit of a better logic if not self.visible: self.updateVisibility(True) @@ -487,7 +491,7 @@ class Section: self.informParentsTextChanged() @staticmethod - def updateAll(): + def updateAll() -> None: """ Process all sections for text size changes """ @@ -500,7 +504,7 @@ class Section: Section.somethingChanged.clear() @staticmethod - def ramp(p, ramp=" ▁▂▃▄▅▆▇█"): + def ramp(p: float, ramp: str = " ▁▂▃▄▅▆▇█") -> str: if p > 1: return ramp[-1] elif p < 0: @@ -511,11 +515,11 @@ class Section: class StatefulSection(Section): # TODO FEAT Allow to temporary expand the section (e.g. when important change) - NUMBER_STATES = None + NUMBER_STATES: int DEFAULT_STATE = 0 - def __init__(self, *args, **kwargs): - Section.__init__(self, *args, **kwargs) + def __init__(self, theme: int | None) -> None: + Section.__init__(self, theme=theme) self.state = self.DEFAULT_STATE if hasattr(self, "onChangeState"): self.onChangeState(self.state) @@ -523,20 +527,22 @@ class StatefulSection(Section): clickLeft=self.incrementState, clickRight=self.decrementState ) - def incrementState(self): + def incrementState(self) -> None: newState = min(self.state + 1, self.NUMBER_STATES - 1) self.changeState(newState) - def decrementState(self): + def decrementState(self) -> None: newState = max(self.state - 1, 0) self.changeState(newState) - def changeState(self, state): - assert isinstance(state, int) + def changeState(self, state: int) -> None: assert state < self.NUMBER_STATES self.state = state if hasattr(self, "onChangeState"): self.onChangeState(state) + assert hasattr( + self, "refreshData" + ), "StatefulSection should be paired with some Updater" self.refreshData() @@ -547,10 +553,13 @@ class ColorCountsSection(StatefulSection): NUMBER_STATES = 3 COLORABLE_ICON = "?" - def __init__(self, theme=None): + def __init__(self, theme: None | int = None) -> None: StatefulSection.__init__(self, theme=theme) - def fetcher(self): + def subfetcher(self) -> list[tuple[int, str]]: + raise NotImplementedError("Interface must be implemented") + + def fetcher(self) -> typing.Union[None, "Text"]: counts = self.subfetcher() # Nothing if not len(counts): @@ -565,67 +574,66 @@ class ColorCountsSection(StatefulSection): # Icon + Total elif self.state == 1 and len(counts) > 1: total = sum([count for count, color in counts]) - return Text(self.COLORABLE_ICON, " ", total) + return Text(self.COLORABLE_ICON, " ", str(total)) # Icon + Counts else: text = Text(self.COLORABLE_ICON) for count, color in counts: - text.append(" ", Text(count, fg=color)) + text.append(" ", Text(str(count), fg=color)) return text class Text: - def _setElements(self, elements): - # TODO OPTI Concatenate consecutrive string - self.elements = list(elements) - - def _setDecorators(self, decorators): + def _setDecorators(self, decorators: dict[str, Decorator]) -> None: # TODO OPTI Convert no decorator to strings self.decorators = decorators - self.prefix = None - self.suffix = None + self.prefix: str | None = None + self.suffix: str | None = None + + def __init__(self, *args: Element, **kwargs: Decorator) -> None: + # TODO OPTI Concatenate consecutrive string + self.elements = list(args) - def __init__(self, *args, **kwargs): - self._setElements(args) self._setDecorators(kwargs) - self.section = None + self.section: Section - def append(self, *args): - self._setElements(self.elements + list(args)) + def append(self, *args: Element) -> None: + self.elements += list(args) - def prepend(self, *args): - self._setElements(list(args) + self.elements) + def prepend(self, *args: Element) -> None: + self.elements = list(args) + self.elements - def setElements(self, *args): - self._setElements(args) + def setElements(self, *args: Element) -> None: + self.elements = list(args) - def setDecorators(self, **kwargs): + def setDecorators(self, **kwargs: Decorator) -> None: self._setDecorators(kwargs) - def setSection(self, section): - assert isinstance(section, Section) + def setSection(self, section: Section) -> None: self.section = section for element in self.elements: if isinstance(element, Text): element.setSection(section) - def _genFixs(self): + def _genFixs(self) -> None: if self.prefix is not None and self.suffix is not None: return self.prefix = "" self.suffix = "" - def nest(prefix, suffix): + def nest(prefix: str, suffix: str) -> None: + assert self.prefix is not None + assert self.suffix is not None self.prefix = self.prefix + "%{" + prefix + "}" self.suffix = "%{" + suffix + "}" + self.suffix - def getColor(val): + def getColor(val: str) -> str: # TODO Allow themes - assert isinstance(val, str) and len(val) == 7 + assert len(val) == 7 return val - def button(number, function): + def button(number: str, function: Handle) -> None: handle = Bar.getFunctionHandle(function) nest("A" + number + ":" + handle.decode() + ":", "A" + number) @@ -634,25 +642,34 @@ class Text: continue if key == "fg": reset = self.section.THEMES[self.section.theme][0] + assert isinstance(val, str) nest("F" + getColor(val), "F" + reset) elif key == "bg": reset = self.section.THEMES[self.section.theme][1] + assert isinstance(val, str) nest("B" + getColor(val), "B" + reset) elif key == "clickLeft": + assert callable(val) button("1", val) elif key == "clickMiddle": + assert callable(val) button("2", val) elif key == "clickRight": + assert callable(val) button("3", val) elif key == "scrollUp": + assert callable(val) button("4", val) elif key == "scrollDown": + assert callable(val) button("5", val) else: log.warn("Unkown decorator: {}".format(key)) - def _text(self, size=None, pad=False): + def _text(self, size: int | None = None, pad: bool = False) -> tuple[str, int]: self._genFixs() + assert self.prefix is not None + assert self.suffix is not None curString = self.prefix curSize = 0 remSize = size @@ -678,9 +695,11 @@ class Text: curString += self.suffix - if pad and remSize > 0: - curString += " " * remSize - curSize += remSize + if pad: + assert remSize is not None + if remSize > 0: + curString += " " * remSize + curSize += remSize if size is not None: if pad: @@ -689,12 +708,14 @@ class Text: assert size >= curSize return curString, curSize - def text(self, *args, **kwargs): - string, size = self._text(*args, **kwargs) + def text(self, size: int | None = None, pad: bool = False) -> str: + string, size = self._text(size=size, pad=pad) return string - def __str__(self): + def __str__(self) -> str: self._genFixs() + assert self.prefix is not None + assert self.suffix is not None curString = self.prefix for element in self.elements: if element is None: @@ -704,7 +725,7 @@ class Text: curString += self.suffix return curString - def __len__(self): + def __len__(self) -> int: curSize = 0 for element in self.elements: if element is None: @@ -715,8 +736,8 @@ class Text: curSize += len(str(element)) return curSize - def __getitem__(self, index): + def __getitem__(self, index: int) -> Element: return self.elements[index] - def __setitem__(self, index, data): + def __setitem__(self, index: int, data: Element) -> None: self.elements[index] = data diff --git a/hm/desktop/frobar/frobar/providers.py b/hm/desktop/frobar/frobar/providers.py index 691ecb6..d2a040b 100644 --- a/hm/desktop/frobar/frobar/providers.py +++ b/hm/desktop/frobar/frobar/providers.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import datetime +import enum import ipaddress import json import logging @@ -8,15 +9,19 @@ import os import random import socket import subprocess +import time import coloredlogs +import i3ipc import mpd import notmuch import psutil import pulsectl -from frobar.display import * -from frobar.updaters import * +from frobar.display import (BarGroup, ColorCountsSection, Element, Section, + StatefulSection, Text) +from frobar.updaters import (I3Updater, InotifyUpdater, MergedUpdater, + PeriodicUpdater, ThreadedUpdater, Updater) coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s") log = logging.getLogger() @@ -25,7 +30,7 @@ log = logging.getLogger() # PulseaudioProvider and MpdProvider) -def humanSize(num): +def humanSize(num: int) -> str: """ Returns a string of width 3+3 """ @@ -35,11 +40,11 @@ def humanSize(num): return "{:3d}{}".format(int(num), unit) else: return "{:.1f}{}".format(num, unit) - num /= 1024.0 + num //= 1024 return "{:d}YiB".format(num) -def randomColor(seed=0): +def randomColor(seed: int | bytes = 0) -> str: random.seed(seed) return "#{:02x}{:02x}{:02x}".format(*[random.randint(0, 255) for _ in range(3)]) @@ -49,11 +54,11 @@ class TimeProvider(StatefulSection, PeriodicUpdater): NUMBER_STATES = len(FORMATS) DEFAULT_STATE = 1 - def fetcher(self): + def fetcher(self) -> str: now = datetime.datetime.now() return now.strftime(self.FORMATS[self.state]) - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): PeriodicUpdater.__init__(self) StatefulSection.__init__(self, theme) self.changeInterval(1) # TODO OPTI When state < 1 @@ -67,10 +72,10 @@ class AlertLevel(enum.Enum): class AlertingSection(StatefulSection): # TODO EASE Correct settings for themes - THEMES = {AlertLevel.NORMAL: 3, AlertLevel.WARNING: 1, AlertLevel.DANGER: 0} + ALERT_THEMES = {AlertLevel.NORMAL: 3, AlertLevel.WARNING: 1, AlertLevel.DANGER: 0} PERSISTENT = True - def getLevel(self, quantity): + def getLevel(self, quantity: float) -> AlertLevel: if quantity > self.dangerThresold: return AlertLevel.DANGER elif quantity > self.warningThresold: @@ -78,14 +83,14 @@ class AlertingSection(StatefulSection): else: return AlertLevel.NORMAL - def updateLevel(self, quantity): + def updateLevel(self, quantity: float) -> None: self.level = self.getLevel(quantity) - self.updateTheme(self.THEMES[self.level]) + self.updateTheme(self.ALERT_THEMES[self.level]) if self.level == AlertLevel.NORMAL: return # TODO Temporary update state - def __init__(self, theme): + def __init__(self, theme: int | None = None): StatefulSection.__init__(self, theme) self.dangerThresold = 0.90 self.warningThresold = 0.75 @@ -95,7 +100,7 @@ class CpuProvider(AlertingSection, PeriodicUpdater): NUMBER_STATES = 3 ICON = "" - def fetcher(self): + def fetcher(self) -> Element: percent = psutil.cpu_percent(percpu=False) self.updateLevel(percent / 100) if self.state >= 2: @@ -103,31 +108,35 @@ class CpuProvider(AlertingSection, PeriodicUpdater): return "".join([Section.ramp(p / 100) for p in percents]) elif self.state >= 1: return Section.ramp(percent / 100) + return "" - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): AlertingSection.__init__(self, theme) PeriodicUpdater.__init__(self) self.changeInterval(1) + class LoadProvider(AlertingSection, PeriodicUpdater): NUMBER_STATES = 3 ICON = "" - def fetcher(self): + def fetcher(self) -> Element: load = os.getloadavg() self.updateLevel(load[0]) if self.state >= 2: return " ".join(f"{load[i]:.2f}" for i in range(3)) elif self.state >= 1: return f"{load[0]:.2f}" + return "" - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): AlertingSection.__init__(self, theme) PeriodicUpdater.__init__(self) self.changeInterval(5) self.warningThresold = 5 self.dangerThresold = 10 + class RamProvider(AlertingSection, PeriodicUpdater): """ Shows free RAM @@ -136,7 +145,7 @@ class RamProvider(AlertingSection, PeriodicUpdater): NUMBER_STATES = 4 ICON = "" - def fetcher(self): + def fetcher(self) -> Element: mem = psutil.virtual_memory() freePerc = mem.percent / 100 self.updateLevel(freePerc) @@ -154,7 +163,7 @@ class RamProvider(AlertingSection, PeriodicUpdater): return text - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): AlertingSection.__init__(self, theme) PeriodicUpdater.__init__(self) self.changeInterval(1) @@ -166,7 +175,7 @@ class TemperatureProvider(AlertingSection, PeriodicUpdater): MAIN_TEMPS = ["coretemp", "amdgpu", "cpu_thermal"] # For Intel, AMD and ARM respectively. - def fetcher(self): + def fetcher(self) -> Element: allTemp = psutil.sensors_temperatures() for main in self.MAIN_TEMPS: if main in allTemp: @@ -182,8 +191,9 @@ class TemperatureProvider(AlertingSection, PeriodicUpdater): self.icon = Section.ramp(temp.current / temp.high, self.RAMP) if self.state >= 1: return "{:.0f}°C".format(temp.current) + return "" - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): AlertingSection.__init__(self, theme) PeriodicUpdater.__init__(self) self.changeInterval(5) @@ -194,10 +204,9 @@ class BatteryProvider(AlertingSection, PeriodicUpdater): NUMBER_STATES = 3 RAMP = "" - def fetcher(self): + def fetcher(self) -> Element: bat = psutil.sensors_battery() if not bat: - self.icon = None return None self.icon = ("" if bat.power_plugged else "") + Section.ramp( @@ -207,7 +216,7 @@ class BatteryProvider(AlertingSection, PeriodicUpdater): self.updateLevel(1 - bat.percent / 100) if self.state < 1: - return + return "" t = Text("{:.0f}%".format(bat.percent)) @@ -219,7 +228,7 @@ class BatteryProvider(AlertingSection, PeriodicUpdater): t.append(" ({:d}:{:02d})".format(h, m)) return t - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): AlertingSection.__init__(self, theme) PeriodicUpdater.__init__(self) self.changeInterval(5) @@ -238,7 +247,7 @@ class XautolockProvider(Section, InotifyUpdater): else: return "?" - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): Section.__init__(self, theme=theme) InotifyUpdater.__init__(self) # TODO XDG @@ -250,7 +259,7 @@ class PulseaudioProvider(StatefulSection, ThreadedUpdater): NUMBER_STATES = 3 DEFAULT_STATE = 1 - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): ThreadedUpdater.__init__(self) StatefulSection.__init__(self, theme) self.pulseEvents = pulsectl.Pulse("event-handler") @@ -260,7 +269,7 @@ class PulseaudioProvider(StatefulSection, ThreadedUpdater): self.start() self.refreshData() - def fetcher(self): + def fetcher(self) -> Element: sinks = [] with pulsectl.Pulse("list-sinks") as pulse: for sink in pulse.sink_list(): @@ -293,10 +302,10 @@ class PulseaudioProvider(StatefulSection, ThreadedUpdater): return Text(*sinks) - def loop(self): + def loop(self) -> None: self.pulseEvents.event_listen() - def handleEvent(self, ev): + def handleEvent(self, ev: pulsectl.PulseEventInfo) -> None: self.refreshData() @@ -304,7 +313,7 @@ class NetworkProviderSection(StatefulSection, Updater): NUMBER_STATES = 5 DEFAULT_STATE = 1 - def actType(self): + def actType(self) -> None: self.ssid = None if self.iface.startswith("eth") or self.iface.startswith("enp"): if "u" in self.iface: @@ -325,10 +334,10 @@ class NetworkProviderSection(StatefulSection, Updater): self.icon = "" elif self.iface.startswith("vboxnet"): self.icon = "" - else: - self.icon = "?" - def getAddresses(self): + def getAddresses( + self, + ) -> tuple[psutil._common.snicaddr, psutil._common.snicaddr]: ipv4 = None ipv6 = None for address in self.parent.addrs[self.iface]: @@ -338,8 +347,8 @@ class NetworkProviderSection(StatefulSection, Updater): ipv6 = address return ipv4, ipv6 - def fetcher(self): - self.icon = None + def fetcher(self) -> Element: + self.icon = "?" self.persistent = False if ( self.iface not in self.parent.stats @@ -393,13 +402,13 @@ class NetworkProviderSection(StatefulSection, Updater): return " ".join(text) - def onChangeState(self, state): + def onChangeState(self, state: int) -> None: self.showSsid = state >= 1 self.showAddress = state >= 2 self.showSpeed = state >= 3 self.showTransfer = state >= 4 - def __init__(self, iface, parent): + def __init__(self, iface: str, parent: "NetworkProvider"): Updater.__init__(self) StatefulSection.__init__(self, theme=parent.theme) self.iface = iface @@ -407,23 +416,23 @@ class NetworkProviderSection(StatefulSection, Updater): class NetworkProvider(Section, PeriodicUpdater): - def fetchData(self): + def fetchData(self) -> None: self.prev = self.last self.prevIO = self.IO self.stats = psutil.net_if_stats() - self.addrs = psutil.net_if_addrs() - self.IO = psutil.net_io_counters(pernic=True) + self.addrs: dict[str, list[psutil._common.snicaddr]] = psutil.net_if_addrs() + self.IO: dict[str, psutil._common.snetio] = psutil.net_io_counters(pernic=True) self.ifaces = self.stats.keys() - self.last = time.perf_counter() + self.last: float = time.perf_counter() self.dt = self.last - self.prev - def fetcher(self): + def fetcher(self) -> None: self.fetchData() # Add missing sections - lastSection = self + lastSection: NetworkProvider | NetworkProviderSection = self for iface in sorted(list(self.ifaces)): if iface not in self.sections.keys(): section = NetworkProviderSection(iface, self) @@ -439,15 +448,15 @@ class NetworkProvider(Section, PeriodicUpdater): return None - def addParent(self, parent): + def addParent(self, parent: BarGroup) -> None: self.parents.add(parent) self.refreshData() - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): PeriodicUpdater.__init__(self) Section.__init__(self, theme) - self.sections = dict() + self.sections: dict[str, NetworkProviderSection] = dict() self.last = 0 self.IO = dict() self.fetchData() @@ -459,7 +468,7 @@ class RfkillProvider(Section, PeriodicUpdater): # toggled PATH = "/sys/class/rfkill" - def fetcher(self): + def fetcher(self) -> Element: t = Text() for device in os.listdir(self.PATH): with open(os.path.join(self.PATH, device, "soft"), "rb") as f: @@ -473,7 +482,7 @@ class RfkillProvider(Section, PeriodicUpdater): with open(os.path.join(self.PATH, device, "type"), "rb") as f: typ = f.read().strip() - fg = (hardBlocked and "#CCCCCC") or (softBlocked and "#FF0000") + fg = (hardBlocked and "#CCCCCC") or (softBlocked and "#FF0000") or None if typ == b"wlan": icon = "" elif typ == b"bluetooth": @@ -484,14 +493,14 @@ class RfkillProvider(Section, PeriodicUpdater): t.append(Text(icon, fg=fg)) return t - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): PeriodicUpdater.__init__(self) Section.__init__(self, theme) self.changeInterval(5) class SshAgentProvider(PeriodicUpdater): - def fetcher(self): + def fetcher(self) -> Element: cmd = ["ssh-add", "-l"] proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) if proc.returncode != 0: @@ -504,13 +513,13 @@ class SshAgentProvider(PeriodicUpdater): text.append(Text("", fg=randomColor(seed=fingerprint))) return text - def __init__(self): + def __init__(self) -> None: PeriodicUpdater.__init__(self) self.changeInterval(5) class GpgAgentProvider(PeriodicUpdater): - def fetcher(self): + def fetcher(self) -> Element: cmd = ["gpg-connect-agent", "keyinfo --list", "/bye"] proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) # proc = subprocess.run(cmd) @@ -527,7 +536,7 @@ class GpgAgentProvider(PeriodicUpdater): text.append(Text("", fg=randomColor(seed=keygrip))) return text - def __init__(self): + def __init__(self) -> None: PeriodicUpdater.__init__(self) self.changeInterval(5) @@ -536,7 +545,7 @@ class KeystoreProvider(Section, MergedUpdater): # TODO OPTI+FEAT Use ColorCountsSection and not MergedUpdater, this is useless ICON = "" - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): MergedUpdater.__init__(self, SshAgentProvider(), GpgAgentProvider()) Section.__init__(self, theme) @@ -544,24 +553,21 @@ class KeystoreProvider(Section, MergedUpdater): class NotmuchUnreadProvider(ColorCountsSection, InotifyUpdater): COLORABLE_ICON = "" - def subfetcher(self): + def subfetcher(self) -> list[tuple[int, str]]: db = notmuch.Database(mode=notmuch.Database.MODE.READ_ONLY, path=self.dir) counts = [] for account in self.accounts: queryStr = "folder:/{}/ and tag:unread".format(account) query = notmuch.Query(db, queryStr) nbMsgs = query.count_messages() - if account == "frogeye": - global q - q = query if nbMsgs < 1: continue counts.append((nbMsgs, self.colors[account])) # db.close() return counts - def __init__(self, dir="~/.mail/", theme=None): - PeriodicUpdater.__init__(self) + def __init__(self, dir: str = "~/.mail/", theme: int | None = None): + InotifyUpdater.__init__(self) ColorCountsSection.__init__(self, theme) self.dir = os.path.realpath(os.path.expanduser(dir)) @@ -587,7 +593,7 @@ class TodoProvider(ColorCountsSection, InotifyUpdater): # TODO OPT Specific callback for specific directory COLORABLE_ICON = "" - def updateCalendarList(self): + def updateCalendarList(self) -> None: calendars = sorted(os.listdir(self.dir)) for calendar in calendars: # If the calendar wasn't in the list @@ -603,9 +609,9 @@ class TodoProvider(ColorCountsSection, InotifyUpdater): path = os.path.join(self.dir, calendar, "color") with open(path, "r") as f: self.colors[calendar] = f.read().strip() - self.calendars = calendars + self.calendars: list[str] = calendars - def __init__(self, dir, theme=None): + def __init__(self, dir: str, theme: int | None = None): """ :parm str dir: [main]path value in todoman.conf """ @@ -615,12 +621,12 @@ class TodoProvider(ColorCountsSection, InotifyUpdater): assert os.path.isdir(self.dir) self.calendars = [] - self.colors = dict() - self.names = dict() + self.colors: dict[str, str] = dict() + self.names: dict[str, str] = dict() self.updateCalendarList() self.refreshData() - def countUndone(self, calendar): + def countUndone(self, calendar: str | None) -> int: cmd = ["todo", "--porcelain", "list"] if calendar: cmd.append(self.names[calendar]) @@ -628,7 +634,7 @@ class TodoProvider(ColorCountsSection, InotifyUpdater): data = json.loads(proc.stdout) return len(data) - def subfetcher(self): + def subfetcher(self) -> list[tuple[int, str]]: counts = [] # TODO This an ugly optimisation that cuts on features, but todoman @@ -653,17 +659,17 @@ 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): + def on_window(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None: self.updateText(e.container.name) - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): I3Updater.__init__(self) Section.__init__(self, theme=theme) self.on("window", self.on_window) class I3WorkspacesProviderSection(Section): - def selectTheme(self): + def selectTheme(self) -> int: if self.urgent: return self.parent.themeUrgent elif self.focused: @@ -674,39 +680,39 @@ class I3WorkspacesProviderSection(Section): # TODO On mode change the state (shown / hidden) gets overriden so every # tab is shown - def show(self): + def show(self) -> None: self.updateTheme(self.selectTheme()) self.updateText(self.fullName if self.focused else self.shortName) - def changeState(self, focused, urgent): + def changeState(self, focused: bool, urgent: bool) -> None: self.focused = focused self.urgent = urgent self.show() - def setName(self, name): + def setName(self, name: str) -> None: self.shortName = name - self.fullName = ( + self.fullName: str = ( self.parent.customNames[name] if name in self.parent.customNames else name ) - def switchTo(self): + def switchTo(self) -> None: self.parent.i3.command("workspace {}".format(self.shortName)) - def __init__(self, name, parent): + def __init__(self, name: str, parent: "I3WorkspacesProvider"): Section.__init__(self) self.parent = parent self.setName(name) self.setDecorators(clickLeft=self.switchTo) - self.tempText = None + self.tempText: Element = None - def empty(self): + def empty(self) -> None: self.updateTheme(self.parent.themeNormal) self.updateText(None) - def tempShow(self): + def tempShow(self) -> None: self.updateText(self.tempText) - def tempEmpty(self): + def tempEmpty(self) -> None: self.tempText = self.dstText[1] self.updateText(None) @@ -714,7 +720,7 @@ class I3WorkspacesProviderSection(Section): class I3WorkspacesProvider(Section, I3Updater): # TODO FEAT Multi-screen - def initialPopulation(self, parent): + def initialPopulation(self, parent: BarGroup) -> None: """ Called on init Can't reuse addWorkspace since i3.get_workspaces() gives dict and not @@ -735,7 +741,7 @@ class I3WorkspacesProvider(Section, I3Updater): lastSection = section - def on_workspace_init(self, i3, e): + def on_workspace_init(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None: workspace = e.current i = workspace.num if i in self.sections: @@ -753,24 +759,24 @@ class I3WorkspacesProvider(Section, I3Updater): section.urgent = workspace.urgent section.show() - def on_workspace_empty(self, i3, e): + def on_workspace_empty(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None: self.sections[e.current.num].empty() - def on_workspace_focus(self, i3, e): + def on_workspace_focus(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None: self.sections[e.old.num].focused = False self.sections[e.old.num].show() self.sections[e.current.num].focused = True self.sections[e.current.num].show() - def on_workspace_urgent(self, i3, e): + def on_workspace_urgent(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None: self.sections[e.current.num].urgent = e.current.urgent self.sections[e.current.num].show() - def on_workspace_rename(self, i3, e): + def on_workspace_rename(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None: self.sections[e.current.num].setName(e.name) self.sections[e.current.num].show() - def on_mode(self, i3, e): + def on_mode(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None: if e.change == "default": self.modeSection.updateText(None) for section in self.sections.values(): @@ -781,7 +787,12 @@ class I3WorkspacesProvider(Section, I3Updater): section.tempEmpty() def __init__( - self, theme=0, themeFocus=3, themeUrgent=1, themeMode=2, customNames=dict() + self, + theme: int = 0, + themeFocus: int = 3, + themeUrgent: int = 1, + themeMode: int = 2, + customNames: dict[str, str] = dict(), ): I3Updater.__init__(self) Section.__init__(self) @@ -790,7 +801,7 @@ class I3WorkspacesProvider(Section, I3Updater): self.themeUrgent = themeUrgent self.customNames = customNames - self.sections = dict() + self.sections: dict[str, I3WorkspacesProviderSection] = dict() self.on("workspace::init", self.on_workspace_init) self.on("workspace::focus", self.on_workspace_focus) self.on("workspace::empty", self.on_workspace_empty) @@ -801,7 +812,7 @@ class I3WorkspacesProvider(Section, I3Updater): self.on("mode", self.on_mode) self.modeSection = Section(theme=themeMode) - def addParent(self, parent): + def addParent(self, parent: BarGroup) -> None: self.parents.add(parent) parent.addSection(self.modeSection) self.initialPopulation(parent) @@ -812,10 +823,10 @@ class MpdProvider(Section, ThreadedUpdater): MAX_LENGTH = 50 - def connect(self): + def connect(self) -> None: self.mpd.connect("localhost", 6600) - def __init__(self, theme=None): + def __init__(self, theme: int | None = None): ThreadedUpdater.__init__(self) Section.__init__(self, theme) @@ -824,7 +835,7 @@ class MpdProvider(Section, ThreadedUpdater): self.refreshData() self.start() - def fetcher(self): + def fetcher(self) -> Element: stat = self.mpd.status() if not len(stat) or stat["state"] == "stop": return None @@ -835,7 +846,7 @@ class MpdProvider(Section, ThreadedUpdater): infos = [] - def tryAdd(field): + def tryAdd(field: str) -> None: if field in cur: infos.append(cur[field]) @@ -849,7 +860,7 @@ class MpdProvider(Section, ThreadedUpdater): return " {}".format(infosStr) - def loop(self): + def loop(self) -> None: try: self.mpd.idle("player") self.refreshData() diff --git a/hm/desktop/frobar/frobar/updaters.py b/hm/desktop/frobar/frobar/updaters.py index 06b65f3..e2a735f 100644 --- a/hm/desktop/frobar/frobar/updaters.py +++ b/hm/desktop/frobar/frobar/updaters.py @@ -11,7 +11,7 @@ import coloredlogs import i3ipc import pyinotify -from frobar.display import Text +from frobar.display import Element from frobar.notbusy import notBusy coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s") @@ -20,24 +20,23 @@ log = logging.getLogger() # TODO Sync bar update with PeriodicUpdater updates - class Updater: @staticmethod - def init(): + def init() -> None: PeriodicUpdater.init() InotifyUpdater.init() notBusy.set() - def updateText(self, text): + def updateText(self, text: Element) -> None: print(text) - def fetcher(self): + def fetcher(self) -> Element: return "{} refreshed".format(self) - def __init__(self): + def __init__(self) -> None: self.lock = threading.Lock() - def refreshData(self): + def refreshData(self) -> None: # TODO OPTI Maybe discard the refresh if there's already another one? self.lock.acquire() try: @@ -50,7 +49,7 @@ class Updater: class PeriodicUpdaterThread(threading.Thread): - def run(self): + def run(self) -> None: # TODO Sync with system clock counter = 0 while True: @@ -67,6 +66,7 @@ class PeriodicUpdaterThread(threading.Thread): provider.refreshData() else: notBusy.clear() + assert PeriodicUpdater.intervalStep is not None counter += PeriodicUpdater.intervalStep counter = counter % PeriodicUpdater.intervalLoop for interval in PeriodicUpdater.intervals.keys(): @@ -80,43 +80,42 @@ class PeriodicUpdater(Updater): Needs to call :func:`PeriodicUpdater.changeInterval` in `__init__` """ - intervals = dict() - intervalStep = None - intervalLoop = None - updateThread = PeriodicUpdaterThread(daemon=True) + intervals: dict[int, set["PeriodicUpdater"]] = dict() + intervalStep: int | None = None + intervalLoop: int + updateThread: threading.Thread = PeriodicUpdaterThread(daemon=True) intervalsChanged = threading.Event() @staticmethod - def gcds(*args): + def gcds(*args: int) -> int: return functools.reduce(math.gcd, args) @staticmethod - def lcm(a, b): + def lcm(a: int, b: int) -> int: """Return lowest common multiple.""" return a * b // math.gcd(a, b) @staticmethod - def lcms(*args): + def lcms(*args: int) -> int: """Return lowest common multiple.""" return functools.reduce(PeriodicUpdater.lcm, args) @staticmethod - def updateIntervals(): + def updateIntervals() -> None: intervalsList = list(PeriodicUpdater.intervals.keys()) PeriodicUpdater.intervalStep = PeriodicUpdater.gcds(*intervalsList) PeriodicUpdater.intervalLoop = PeriodicUpdater.lcms(*intervalsList) PeriodicUpdater.intervalsChanged.set() @staticmethod - def init(): + def init() -> None: PeriodicUpdater.updateThread.start() - def __init__(self): + def __init__(self) -> None: Updater.__init__(self) - self.interval = None + self.interval: int | None = None - def changeInterval(self, interval): - assert isinstance(interval, int) + def changeInterval(self, interval: int) -> None: if self.interval is not None: PeriodicUpdater.intervals[self.interval].remove(self) @@ -131,7 +130,7 @@ class PeriodicUpdater(Updater): class InotifyUpdaterEventHandler(pyinotify.ProcessEvent): - def process_default(self, event): + def process_default(self, event: pyinotify.Event) -> None: # DEBUG # from pprint import pprint # pprint(event.__dict__) @@ -154,10 +153,10 @@ class InotifyUpdater(Updater): """ wm = pyinotify.WatchManager() - paths = dict() + paths: dict[str, dict[str | int, set["InotifyUpdater"]]] = dict() @staticmethod - def init(): + def init() -> None: notifier = pyinotify.ThreadedNotifier( InotifyUpdater.wm, InotifyUpdaterEventHandler() ) @@ -166,14 +165,14 @@ class InotifyUpdater(Updater): # TODO Mask for folders MASK = pyinotify.IN_CREATE | pyinotify.IN_MODIFY | pyinotify.IN_DELETE - def addPath(self, path, refresh=True): + def addPath(self, path: str, refresh: bool = True) -> None: path = os.path.realpath(os.path.expanduser(path)) # Detect if file or folder if os.path.isdir(path): - self.dirpath = path + self.dirpath: str = path # 0: Directory watcher - self.filename = 0 + self.filename: str | int = 0 elif os.path.isfile(path): self.dirpath = os.path.dirname(path) self.filename = os.path.basename(path) @@ -195,12 +194,12 @@ class InotifyUpdater(Updater): class ThreadedUpdaterThread(threading.Thread): - def __init__(self, updater, *args, **kwargs): + def __init__(self, updater: "ThreadedUpdater") -> None: self.updater = updater - threading.Thread.__init__(self, *args, **kwargs) + threading.Thread.__init__(self, daemon=True) self.looping = True - def run(self): + def run(self) -> None: try: while self.looping: self.updater.loop() @@ -215,57 +214,31 @@ class ThreadedUpdater(Updater): Must implement loop(), and call start() """ - def __init__(self): + def __init__(self) -> None: Updater.__init__(self) - self.thread = ThreadedUpdaterThread(self, daemon=True) + self.thread = ThreadedUpdaterThread(self) - def loop(self): + def loop(self) -> None: self.refreshData() time.sleep(10) - def start(self): + def start(self) -> None: self.thread.start() class I3Updater(ThreadedUpdater): # TODO OPTI One i3 connection for all - def __init__(self): + def __init__(self) -> None: ThreadedUpdater.__init__(self) self.i3 = i3ipc.Connection() + self.on = self.i3.on self.start() - def on(self, event, function): - self.i3.on(event, function) - - def loop(self): + def loop(self) -> None: self.i3.main() class MergedUpdater(Updater): - # TODO OPTI Do not update until end of periodic batch - def fetcher(self): - text = Text() - for updater in self.updaters: - text.append(self.texts[updater]) - if not len(text): - return None - return text - - def __init__(self, *args): - 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) - self.texts[updater] = "" + def __init__(self, *args: Updater) -> None: + raise NotImplementedError("Deprecated, as hacky and currently unused")