geoffrey/dotfiles
geoffrey/dotfiles
1
0
Fork 0
Code Issues 7 Pull requests Releases Activity

frobarng: Some refactor

Browse source
This commit is contained in:
Geoffrey Frogeye 2024-09-27 23:06:13 +02:00
parent 1ae7d6b447
commit 6644e85c30
Signed by: geoffrey
GPG key ID: C72403E7F82E6AD8
1 changed files with 103 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

173
hm/desktop/frobar/.dev/new.py
View file

@ -292,6 +292,8 @@ class Bar(ComposableText):
self.actionIndex = 0 self.actionIndex = 0
self.actions: dict[str, typing.Callable] = dict() self.actions: dict[str, typing.Callable] = dict()
self.periodicProviderTask: typing.Coroutine | None = None
i3 = i3ipc.Connection() i3 = i3ipc.Connection()
for output in i3.get_outputs(): for output in i3.get_outputs():
if not output.active: if not output.active:
@ -382,7 +384,8 @@ class Provider:
self.modules: list[Module] = list() self.modules: list[Module] = list()
async def run(self) -> None: async def run(self) -> None:
raise NotImplementedError() # Not a NotImplementedError, otherwise can't combine all classes
pass
class MirrorProvider(Provider): class MirrorProvider(Provider):
@ -391,17 +394,16 @@ class MirrorProvider(Provider):
self.module: Module self.module: Module
async def run(self) -> None: async def run(self) -> None:
await super().run()
self.module = self.modules[0] self.module = self.modules[0]
for module in self.modules[1:]: for module in self.modules[1:]:
module.mirror(self.module) module.mirror(self.module)
class SingleSectionProvider(MirrorProvider): class SingleSectionProvider(MirrorProvider):
SECTION_CLASS = Section
async def run(self) -> None: async def run(self) -> None:
await super().run() await super().run()
self.section = self.SECTION_CLASS(parent=self.module) self.section = Section(parent=self.module)
class StaticProvider(SingleSectionProvider): class StaticProvider(SingleSectionProvider):
@ -419,7 +421,6 @@ class StatefulSection(Section):
super().__init__(parent=parent, sortKey=sortKey) super().__init__(parent=parent, sortKey=sortKey)
self.state = 0 self.state = 0
self.numberStates: int self.numberStates: int
self.stateChanged = asyncio.Event()
self.setAction(Button.CLICK_LEFT, self.incrementState) self.setAction(Button.CLICK_LEFT, self.incrementState)
self.setAction(Button.CLICK_RIGHT, self.decrementState) self.setAction(Button.CLICK_RIGHT, self.decrementState)
@ -432,14 +433,61 @@ class StatefulSection(Section):
self.state -= 1 self.state -= 1
self.changeState() self.changeState()
def setChangedState(self, callback: typing.Callable) -> None:
self.callback = callback
def changeState(self) -> None: def changeState(self) -> None:
self.state %= self.numberStates self.state %= self.numberStates
self.stateChanged.set() self.bar.taskGroup.create_task(self.callback())
self.stateChanged.clear()
class StatefulProvider(SingleSectionProvider): class SingleStatefulSectionProvider(MirrorProvider):
SECTION_CLASS = StatefulSection async def run(self) -> None:
await super().run()
self.section = StatefulSection(parent=self.module)
class PeriodicProvider(Provider):
async def init(self) -> None:
pass
async def loop(self) -> None:
raise NotImplementedError()
@classmethod
async def task(cls, bar: Bar) -> None:
providers = list()
for provider in bar.providers:
if isinstance(provider, PeriodicProvider):
providers.append(provider)
await provider.init()
while True:
# TODO Block bar update during the periodic update of the loops
loops = [provider.loop() for provider in providers]
asyncio.gather(*loops)
now = datetime.datetime.now()
# Hardcoded to 1 second... not sure if we want some more than that,
# and if the logic to check if a task should run would be a win
# compared to the task itself
remaining = 1 - now.microsecond / 1000000
await asyncio.sleep(remaining)
async def run(self) -> None:
await super().run()
for module in self.modules:
bar = module.getFirstParentOfType(Bar)
assert bar
if not bar.periodicProviderTask:
bar.periodicProviderTask = PeriodicProvider.task(bar)
bar.addLongRunningTask(bar.periodicProviderTask)
class PeriodicStatefulProvider(SingleStatefulSectionProvider, PeriodicProvider):
async def run(self) -> None:
await super().run()
self.section.setChangedState(self.loop)
# Providers # Providers
@ -470,8 +518,8 @@ class I3WindowTitleProvider(SingleSectionProvider):
class I3WorkspacesProvider(Provider): class I3WorkspacesProvider(Provider):
# FIXME Custom names # TODO Custom names
# FIXME Colors # TODO Colors
async def updateWorkspaces(self, i3: i3ipc.Connection) -> None: async def updateWorkspaces(self, i3: i3ipc.Connection) -> None:
""" """
@ -578,14 +626,12 @@ class NetworkProviderSection(StatefulSection):
self.numberStates = 5 if self.wifi else 4 self.numberStates = 5 if self.wifi else 4
self.state = 1 if self.wifi else 0 self.state = 1 if self.wifi else 0
async def updateOnStateChange(self) -> None: self.setChangedState(self.update)
while True:
await self.stateChanged.wait()
await self.provider.updateIface(self.iface)
async def getText(self) -> str | None: async def update(self) -> None:
if self.ignore or not self.provider.if_stats[self.iface].isup: if self.ignore or not self.provider.if_stats[self.iface].isup:
return None self.setText(None)
return
text = self.icon text = self.icon
state = self.state + (0 if self.wifi else 1) # SSID state = self.state + (0 if self.wifi else 1) # SSID
@ -611,84 +657,71 @@ class NetworkProviderSection(StatefulSection):
recv = self.provider.io_counters[self.iface].bytes_recv recv = self.provider.io_counters[self.iface].bytes_recv
prevSent = self.provider.prev_io_counters[self.iface].bytes_sent prevSent = self.provider.prev_io_counters[self.iface].bytes_sent
sent = self.provider.io_counters[self.iface].bytes_sent sent = self.provider.io_counters[self.iface].bytes_sent
recvDiff = recv - prevRecv
sentDiff = sent - prevSent
dt = self.provider.time - self.provider.prev_time dt = self.provider.time - self.provider.prev_time
recvDiff /= dt
sentDiff /= dt
recvDiff = (recv - prevRecv) / dt
sentDiff = (sent - prevSent) / dt
text += f" ↓{humanSize(recvDiff)}↑{humanSize(sentDiff)}" text += f" ↓{humanSize(recvDiff)}↑{humanSize(sentDiff)}"
if state >= 4: # Counter if state >= 4: # Counter
text += f" ⇓{humanSize(recv)}⇑{humanSize(sent)}" text += f" ⇓{humanSize(recv)}⇑{humanSize(sent)}"
return text self.setText(text)
class NetworkProvider(MirrorProvider): class NetworkProvider(MirrorProvider, PeriodicProvider):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__()
self.sections: dict[str, NetworkProviderSection] = dict() self.sections: dict[str, NetworkProviderSection] = dict()
async def updateIface(self, iface: str) -> None: async def init(self) -> None:
section = self.sections[iface]
section.setText(await section.getText())
async def run(self) -> None:
await super().run()
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
self.time = loop.time() self.time = loop.time()
self.io_counters = psutil.net_io_counters(pernic=True) self.io_counters = psutil.net_io_counters(pernic=True)
while True: async def loop(self) -> None:
# Separate TaskGroup in case it takes longer than one second, loop = asyncio.get_running_loop()
# it doesn't fill the main TaskGroup async with asyncio.TaskGroup() as tg:
async with asyncio.TaskGroup() as tg: self.prev_io_counters = self.io_counters
self.prev_io_counters = self.io_counters self.prev_time = self.time
self.prev_time = self.time # On-demand would only benefit if_addrs:
# On-demand would only benefit if_addrs: # stats are used to determine display,
# stats are used to determine display, # and we want to keep previous io_counters
# and we want to keep previous io_counters # so displaying stats is ~instant.
# so displaying stats is ~instant. self.time = loop.time()
self.time = loop.time() self.if_stats = psutil.net_if_stats()
self.if_stats = psutil.net_if_stats() self.if_addrs = psutil.net_if_addrs()
self.if_addrs = psutil.net_if_addrs() self.io_counters = psutil.net_io_counters(pernic=True)
self.io_counters = psutil.net_io_counters(pernic=True)
for iface in self.if_stats: for iface in self.if_stats:
if iface not in self.sections: section = self.sections.get(iface)
section = NetworkProviderSection( if not section:
parent=self.module, iface=iface, provider=self section = NetworkProviderSection(
) parent=self.module, iface=iface, provider=self
self.sections[iface] = section )
self.sections[iface] = section
self.module.bar.addLongRunningTask(section.updateOnStateChange()) tg.create_task(section.update())
tg.create_task(self.updateIface(iface)) for iface, section in self.sections.items():
for iface, section in self.sections.items(): if iface not in self.if_stats:
if iface not in self.if_stats: section.setText(None)
section.setText(None)
tg.create_task(asyncio.sleep(1)) async def onStateChange(self, section: StatefulSection) -> None:
assert isinstance(section, NetworkProviderSection)
await section.update()
class TimeProvider(StatefulProvider): class TimeProvider(PeriodicStatefulProvider):
FORMATS = ["%H:%M", "%m-%d %H:%M:%S", "%a %y-%m-%d %H:%M:%S"] FORMATS = ["%H:%M", "%m-%d %H:%M:%S", "%a %y-%m-%d %H:%M:%S"]
async def run(self) -> None: async def init(self) -> None:
await super().run()
assert isinstance(self.section, StatefulSection)
self.section.state = 1 self.section.state = 1
self.section.numberStates = len(self.FORMATS) self.section.numberStates = len(self.FORMATS)
while True: async def loop(self) -> None:
now = datetime.datetime.now() now = datetime.datetime.now()
format = self.FORMATS[self.section.state] format = self.FORMATS[self.section.state]
self.section.setText(now.strftime(format)) self.section.setText(now.strftime(format))
remaining = 1 - now.microsecond / 1000000
try:
await asyncio.wait_for(self.section.stateChanged.wait(), remaining)
except TimeoutError:
pass
async def main() -> None: async def main() -> None: