display.py

"""HT16K33 7-segment display driver over raw I2C (smbus2).

The Adafruit 1.2" 4-digit backpack is an HT16K33 driving a common-cathode
7-segment display. The protocol is tiny and stable, so we talk to it directly
rather than pulling in the Blinka/Adafruit stack.

Display RAM layout for the 4-digit 7-segment backpack (one byte per even
address; segment bit order is gfedcba with bit7 = decimal point):

    0x00  digit 0 (leftmost)
    0x02  digit 1
    0x04  colon + indicator dots (bit 1 = centre colon, bit 2 = upper-left
          dot, bit 3 = lower-left dot, bit 4 = upper-right dot)
    0x06  digit 2
    0x08  digit 3 (rightmost)

Two of the indicator dots double as status lights: the lower-left dot is the
NTP-sync indicator (lit while syncs are arriving, dark while the clock
free-runs) and the upper-left dot is the Wi-Fi indicator (lit while the link
is up, dark when it drops).
"""

from __future__ import annotations

import logging
import time
from typing import Callable, Protocol

log = logging.getLogger("piclock")

# Segment bitmaps, bit0 = segment a ... bit6 = segment g.
_FONT: dict[str, int] = {
    "0": 0x3F,
    "1": 0x06,
    "2": 0x5B,
    "3": 0x4F,
    "4": 0x66,
    "5": 0x6D,
    "6": 0x7D,
    "7": 0x07,
    "8": 0x7F,
    "9": 0x6F,
    " ": 0x00,
    "-": 0x40,
}

_DIGIT_ADDR = (0, 2, 6, 8)  # RAM offset of each of the four digit positions
_COLON_ADDR = 4
_COLON_BIT = 0x02
_SYNC_DOT_BIT = 0x08  # lower-left indicator dot, shares the colon's RAM byte
_WIFI_DOT_BIT = 0x04  # upper-left indicator dot, same byte

# HT16K33 command bytes.
_CMD_OSCILLATOR_ON = 0x21  # system setup: oscillator on
_CMD_DISPLAY_ON = 0x81  # display setup: on, no blink
_CMD_DIMMING = 0xE0  # | level (0-15)


def build_buffer(
    text: str, colon: bool, sync_dot: bool = False, wifi_dot: bool = False
) -> list[int]:
    """Encode four characters + colon + indicator dots into the 16-byte HT16K33 RAM buffer.

    Pure and hardware-free so it can be unit-tested. Raises ValueError on a
    wrong-length string or an unsupported character.
    """
    if len(text) != 4:
        raise ValueError(f"text must be exactly 4 characters, got {text!r}")
    buf = [0] * 16
    for i, char in enumerate(text):
        try:
            buf[_DIGIT_ADDR[i]] = _FONT[char]
        except KeyError:
            raise ValueError(f"unsupported character {char!r} in {text!r}") from None
    buf[_COLON_ADDR] = (
        (_COLON_BIT if colon else 0)
        | (_SYNC_DOT_BIT if sync_dot else 0)
        | (_WIFI_DOT_BIT if wifi_dot else 0)
    )
    return buf


class Display(Protocol):
    """Minimal display interface the app loop depends on."""

    def show(
        self, text: str, colon: bool, sync_dot: bool = False, wifi_dot: bool = False
    ) -> None: ...
    def set_brightness(self, level: int) -> None: ...
    def clear(self) -> None: ...
    def close(self) -> None: ...


class NullDisplay:
    """No-op display for development and tests off the Pi."""

    def show(
        self, text: str, colon: bool, sync_dot: bool = False, wifi_dot: bool = False
    ) -> None:
        pass

    def set_brightness(self, level: int) -> None:
        pass

    def clear(self) -> None:
        pass

    def close(self) -> None:
        pass


