Compare commits
6 commits
2951280faa
...
445c2b8a99
Author | SHA1 | Date | |
---|---|---|---|
Geoffrey Frogeye | 445c2b8a99 | ||
Geoffrey Frogeye | e09774c4ca | ||
Geoffrey Frogeye | a489830949 | ||
Geoffrey Frogeye | 42034eb5d8 | ||
Geoffrey Frogeye | d6d3df65df | ||
Geoffrey Frogeye | a3fcaf9d27 |
|
@ -5,7 +5,7 @@
|
|||
./autorandr
|
||||
./background
|
||||
./browser
|
||||
./frobar
|
||||
./frobar/module.nix
|
||||
./i3.nix
|
||||
./lock
|
||||
./mpd
|
||||
|
|
|
@ -1,48 +1,31 @@
|
|||
{ pkgs ? import <nixpkgs> { config = { }; overlays = [ ]; }, lib, config, ... }:
|
||||
# Tried using pyproject.nix but mpd2 dependency wouldn't resolve,
|
||||
# is called pyton-mpd2 on PyPi but mpd2 in nixpkgs.
|
||||
{ pkgs ? import <nixpkgs> { config = { }; overlays = [ ]; }, ... }:
|
||||
let
|
||||
frobar = pkgs.python3Packages.buildPythonApplication {
|
||||
pname = "frobar";
|
||||
version = "2.0";
|
||||
|
||||
runtimeInputs = with pkgs; [ lemonbar-xft wirelesstools ];
|
||||
propagatedBuildInputs = with pkgs.python3Packages; [
|
||||
coloredlogs
|
||||
notmuch
|
||||
i3ipc
|
||||
mpd2
|
||||
psutil
|
||||
pulsectl
|
||||
pyinotify
|
||||
];
|
||||
makeWrapperArgs = [ "--prefix PATH : ${pkgs.lib.makeBinPath (with pkgs; [ lemonbar-xft wirelesstools ])}" ];
|
||||
|
||||
src = ./.;
|
||||
};
|
||||
lemonbar = (pkgs.lemonbar-xft.overrideAttrs (old: {
|
||||
src = pkgs.fetchFromGitHub {
|
||||
owner = "drscream";
|
||||
repo = "lemonbar-xft";
|
||||
rev = "a64a2a6a6d643f4d92f9d7600722710eebce7bdb";
|
||||
sha256 = "sha256-T5FhEPIiDt/9paJwL9Sj84CBtA0YFi1hZz0+87Hd6jU=";
|
||||
# https://github.com/drscream/lemonbar-xft/pull/2
|
||||
};
|
||||
}));
|
||||
in
|
||||
{
|
||||
config = lib.mkIf config.frogeye.desktop.xorg {
|
||||
xsession.windowManager.i3.config.bars = [ ];
|
||||
programs.autorandr.hooks.postswitch = {
|
||||
frobar = "${pkgs.systemd}/bin/systemctl --user restart frobar";
|
||||
};
|
||||
systemd.user.services.frobar = {
|
||||
Unit = {
|
||||
Description = "frobar";
|
||||
After = [ "graphical-session-pre.target" ];
|
||||
PartOf = [ "graphical-session.target" ];
|
||||
};
|
||||
# Tried using pyproject.nix but mpd2 dependency wouldn't resolve,
|
||||
# is called pyton-mpd2 on PyPi but mpd2 in nixpkgs.
|
||||
pkgs.python3Packages.buildPythonApplication {
|
||||
pname = "frobar";
|
||||
version = "2.0";
|
||||
|
||||
Service = {
|
||||
# Wait for i3 to start. Can't use ExecStartPre because otherwise it blocks graphical-session.target, and there's nothing i3/systemd
|
||||
# TODO Do that better
|
||||
ExecStart = ''${pkgs.bash}/bin/bash -c "while ! ${pkgs.i3}/bin/i3-msg; do ${pkgs.coreutils}/bin/sleep 1; done; ${frobar}/bin/frobar"'';
|
||||
};
|
||||
propagatedBuildInputs = with pkgs.python3Packages; [
|
||||
coloredlogs
|
||||
notmuch
|
||||
i3ipc
|
||||
mpd2
|
||||
psutil
|
||||
pulsectl
|
||||
pyinotify
|
||||
];
|
||||
makeWrapperArgs = [ "--prefix PATH : ${pkgs.lib.makeBinPath ([ lemonbar ] ++ (with pkgs; [ wirelesstools ]))}" ];
|
||||
|
||||
Install = { WantedBy = [ "graphical-session.target" ]; };
|
||||
};
|
||||
};
|
||||
src = ./.;
|
||||
}
|
||||
# TODO Connection with i3 is lost on start sometimes, more often than with Arch?
|
||||
# TODO Restore ability to build frobar with nix-build
|
||||
|
|
|
@ -1,13 +1,20 @@
|
|||
#!/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()
|
||||
|
||||
# Bar.addSectionAll(fp.CpuProvider(), BarGroupType.RIGHT)
|
||||
# Bar.addSectionAll(fp.NetworkProvider(theme=2), BarGroupType.RIGHT)
|
||||
|
||||
WORKSPACE_THEME = 8
|
||||
FOCUS_THEME = 2
|
||||
URGENT_THEME = 0
|
||||
|
@ -19,7 +26,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,35 +37,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(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(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()
|
||||
|
|
|
@ -1,4 +1,4 @@
|
|||
#!/usr/bin/env python3
|
||||
#!/usr/bin/env python3init
|
||||
|
||||
import enum
|
||||
import logging
|
||||
|
@ -7,11 +7,12 @@ import signal
|
|||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
import typing
|
||||
|
||||
import coloredlogs
|
||||
import i3ipc
|
||||
|
||||
from frobar.notbusy import notBusy
|
||||
from frobar.common import notBusy
|
||||
|
||||
coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s")
|
||||
log = logging.getLogger()
|
||||
|
@ -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()
|
||||
|
@ -62,20 +70,31 @@ class Bar:
|
|||
@staticmethod
|
||||
def init() -> None:
|
||||
Bar.running = True
|
||||
Bar.everyone = set()
|
||||
Section.init()
|
||||
|
||||
cmd = ["lemonbar", "-b", "-a", "64"]
|
||||
cmd = [
|
||||
"lemonbar",
|
||||
"-b",
|
||||
"-a",
|
||||
"64",
|
||||
"-F",
|
||||
Section.FGCOLOR,
|
||||
"-B",
|
||||
Section.BGCOLOR,
|
||||
]
|
||||
for font in Bar.FONTS:
|
||||
cmd += ["-f", "{}:size={}".format(font, Bar.FONTSIZE)]
|
||||
Bar.process = subprocess.Popen(
|
||||
cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE
|
||||
)
|
||||
Bar.stdoutThread = BarStdoutThread()
|
||||
Bar.stdoutThread.start()
|
||||
BarStdoutThread().start()
|
||||
|
||||
# Debug
|
||||
Bar(0)
|
||||
# Bar(1)
|
||||
i3 = i3ipc.Connection()
|
||||
for output in i3.get_outputs():
|
||||
if not output.active:
|
||||
continue
|
||||
Bar(output.name)
|
||||
|
||||
@staticmethod
|
||||
def stop() -> None:
|
||||
|
@ -90,29 +109,27 @@ class Bar:
|
|||
Bar.forever()
|
||||
i3 = i3ipc.Connection()
|
||||
|
||||
def doStop(*args) -> None:
|
||||
def doStop(*args: list) -> None:
|
||||
Bar.stop()
|
||||
print(88)
|
||||
|
||||
try:
|
||||
i3.on("ipc_shutdown", doStop)
|
||||
i3.main()
|
||||
except BaseException:
|
||||
print(93)
|
||||
Bar.stop()
|
||||
|
||||
# Class globals
|
||||
everyone = set()
|
||||
everyone: set["Bar"]
|
||||
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,13 +143,12 @@ class Bar:
|
|||
return handle
|
||||
|
||||
@staticmethod
|
||||
def forever():
|
||||
def forever() -> None:
|
||||
Bar.process.wait()
|
||||
Bar.stop()
|
||||
|
||||
def __init__(self, screen):
|
||||
assert isinstance(screen, int)
|
||||
self.screen = "%{S" + str(screen) + "}"
|
||||
def __init__(self, output: str) -> None:
|
||||
self.output = output
|
||||
self.groups = dict()
|
||||
|
||||
for groupType in BarGroupType:
|
||||
|
@ -140,36 +156,33 @@ class Bar:
|
|||
self.groups[groupType] = group
|
||||
|
||||
self.childsChanged = False
|
||||
|
||||
self.everyone.add(self)
|
||||
Bar.everyone.add(self)
|
||||
|
||||
@staticmethod
|
||||
def addSectionAll(section, group, screens=None):
|
||||
def addSectionAll(
|
||||
section: "Section", group: "BarGroupType"
|
||||
) -> 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)
|
||||
section.added()
|
||||
|
||||
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 = "%{Sn" + self.output + "}"
|
||||
self.string += self.groups[BarGroupType.LEFT].string
|
||||
self.string += self.groups[BarGroupType.RIGHT].string
|
||||
|
||||
self.childsChanged = False
|
||||
|
||||
@staticmethod
|
||||
def updateAll():
|
||||
def updateAll() -> None:
|
||||
if Bar.running:
|
||||
Bar.string = ""
|
||||
for bar in Bar.everyone:
|
||||
|
@ -178,8 +191,10 @@ class Bar:
|
|||
# Color for empty sections
|
||||
Bar.string += BarGroup.color(*Section.EMPTY)
|
||||
|
||||
# print(Bar.string)
|
||||
Bar.process.stdin.write(bytes(Bar.string + "\n", "utf-8"))
|
||||
string = Bar.string + "\n"
|
||||
# print(string)
|
||||
assert Bar.process.stdin
|
||||
Bar.process.stdin.write(string.encode())
|
||||
Bar.process.stdin.flush()
|
||||
|
||||
|
||||
|
@ -188,18 +203,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 +222,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 +234,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 +296,7 @@ class BarGroup:
|
|||
self.childsTextChanged = False
|
||||
|
||||
@staticmethod
|
||||
def updateAll():
|
||||
def updateAll() -> None:
|
||||
for group in BarGroup.everyone:
|
||||
group.update()
|
||||
Bar.updateAll()
|
||||
|
@ -294,7 +307,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 +324,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 +350,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 +371,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 +394,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 +409,29 @@ 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 added(self) -> None:
|
||||
pass
|
||||
|
||||
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 +458,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 +472,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 +504,7 @@ class Section:
|
|||
self.informParentsTextChanged()
|
||||
|
||||
@staticmethod
|
||||
def updateAll():
|
||||
def updateAll() -> None:
|
||||
"""
|
||||
Process all sections for text size changes
|
||||
"""
|
||||
|
@ -500,7 +517,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 +528,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 +540,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 +566,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 +587,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 +655,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 +708,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 +721,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 +738,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 +749,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
|
||||
|
|
|
@ -1,21 +1,27 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
import datetime
|
||||
import enum
|
||||
import ipaddress
|
||||
import json
|
||||
import logging
|
||||
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 (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()
|
||||
|
@ -24,7 +30,7 @@ log = logging.getLogger()
|
|||
# PulseaudioProvider and MpdProvider)
|
||||
|
||||
|
||||
def humanSize(num):
|
||||
def humanSize(num: int) -> str:
|
||||
"""
|
||||
Returns a string of width 3+3
|
||||
"""
|
||||
|
@ -34,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)])
|
||||
|
||||
|
@ -48,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
|
||||
|
@ -66,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:
|
||||
|
@ -77,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
|
||||
|
@ -92,9 +98,9 @@ class AlertingSection(StatefulSection):
|
|||
|
||||
class CpuProvider(AlertingSection, PeriodicUpdater):
|
||||
NUMBER_STATES = 3
|
||||
ICON = ""
|
||||
ICON = ""
|
||||
|
||||
def fetcher(self):
|
||||
def fetcher(self) -> Element:
|
||||
percent = psutil.cpu_percent(percpu=False)
|
||||
self.updateLevel(percent / 100)
|
||||
if self.state >= 2:
|
||||
|
@ -102,22 +108,44 @@ 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) -> 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: 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
|
||||
"""
|
||||
|
||||
NUMBER_STATES = 4
|
||||
ICON = ""
|
||||
ICON = ""
|
||||
|
||||
def fetcher(self):
|
||||
def fetcher(self) -> Element:
|
||||
mem = psutil.virtual_memory()
|
||||
freePerc = mem.percent / 100
|
||||
self.updateLevel(freePerc)
|
||||
|
@ -135,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)
|
||||
|
@ -144,23 +172,28 @@ class RamProvider(AlertingSection, PeriodicUpdater):
|
|||
class TemperatureProvider(AlertingSection, PeriodicUpdater):
|
||||
NUMBER_STATES = 2
|
||||
RAMP = ""
|
||||
MAIN_TEMPS = ["coretemp", "amdgpu", "cpu_thermal"]
|
||||
# For Intel, AMD and ARM respectively.
|
||||
|
||||
def fetcher(self):
|
||||
def fetcher(self) -> Element:
|
||||
allTemp = psutil.sensors_temperatures()
|
||||
if "coretemp" not in allTemp:
|
||||
# TODO Opti Remove interval
|
||||
return ""
|
||||
temp = allTemp["coretemp"][0]
|
||||
for main in self.MAIN_TEMPS:
|
||||
if main in allTemp:
|
||||
break
|
||||
else:
|
||||
return "?"
|
||||
temp = allTemp[main][0]
|
||||
|
||||
self.warningThresold = temp.high
|
||||
self.dangerThresold = temp.critical
|
||||
self.warningThresold = temp.high or 90.0
|
||||
self.dangerThresold = temp.critical or 100.0
|
||||
self.updateLevel(temp.current)
|
||||
|
||||
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)
|
||||
|
@ -171,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(
|
||||
|
@ -184,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))
|
||||
|
||||
|
@ -196,17 +228,38 @@ 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)
|
||||
|
||||
|
||||
class XautolockProvider(Section, InotifyUpdater):
|
||||
ICON = ""
|
||||
|
||||
def fetcher(self) -> str | None:
|
||||
with open(self.path) as fd:
|
||||
state = fd.read().strip()
|
||||
if state == "enabled":
|
||||
return None
|
||||
elif state == "disabled":
|
||||
return ""
|
||||
else:
|
||||
return "?"
|
||||
|
||||
def __init__(self, theme: int | None = None):
|
||||
Section.__init__(self, theme=theme)
|
||||
InotifyUpdater.__init__(self)
|
||||
# TODO XDG
|
||||
self.path = os.path.realpath(os.path.expanduser("~/.cache/xautolock"))
|
||||
self.addPath(self.path)
|
||||
|
||||
|
||||
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")
|
||||
|
@ -216,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():
|
||||
|
@ -249,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()
|
||||
|
||||
|
||||
|
@ -260,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:
|
||||
|
@ -281,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]:
|
||||
|
@ -294,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
|
||||
|
@ -349,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
|
||||
|
@ -363,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)
|
||||
|
@ -395,15 +448,11 @@ class NetworkProvider(Section, PeriodicUpdater):
|
|||
|
||||
return None
|
||||
|
||||
def addParent(self, parent):
|
||||
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()
|
||||
|
@ -415,7 +464,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:
|
||||
|
@ -429,7 +478,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":
|
||||
|
@ -440,14 +489,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:
|
||||
|
@ -460,13 +509,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)
|
||||
|
@ -483,7 +532,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)
|
||||
|
||||
|
@ -492,7 +541,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)
|
||||
|
||||
|
@ -500,24 +549,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))
|
||||
|
@ -543,7 +589,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
|
||||
|
@ -559,9 +605,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
|
||||
"""
|
||||
|
@ -571,12 +617,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])
|
||||
|
@ -584,7 +630,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
|
||||
|
@ -609,124 +655,107 @@ 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):
|
||||
if self.urgent:
|
||||
def selectTheme(self) -> int:
|
||||
if self.workspace.urgent:
|
||||
return self.parent.themeUrgent
|
||||
elif self.focused:
|
||||
elif self.workspace.focused:
|
||||
return self.parent.themeFocus
|
||||
elif self.workspace.visible:
|
||||
return self.parent.themeVisible
|
||||
else:
|
||||
return self.parent.themeNormal
|
||||
|
||||
# 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):
|
||||
self.focused = focused
|
||||
self.urgent = urgent
|
||||
self.show()
|
||||
|
||||
def setName(self, name):
|
||||
self.shortName = name
|
||||
self.fullName = (
|
||||
self.parent.customNames[name] if name in self.parent.customNames else name
|
||||
self.updateText(
|
||||
self.fullName if self.workspace.focused else self.workspace.name
|
||||
)
|
||||
|
||||
def switchTo(self):
|
||||
self.parent.i3.command("workspace {}".format(self.shortName))
|
||||
def switchTo(self) -> None:
|
||||
self.parent.i3.command("workspace {}".format(self.workspace.name))
|
||||
|
||||
def __init__(self, name, parent):
|
||||
def updateWorkspace(self, workspace: i3ipc.WorkspaceReply) -> None:
|
||||
self.workspace = workspace
|
||||
self.fullName: str = self.parent.customNames.get(workspace.name, workspace.name)
|
||||
self.show()
|
||||
|
||||
def __init__(self, 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)
|
||||
|
||||
|
||||
class I3WorkspacesProvider(Section, I3Updater):
|
||||
# TODO FEAT Multi-screen
|
||||
|
||||
def initialPopulation(self, parent):
|
||||
"""
|
||||
Called on init
|
||||
Can't reuse addWorkspace since i3.get_workspaces() gives dict and not
|
||||
ConObjects
|
||||
"""
|
||||
workspaces = self.i3.get_workspaces()
|
||||
lastSection = self.modeSection
|
||||
for workspace in workspaces:
|
||||
# if parent.display != workspace["display"]:
|
||||
# continue
|
||||
|
||||
section = I3WorkspacesProviderSection(workspace.name, self)
|
||||
section.focused = workspace.focused
|
||||
section.urgent = workspace.urgent
|
||||
section.show()
|
||||
parent.addSectionAfter(lastSection, section)
|
||||
self.sections[workspace.num] = section
|
||||
|
||||
lastSection = section
|
||||
|
||||
def on_workspace_init(self, i3, e):
|
||||
workspace = e.current
|
||||
i = workspace.num
|
||||
if i in self.sections:
|
||||
section = self.sections[i]
|
||||
def updateWorkspace(self, workspace: i3ipc.WorkspaceReply) -> None:
|
||||
section: Section | None = None
|
||||
lastSectionOnOutput = self.modeSection
|
||||
highestNumOnOutput = -1
|
||||
for sect in self.sections.values():
|
||||
if sect.workspace.num == workspace.num:
|
||||
section = sect
|
||||
break
|
||||
elif (
|
||||
sect.workspace.num > highestNumOnOutput
|
||||
and sect.workspace.num < workspace.num
|
||||
and sect.workspace.output == workspace.output
|
||||
):
|
||||
lastSectionOnOutput = sect
|
||||
highestNumOnOutput = sect.workspace.num
|
||||
else:
|
||||
# Find the section just before
|
||||
while i not in self.sections.keys() and i > 0:
|
||||
i -= 1
|
||||
prevSection = self.sections[i] if i != 0 else self.modeSection
|
||||
|
||||
section = I3WorkspacesProviderSection(workspace.name, self)
|
||||
prevSection.appendAfter(section)
|
||||
section = I3WorkspacesProviderSection(self)
|
||||
self.sections[workspace.num] = section
|
||||
section.focused = workspace.focused
|
||||
section.urgent = workspace.urgent
|
||||
section.show()
|
||||
|
||||
def on_workspace_empty(self, i3, e):
|
||||
for bargroup in self.parents:
|
||||
if bargroup.parent.output == workspace.output:
|
||||
break
|
||||
else:
|
||||
bargroup = list(self.parents)[0]
|
||||
bargroup.addSectionAfter(lastSectionOnOutput, section)
|
||||
section.updateWorkspace(workspace)
|
||||
|
||||
def updateWorkspaces(self) -> None:
|
||||
workspaces = self.i3.get_workspaces()
|
||||
for workspace in workspaces:
|
||||
self.updateWorkspace(workspace)
|
||||
|
||||
def added(self) -> None:
|
||||
super().added()
|
||||
self.appendAfter(self.modeSection)
|
||||
self.updateWorkspaces()
|
||||
|
||||
def on_workspace_change(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None:
|
||||
self.updateWorkspaces()
|
||||
|
||||
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):
|
||||
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):
|
||||
self.sections[e.current.num].urgent = e.current.urgent
|
||||
self.sections[e.current.num].show()
|
||||
|
||||
def on_workspace_rename(self, i3, e):
|
||||
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():
|
||||
|
@ -737,41 +766,46 @@ class I3WorkspacesProvider(Section, I3Updater):
|
|||
section.tempEmpty()
|
||||
|
||||
def __init__(
|
||||
self, theme=0, themeFocus=3, themeUrgent=1, themeMode=2, customNames=dict()
|
||||
self,
|
||||
theme: int = 0,
|
||||
themeVisible: int = 4,
|
||||
themeFocus: int = 3,
|
||||
themeUrgent: int = 1,
|
||||
themeMode: int = 2,
|
||||
customNames: dict[str, str] = dict(),
|
||||
):
|
||||
I3Updater.__init__(self)
|
||||
Section.__init__(self)
|
||||
self.themeNormal = theme
|
||||
self.themeFocus = themeFocus
|
||||
self.themeUrgent = themeUrgent
|
||||
self.themeVisible = themeVisible
|
||||
self.customNames = customNames
|
||||
|
||||
self.sections = dict()
|
||||
self.on("workspace::init", self.on_workspace_init)
|
||||
self.on("workspace::focus", self.on_workspace_focus)
|
||||
self.sections: dict[int, I3WorkspacesProviderSection] = dict()
|
||||
# The event object doesn't have the visible property,
|
||||
# so we have to fetch the list of workspaces anyways.
|
||||
# This sacrifices a bit of performance for code simplicity.
|
||||
self.on("workspace::init", self.on_workspace_change)
|
||||
self.on("workspace::focus", self.on_workspace_change)
|
||||
self.on("workspace::empty", self.on_workspace_empty)
|
||||
self.on("workspace::urgent", self.on_workspace_urgent)
|
||||
self.on("workspace::rename", self.on_workspace_rename)
|
||||
self.on("workspace::urgent", self.on_workspace_change)
|
||||
self.on("workspace::rename", self.on_workspace_change)
|
||||
# TODO Un-handled/tested: reload, rename, restored, move
|
||||
|
||||
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):
|
||||
# TODO FEAT More informations and controls
|
||||
|
||||
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)
|
||||
|
||||
|
@ -780,7 +814,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
|
||||
|
@ -791,7 +825,7 @@ class MpdProvider(Section, ThreadedUpdater):
|
|||
|
||||
infos = []
|
||||
|
||||
def tryAdd(field):
|
||||
def tryAdd(field: str) -> None:
|
||||
if field in cur:
|
||||
infos.append(cur[field])
|
||||
|
||||
|
@ -805,7 +839,7 @@ class MpdProvider(Section, ThreadedUpdater):
|
|||
|
||||
return " {}".format(infosStr)
|
||||
|
||||
def loop(self):
|
||||
def loop(self) -> None:
|
||||
try:
|
||||
self.mpd.idle("player")
|
||||
self.refreshData()
|
||||
|
|
|
@ -11,8 +11,8 @@ import coloredlogs
|
|||
import i3ipc
|
||||
import pyinotify
|
||||
|
||||
from frobar.display import Text
|
||||
from frobar.notbusy import notBusy
|
||||
from frobar.display import Element
|
||||
from frobar.common import notBusy
|
||||
|
||||
coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s")
|
||||
log = logging.getLogger()
|
||||
|
@ -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")
|
||||
|
|
25
hm/desktop/frobar/module.nix
Normal file
25
hm/desktop/frobar/module.nix
Normal file
|
@ -0,0 +1,25 @@
|
|||
{ pkgs, lib, config, ... }:
|
||||
{
|
||||
config = lib.mkIf config.frogeye.desktop.xorg {
|
||||
xsession.windowManager.i3.config.bars = [ ];
|
||||
programs.autorandr.hooks.postswitch = {
|
||||
frobar = "${pkgs.systemd}/bin/systemctl --user restart frobar";
|
||||
};
|
||||
systemd.user.services.frobar = {
|
||||
Unit = {
|
||||
Description = "frobar";
|
||||
After = [ "graphical-session-pre.target" ];
|
||||
PartOf = [ "graphical-session.target" ];
|
||||
};
|
||||
|
||||
Service = {
|
||||
# Wait for i3 to start. Can't use ExecStartPre because otherwise it blocks graphical-session.target, and there's nothing i3/systemd
|
||||
# TODO Do that better
|
||||
ExecStart = ''${pkgs.bash}/bin/bash -c "while ! ${pkgs.i3}/bin/i3-msg; do ${pkgs.coreutils}/bin/sleep 1; done; ${pkgs.callPackage ./. {}}/bin/frobar"'';
|
||||
};
|
||||
|
||||
Install = { WantedBy = [ "graphical-session.target" ]; };
|
||||
};
|
||||
};
|
||||
}
|
||||
# TODO Connection with i3 is lost on start sometimes, more often than with Arch?
|
|
@ -11,6 +11,7 @@ let
|
|||
'';
|
||||
lockPng = pkgs.runCommand "lock.png" { } "${pkgs.imagemagick}/bin/convert ${lockSvg} $out";
|
||||
mod = config.xsession.windowManager.i3.config.modifier;
|
||||
xautolockState = "${config.xdg.cacheHome}/xautolock";
|
||||
in
|
||||
{
|
||||
config = lib.mkIf config.frogeye.desktop.xorg {
|
||||
|
@ -43,12 +44,27 @@ in
|
|||
keybindings = {
|
||||
# Screen off commands
|
||||
"${mod}+F1" = "--release exec --no-startup-id ${pkgs.xorg.xset}/bin/xset dpms force off";
|
||||
"${mod}+F4" = "exec --no-startup-id ${pkgs.xautolock}/bin/xautolock -disable";
|
||||
"${mod}+F5" = "exec --no-startup-id ${pkgs.xautolock}/bin/xautolock -enable";
|
||||
# Toggle to save on buttons
|
||||
# xautolock -toggle doesn't allow to read state.
|
||||
# Writing into a file also allows frobar to display a lock icon
|
||||
"${mod}+F5" = "exec --no-startup-id ${pkgs.writeShellScript "xautolock-toggle" ''
|
||||
state="$(cat "${xautolockState}")"
|
||||
if [ "$state" = "disabled" ]
|
||||
then
|
||||
${pkgs.xautolock}/bin/xautolock -enable
|
||||
echo enabled > ${xautolockState}
|
||||
else
|
||||
${pkgs.xautolock}/bin/xautolock -disable
|
||||
echo disabled > ${xautolockState}
|
||||
fi
|
||||
''}";
|
||||
};
|
||||
startup = [
|
||||
# Stop screen after 10 minutes, 1 minutes after lock it
|
||||
{ notification = false; command = "${pkgs.xautolock}/bin/xautolock -time 10 -locker '${pkgs.xorg.xset}/bin/xset dpms force standby' -killtime 1 -killer xlock"; }
|
||||
{ notification = false; command = "${pkgs.writeShellScript "xautolock-start" ''
|
||||
echo enabled > ${xautolockState}
|
||||
${pkgs.xautolock}/bin/xautolock -time 10 -locker '${pkgs.xorg.xset}/bin/xset dpms force standby' -killtime 1 -killer xlock
|
||||
''}"; }
|
||||
# services.screen-locker.xautolock is hardcoded to use systemd for -locker (doesn't even work...)
|
||||
];
|
||||
};
|
||||
|
|
|
@ -4,7 +4,7 @@
|
|||
programs.powerline-go = {
|
||||
enable = true;
|
||||
modules = [ "user" "host" "venv" "cwd" "perms" "nix-shell" "git" ];
|
||||
modulesRight = [ "jobs" "exit" "duration" "load" ];
|
||||
modulesRight = [ "jobs" "exit" "duration" ];
|
||||
settings = {
|
||||
colorize-hostname = true;
|
||||
hostname-only-if-ssh = true;
|
||||
|
@ -20,4 +20,3 @@
|
|||
};
|
||||
};
|
||||
}
|
||||
# TODO Replace load with a frobar indicator
|
||||
|
|
|
@ -1,10 +0,0 @@
|
|||
# Automatrop
|
||||
|
||||
Because I'm getting tired of too many bash scripts and yet using Ansible seems
|
||||
overkill at the same time.
|
||||
|
||||
## Dependencies
|
||||
|
||||
```bash
|
||||
ansible-galaxy install mnussbaum.base16-builder-ansible
|
||||
```
|
|
@ -1,9 +0,0 @@
|
|||
[defaults]
|
||||
inventory=hosts
|
||||
roles_path=roles
|
||||
interpreter_python=auto
|
||||
library=plugins/modules
|
||||
|
||||
[ssh_connection]
|
||||
pipelining = True # does not work with requiretty in /etc/sudoers
|
||||
ssh_args=-o ForwardAgent=yes # no need for installing/configuring/unlocking SSH/GPG keys on the host to be able to git clone extensions
|
|
@ -1,34 +0,0 @@
|
|||
# Default values
|
||||
|
||||
# If you have root access on the machine (via sudo)
|
||||
root_access: no
|
||||
|
||||
# Display server (no, "x11", "wayland")
|
||||
display_server: no
|
||||
|
||||
# What development work will I do on this machine
|
||||
dev_stuffs: []
|
||||
|
||||
# Install software that is rarely used
|
||||
software_full: no
|
||||
|
||||
# Which additional software to install
|
||||
software_snippets: []
|
||||
|
||||
# If the computer has a battery and we want to use it
|
||||
has_battery: no
|
||||
|
||||
# Activate numlock by default
|
||||
auto_numlock: no
|
||||
|
||||
# Machine has SSH key to access git.frogeye.fr
|
||||
has_forge_access: no
|
||||
|
||||
# Wether to permit /home/$USER to be encrypted
|
||||
# with stacked filesystem encryption
|
||||
encrypt_home_stacked_fs: no
|
||||
|
||||
# Which extensions to load
|
||||
extensions: []
|
||||
|
||||
# TODO Make role/playbook defaults instead
|
|
@ -1,30 +0,0 @@
|
|||
root_access: yes
|
||||
display_server: "x11"
|
||||
dev_stuffs:
|
||||
- ansible
|
||||
- docker
|
||||
- network
|
||||
- nix
|
||||
- perl
|
||||
- php
|
||||
- python
|
||||
- shell
|
||||
- sql
|
||||
software_full: yes
|
||||
has_battery: yes
|
||||
auto_numlock: yes
|
||||
has_forge_access: yes
|
||||
extensions:
|
||||
- g
|
||||
- gh
|
||||
x11_screens:
|
||||
# nvidia-xrun
|
||||
# - HDMI-0
|
||||
# - eDP-1-1
|
||||
# mesa + nouveau
|
||||
# - HDMI-1-3
|
||||
# - eDP1
|
||||
# mesa + nvidia
|
||||
- HDMI-1-0
|
||||
- eDP1
|
||||
max_video_height: 1440
|
|
@ -1,14 +0,0 @@
|
|||
root_access: no
|
||||
display_server: "x11"
|
||||
dev_stuffs:
|
||||
- shell
|
||||
- network
|
||||
- ansible
|
||||
- perl
|
||||
- python
|
||||
extensions:
|
||||
- gh
|
||||
x11_screens:
|
||||
- HDMI-1
|
||||
- HDMI-2
|
||||
base16_scheme: solarized-light
|
|
@ -1,4 +0,0 @@
|
|||
curacao.geoffrey.frogeye.fr
|
||||
# triffle.geoffrey.frogeye.fr
|
||||
pindakaas.geoffrey.frogeye.fr
|
||||
gho.geoffrey.frogeye.fr ansible_host=localhost ansible_port=2222
|
|
@ -1,10 +0,0 @@
|
|||
---
|
||||
- name: Default
|
||||
hosts: all
|
||||
roles:
|
||||
- role: system
|
||||
tags: system
|
||||
when: root_access
|
||||
- role: termux
|
||||
tags: termux
|
||||
when: termux
|
|
@ -1,44 +0,0 @@
|
|||
[general]
|
||||
status_path = "~/.cache/vdirsyncer/status/"
|
||||
|
||||
{% for config in configs %}
|
||||
|
||||
# CarDAV
|
||||
|
||||
[pair geoffrey_contacts]
|
||||
a = "geoffrey_contacts_local"
|
||||
b = "geoffrey_contacts_remote"
|
||||
collections = ["from a", "from b"]
|
||||
metadata = ["displayname"]
|
||||
|
||||
[storage geoffrey_contacts_local]
|
||||
type = "filesystem"
|
||||
path = "~/.cache/vdirsyncer/contacts/"
|
||||
fileext = ".vcf"
|
||||
|
||||
[storage geoffrey_contacts_remote]
|
||||
type = "carddav"
|
||||
url = "https://cloud.frogeye.fr/remote.php/dav"
|
||||
username = "geoffrey"
|
||||
password.fetch = ["command", "sh", "-c", "cat ~/.config/vdirsyncer/pass"]
|
||||
|
||||
# CalDAV
|
||||
|
||||
[pair geoffrey_calendar]
|
||||
a = "geoffrey_calendar_local"
|
||||
b = "geoffrey_calendar_remote"
|
||||
collections = ["from a", "from b"]
|
||||
metadata = ["displayname", "color"]
|
||||
|
||||
[storage geoffrey_calendar_local]
|
||||
type = "filesystem"
|
||||
path = "~/.cache/vdirsyncer/calendars/"
|
||||
fileext = ".ics"
|
||||
|
||||
[storage geoffrey_calendar_remote]
|
||||
type = "caldav"
|
||||
url = "https://cloud.frogeye.fr/remote.php/dav"
|
||||
username = "geoffrey"
|
||||
password.fetch = ["command", "sh", "-c", "cat ~/.config/vdirsyncer/pass"]
|
||||
|
||||
{% endfor %}
|
|
@ -1,4 +0,0 @@
|
|||
# https://i.imgur.com/yVtVucs.jpg # Doctor Who Series 11
|
||||
# Derivate of these ones https://wallpapers.wallhaven.cc/wallpapers/full/wallhaven-230622.png
|
||||
# https://geoffrey.frogeye.fr/files/backgrounds/VertBleu.png
|
||||
https://geoffrey.frogeye.fr/files/backgrounds/BleuVert.png
|
|
@ -1,32 +0,0 @@
|
|||
#!/usr/bin/env bash
|
||||
|
||||
DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd )"
|
||||
# From https://stackoverflow.com/a/246128
|
||||
|
||||
# Relaunch the bars
|
||||
# i3-msg exec ~/.config/polybar/launch.sh
|
||||
# TODO Make something better with that
|
||||
i3-msg exec ~/.config/lemonbar/launch.sh
|
||||
|
||||
# Resize background
|
||||
BGDIR="$HOME/.cache/background"
|
||||
mkdir -p "$BGDIR"
|
||||
|
||||
list="$DIR/bg"
|
||||
url="$(cat "$list" | sed -e 's/#.*$//' -e 's/ \+$//' -e '/^$/d' | sort -R | head -1)"
|
||||
|
||||
hash="$(printf "$url" | md5sum | cut -f1 -d' ')"
|
||||
filepath="$BGDIR/$hash"
|
||||
|
||||
if [ ! -e "$filepath" ]; then
|
||||
wget -c "$url" -O "$filepath"
|
||||
fi
|
||||
|
||||
feh --no-fehbg --bg-fill "$filepath"
|
||||
|
||||
# Make i3 distribute the workspaces on all screens
|
||||
monitors_json="$(xrandr --listmonitors | tail -n+2 | awk '{ print $4 }' | sed 's|.\+|"\0"|' | tr '\n' ',')"
|
||||
automatrop -e '{"x11_screens":['"$monitors_json"']}' --tags i3
|
||||
|
||||
# TODO Make sure it goes from left to right
|
||||
# Either with the "main" display or using the geometry data
|
|
@ -1,26 +0,0 @@
|
|||
[Plugins]
|
||||
Output=i3bar
|
||||
Input=nm;pulseaudio;upower;time
|
||||
Order=nm;pulseaudio;upower;time
|
||||
|
||||
[Time]
|
||||
Zones=Europe/Paris;
|
||||
|
||||
[PulseAudio]
|
||||
#Actions=raise
|
||||
|
||||
[NetworkManager]
|
||||
Interfaces=enp3s0;wlp2s0
|
||||
HideUnavailable=true
|
||||
|
||||
[Override pulseaudio:auto_null]
|
||||
Label=🔊
|
||||
|
||||
[Override nm-wifi:wlp2s0]
|
||||
Label=📶
|
||||
|
||||
[Override time:Europe/Paris]
|
||||
Label=🕗
|
||||
|
||||
[Override upower-battery:BAT0]
|
||||
Label=🔋
|
|
@ -1,24 +0,0 @@
|
|||
#!/bin/sh
|
||||
if [ "$TERM" = "linux" ]; then
|
||||
/bin/echo -e "
|
||||
\e]P048483e
|
||||
\e]P1dc2566
|
||||
\e]P28fc029
|
||||
\e]P3d4c96e
|
||||
\e]P455bcce
|
||||
\e]P59358fe
|
||||
\e]P656b7a5
|
||||
\e]P7acada1
|
||||
\e]P876715e
|
||||
\e]P9fa2772
|
||||
\e]PAa7e22e
|
||||
\e]PBe7db75
|
||||
\e]PC66d9ee
|
||||
\e]PDae82ff
|
||||
\e]PE66efd5
|
||||
\e]PFcfd0c2
|
||||
"
|
||||
# get rid of artifacts
|
||||
# clear
|
||||
fi
|
||||
|
|
@ -1 +0,0 @@
|
|||
theme.css
|
Loading…
Reference in a new issue