Compare commits
No commits in common. "6690f3aa0d6804299db7a6da60a37e9a850b22b7" and "139d3a3ea8e10f1424e940c22b79f28458c8cc6c" have entirely different histories.
6690f3aa0d
...
139d3a3ea8
|
|
@ -20,15 +20,27 @@ class ComposableText:
|
|||
assert isinstance(prevParent, Bar)
|
||||
self.bar: Bar = prevParent
|
||||
|
||||
def updateMarkup(self) -> None:
|
||||
self.text: str
|
||||
self.needsComposition = True
|
||||
|
||||
def composeText(self) -> str:
|
||||
raise NotImplementedError(f"{self} cannot compose text")
|
||||
|
||||
def getText(self) -> str:
|
||||
if self.needsComposition:
|
||||
self.text = self.composeText()
|
||||
print(f"{self} composed {self.text}")
|
||||
self.needsComposition = False
|
||||
return self.text
|
||||
|
||||
def setText(self, text: str) -> None:
|
||||
self.text = text
|
||||
self.needsComposition = False
|
||||
parent = self.parent
|
||||
while parent and not parent.needsComposition:
|
||||
parent.needsComposition = True
|
||||
parent = parent.parent
|
||||
self.bar.refresh.set()
|
||||
# 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()
|
||||
|
||||
|
||||
def randomColor(seed: int | bytes | None = None) -> str:
|
||||
|
|
@ -45,53 +57,6 @@ class Section(ComposableText):
|
|||
def __init__(self, parent: "Module") -> None:
|
||||
super().__init__(parent=parent)
|
||||
self.color = randomColor()
|
||||
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
|
||||
|
||||
|
||||
class Module(ComposableText):
|
||||
|
|
@ -102,23 +67,9 @@ class Module(ComposableText):
|
|||
def __init__(self, parent: "Side") -> None:
|
||||
super().__init__(parent=parent)
|
||||
self.sections: list[Section] = []
|
||||
self.mirroring: Module | None = None
|
||||
self.mirrors: list[Module] = list()
|
||||
|
||||
def mirror(self, module: "Module") -> None:
|
||||
self.mirroring = module
|
||||
module.mirrors.append(self)
|
||||
|
||||
def getSections(self) -> list[Section]:
|
||||
if self.mirroring:
|
||||
return self.mirroring.sections
|
||||
else:
|
||||
return self.sections
|
||||
|
||||
def updateMarkup(self) -> None:
|
||||
super().updateMarkup()
|
||||
for mirror in self.mirrors:
|
||||
mirror.updateMarkup()
|
||||
def appendSection(self, section: Section) -> None:
|
||||
self.sections.append(section)
|
||||
|
||||
|
||||
class Alignment(enum.Enum):
|
||||
|
|
@ -133,15 +84,13 @@ class Side(ComposableText):
|
|||
self.alignment = alignment
|
||||
self.modules: list[Module] = []
|
||||
|
||||
def generateMarkup(self) -> str:
|
||||
def composeText(self) -> str:
|
||||
if not self.modules:
|
||||
return ""
|
||||
text = "%{" + self.alignment.value + "}"
|
||||
lastSection: Section | None = None
|
||||
for module in self.modules:
|
||||
for section in module.getSections():
|
||||
if section.isHidden():
|
||||
continue
|
||||
for section in module.sections:
|
||||
if lastSection is None:
|
||||
if self.alignment == Alignment.LEFT:
|
||||
text += "%{B" + section.color + "}%{F-}"
|
||||
|
|
@ -159,7 +108,9 @@ class Side(ComposableText):
|
|||
else:
|
||||
text += "%{R}%{B" + section.color + "}"
|
||||
text += "%{F-}"
|
||||
text += section.getMarkup()
|
||||
text += " " + section.getText() + " "
|
||||
# FIXME Should be handled by animation
|
||||
# FIXME Support hidden sections
|
||||
lastSection = section
|
||||
if self.alignment != Alignment.RIGHT:
|
||||
text += "%{R}%{B-}"
|
||||
|
|
@ -175,12 +126,15 @@ class Screen(ComposableText):
|
|||
for alignment in Alignment:
|
||||
self.sides[alignment] = Side(parent=self, alignment=alignment)
|
||||
|
||||
def generateMarkup(self) -> str:
|
||||
def composeText(self) -> str:
|
||||
return ("%{Sn" + self.output + "}") + "".join(
|
||||
side.getMarkup() for side in self.sides.values()
|
||||
side.getText() for side in self.sides.values()
|
||||
)
|
||||
|
||||
|
||||
ScreenSelection = enum.Enum("ScreenSelection", ["ALL", "PRIMARY", "SECONDARY"])
|
||||
|
||||
|
||||
class Bar(ComposableText):
|
||||
"""
|
||||
Top-level
|
||||
|
|
@ -188,9 +142,8 @@ class Bar(ComposableText):
|
|||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.providers: list["Provider"] = []
|
||||
self.refresh = asyncio.Event()
|
||||
self.taskGroup = asyncio.TaskGroup()
|
||||
self.providers: list["Provider"] = list()
|
||||
|
||||
self.screens = []
|
||||
i3 = i3ipc.Connection()
|
||||
|
|
@ -201,43 +154,45 @@ class Bar(ComposableText):
|
|||
self.screens.append(screen)
|
||||
|
||||
async def run(self) -> None:
|
||||
cmd = [
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
"lemonbar",
|
||||
"-b",
|
||||
"-a",
|
||||
"64",
|
||||
"-f",
|
||||
"DejaVuSansM Nerd Font:size=10",
|
||||
]
|
||||
proc = await asyncio.create_subprocess_exec(*cmd, stdin=asyncio.subprocess.PIPE)
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
)
|
||||
|
||||
async def refresher() -> None:
|
||||
assert proc.stdin
|
||||
while True:
|
||||
await self.refresh.wait()
|
||||
self.refresh.clear()
|
||||
proc.stdin.write(self.getMarkup().encode())
|
||||
proc.stdin.write(self.getText().encode())
|
||||
|
||||
async with self.taskGroup as tg:
|
||||
async with asyncio.TaskGroup() as tg:
|
||||
tg.create_task(refresher())
|
||||
for provider in self.providers:
|
||||
tg.create_task(provider.run())
|
||||
for updater in self.providers:
|
||||
tg.create_task(updater.run())
|
||||
|
||||
def generateMarkup(self) -> str:
|
||||
return "".join(section.getMarkup() for section in self.screens) + "\n"
|
||||
def composeText(self) -> str:
|
||||
return "".join(section.getText() for section in self.screens) + "\n"
|
||||
|
||||
def addProvider(
|
||||
self,
|
||||
provider: "Provider",
|
||||
alignment: Alignment = Alignment.LEFT,
|
||||
screenNum: int | None = None,
|
||||
screenSelection: ScreenSelection = ScreenSelection.ALL,
|
||||
) -> None:
|
||||
"""
|
||||
screenNum: the provider will be added on this screen if set, all otherwise
|
||||
"""
|
||||
# FIXME Actually have a screenNum and screenCount args
|
||||
modules = list()
|
||||
for s, screen in enumerate(self.screens):
|
||||
if screenNum is None or s == screenNum:
|
||||
if (
|
||||
screenSelection == ScreenSelection.ALL
|
||||
or (screenSelection == ScreenSelection.PRIMARY and s == 0)
|
||||
or (screenSelection == ScreenSelection.SECONDARY and s == 1)
|
||||
):
|
||||
side = screen.sides[alignment]
|
||||
module = Module(parent=side)
|
||||
side.modules.append(module)
|
||||
|
|
@ -255,79 +210,32 @@ class Provider:
|
|||
raise NotImplementedError()
|
||||
|
||||
|
||||
class MirrorProvider(Provider):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.module: Module
|
||||
|
||||
class TimeProvider(Provider):
|
||||
async def run(self) -> None:
|
||||
self.module = self.modules[0]
|
||||
for module in self.modules[1:]:
|
||||
module.mirror(self.module)
|
||||
|
||||
|
||||
class SingleSectionProvider(MirrorProvider):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.section: Section
|
||||
|
||||
async def run(self) -> None:
|
||||
await super().run()
|
||||
self.section = Section(parent=self.module)
|
||||
self.module.sections.append(self.section)
|
||||
|
||||
|
||||
class StaticProvider(SingleSectionProvider):
|
||||
def __init__(self, text: str) -> None:
|
||||
self.text = text
|
||||
|
||||
async def run(self) -> None:
|
||||
await super().run()
|
||||
self.section.setText(self.text)
|
||||
|
||||
|
||||
class TimeProvider(SingleSectionProvider):
|
||||
async def run(self) -> None:
|
||||
await super().run()
|
||||
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:
|
||||
now = datetime.datetime.now()
|
||||
# section.setText(now.strftime("%a %y-%m-%d %H:%M:%S.%f"))
|
||||
self.section.setText("-" * (now.second % 10))
|
||||
remaining = 1 - now.microsecond / 1000000
|
||||
await asyncio.sleep(remaining)
|
||||
for s, section in enumerate(sections):
|
||||
now = datetime.datetime.now()
|
||||
section.setText(now.strftime("%a %y-%m-%d %H:%M:%S"))
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
bar = Bar()
|
||||
dualScreen = len(bar.screens) > 1
|
||||
|
||||
bar.addProvider(StaticProvider(text="i3 workspaces"), alignment=Alignment.LEFT)
|
||||
if dualScreen:
|
||||
bar.addProvider(
|
||||
StaticProvider(text="i3 title"), screenNum=0, alignment=Alignment.CENTER
|
||||
)
|
||||
bar.addProvider(
|
||||
StaticProvider(text="mpris"),
|
||||
screenNum=1 if dualScreen else None,
|
||||
alignment=Alignment.CENTER,
|
||||
)
|
||||
|
||||
bar.addProvider(StaticProvider("C L M T B"), alignment=Alignment.RIGHT)
|
||||
bar.addProvider(
|
||||
StaticProvider("pulse"),
|
||||
screenNum=1 if dualScreen else None,
|
||||
alignment=Alignment.RIGHT,
|
||||
)
|
||||
bar.addProvider(
|
||||
StaticProvider("network"),
|
||||
screenNum=0 if dualScreen else None,
|
||||
alignment=Alignment.RIGHT,
|
||||
)
|
||||
bar.addProvider(TimeProvider(), alignment=Alignment.RIGHT)
|
||||
|
||||
bar.addProvider(TimeProvider())
|
||||
# bar.addProvider(TimeProvider(), screenSelection=ScreenSelection.PRIMARY)
|
||||
# bar.addProvider(TimeProvider(), alignment=Alignment.CENTER)
|
||||
# bar.addProvider(TimeProvider(), alignment=Alignment.CENTER)
|
||||
# bar.addProvider(TimeProvider(), alignment=Alignment.RIGHT)
|
||||
# bar.addProvider(TimeProvider(), alignment=Alignment.RIGHT)
|
||||
await bar.run()
|
||||
# FIXME Why time not updating?
|
||||
|
||||
|
||||
asyncio.run(main())
|
||||
# TODO Replace while True with while bar.running or something
|
||||
|
|
|
|||
Loading…
Reference in a new issue