class HT16K33Display:
    """Drives the HT16K33 backpack over an I2C bus via smbus2."""

    def __init__(self, bus_num: int = 1, address: int = 0x70, brightness: int = 3) -> None:
        from smbus2 import SMBus  # imported lazily so the module loads off-Pi

        self._address = address
        self._bus = SMBus(bus_num)
        self._last: tuple[str, bool, bool, bool] | None = None
        self._last_brightness: int | None = None

        try:
            self._bus.write_byte(address, _CMD_OSCILLATOR_ON)
            self.set_brightness(brightness)
            self._bus.write_byte(address, _CMD_DISPLAY_ON)
        except OSError:
            self._bus.close()  # don't leak the fd if the device isn't there
            raise

    def set_brightness(self, level: int) -> None:
        level = max(0, min(15, int(level)))
        if level == self._last_brightness:
            return  # skip redundant I2C traffic (the loop pushes every tick)
        self._bus.write_byte(self._address, _CMD_DIMMING | level)
        self._last_brightness = level

    def show(
        self, text: str, colon: bool, sync_dot: bool = False, wifi_dot: bool = False
    ) -> None:
        key = (text, colon, sync_dot, wifi_dot)
        if key == self._last:
            return  # avoid redundant I2C traffic (e.g. unchanged minute)
        self._bus.write_i2c_block_data(
            self._address, 0x00, build_buffer(text, colon, sync_dot, wifi_dot)
        )
        self._last = key

    def clear(self) -> None:
        self._bus.write_i2c_block_data(self._address, 0x00, [0] * 16)
        self._last = None

    def close(self) -> None:
        try:
            self.clear()
        finally:
            self._bus.close()


class ResilientDisplay:
    """Wraps a hardware-display factory and tolerates the device being absent.

    A bedside clock should never crash-loop just because the display isn't
    plugged in (or a connector got bumped). This never raises from show()/
    clear(): if the device can't be reached it logs once and retries building it
    at most every ``retry_interval`` seconds. The clock keeps ticking meanwhile,
    so the correct time appears the moment the device (re)connects.
    """

    def __init__(
        self,
        factory: Callable[[], Display],
        *,
        retry_interval: float = 10.0,
        monotonic: Callable[[], float] = time.monotonic,
    ) -> None:
        self._factory = factory
        self._retry_interval = retry_interval
        self._monotonic = monotonic
        self._dev: Display | None = None
        self._next_retry = 0.0
        self._reported_down = False
        self._brightness: int | None = None

    def _connect(self) -> Display | None:
        if self._dev is not None:
            return self._dev
        now = self._monotonic()
        if now < self._next_retry:
            return None
        try:
            self._dev = self._factory()
        except OSError as exc:
            self._next_retry = now + self._retry_interval
            if not self._reported_down:
                log.warning(
                    "display unavailable, retrying every %gs (%s)",
                    self._retry_interval,
                    exc,
                )
                self._reported_down = True
            return None
        log.info("display reconnected" if self._reported_down else "display connected")
        self._reported_down = False
        return self._dev

    def _drop(self, exc: OSError) -> None:
        if not self._reported_down:
            log.warning("display write failed, will reconnect (%s)", exc)
            self._reported_down = True
        dev = self._dev
        self._dev = None
        self._next_retry = self._monotonic() + self._retry_interval
        if dev is not None:
            try:
                dev.close()
            except Exception:  # best effort after an I/O failure
                pass

    def show(
        self, text: str, colon: bool, sync_dot: bool = False, wifi_dot: bool = False
    ) -> None:
        dev = self._connect()
        if dev is None:
            return
        try:
            if self._brightness is not None:
                dev.set_brightness(self._brightness)
            dev.show(text, colon, sync_dot, wifi_dot)
        except OSError as exc:
            self._drop(exc)

    def set_brightness(self, level: int) -> None:
        # Remember changes while disconnected and apply before the first frame.
        # show() still drives connection attempts; the hardware deduplicates writes.
        self._brightness = level
        if self._dev is None:
            return
        try:
            self._dev.set_brightness(level)
        except OSError as exc:
            self._drop(exc)

    def clear(self) -> None:
        if self._dev is None:
            return
        try:
            self._dev.clear()
        except OSError as exc:
            self._drop(exc)

    def close(self) -> None:
        if self._dev is None:
            return
        try:
            self._dev.close()
        except Exception:  # best effort on the way out
            pass
        self._dev = None


def make_display(config) -> Display:
    """Construct the display backend named by the config.

    Hardware is wrapped in ResilientDisplay so a missing/flaky display degrades
    gracefully instead of crashing the service.
    """
    if config.backend == "null":
        return NullDisplay()

    def factory() -> Display:
        return HT16K33Display(
            bus_num=config.i2c_bus,
            address=config.i2c_address,
            brightness=config.brightness,
        )

    return ResilientDisplay(factory)