dotfiles/hm/desktop/frobar/.dev/new.py

277 lines
8.2 KiB
Python
Raw Normal View History

2024-08-13 01:58:51 +02:00
#!/usr/bin/env python3
import asyncio
import datetime
import enum
import random
import typing
import i3ipc
class ComposableText:
def __init__(self, parent: typing.Optional["ComposableText"] = None) -> None:
self.parent = parent
prevParent = self
while parent:
prevParent = parent
parent = parent.parent
assert isinstance(prevParent, Bar)
self.bar: Bar = prevParent
2024-08-14 00:28:17 +02:00
def updateMarkup(self) -> None:
2024-08-13 01:58:51 +02:00
self.bar.refresh.set()
2024-08-14 00:28:17 +02:00
# OPTI See if worth caching the output
def generateMarkup(self) -> str:
raise NotImplementedError(f"{self} cannot generate markup")
def getMarkup(self) -> str:
return self.generateMarkup()
2024-08-13 01:58:51 +02:00
def randomColor(seed: int | bytes | None = None) -> str:
if seed is not None:
random.seed(seed)
return "#" + "".join(f"{random.randint(0, 0xff):02x}" for _ in range(3))
class Section(ComposableText):
"""
Colorable block separated by chevrons
"""
def __init__(self, parent: "Module") -> None:
super().__init__(parent=parent)
self.color = randomColor()
2024-08-14 00:28:17 +02:00
self.text: str = ""
self.size = 0
self.animationTask: asyncio.Task | None = None
def isHidden(self) -> bool:
return self.text is None
# Geometric series
ANIM_A = 0.025
ANIM_R = 0.9
async def animate(self) -> None:
targetSize = len(self.text)
increment = 1 if self.size < targetSize else -1
loop = asyncio.get_running_loop()
frameTime = loop.time()
animTime = self.ANIM_A
while self.size != targetSize:
self.size += increment
self.updateMarkup()
animTime *= self.ANIM_R
frameTime += animTime
sleepTime = frameTime - loop.time()
# In case of stress, skip refreshing by not awaiting
if sleepTime > 0:
await asyncio.sleep(sleepTime)
def setText(self, text: str | None) -> None:
# OPTI Skip if same text
oldText = self.text
self.text = f" {text} "
if oldText == self.text:
return
if len(oldText) == len(self.text):
self.updateMarkup()
else:
if self.animationTask:
self.animationTask.cancel()
self.animationTask = self.bar.taskGroup.create_task(self.animate())
def generateMarkup(self) -> str:
pad = max(0, self.size - len(self.text))
return self.text[: self.size] + " " * pad
2024-08-13 01:58:51 +02:00
class Module(ComposableText):
"""
Sections handled by a same updater
"""
def __init__(self, parent: "Side") -> None:
super().__init__(parent=parent)
self.sections: list[Section] = []
class Alignment(enum.Enum):
LEFT = "l"
RIGHT = "r"
CENTER = "c"
class Side(ComposableText):
def __init__(self, parent: "Screen", alignment: Alignment) -> None:
super().__init__(parent=parent)
self.alignment = alignment
self.modules: list[Module] = []
2024-08-14 00:28:17 +02:00
def generateMarkup(self) -> str:
2024-08-13 01:58:51 +02:00
if not self.modules:
return ""
text = "%{" + self.alignment.value + "}"
lastSection: Section | None = None
for module in self.modules:
for section in module.sections:
2024-08-14 00:28:17 +02:00
if section.isHidden():
continue
2024-08-13 01:58:51 +02:00
if lastSection is None:
if self.alignment == Alignment.LEFT:
text += "%{B" + section.color + "}%{F-}"
else:
text += "%{B-}%{F" + section.color + "}%{R}%{F-}"
else:
if self.alignment == Alignment.RIGHT:
if lastSection.color == section.color:
text += ""
else:
text += "%{F" + section.color + "}%{R}"
else:
if lastSection.color == section.color:
text += ""
else:
text += "%{R}%{B" + section.color + "}"
text += "%{F-}"
2024-08-14 00:28:17 +02:00
text += section.getMarkup()
2024-08-13 01:58:51 +02:00
lastSection = section
if self.alignment != Alignment.RIGHT:
text += "%{R}%{B-}"
return text
class Screen(ComposableText):
def __init__(self, parent: "Bar", output: str) -> None:
super().__init__(parent=parent)
self.output = output
self.sides = dict()
for alignment in Alignment:
self.sides[alignment] = Side(parent=self, alignment=alignment)
2024-08-14 00:28:17 +02:00
def generateMarkup(self) -> str:
2024-08-13 01:58:51 +02:00
return ("%{Sn" + self.output + "}") + "".join(
2024-08-14 00:28:17 +02:00
side.getMarkup() for side in self.sides.values()
2024-08-13 01:58:51 +02:00
)
class Bar(ComposableText):
"""
Top-level
"""
def __init__(self) -> None:
super().__init__()
self.refresh = asyncio.Event()
2024-08-14 00:28:17 +02:00
self.taskGroup = asyncio.TaskGroup()
self.providers: list["Provider"] = list()
2024-08-13 01:58:51 +02:00
self.screens = []
i3 = i3ipc.Connection()
for output in i3.get_outputs():
if not output.active:
continue
screen = Screen(parent=self, output=output.name)
self.screens.append(screen)
async def run(self) -> None:
proc = await asyncio.create_subprocess_exec(
"lemonbar",
"-b",
"-a",
"64",
"-f",
"DejaVuSansM Nerd Font:size=10",
stdin=asyncio.subprocess.PIPE,
)
async def refresher() -> None:
assert proc.stdin
while True:
await self.refresh.wait()
self.refresh.clear()
2024-08-14 00:28:17 +02:00
proc.stdin.write(self.getMarkup().encode())
2024-08-13 01:58:51 +02:00
2024-08-14 00:28:17 +02:00
async with self.taskGroup as tg:
2024-08-13 01:58:51 +02:00
tg.create_task(refresher())
2024-08-14 00:28:17 +02:00
for provider in self.providers:
tg.create_task(provider.run())
2024-08-13 01:58:51 +02:00
2024-08-14 00:28:17 +02:00
def generateMarkup(self) -> str:
return "".join(section.getMarkup() for section in self.screens) + "\n"
2024-08-13 01:58:51 +02:00
def addProvider(
self,
provider: "Provider",
alignment: Alignment = Alignment.LEFT,
2024-08-14 00:28:17 +02:00
screenNum: int | None = None,
screenCount: int | None = None,
2024-08-13 01:58:51 +02:00
) -> None:
2024-08-14 00:28:17 +02:00
"""
screenNum: the provider will be added on this screen if set, all otherwise
screenCount: the provider will be added if there is this many screens,
always otherwise
"""
2024-08-13 01:58:51 +02:00
modules = list()
for s, screen in enumerate(self.screens):
2024-08-14 00:28:17 +02:00
if (screenCount is None or len(self.screens) == screenCount) and (
screenNum is None or s == screenNum
2024-08-13 01:58:51 +02:00
):
side = screen.sides[alignment]
module = Module(parent=side)
side.modules.append(module)
modules.append(module)
provider.modules = modules
if modules:
self.providers.append(provider)
class Provider:
def __init__(self) -> None:
self.modules: list[Module] = list()
async def run(self) -> None:
raise NotImplementedError()
class TimeProvider(Provider):
async def run(self) -> None:
sections = list()
for module in self.modules:
section = Section(parent=module)
module.sections.append(section)
sections.append(section)
# FIXME Allow for mirror(ed) modules so no need for updaters to handle all
while True:
2024-08-14 00:28:17 +02:00
now = datetime.datetime.now()
2024-08-13 01:58:51 +02:00
for s, section in enumerate(sections):
2024-08-14 00:28:17 +02:00
# section.setText(now.strftime("%a %y-%m-%d %H:%M:%S.%f"))
section.setText("-" * (now.second % 10))
remaining = 1 - now.microsecond / 1000000
await asyncio.sleep(remaining)
2024-08-13 01:58:51 +02:00
async def main() -> None:
bar = Bar()
bar.addProvider(TimeProvider())
2024-08-14 00:28:17 +02:00
bar.addProvider(TimeProvider(), screenNum=1)
bar.addProvider(TimeProvider(), alignment=Alignment.CENTER, screenCount=2)
bar.addProvider(TimeProvider(), alignment=Alignment.CENTER)
bar.addProvider(TimeProvider(), alignment=Alignment.RIGHT)
bar.addProvider(TimeProvider(), alignment=Alignment.RIGHT)
2024-08-13 01:58:51 +02:00
await bar.run()
asyncio.run(main())
2024-08-14 00:28:17 +02:00
# TODO Replace while True with while bar.running or something