ohmu/dev/varvusensor/collector/live.py

"""Reaalajas AS7341 värvinäitaja.

Loeb ClickHouse'ist viimase mõõtmise ja kuvab terminalis värvinäidise (24-bitine
ANSI) + kanalite spektri. Mõeldud katseteks (eri värvi pindade mõõtmine).

Käivitamine (projekti .venv):
  CH_URL=http://100.87.1.24:8125 CH_USER=zigbee CH_PASS=<parool> \
    .venv/bin/python varvusensor/collector/live.py [--interval 2] [--once]
"""
from __future__ import annotations

import argparse
import json
import os
import time

import requests

import color
from settings import CH_URL, CH_DB, CH_TABLE, CH_USER, CH_PASS

BANDS = color.CHANNEL_WAVELENGTHS  # {f1: 415, ...}
RESET = "\x1b[0m"


def fetch_latest() -> dict | None:
    cols = ", ".join(BANDS)
    sql = (
        f"SELECT ts, {cols}, rgb_hex, x_chrom, y_chrom, r, g, b "
        f"FROM {CH_DB}.{CH_TABLE} ORDER BY ts DESC LIMIT 1 FORMAT JSONEachRow"
    )
    resp = requests.post(
        f"{CH_URL}/",
        params={"database": CH_DB, "query": sql},
        auth=(CH_USER, CH_PASS),
        timeout=10,
    )
    resp.raise_for_status()
    lines = [line for line in resp.text.strip().splitlines() if line]
    return json.loads(lines[0]) if lines else None


def swatch(r: int, g: int, b: int, width: int = 28, height: int = 3) -> str:
    block = f"\x1b[48;2;{r};{g};{b}m" + " " * width + RESET
    return "\n".join([block] * height)


def spectrum_bar(channels: dict[str, float]) -> str:
    vmax = max((channels.get(band) or 0.0) for band in BANDS) or 1.0
    lines = []
    for band, wavelength in BANDS.items():
        value = channels.get(band) or 0.0
        bars = int(round(40 * value / vmax))
        lines.append(f"  {band} {wavelength:3d}nm |{'█' * bars:<40}| {value:8.1f}")
    return "\n".join(lines)


def render(row: dict) -> str:
    r, g, b = int(row["r"]), int(row["g"]), int(row["b"])
    channels = {band: row.get(band) for band in BANDS}
    return "\n".join(
        [
            f"  Aeg:     {row['ts']}",
            f"  Värvus:  {row['rgb_hex']}  x={row['x_chrom']:.3f} y={row['y_chrom']:.3f}",
            "",
            swatch(r, g, b),
            "",
            "  Spekter (kanalid):",
            spectrum_bar(channels),
        ]
    )


def main() -> int:
    parser = argparse.ArgumentParser(description="AS7341 reaalajas värvinäitaja")
    parser.add_argument("--interval", type=float, default=2.0, help="värskendus sekundites")
    parser.add_argument("--once", action="store_true", help="kuva üks kord ja välju")
    args = parser.parse_args()
    try:
        while True:
            row = fetch_latest()
            os.system("clear")
            print("AS7341 värvinäitaja  (Ctrl+C väljumiseks)")
            print("=" * 54)
            print(render(row) if row else "  Andmeid pole.")
            if args.once:
                break
            time.sleep(args.interval)
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    raise SystemExit(main())