Compare commits

...

6 commits

Author SHA1 Message Date
Geoffrey Frogeye 445c2b8a99
frobar: Mutli-display support
Freaking finally
2024-06-18 04:09:59 +02:00
Geoffrey Frogeye e09774c4ca
frobar: Type hint
That was TEDIOUS
2024-06-18 00:31:29 +02:00
Geoffrey Frogeye a489830949
xautolock: Key is now a toggle 2024-06-17 20:36:50 +02:00
Geoffrey Frogeye 42034eb5d8
frobar: Add load provider 2024-06-17 19:31:16 +02:00
Geoffrey Frogeye d6d3df65df
frobar: Display temperature for other CPU types 2024-06-17 19:30:48 +02:00
Geoffrey Frogeye a3fcaf9d27
unprocessed: Cleanup 2024-06-17 18:45:35 +02:00
25 changed files with 498 additions and 664 deletions

View file

@ -5,7 +5,7 @@
./autorandr
./background
./browser
./frobar
./frobar/module.nix
./i3.nix
./lock
./mpd

View file

@ -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

View file

@ -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()

View file

@ -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

View file

@ -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()

View file

@ -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")

View 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?

View file

@ -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...)
];
};

View file

@ -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

View file

@ -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
```

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -1,10 +0,0 @@
---
- name: Default
hosts: all
roles:
- role: system
tags: system
when: root_access
- role: termux
tags: termux
when: termux

View file

@ -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 %}

View file

@ -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

View file

@ -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

View file

@ -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=🔋

View file

@ -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

View file

@ -1 +0,0 @@
theme.css