Compare commits

..

No commits in common. "445c2b8a99dd4b3d0e9b157941d10fce734aa1ca" and "2951280faaf5d3c036f0942584937872db933e00" have entirely different histories.

25 changed files with 663 additions and 497 deletions

View file

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

View file

@ -1,21 +1,12 @@
{ pkgs ? import <nixpkgs> { config = { }; overlays = [ ]; }, ... }:
let
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
{ 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.python3Packages.buildPythonApplication {
# is called pyton-mpd2 on PyPi but mpd2 in nixpkgs.
let
frobar = pkgs.python3Packages.buildPythonApplication {
pname = "frobar";
version = "2.0";
runtimeInputs = with pkgs; [ lemonbar-xft wirelesstools ];
propagatedBuildInputs = with pkgs.python3Packages; [
coloredlogs
notmuch
@ -25,7 +16,33 @@ pkgs.python3Packages.buildPythonApplication {
pulsectl
pyinotify
];
makeWrapperArgs = [ "--prefix PATH : ${pkgs.lib.makeBinPath ([ lemonbar ] ++ (with pkgs; [ wirelesstools ]))}" ];
makeWrapperArgs = [ "--prefix PATH : ${pkgs.lib.makeBinPath (with pkgs; [ lemonbar-xft wirelesstools ])}" ];
src = ./.;
};
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" ];
};
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"'';
};
Install = { WantedBy = [ "graphical-session.target" ]; };
};
};
}
# 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,20 +1,13 @@
#!/usr/bin/env python3
from frobar import providers as fp
from frobar.display import Bar, BarGroupType
from frobar.updaters import Updater
from frobar.providers import *
# TODO If multiple screen, expand the sections and share them
# TODO Graceful exit
def run() -> None:
def run():
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
@ -26,7 +19,7 @@ def run() -> None:
full = short + " " + CUSTOM_SUFFIXES[i]
customNames[short] = full
Bar.addSectionAll(
fp.I3WorkspacesProvider(
I3WorkspacesProvider(
theme=WORKSPACE_THEME,
themeFocus=FOCUS_THEME,
themeUrgent=URGENT_THEME,
@ -37,40 +30,35 @@ def run() -> None:
)
# TODO Middle
Bar.addSectionAll(fp.MpdProvider(theme=9), BarGroupType.LEFT)
Bar.addSectionAll(MpdProvider(theme=9), BarGroupType.LEFT)
# Bar.addSectionAll(I3WindowTitleProvider(), BarGroupType.LEFT)
# TODO Computer modes
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)
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)
# Peripherals
PERIPHERAL_THEME = 6
NETWORK_THEME = 5
# TODO Disk space provider
# TODO Screen (connected, autorandr configuration, bbswitch) provider
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)
Bar.addSectionAll(PulseaudioProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(RfkillProvider(theme=PERIPHERAL_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(NetworkProvider(theme=NETWORK_THEME), BarGroupType.RIGHT)
# Personal
# 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,
# )
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)
TIME_THEME = 4
Bar.addSectionAll(fp.TimeProvider(theme=TIME_THEME), BarGroupType.RIGHT)
Bar.addSectionAll(TimeProvider(theme=TIME_THEME), BarGroupType.RIGHT)
# Bar.run()

View file

@ -1,4 +1,4 @@
#!/usr/bin/env python3init
#!/usr/bin/env python3
import enum
import logging
@ -7,12 +7,11 @@ import signal
import subprocess
import threading
import time
import typing
import coloredlogs
import i3ipc
from frobar.common import notBusy
from frobar.notbusy import notBusy
coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s")
log = logging.getLogger()
@ -30,12 +29,6 @@ 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
@ -47,7 +40,6 @@ class BarGroupType(enum.Enum):
class BarStdoutThread(threading.Thread):
def run(self) -> None:
while Bar.running:
assert Bar.process.stdout
handle = Bar.process.stdout.readline().strip()
if not len(handle):
Bar.stop()
@ -70,31 +62,20 @@ class Bar:
@staticmethod
def init() -> None:
Bar.running = True
Bar.everyone = set()
Section.init()
cmd = [
"lemonbar",
"-b",
"-a",
"64",
"-F",
Section.FGCOLOR,
"-B",
Section.BGCOLOR,
]
cmd = ["lemonbar", "-b", "-a", "64"]
for font in Bar.FONTS:
cmd += ["-f", "{}:size={}".format(font, Bar.FONTSIZE)]
Bar.process = subprocess.Popen(
cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE
)
BarStdoutThread().start()
Bar.stdoutThread = BarStdoutThread()
Bar.stdoutThread.start()
i3 = i3ipc.Connection()
for output in i3.get_outputs():
if not output.active:
continue
Bar(output.name)
# Debug
Bar(0)
# Bar(1)
@staticmethod
def stop() -> None:
@ -109,27 +90,29 @@ class Bar:
Bar.forever()
i3 = i3ipc.Connection()
def doStop(*args: list) -> None:
def doStop(*args) -> None:
Bar.stop()
print(88)
try:
i3.on("ipc_shutdown", doStop)
i3.main()
except BaseException:
print(93)
Bar.stop()
# Class globals
everyone: set["Bar"]
everyone = set()
string = ""
process: subprocess.Popen
process = None
running = False
nextHandle = 0
actionsF2H: dict[Handle, bytes] = dict()
actionsH2F: dict[bytes, Handle] = dict()
actionsF2H = dict()
actionsH2F = dict()
@staticmethod
def getFunctionHandle(function: typing.Callable[[], None]) -> bytes:
def getFunctionHandle(function):
assert callable(function)
if function in Bar.actionsF2H.keys():
return Bar.actionsF2H[function]
@ -143,12 +126,13 @@ class Bar:
return handle
@staticmethod
def forever() -> None:
def forever():
Bar.process.wait()
Bar.stop()
def __init__(self, output: str) -> None:
self.output = output
def __init__(self, screen):
assert isinstance(screen, int)
self.screen = "%{S" + str(screen) + "}"
self.groups = dict()
for groupType in BarGroupType:
@ -156,33 +140,36 @@ class Bar:
self.groups[groupType] = group
self.childsChanged = False
Bar.everyone.add(self)
self.everyone.add(self)
@staticmethod
def addSectionAll(
section: "Section", group: "BarGroupType"
) -> None:
def addSectionAll(section, group, screens=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: "Section", group: "BarGroupType") -> None:
def addSection(self, section, group):
assert isinstance(section, Section)
assert isinstance(group, BarGroupType)
self.groups[group].addSection(section)
def update(self) -> None:
def update(self):
if self.childsChanged:
self.string = "%{Sn" + self.output + "}"
self.string = self.screen
self.string += self.groups[BarGroupType.LEFT].string
self.string += self.groups[BarGroupType.RIGHT].string
self.childsChanged = False
@staticmethod
def updateAll() -> None:
def updateAll():
if Bar.running:
Bar.string = ""
for bar in Bar.everyone:
@ -191,10 +178,8 @@ class Bar:
# Color for empty sections
Bar.string += BarGroup.color(*Section.EMPTY)
string = Bar.string + "\n"
# print(string)
assert Bar.process.stdin
Bar.process.stdin.write(string.encode())
# print(Bar.string)
Bar.process.stdin.write(bytes(Bar.string + "\n", "utf-8"))
Bar.process.stdin.flush()
@ -203,16 +188,18 @@ class BarGroup:
One for each group of each bar
"""
everyone: set["BarGroup"] = set()
everyone = set()
def __init__(self, groupType: BarGroupType, parent: Bar):
def __init__(self, groupType, parent):
assert isinstance(groupType, BarGroupType)
assert isinstance(parent, Bar)
self.groupType = groupType
self.parent = parent
self.sections: list["Section"] = list()
self.sections = list()
self.string = ""
self.parts: list[Part] = []
self.parts = []
#: One of the sections that had their theme or visibility changed
self.childsThemeChanged = False
@ -222,11 +209,11 @@ class BarGroup:
BarGroup.everyone.add(self)
def addSection(self, section: "Section") -> None:
def addSection(self, section):
self.sections.append(section)
section.addParent(self)
def addSectionAfter(self, sectionRef: "Section", section: "Section") -> None:
def addSectionAfter(self, sectionRef, section):
index = self.sections.index(sectionRef)
self.sections.insert(index + 1, section)
section.addParent(self)
@ -234,20 +221,20 @@ class BarGroup:
ALIGNS = {BarGroupType.LEFT: "%{l}", BarGroupType.RIGHT: "%{r}"}
@staticmethod
def fgColor(color: str) -> str:
def fgColor(color):
return "%{F" + (color or "-") + "}"
@staticmethod
def bgColor(color: str) -> str:
def bgColor(color):
return "%{B" + (color or "-") + "}"
@staticmethod
def color(fg: str, bg: str) -> str:
def color(fg, bg):
return BarGroup.fgColor(fg) + BarGroup.bgColor(bg)
def update(self) -> None:
def update(self):
if self.childsThemeChanged:
parts: list[Part] = [BarGroup.ALIGNS[self.groupType]]
parts = [BarGroup.ALIGNS[self.groupType]]
secs = [sec for sec in self.sections if sec.visible]
lenS = len(secs)
@ -296,7 +283,7 @@ class BarGroup:
self.childsTextChanged = False
@staticmethod
def updateAll() -> None:
def updateAll():
for group in BarGroup.everyone:
group.update()
Bar.updateAll()
@ -307,7 +294,7 @@ class SectionThread(threading.Thread):
ANIMATION_STOP = 0.001
ANIMATION_EVOLUTION = 0.9
def run(self) -> None:
def run(self):
while Section.somethingChanged.wait():
notBusy.wait()
Section.updateAll()
@ -324,9 +311,6 @@ class SectionThread(threading.Thread):
animTime = self.ANIMATION_STOP
Theme = tuple[str, str]
class Section:
# TODO Update all of that to base16
COLORS = [
@ -350,20 +334,20 @@ class Section:
FGCOLOR = "#fff0f1"
BGCOLOR = "#092c0e"
THEMES: list[Theme] = list()
EMPTY: Theme = (FGCOLOR, BGCOLOR)
THEMES = list()
EMPTY = (FGCOLOR, BGCOLOR)
ICON: str | None = None
ICON = None
PERSISTENT = False
#: Sections that do not have their destination size
sizeChanging: set["Section"] = set()
updateThread: threading.Thread = SectionThread(daemon=True)
sizeChanging = set()
updateThread = SectionThread(daemon=True)
somethingChanged = threading.Event()
lastChosenTheme = 0
@staticmethod
def init() -> None:
def init():
for t in range(8, 16):
Section.THEMES.append((Section.COLORS[0], Section.COLORS[t]))
Section.THEMES.append((Section.COLORS[0], Section.COLORS[3]))
@ -371,7 +355,7 @@ class Section:
Section.updateThread.start()
def __init__(self, theme: int | None = None) -> None:
def __init__(self, theme=None):
#: Displayed section
#: Note: A section can be empty and displayed!
self.visible = False
@ -394,12 +378,12 @@ class Section:
self.dstSize = 0
#: Groups that have this section
self.parents: set[BarGroup] = set()
self.parents = set()
self.icon = self.ICON
self.persistent = self.PERSISTENT
def __str__(self) -> str:
def __str__(self):
try:
return "<{}><{}>{:01d}{}{:02d}/{:02d}".format(
self.curText,
@ -409,29 +393,26 @@ class Section:
self.curSize,
self.dstSize,
)
except Exception:
except:
return super().__str__()
def addParent(self, parent: BarGroup) -> None:
def addParent(self, parent):
self.parents.add(parent)
def appendAfter(self, section: "Section") -> None:
def appendAfter(self, section):
assert len(self.parents)
for parent in self.parents:
parent.addSectionAfter(self, section)
def added(self) -> None:
pass
def informParentsThemeChanged(self) -> None:
def informParentsThemeChanged(self):
for parent in self.parents:
parent.childsThemeChanged = True
def informParentsTextChanged(self) -> None:
def informParentsTextChanged(self):
for parent in self.parents:
parent.childsTextChanged = True
def updateText(self, text: Element) -> None:
def updateText(self, text):
if isinstance(text, str):
text = Text(text)
elif isinstance(text, Text) and not len(text.elements):
@ -458,13 +439,14 @@ class Section:
Section.sizeChanging.add(self)
Section.somethingChanged.set()
def setDecorators(self, **kwargs: Handle) -> None:
def setDecorators(self, **kwargs):
self.dstText.setDecorators(**kwargs)
self.curText = str(self.dstText)
self.informParentsTextChanged()
Section.somethingChanged.set()
def updateTheme(self, theme: int) -> None:
def updateTheme(self, theme):
assert isinstance(theme, int)
assert theme < len(Section.THEMES)
if theme == self.theme:
return
@ -472,18 +454,19 @@ class Section:
self.informParentsThemeChanged()
Section.somethingChanged.set()
def updateVisibility(self, visibility: bool) -> None:
def updateVisibility(self, visibility):
assert isinstance(visibility, bool)
self.visible = visibility
self.informParentsThemeChanged()
Section.somethingChanged.set()
@staticmethod
def fit(text: str, size: int) -> str:
def fit(text, size):
t = len(text)
return text[:size] if t >= size else text + " " * (size - t)
return text[:size] if t >= size else text + [" "] * (size - t)
def update(self) -> None:
def update(self):
# TODO Might profit of a better logic
if not self.visible:
self.updateVisibility(True)
@ -504,7 +487,7 @@ class Section:
self.informParentsTextChanged()
@staticmethod
def updateAll() -> None:
def updateAll():
"""
Process all sections for text size changes
"""
@ -517,7 +500,7 @@ class Section:
Section.somethingChanged.clear()
@staticmethod
def ramp(p: float, ramp: str = " ▁▂▃▄▅▆▇█") -> str:
def ramp(p, ramp=" ▁▂▃▄▅▆▇█"):
if p > 1:
return ramp[-1]
elif p < 0:
@ -528,11 +511,11 @@ class Section:
class StatefulSection(Section):
# TODO FEAT Allow to temporary expand the section (e.g. when important change)
NUMBER_STATES: int
NUMBER_STATES = None
DEFAULT_STATE = 0
def __init__(self, theme: int | None) -> None:
Section.__init__(self, theme=theme)
def __init__(self, *args, **kwargs):
Section.__init__(self, *args, **kwargs)
self.state = self.DEFAULT_STATE
if hasattr(self, "onChangeState"):
self.onChangeState(self.state)
@ -540,22 +523,20 @@ class StatefulSection(Section):
clickLeft=self.incrementState, clickRight=self.decrementState
)
def incrementState(self) -> None:
def incrementState(self):
newState = min(self.state + 1, self.NUMBER_STATES - 1)
self.changeState(newState)
def decrementState(self) -> None:
def decrementState(self):
newState = max(self.state - 1, 0)
self.changeState(newState)
def changeState(self, state: int) -> None:
def changeState(self, state):
assert isinstance(state, int)
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()
@ -566,13 +547,10 @@ class ColorCountsSection(StatefulSection):
NUMBER_STATES = 3
COLORABLE_ICON = "?"
def __init__(self, theme: None | int = None) -> None:
def __init__(self, theme=None):
StatefulSection.__init__(self, theme=theme)
def subfetcher(self) -> list[tuple[int, str]]:
raise NotImplementedError("Interface must be implemented")
def fetcher(self) -> typing.Union[None, "Text"]:
def fetcher(self):
counts = self.subfetcher()
# Nothing
if not len(counts):
@ -587,66 +565,67 @@ 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, " ", str(total))
return Text(self.COLORABLE_ICON, " ", total)
# Icon + Counts
else:
text = Text(self.COLORABLE_ICON)
for count, color in counts:
text.append(" ", Text(str(count), fg=color))
text.append(" ", Text(count, fg=color))
return text
class Text:
def _setDecorators(self, decorators: dict[str, Decorator]) -> None:
def _setElements(self, elements):
# TODO OPTI Concatenate consecutrive string
self.elements = list(elements)
def _setDecorators(self, decorators):
# TODO OPTI Convert no decorator to strings
self.decorators = decorators
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)
self.prefix = None
self.suffix = None
def __init__(self, *args, **kwargs):
self._setElements(args)
self._setDecorators(kwargs)
self.section: Section
self.section = None
def append(self, *args: Element) -> None:
self.elements += list(args)
def append(self, *args):
self._setElements(self.elements + list(args))
def prepend(self, *args: Element) -> None:
self.elements = list(args) + self.elements
def prepend(self, *args):
self._setElements(list(args) + self.elements)
def setElements(self, *args: Element) -> None:
self.elements = list(args)
def setElements(self, *args):
self._setElements(args)
def setDecorators(self, **kwargs: Decorator) -> None:
def setDecorators(self, **kwargs):
self._setDecorators(kwargs)
def setSection(self, section: Section) -> None:
def setSection(self, section):
assert isinstance(section, Section)
self.section = section
for element in self.elements:
if isinstance(element, Text):
element.setSection(section)
def _genFixs(self) -> None:
def _genFixs(self):
if self.prefix is not None and self.suffix is not None:
return
self.prefix = ""
self.suffix = ""
def nest(prefix: str, suffix: str) -> None:
assert self.prefix is not None
assert self.suffix is not None
def nest(prefix, suffix):
self.prefix = self.prefix + "%{" + prefix + "}"
self.suffix = "%{" + suffix + "}" + self.suffix
def getColor(val: str) -> str:
def getColor(val):
# TODO Allow themes
assert len(val) == 7
assert isinstance(val, str) and len(val) == 7
return val
def button(number: str, function: Handle) -> None:
def button(number, function):
handle = Bar.getFunctionHandle(function)
nest("A" + number + ":" + handle.decode() + ":", "A" + number)
@ -655,34 +634,25 @@ 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: int | None = None, pad: bool = False) -> tuple[str, int]:
def _text(self, size=None, pad=False):
self._genFixs()
assert self.prefix is not None
assert self.suffix is not None
curString = self.prefix
curSize = 0
remSize = size
@ -708,9 +678,7 @@ class Text:
curString += self.suffix
if pad:
assert remSize is not None
if remSize > 0:
if pad and remSize > 0:
curString += " " * remSize
curSize += remSize
@ -721,14 +689,12 @@ class Text:
assert size >= curSize
return curString, curSize
def text(self, size: int | None = None, pad: bool = False) -> str:
string, size = self._text(size=size, pad=pad)
def text(self, *args, **kwargs):
string, size = self._text(*args, **kwargs)
return string
def __str__(self) -> str:
def __str__(self):
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:
@ -738,7 +704,7 @@ class Text:
curString += self.suffix
return curString
def __len__(self) -> int:
def __len__(self):
curSize = 0
for element in self.elements:
if element is None:
@ -749,8 +715,8 @@ class Text:
curSize += len(str(element))
return curSize
def __getitem__(self, index: int) -> Element:
def __getitem__(self, index):
return self.elements[index]
def __setitem__(self, index: int, data: Element) -> None:
def __setitem__(self, index, data):
self.elements[index] = data

View file

@ -1,27 +1,21 @@
#!/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 (ColorCountsSection, Element, Section,
StatefulSection, Text)
from frobar.updaters import (I3Updater, InotifyUpdater, MergedUpdater,
PeriodicUpdater, ThreadedUpdater, Updater)
from frobar.display import *
from frobar.updaters import *
coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s")
log = logging.getLogger()
@ -30,7 +24,7 @@ log = logging.getLogger()
# PulseaudioProvider and MpdProvider)
def humanSize(num: int) -> str:
def humanSize(num):
"""
Returns a string of width 3+3
"""
@ -40,11 +34,11 @@ def humanSize(num: int) -> str:
return "{:3d}{}".format(int(num), unit)
else:
return "{:.1f}{}".format(num, unit)
num //= 1024
num /= 1024.0
return "{:d}YiB".format(num)
def randomColor(seed: int | bytes = 0) -> str:
def randomColor(seed=0):
random.seed(seed)
return "#{:02x}{:02x}{:02x}".format(*[random.randint(0, 255) for _ in range(3)])
@ -54,11 +48,11 @@ class TimeProvider(StatefulSection, PeriodicUpdater):
NUMBER_STATES = len(FORMATS)
DEFAULT_STATE = 1
def fetcher(self) -> str:
def fetcher(self):
now = datetime.datetime.now()
return now.strftime(self.FORMATS[self.state])
def __init__(self, theme: int | None = None):
def __init__(self, theme=None):
PeriodicUpdater.__init__(self)
StatefulSection.__init__(self, theme)
self.changeInterval(1) # TODO OPTI When state < 1
@ -72,10 +66,10 @@ class AlertLevel(enum.Enum):
class AlertingSection(StatefulSection):
# TODO EASE Correct settings for themes
ALERT_THEMES = {AlertLevel.NORMAL: 3, AlertLevel.WARNING: 1, AlertLevel.DANGER: 0}
THEMES = {AlertLevel.NORMAL: 3, AlertLevel.WARNING: 1, AlertLevel.DANGER: 0}
PERSISTENT = True
def getLevel(self, quantity: float) -> AlertLevel:
def getLevel(self, quantity):
if quantity > self.dangerThresold:
return AlertLevel.DANGER
elif quantity > self.warningThresold:
@ -83,14 +77,14 @@ class AlertingSection(StatefulSection):
else:
return AlertLevel.NORMAL
def updateLevel(self, quantity: float) -> None:
def updateLevel(self, quantity):
self.level = self.getLevel(quantity)
self.updateTheme(self.ALERT_THEMES[self.level])
self.updateTheme(self.THEMES[self.level])
if self.level == AlertLevel.NORMAL:
return
# TODO Temporary update state
def __init__(self, theme: int | None = None):
def __init__(self, theme):
StatefulSection.__init__(self, theme)
self.dangerThresold = 0.90
self.warningThresold = 0.75
@ -98,9 +92,9 @@ class AlertingSection(StatefulSection):
class CpuProvider(AlertingSection, PeriodicUpdater):
NUMBER_STATES = 3
ICON = ""
ICON = ""
def fetcher(self) -> Element:
def fetcher(self):
percent = psutil.cpu_percent(percpu=False)
self.updateLevel(percent / 100)
if self.state >= 2:
@ -108,44 +102,22 @@ 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: int | None = None):
def __init__(self, theme=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) -> Element:
def fetcher(self):
mem = psutil.virtual_memory()
freePerc = mem.percent / 100
self.updateLevel(freePerc)
@ -163,7 +135,7 @@ class RamProvider(AlertingSection, PeriodicUpdater):
return text
def __init__(self, theme: int | None = None):
def __init__(self, theme=None):
AlertingSection.__init__(self, theme)
PeriodicUpdater.__init__(self)
self.changeInterval(1)
@ -172,28 +144,23 @@ 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) -> Element:
def fetcher(self):
allTemp = psutil.sensors_temperatures()
for main in self.MAIN_TEMPS:
if main in allTemp:
break
else:
return "?"
temp = allTemp[main][0]
if "coretemp" not in allTemp:
# TODO Opti Remove interval
return ""
temp = allTemp["coretemp"][0]
self.warningThresold = temp.high or 90.0
self.dangerThresold = temp.critical or 100.0
self.warningThresold = temp.high
self.dangerThresold = temp.critical
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: int | None = None):
def __init__(self, theme=None):
AlertingSection.__init__(self, theme)
PeriodicUpdater.__init__(self)
self.changeInterval(5)
@ -204,9 +171,10 @@ class BatteryProvider(AlertingSection, PeriodicUpdater):
NUMBER_STATES = 3
RAMP = ""
def fetcher(self) -> Element:
def fetcher(self):
bat = psutil.sensors_battery()
if not bat:
self.icon = None
return None
self.icon = ("" if bat.power_plugged else "") + Section.ramp(
@ -216,7 +184,7 @@ class BatteryProvider(AlertingSection, PeriodicUpdater):
self.updateLevel(1 - bat.percent / 100)
if self.state < 1:
return ""
return
t = Text("{:.0f}%".format(bat.percent))
@ -228,38 +196,17 @@ class BatteryProvider(AlertingSection, PeriodicUpdater):
t.append(" ({:d}:{:02d})".format(h, m))
return t
def __init__(self, theme: int | None = None):
def __init__(self, theme=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: int | None = None):
def __init__(self, theme=None):
ThreadedUpdater.__init__(self)
StatefulSection.__init__(self, theme)
self.pulseEvents = pulsectl.Pulse("event-handler")
@ -269,7 +216,7 @@ class PulseaudioProvider(StatefulSection, ThreadedUpdater):
self.start()
self.refreshData()
def fetcher(self) -> Element:
def fetcher(self):
sinks = []
with pulsectl.Pulse("list-sinks") as pulse:
for sink in pulse.sink_list():
@ -302,10 +249,10 @@ class PulseaudioProvider(StatefulSection, ThreadedUpdater):
return Text(*sinks)
def loop(self) -> None:
def loop(self):
self.pulseEvents.event_listen()
def handleEvent(self, ev: pulsectl.PulseEventInfo) -> None:
def handleEvent(self, ev):
self.refreshData()
@ -313,7 +260,7 @@ class NetworkProviderSection(StatefulSection, Updater):
NUMBER_STATES = 5
DEFAULT_STATE = 1
def actType(self) -> None:
def actType(self):
self.ssid = None
if self.iface.startswith("eth") or self.iface.startswith("enp"):
if "u" in self.iface:
@ -334,10 +281,10 @@ class NetworkProviderSection(StatefulSection, Updater):
self.icon = ""
elif self.iface.startswith("vboxnet"):
self.icon = ""
else:
self.icon = "?"
def getAddresses(
self,
) -> tuple[psutil._common.snicaddr, psutil._common.snicaddr]:
def getAddresses(self):
ipv4 = None
ipv6 = None
for address in self.parent.addrs[self.iface]:
@ -347,8 +294,8 @@ class NetworkProviderSection(StatefulSection, Updater):
ipv6 = address
return ipv4, ipv6
def fetcher(self) -> Element:
self.icon = "?"
def fetcher(self):
self.icon = None
self.persistent = False
if (
self.iface not in self.parent.stats
@ -402,13 +349,13 @@ class NetworkProviderSection(StatefulSection, Updater):
return " ".join(text)
def onChangeState(self, state: int) -> None:
def onChangeState(self, state):
self.showSsid = state >= 1
self.showAddress = state >= 2
self.showSpeed = state >= 3
self.showTransfer = state >= 4
def __init__(self, iface: str, parent: "NetworkProvider"):
def __init__(self, iface, parent):
Updater.__init__(self)
StatefulSection.__init__(self, theme=parent.theme)
self.iface = iface
@ -416,23 +363,23 @@ class NetworkProviderSection(StatefulSection, Updater):
class NetworkProvider(Section, PeriodicUpdater):
def fetchData(self) -> None:
def fetchData(self):
self.prev = self.last
self.prevIO = self.IO
self.stats = psutil.net_if_stats()
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.addrs = psutil.net_if_addrs()
self.IO = psutil.net_io_counters(pernic=True)
self.ifaces = self.stats.keys()
self.last: float = time.perf_counter()
self.last = time.perf_counter()
self.dt = self.last - self.prev
def fetcher(self) -> None:
def fetcher(self):
self.fetchData()
# Add missing sections
lastSection: NetworkProvider | NetworkProviderSection = self
lastSection = self
for iface in sorted(list(self.ifaces)):
if iface not in self.sections.keys():
section = NetworkProviderSection(iface, self)
@ -448,11 +395,15 @@ class NetworkProvider(Section, PeriodicUpdater):
return None
def __init__(self, theme: int | None = None):
def addParent(self, parent):
self.parents.add(parent)
self.refreshData()
def __init__(self, theme=None):
PeriodicUpdater.__init__(self)
Section.__init__(self, theme)
self.sections: dict[str, NetworkProviderSection] = dict()
self.sections = dict()
self.last = 0
self.IO = dict()
self.fetchData()
@ -464,7 +415,7 @@ class RfkillProvider(Section, PeriodicUpdater):
# toggled
PATH = "/sys/class/rfkill"
def fetcher(self) -> Element:
def fetcher(self):
t = Text()
for device in os.listdir(self.PATH):
with open(os.path.join(self.PATH, device, "soft"), "rb") as f:
@ -478,7 +429,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") or None
fg = (hardBlocked and "#CCCCCC") or (softBlocked and "#FF0000")
if typ == b"wlan":
icon = ""
elif typ == b"bluetooth":
@ -489,14 +440,14 @@ class RfkillProvider(Section, PeriodicUpdater):
t.append(Text(icon, fg=fg))
return t
def __init__(self, theme: int | None = None):
def __init__(self, theme=None):
PeriodicUpdater.__init__(self)
Section.__init__(self, theme)
self.changeInterval(5)
class SshAgentProvider(PeriodicUpdater):
def fetcher(self) -> Element:
def fetcher(self):
cmd = ["ssh-add", "-l"]
proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
if proc.returncode != 0:
@ -509,13 +460,13 @@ class SshAgentProvider(PeriodicUpdater):
text.append(Text("", fg=randomColor(seed=fingerprint)))
return text
def __init__(self) -> None:
def __init__(self):
PeriodicUpdater.__init__(self)
self.changeInterval(5)
class GpgAgentProvider(PeriodicUpdater):
def fetcher(self) -> Element:
def fetcher(self):
cmd = ["gpg-connect-agent", "keyinfo --list", "/bye"]
proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
# proc = subprocess.run(cmd)
@ -532,7 +483,7 @@ class GpgAgentProvider(PeriodicUpdater):
text.append(Text("", fg=randomColor(seed=keygrip)))
return text
def __init__(self) -> None:
def __init__(self):
PeriodicUpdater.__init__(self)
self.changeInterval(5)
@ -541,7 +492,7 @@ class KeystoreProvider(Section, MergedUpdater):
# TODO OPTI+FEAT Use ColorCountsSection and not MergedUpdater, this is useless
ICON = ""
def __init__(self, theme: int | None = None):
def __init__(self, theme=None):
MergedUpdater.__init__(self, SshAgentProvider(), GpgAgentProvider())
Section.__init__(self, theme)
@ -549,21 +500,24 @@ class KeystoreProvider(Section, MergedUpdater):
class NotmuchUnreadProvider(ColorCountsSection, InotifyUpdater):
COLORABLE_ICON = ""
def subfetcher(self) -> list[tuple[int, str]]:
def subfetcher(self):
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: str = "~/.mail/", theme: int | None = None):
InotifyUpdater.__init__(self)
def __init__(self, dir="~/.mail/", theme=None):
PeriodicUpdater.__init__(self)
ColorCountsSection.__init__(self, theme)
self.dir = os.path.realpath(os.path.expanduser(dir))
@ -589,7 +543,7 @@ class TodoProvider(ColorCountsSection, InotifyUpdater):
# TODO OPT Specific callback for specific directory
COLORABLE_ICON = ""
def updateCalendarList(self) -> None:
def updateCalendarList(self):
calendars = sorted(os.listdir(self.dir))
for calendar in calendars:
# If the calendar wasn't in the list
@ -605,9 +559,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: list[str] = calendars
self.calendars = calendars
def __init__(self, dir: str, theme: int | None = None):
def __init__(self, dir, theme=None):
"""
:parm str dir: [main]path value in todoman.conf
"""
@ -617,12 +571,12 @@ class TodoProvider(ColorCountsSection, InotifyUpdater):
assert os.path.isdir(self.dir)
self.calendars = []
self.colors: dict[str, str] = dict()
self.names: dict[str, str] = dict()
self.colors = dict()
self.names = dict()
self.updateCalendarList()
self.refreshData()
def countUndone(self, calendar: str | None) -> int:
def countUndone(self, calendar):
cmd = ["todo", "--porcelain", "list"]
if calendar:
cmd.append(self.names[calendar])
@ -630,7 +584,7 @@ class TodoProvider(ColorCountsSection, InotifyUpdater):
data = json.loads(proc.stdout)
return len(data)
def subfetcher(self) -> list[tuple[int, str]]:
def subfetcher(self):
counts = []
# TODO This an ugly optimisation that cuts on features, but todoman
@ -655,107 +609,124 @@ 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: i3ipc.Connection, e: i3ipc.Event) -> None:
def on_window(self, i3, e):
self.updateText(e.container.name)
def __init__(self, theme: int | None = None):
def __init__(self, theme=None):
I3Updater.__init__(self)
Section.__init__(self, theme=theme)
self.on("window", self.on_window)
class I3WorkspacesProviderSection(Section):
def selectTheme(self) -> int:
if self.workspace.urgent:
def selectTheme(self):
if self.urgent:
return self.parent.themeUrgent
elif self.workspace.focused:
elif self.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) -> None:
def show(self):
self.updateTheme(self.selectTheme())
self.updateText(
self.fullName if self.workspace.focused else self.workspace.name
)
self.updateText(self.fullName if self.focused else self.shortName)
def switchTo(self) -> None:
self.parent.i3.command("workspace {}".format(self.workspace.name))
def updateWorkspace(self, workspace: i3ipc.WorkspaceReply) -> None:
self.workspace = workspace
self.fullName: str = self.parent.customNames.get(workspace.name, workspace.name)
def changeState(self, focused, urgent):
self.focused = focused
self.urgent = urgent
self.show()
def __init__(self, parent: "I3WorkspacesProvider"):
def setName(self, name):
self.shortName = name
self.fullName = (
self.parent.customNames[name] if name in self.parent.customNames else name
)
def switchTo(self):
self.parent.i3.command("workspace {}".format(self.shortName))
def __init__(self, name, parent):
Section.__init__(self)
self.parent = parent
self.setName(name)
self.setDecorators(clickLeft=self.switchTo)
self.tempText: Element = None
self.tempText = None
def empty(self) -> None:
def empty(self):
self.updateTheme(self.parent.themeNormal)
self.updateText(None)
def tempShow(self) -> None:
def tempShow(self):
self.updateText(self.tempText)
def tempEmpty(self) -> None:
def tempEmpty(self):
self.tempText = self.dstText[1]
self.updateText(None)
class I3WorkspacesProvider(Section, I3Updater):
# TODO FEAT Multi-screen
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:
section = I3WorkspacesProviderSection(self)
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
for bargroup in self.parents:
if bargroup.parent.output == workspace.output:
break
lastSection = section
def on_workspace_init(self, i3, e):
workspace = e.current
i = workspace.num
if i in self.sections:
section = self.sections[i]
else:
bargroup = list(self.parents)[0]
bargroup.addSectionAfter(lastSectionOnOutput, section)
section.updateWorkspace(workspace)
# 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
def updateWorkspaces(self) -> None:
workspaces = self.i3.get_workspaces()
for workspace in workspaces:
self.updateWorkspace(workspace)
section = I3WorkspacesProviderSection(workspace.name, self)
prevSection.appendAfter(section)
self.sections[workspace.num] = section
section.focused = workspace.focused
section.urgent = workspace.urgent
section.show()
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:
def on_workspace_empty(self, i3, e):
self.sections[e.current.num].empty()
def on_mode(self, i3: i3ipc.Connection, e: i3ipc.Event) -> None:
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):
if e.change == "default":
self.modeSection.updateText(None)
for section in self.sections.values():
@ -766,46 +737,41 @@ class I3WorkspacesProvider(Section, I3Updater):
section.tempEmpty()
def __init__(
self,
theme: int = 0,
themeVisible: int = 4,
themeFocus: int = 3,
themeUrgent: int = 1,
themeMode: int = 2,
customNames: dict[str, str] = dict(),
self, theme=0, themeFocus=3, themeUrgent=1, themeMode=2, customNames=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[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.sections = dict()
self.on("workspace::init", self.on_workspace_init)
self.on("workspace::focus", self.on_workspace_focus)
self.on("workspace::empty", self.on_workspace_empty)
self.on("workspace::urgent", self.on_workspace_change)
self.on("workspace::rename", self.on_workspace_change)
self.on("workspace::urgent", self.on_workspace_urgent)
self.on("workspace::rename", self.on_workspace_rename)
# TODO Un-handled/tested: reload, rename, restored, move
self.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) -> None:
def connect(self):
self.mpd.connect("localhost", 6600)
def __init__(self, theme: int | None = None):
def __init__(self, theme=None):
ThreadedUpdater.__init__(self)
Section.__init__(self, theme)
@ -814,7 +780,7 @@ class MpdProvider(Section, ThreadedUpdater):
self.refreshData()
self.start()
def fetcher(self) -> Element:
def fetcher(self):
stat = self.mpd.status()
if not len(stat) or stat["state"] == "stop":
return None
@ -825,7 +791,7 @@ class MpdProvider(Section, ThreadedUpdater):
infos = []
def tryAdd(field: str) -> None:
def tryAdd(field):
if field in cur:
infos.append(cur[field])
@ -839,7 +805,7 @@ class MpdProvider(Section, ThreadedUpdater):
return "{}".format(infosStr)
def loop(self) -> None:
def loop(self):
try:
self.mpd.idle("player")
self.refreshData()

View file

@ -11,8 +11,8 @@ import coloredlogs
import i3ipc
import pyinotify
from frobar.display import Element
from frobar.common import notBusy
from frobar.display import Text
from frobar.notbusy import notBusy
coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s")
log = logging.getLogger()
@ -20,23 +20,24 @@ log = logging.getLogger()
# TODO Sync bar update with PeriodicUpdater updates
class Updater:
@staticmethod
def init() -> None:
def init():
PeriodicUpdater.init()
InotifyUpdater.init()
notBusy.set()
def updateText(self, text: Element) -> None:
def updateText(self, text):
print(text)
def fetcher(self) -> Element:
def fetcher(self):
return "{} refreshed".format(self)
def __init__(self) -> None:
def __init__(self):
self.lock = threading.Lock()
def refreshData(self) -> None:
def refreshData(self):
# TODO OPTI Maybe discard the refresh if there's already another one?
self.lock.acquire()
try:
@ -49,7 +50,7 @@ class Updater:
class PeriodicUpdaterThread(threading.Thread):
def run(self) -> None:
def run(self):
# TODO Sync with system clock
counter = 0
while True:
@ -66,7 +67,6 @@ 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,42 +80,43 @@ class PeriodicUpdater(Updater):
Needs to call :func:`PeriodicUpdater.changeInterval` in `__init__`
"""
intervals: dict[int, set["PeriodicUpdater"]] = dict()
intervalStep: int | None = None
intervalLoop: int
updateThread: threading.Thread = PeriodicUpdaterThread(daemon=True)
intervals = dict()
intervalStep = None
intervalLoop = None
updateThread = PeriodicUpdaterThread(daemon=True)
intervalsChanged = threading.Event()
@staticmethod
def gcds(*args: int) -> int:
def gcds(*args):
return functools.reduce(math.gcd, args)
@staticmethod
def lcm(a: int, b: int) -> int:
def lcm(a, b):
"""Return lowest common multiple."""
return a * b // math.gcd(a, b)
@staticmethod
def lcms(*args: int) -> int:
def lcms(*args):
"""Return lowest common multiple."""
return functools.reduce(PeriodicUpdater.lcm, args)
@staticmethod
def updateIntervals() -> None:
def updateIntervals():
intervalsList = list(PeriodicUpdater.intervals.keys())
PeriodicUpdater.intervalStep = PeriodicUpdater.gcds(*intervalsList)
PeriodicUpdater.intervalLoop = PeriodicUpdater.lcms(*intervalsList)
PeriodicUpdater.intervalsChanged.set()
@staticmethod
def init() -> None:
def init():
PeriodicUpdater.updateThread.start()
def __init__(self) -> None:
def __init__(self):
Updater.__init__(self)
self.interval: int | None = None
self.interval = None
def changeInterval(self, interval: int) -> None:
def changeInterval(self, interval):
assert isinstance(interval, int)
if self.interval is not None:
PeriodicUpdater.intervals[self.interval].remove(self)
@ -130,7 +131,7 @@ class PeriodicUpdater(Updater):
class InotifyUpdaterEventHandler(pyinotify.ProcessEvent):
def process_default(self, event: pyinotify.Event) -> None:
def process_default(self, event):
# DEBUG
# from pprint import pprint
# pprint(event.__dict__)
@ -153,10 +154,10 @@ class InotifyUpdater(Updater):
"""
wm = pyinotify.WatchManager()
paths: dict[str, dict[str | int, set["InotifyUpdater"]]] = dict()
paths = dict()
@staticmethod
def init() -> None:
def init():
notifier = pyinotify.ThreadedNotifier(
InotifyUpdater.wm, InotifyUpdaterEventHandler()
)
@ -165,14 +166,14 @@ class InotifyUpdater(Updater):
# TODO Mask for folders
MASK = pyinotify.IN_CREATE | pyinotify.IN_MODIFY | pyinotify.IN_DELETE
def addPath(self, path: str, refresh: bool = True) -> None:
def addPath(self, path, refresh=True):
path = os.path.realpath(os.path.expanduser(path))
# Detect if file or folder
if os.path.isdir(path):
self.dirpath: str = path
self.dirpath = path
# 0: Directory watcher
self.filename: str | int = 0
self.filename = 0
elif os.path.isfile(path):
self.dirpath = os.path.dirname(path)
self.filename = os.path.basename(path)
@ -194,12 +195,12 @@ class InotifyUpdater(Updater):
class ThreadedUpdaterThread(threading.Thread):
def __init__(self, updater: "ThreadedUpdater") -> None:
def __init__(self, updater, *args, **kwargs):
self.updater = updater
threading.Thread.__init__(self, daemon=True)
threading.Thread.__init__(self, *args, **kwargs)
self.looping = True
def run(self) -> None:
def run(self):
try:
while self.looping:
self.updater.loop()
@ -214,31 +215,57 @@ class ThreadedUpdater(Updater):
Must implement loop(), and call start()
"""
def __init__(self) -> None:
def __init__(self):
Updater.__init__(self)
self.thread = ThreadedUpdaterThread(self)
self.thread = ThreadedUpdaterThread(self, daemon=True)
def loop(self) -> None:
def loop(self):
self.refreshData()
time.sleep(10)
def start(self) -> None:
def start(self):
self.thread.start()
class I3Updater(ThreadedUpdater):
# TODO OPTI One i3 connection for all
def __init__(self) -> None:
def __init__(self):
ThreadedUpdater.__init__(self)
self.i3 = i3ipc.Connection()
self.on = self.i3.on
self.start()
def loop(self) -> None:
def on(self, event, function):
self.i3.on(event, function)
def loop(self):
self.i3.main()
class MergedUpdater(Updater):
def __init__(self, *args: Updater) -> None:
raise NotImplementedError("Deprecated, as hacky and currently unused")
# 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] = ""

View file

@ -1,25 +0,0 @@
{ 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,7 +11,6 @@ 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 {
@ -44,27 +43,12 @@ in
keybindings = {
# Screen off commands
"${mod}+F1" = "--release exec --no-startup-id ${pkgs.xorg.xset}/bin/xset dpms force off";
# 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
''}";
"${mod}+F4" = "exec --no-startup-id ${pkgs.xautolock}/bin/xautolock -disable";
"${mod}+F5" = "exec --no-startup-id ${pkgs.xautolock}/bin/xautolock -enable";
};
startup = [
# Stop screen after 10 minutes, 1 minutes after lock it
{ 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
''}"; }
{ notification = false; command = "${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" ];
modulesRight = [ "jobs" "exit" "duration" "load" ];
settings = {
colorize-hostname = true;
hostname-only-if-ssh = true;
@ -20,3 +20,4 @@
};
};
}
# TODO Replace load with a frobar indicator

View file

@ -0,0 +1,10 @@
# 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

@ -0,0 +1,9 @@
[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

@ -0,0 +1,34 @@
# 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

@ -0,0 +1,30 @@
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

@ -0,0 +1,14 @@
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

@ -0,0 +1,4 @@
curacao.geoffrey.frogeye.fr
# triffle.geoffrey.frogeye.fr
pindakaas.geoffrey.frogeye.fr
gho.geoffrey.frogeye.fr ansible_host=localhost ansible_port=2222

View file

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

View file

@ -0,0 +1,44 @@
[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

View file

@ -0,0 +1,4 @@
# 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

@ -0,0 +1,32 @@
#!/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

@ -0,0 +1,26 @@
[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

@ -0,0 +1,24 @@
#!/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

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

View file