62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
import asyncio
|
|
import logging
|
|
from random import randint
|
|
|
|
from rio import Text, Component, TextStyle, i_know_what_im_doing
|
|
|
|
|
|
class Ticker(Component):
|
|
@i_know_what_im_doing
|
|
def __init__(self, texts: list[str], refresh_interval: float, style: TextStyle, tick_in_order: bool, **kwargs) -> None:
|
|
super().__init__()
|
|
if refresh_interval <= 0:
|
|
raise ValueError("refresh_interval must be greater than zero")
|
|
self.kwargs = kwargs
|
|
self.kwargs.pop("text", None)
|
|
self.kwargs.pop("style", None)
|
|
|
|
self.texts = texts
|
|
self.refresh_interval = refresh_interval
|
|
self.style = style
|
|
self._tick_in_order = tick_in_order
|
|
self._last_text = -1
|
|
self._task = asyncio.get_event_loop().create_task(self._set_next_text())
|
|
|
|
def stop_ticking(self) -> None:
|
|
self._task.cancel()
|
|
|
|
def start_ticking(self) -> None:
|
|
if not self._task.cancelled():
|
|
self._task.cancel()
|
|
|
|
self._task = asyncio.get_event_loop().create_task(self._set_next_text())
|
|
|
|
def set_new_texts(self, new_texts: list[str]) -> None:
|
|
self.stop_ticking()
|
|
self._last_text = -1
|
|
self.texts = new_texts
|
|
self.start_ticking()
|
|
|
|
def add_text(self, text: str) -> None:
|
|
self.texts.append(text)
|
|
|
|
def build(self) -> Component:
|
|
try:
|
|
self.kwargs["text"] = self.texts[self._last_text]
|
|
return Text(**self.kwargs, style=self.style)
|
|
except Exception as e:
|
|
logging.error(e)
|
|
logging.error("Ticker stopped due to an exception")
|
|
self.stop_ticking()
|
|
|
|
async def _set_next_text(self) -> None:
|
|
while True:
|
|
if self._tick_in_order:
|
|
self._last_text += 1
|
|
if self._last_text >= len(self.texts):
|
|
self._last_text = 0
|
|
else:
|
|
self._last_text = randint(0, len(self.texts) - 1)
|
|
await self.force_refresh()
|
|
await asyncio.sleep(self.refresh_interval)
|