buba/buba/bubaanimation.py

import logging
from datetime import timedelta, datetime
from itertools import islice
from threading import Thread
from time import sleep

from buba.bubaanimator import LineLayoutColumn, WEEKDAYS_DE
from buba.bubacmd import BubaCmd


class BubaAnimation:
    default_layout = [
        LineLayoutColumn(90, BubaCmd.ALIGN_LEFT),
        LineLayoutColumn(30, BubaCmd.ALIGN_RIGHT),
    ]

    def __init__(self, buba: BubaCmd):
        self.log = logging.getLogger(type(self).__name__)
        self.buba = buba
        self.update_interval = timedelta(minutes=1)
        self.page_interval = timedelta(seconds=7)
        self.title = None
        self.layout = self.default_layout
        self.rows = []
        self.thread = Thread(target=self.update_rows, daemon=True)
        self.thread.start()
        pass

    def __repr__(self):
        return f"<{type(self).__name__}>"

    def update(self) -> None:
        """
        Do everything necessary to produce the information to be displayed.
        :return:
        """

    def update_rows(self) -> None:
        """
        Update loop. This will be run continuously in a thread to call update().
        :return:
        """
        sleep(1) # give the code a bit of time to configure stuff
        while True:
            try:
                self.update()
            except Exception as e:
                self.log.warning(f"unable to update: {e}")
            sleep(self.update_interval.total_seconds())

    def show(self) -> None:
        """
        Write information to the display.
        You should also implement any waiting time for people to be able to take the information in.
        """
        self.show_pages()

    @staticmethod
    def chunk(it, size):
        """
        Return list in groups of size.
        :param it: list
        :param size: chunk size
        :return: list of chunks
        """
        it = iter(it)
        return iter(lambda: tuple(islice(it, size)), ())

    @staticmethod
    def humanize_delta(dt, now_delta, day_delta):
        """
        Produce a short text representing a target date and time.
        :param dt: the target date and time
        :param now_delta: time delta of the target time to now
        :param day_delta: delta rounded to the full day
        :return:
        """
        if now_delta < timedelta(seconds=60):
            return "jetzt"
        if now_delta < timedelta(minutes=90):
            return f"{int(now_delta.seconds / 60)}m"
        if day_delta < timedelta(hours=24):
            return f"{int((now_delta.seconds + 3599) / 3600)}h"
        if day_delta < timedelta(days=7):
            # return dt.strftime("%a")  # weekday
            return WEEKDAYS_DE[dt.weekday()]
        return dt.strftime("%d.%m.")

    @staticmethod
    def countdown(dt: datetime):
        """
        Compute a human-readable time specification until the target event starts. The day starts at 04:00.
        :param dt: datetime timezone-aware datetime
        :return:
        """
        now = datetime.now(dt.tzinfo)
        from_day_start = now.replace(hour=4, minute=0, second=0, microsecond=0)
        now_delta = dt - now
        day_delta = dt - from_day_start
        h = BubaAnimation.humanize_delta(dt, now_delta, day_delta)
        return h

    @staticmethod
    def ellipsis(text, max=28):
        """
        If the text is longer that max, shorten it and add ellipsis.
        :param text: to be shortened
        :param max: max length
        :return: shortened text
        """
        if len(text) > max:
            text = text[:max - 2] + "..."  # we can get away with just 2, since the periods are very narrow
        return text

    def _write_row(self, nrow: int, row:list[str], layout: list[LineLayoutColumn]) -> None:
        """
        Write one row to the display.
        :param nrow: row number to write
        :param row: contents of the row
        :param layout: layout to use
        :return:
        """
        col = 0
        for i, ll in enumerate(layout):
            t = row[i] if len(row) > i else ""
            if ll.width > 24:
                t = self.ellipsis(t, int(ll.width / 3.7))
            self.buba.text(page=0, row=nrow, col_start=col, col_end=col + ll.width - 1, text=t, align=ll.align)
            col += ll.width

    def write_title(self, row: list[str], layout: list[LineLayoutColumn] = None) -> None:
        """
        Write the title row. If row has one element, fill the entire row with that; if it has two, the second one
        is right-aligned and shows a page number or the current time.
        :param row: the title info
        :param layout: the layout to use
        :return:
        """
        if len(row) == 1:
            self._write_row(0, row, [LineLayoutColumn(120, BubaCmd.ALIGN_LEFT)])
        else:
            self._write_row(0, row, [
                LineLayoutColumn(100, BubaCmd.ALIGN_LEFT),
                LineLayoutColumn(20, BubaCmd.ALIGN_RIGHT),
            ])

    def write_row(self, nrow: int, row: list[str]) -> None:
        """
        Write one row to the display. Use the layout specification to format the row.
        :param nrow: row number to be written
        :param row: contents of the row
        :param layout: layout to use
        :return:
        """
        if len(row) == 1:
            # if the row only has a single part, simply show that over the entire width
            self._write_row(nrow, row, [LineLayoutColumn(120, BubaCmd.ALIGN_LEFT)])
        else:
            self._write_row(nrow, row, self.layout)

    def show_pages(self) -> None:
        """
        Show rows on the display. Paginate the rows if there are more rows than the display can show.
        :return:
        """
        pages = list(self.chunk(self.rows, 3))
        for n, page in enumerate(pages):
            if len(pages) <= 1:
                self.write_title([self.title])
            else:
                self.write_title([self.title, f"({n + 1}/{len(pages)})"])
            for i in range(3):
                if i >= len(page):
                    self.buba.text(page=0, row=i + 1, col_start=0, col_end=119, text="")
                else:
                    p = page[i]
                    if isinstance(p, str):
                        p = [p, ]
                    self.write_row(i + 1, p)
            sleep(self.page_interval.total_seconds())