music-light/luz-musica.py

#!/usr/bin/env python3
from __future__ import annotations

import colorsys
import json
import math
import os
import subprocess
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from io import BytesIO
from urllib.parse import parse_qs, urlparse

import requests
from PIL import Image

HA_URL = os.environ.get("HA_URL", "http://127.0.0.1:8123")
HA_TOKEN = os.environ.get(
    "HA_TOKEN",
    "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJkZjM1MmRhNjg1ZWM0MGJiYTc4YTZlNWY3ZmQ0YTA1YSIsImlhdCI6MTc3NjI4MTI0MiwiZXhwIjoyMDkxNjQxMjQyfQ.rR0Kfwk5Smr06_9KJSDn3YPXiRlokEZZFiAa_naNs78",
)
LIGHT_ENTITY = os.environ.get("LIGHT_ENTITY", "light.luzsala_luz_sala")

# Ajustes visuais
BRIGHTNESS_PCT = 70
SATURATION_BOOST = 1.15
MIN_SATURATION = 0.20
MIN_VALUE = 0.18
REQUEST_TIMEOUT = 10
MAX_VALUE = 0.82
PALETTE_SIZE = 32
OKLCH_LIGHTNESS = 0.72
OKLCH_CHROMA = 0.24
MAX_BRIGHTNESS_STEP = 0.10
PLAYERCTL_BIN = os.environ.get("PLAYERCTL_BIN", "playerctl")
CONTROL_HOST = os.environ.get("CONTROL_HOST", "0.0.0.0")
CONTROL_PORT = int(os.environ.get("CONTROL_PORT", "8765"))
CONTROL_TOKEN = os.environ.get("CONTROL_TOKEN", "")
YELLOW_HUE_START = 0.12
YELLOW_HUE_END = 0.18
AMBER_TARGET_HUE = 0.10
YELLOW_CHROMA_SCALE = 0.82
YELLOW_LIGHTNESS_SCALE = 0.94
NEUTRAL_CHROMA_THRESHOLD = 0.05
NEUTRAL_SATURATION_THRESHOLD = 0.18
PURPLE_FALLBACK_HUE = 0.78


class ControlState:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._paused = False
        self._force_update = False

    def pause(self) -> None:
        with self._lock:
            self._paused = True

    def resume(self) -> None:
        with self._lock:
            self._paused = False
            self._force_update = True

    def toggle(self) -> bool:
        with self._lock:
            self._paused = not self._paused
            if not self._paused:
                self._force_update = True
            return self._paused

    def is_paused(self) -> bool:
        with self._lock:
            return self._paused

    def consume_force_update(self) -> bool:
        with self._lock:
            force_update = self._force_update
            self._force_update = False
            return force_update

    def snapshot(self) -> dict[str, object]:
        with self._lock:
            return {
                "paused": self._paused,
                "running": True,
            }

    def set_current(self, art_url: str, rgb: tuple[int, int, int]) -> None:
        with self._lock:
            self._art_url = art_url
            self._rgb = rgb

    def current(self) -> dict[str, object]:
        with self._lock:
            return {
                "art_url": getattr(self, "_art_url", None),
                "rgb": list(getattr(self, "_rgb", [0, 200, 255])),
            }


CONTROL_STATE = ControlState()


class ControlHandler(BaseHTTPRequestHandler):
    def log_message(self, format: str, *args: object) -> None:
        return

    def do_GET(self) -> None:
        if self.path == "/status":
            self.write_json(200, CONTROL_STATE.snapshot())
            return
        if self.path == "/current":
            self.write_json(200, CONTROL_STATE.current())
            return
        if self.path.startswith("/art"):
            art_url = CONTROL_STATE.current().get("art_url")
            if not art_url:
                self.write_json(404, {"error": "no art"})
                return
            try:
                resp = requests.get(art_url, timeout=5)
                body = resp.content
                ctype = resp.headers.get("Content-Type", "image/jpeg")
                self.send_response(200)
                self.send_header("Content-Type", ctype)
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Access-Control-Allow-Origin", "*")
                self.end_headers()
                self.wfile.write(body)
            except Exception:
                self.write_json(502, {"error": "fetch failed"})
            return
        self.write_json(404, {"error": "not found"})

    def do_POST(self) -> None:
        if not self.authorized():
            self.write_json(401, {"error": "unauthorized"})
            return

        if self.path == "/pause":
            CONTROL_STATE.pause()
        elif self.path == "/resume":
            CONTROL_STATE.resume()
        elif self.path == "/toggle":
            CONTROL_STATE.toggle()
        else:
            self.write_json(404, {"error": "not found"})
            return

        self.write_json(200, CONTROL_STATE.snapshot())

    def authorized(self) -> bool:
        if not CONTROL_TOKEN:
            return True

        header_token = self.headers.get("X-Music-Light-Token", "")
        bearer_token = self.headers.get("Authorization", "").removeprefix("Bearer ")
        return CONTROL_TOKEN in {header_token, bearer_token}

    def write_json(self, status: int, payload: dict[str, object]) -> None:
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def start_control_server() -> ThreadingHTTPServer:
    server = ThreadingHTTPServer((CONTROL_HOST, CONTROL_PORT), ControlHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    print(f"Controle HTTP ouvindo em {CONTROL_HOST}:{CONTROL_PORT}")
    return server


def run_playerctl(field: str, player: str | None = None) -> str:
    command = [PLAYERCTL_BIN]
    if player:
        command.extend(["--player", player])
    command.extend(["metadata", field])

    result = subprocess.run(
        command,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return ""
    return result.stdout.strip()


def list_players() -> list[str]:
    result = subprocess.run(
        [PLAYERCTL_BIN, "-l"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return []
    return [line.strip() for line in result.stdout.splitlines() if line.strip()]


def player_status(player: str) -> str:
    result = subprocess.run(
        [PLAYERCTL_BIN, "--player", player, "status"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return ""
    return result.stdout.strip()


def choose_active_player() -> str:
    for player in list_players():
        if player_status(player) == "Playing":
            return player
    return ""


def get_track_url(player: str) -> str:
    return run_playerctl("xesam:url", player)


def get_player_name(player: str) -> str:
    result = subprocess.run(
        [PLAYERCTL_BIN, "--player", player, "metadata", "--format", "{{playerName}}"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return player
    return result.stdout.strip() or player


def youtube_video_id(url: str) -> str:
    if not url:
        return ""

    parsed = urlparse(url)
    host = parsed.netloc.lower()
    path = parsed.path.strip("/")

    if host in {"youtu.be", "www.youtu.be"}:
        return path.split("/", 1)[0]

    if "youtube.com" not in host:
        return ""

    if path == "watch":
        return parse_qs(parsed.query).get("v", [""])[0]

    parts = path.split("/")
    if len(parts) >= 2 and parts[0] in {"shorts", "live", "embed", "v"}:
        return parts[1]

    return ""


def youtube_thumbnail_url(player: str) -> str:
    video_id = youtube_video_id(get_track_url(player))
    if not video_id:
        return ""
    return f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg"


def get_current_track_id(player: str) -> str:
    title = run_playerctl("title", player)
    artist = run_playerctl("artist", player)
    art = run_playerctl("mpris:artUrl", player)
    url = get_track_url(player)
    player_name = get_player_name(player)
    return f"{player_name}::{artist}::{title}::{art or url}"


def get_album_art_url(player: str) -> str:
    art_url = run_playerctl("mpris:artUrl", player)
    if art_url:
        return art_url

    track_url = get_track_url(player)
    if youtube_video_id(track_url):
        thumb_url = youtube_thumbnail_url(player)
        if thumb_url:
            print(f"Usando thumbnail do YouTube: {thumb_url}")
            return thumb_url

    return ""


def load_image_from_url(url: str) -> Image.Image:
    if url.startswith("file://"):
        local_path = url[7:]
        return Image.open(local_path).convert("RGB")

    resp = requests.get(url, timeout=REQUEST_TIMEOUT)
    resp.raise_for_status()
    return Image.open(BytesIO(resp.content)).convert("RGB")


def image_pixels(img: Image.Image) -> list[tuple[int, int, int]]:
    if hasattr(img, "get_flattened_data"):
        return list(img.get_flattened_data())
    return list(img.getdata())


def hue_distance(a: float, b: float) -> float:
    diff = abs(a - b)
    return min(diff, 1.0 - diff)


def rgb_to_hsv(rgb: tuple[int, int, int]) -> tuple[float, float, float]:
    r, g, b = rgb
    return colorsys.rgb_to_hsv(r / 255.0, g / 255.0, b / 255.0)


def hsv_to_rgb(h: float, s: float, v: float) -> tuple[int, int, int]:
    r, g, b = colorsys.hsv_to_rgb(h, s, v)
    return (int(r * 255), int(g * 255), int(b * 255))


def tame_yellow_oklch(lightness: float, chroma: float, hue: float) -> tuple[float, float, float]:
    if YELLOW_HUE_START <= hue <= YELLOW_HUE_END:
        return (
            lightness * YELLOW_LIGHTNESS_SCALE,
            chroma * YELLOW_CHROMA_SCALE,
            AMBER_TARGET_HUE,
        )
    return (lightness, chroma, hue)


def srgb_channel_to_linear(channel: float) -> float:
    if channel <= 0.04045:
        return channel / 12.92
    return ((channel + 0.055) / 1.055) ** 2.4


def linear_channel_to_srgb(channel: float) -> float:
    if channel <= 0.0031308:
        return 12.92 * channel
    return 1.055 * (channel ** (1 / 2.4)) - 0.055


def rgb_to_oklab(rgb: tuple[int, int, int]) -> tuple[float, float, float]:
    r, g, b = rgb
    rl = srgb_channel_to_linear(r / 255.0)
    gl = srgb_channel_to_linear(g / 255.0)
    bl = srgb_channel_to_linear(b / 255.0)

    l = 0.4122214708 * rl + 0.5363325363 * gl + 0.0514459929 * bl
    m = 0.2119034982 * rl + 0.6806995451 * gl + 0.1073969566 * bl
    s = 0.0883024619 * rl + 0.2817188376 * gl + 0.6299787005 * bl

    l_ = l ** (1 / 3)
    m_ = m ** (1 / 3)
    s_ = s ** (1 / 3)

    return (
        0.2104542553 * l_ + 0.7936177850 * m_ - 0.0040720468 * s_,
        1.9779984951 * l_ - 2.4285922050 * m_ + 0.4505937099 * s_,
        0.0259040371 * l_ + 0.7827717662 * m_ - 0.8086757660 * s_,
    )


def oklab_to_rgb(lightness: float, a: float, b: float) -> tuple[int, int, int]:
    l_ = lightness + 0.3963377774 * a + 0.2158037573 * b
    m_ = lightness - 0.1055613458 * a - 0.0638541728 * b
    s_ = lightness - 0.0894841775 * a - 1.2914855480 * b

    l = l_ ** 3
    m = m_ ** 3
    s = s_ ** 3

    rl = +4.0767416621 * l - 3.3077115913 * m + 0.2309699292 * s
    gl = -1.2684380046 * l + 2.6097574011 * m - 0.3413193965 * s
    bl = -0.0041960863 * l - 0.7034186147 * m + 1.7076147010 * s

    r = min(1.0, max(0.0, linear_channel_to_srgb(rl)))
    g = min(1.0, max(0.0, linear_channel_to_srgb(gl)))
    b = min(1.0, max(0.0, linear_channel_to_srgb(bl)))
    return (int(r * 255), int(g * 255), int(b * 255))


def rgb_to_oklch(rgb: tuple[int, int, int]) -> tuple[float, float, float]:
    lightness, a, b = rgb_to_oklab(rgb)
    chroma = (a * a + b * b) ** 0.5
    hue = (0.0 if chroma == 0 else (math.atan2(b, a) / (2 * math.pi)) % 1.0)
    return (lightness, chroma, hue)


def oklch_to_rgb(lightness: float, chroma: float, hue: float) -> tuple[int, int, int]:
    angle = 2 * math.pi * hue
    a = chroma * math.cos(angle)
    b = chroma * math.sin(angle)
    return oklab_to_rgb(lightness, a, b)


def pure_tone_palette() -> list[tuple[int, int, int]]:
    return [
        oklch_to_rgb(*tame_yellow_oklch(OKLCH_LIGHTNESS, OKLCH_CHROMA, index / PALETTE_SIZE))
        for index in range(PALETTE_SIZE)
    ]


PURE_TONE_PALETTE = pure_tone_palette()
PURPLE_FALLBACK_RGB = oklch_to_rgb(OKLCH_LIGHTNESS, OKLCH_CHROMA, PURPLE_FALLBACK_HUE)


def clamp_brightness(rgb: tuple[int, int, int], previous_rgb: tuple[int, int, int] | None) -> tuple[int, int, int]:
    if previous_rgb is None:
        return rgb

    h, s, v = rgb_to_hsv(rgb)
    _, _, prev_v = rgb_to_hsv(previous_rgb)

    limited_v = max(prev_v - MAX_BRIGHTNESS_STEP, min(prev_v + MAX_BRIGHTNESS_STEP, v))
    return hsv_to_rgb(h, s, limited_v)


def snap_to_palette(rgb: tuple[int, int, int]) -> tuple[int, int, int]:
    _, _, hue = rgb_to_oklch(rgb)

    best = min(
        PURE_TONE_PALETTE,
        key=lambda candidate: hue_distance(hue, rgb_to_oklch(candidate)[2]),
    )

    best_lightness, _, best_hue = rgb_to_oklch(best)
    return oklch_to_rgb(*tame_yellow_oklch(best_lightness, OKLCH_CHROMA, best_hue))


def good_pixels(img: Image.Image) -> list[tuple[int, int, int]]:
    img = img.resize((80, 80))
    pixels = image_pixels(img)

    filtered: list[tuple[int, int, int]] = []
    for r, g, b in pixels:
        rf, gf, bf = r / 255.0, g / 255.0, b / 255.0
        h, s, v = colorsys.rgb_to_hsv(rf, gf, bf)

        # ignora pixels muito escuros e quase cinza
        if v < MIN_VALUE:
            continue
        if s < MIN_SATURATION:
            continue
        if v > MAX_VALUE and s < 0.45:
            continue

        filtered.append((r, g, b))

    return filtered


def color_information_score(pixels: list[tuple[int, int, int]]) -> tuple[float, float]:
    if not pixels:
        return (0.0, 0.0)

    chroma_total = 0.0
    saturation_total = 0.0
    for pixel in pixels:
        _, chroma, _ = rgb_to_oklch(pixel)
        _, saturation, _ = rgb_to_hsv(pixel)
        chroma_total += chroma
        saturation_total += saturation

    count = len(pixels)
    return (chroma_total / count, saturation_total / count)


def should_use_purple_fallback(pixels: list[tuple[int, int, int]]) -> bool:
    avg_chroma, avg_saturation = color_information_score(pixels)
    return avg_chroma < NEUTRAL_CHROMA_THRESHOLD or avg_saturation < NEUTRAL_SATURATION_THRESHOLD


def dominant_color(img: Image.Image) -> tuple[int, int, int]:
    pixels = good_pixels(img)
    if not pixels:
        # fallback simples
        return PURPLE_FALLBACK_RGB

    if should_use_purple_fallback(pixels):
        return PURPLE_FALLBACK_RGB

    # Quantiza para agrupar tons parecidos
    buckets: dict[tuple[int, int, int], int] = {}
    for r, g, b in pixels:
        key = (r // 16 * 16, g // 16 * 16, b // 16 * 16)
        buckets[key] = buckets.get(key, 0) + 1

    best = max(
        buckets.items(),
        key=lambda item: item[1] * bucket_weight(item[0]),
    )[0]
    return boost_color(best)


def bucket_weight(rgb: tuple[int, int, int]) -> float:
    _, saturation, value = rgb_to_hsv(rgb)

    saturation_score = 0.6 + saturation
    brightness_midpoint = 0.62
    brightness_penalty = abs(value - brightness_midpoint) * 1.2
    return max(0.2, saturation_score - brightness_penalty)


def boost_color(rgb: tuple[int, int, int]) -> tuple[int, int, int]:
    h, s, v = rgb_to_hsv(rgb)

    s = min(1.0, s * SATURATION_BOOST)
    v = min(0.74, max(v, 0.48))

    return hsv_to_rgb(h, s, v)


def set_light_color(rgb: tuple[int, int, int]) -> None:
    response = requests.post(
        f"{HA_URL}/api/services/light/turn_on",
        headers={
            "Authorization": f"Bearer {HA_TOKEN}",
            "Content-Type": "application/json",
        },
        json={
            "entity_id": LIGHT_ENTITY,
            "rgb_color": list(rgb),
            "brightness_pct": BRIGHTNESS_PCT,
        },
        timeout=REQUEST_TIMEOUT,
    )
    response.raise_for_status()


def main() -> None:
    print("Monitorando música atual. Ctrl+C para sair.")
    control_server = start_control_server()
    last_track = ""
    last_rgb: tuple[int, int, int] | None = None

    while True:
        try:
            if CONTROL_STATE.is_paused():
                time.sleep(2)
                continue

            player = choose_active_player()
            if not player:
                time.sleep(2)
                continue

            track_id = get_current_track_id(player)

            if not track_id or track_id == "::":
                time.sleep(2)
                continue

            if CONTROL_STATE.consume_force_update() or track_id != last_track:
                art_url = get_album_art_url(player)
                if not art_url:
                    print("Sem capa disponível para a faixa atual.")
                    last_track = track_id
                    time.sleep(2)
                    continue

                print(f"Nova faixa detectada: {track_id}")
                img = load_image_from_url(art_url)
                raw_rgb = dominant_color(img)
                palette_rgb = snap_to_palette(raw_rgb)
                rgb = clamp_brightness(palette_rgb, last_rgb)
                print(f"Cor base: {raw_rgb} -> paleta: {palette_rgb}")
                print(f"Aplicando cor: {rgb}")
                set_light_color(rgb)
                CONTROL_STATE.set_current(art_url, rgb)
                last_track = track_id
                last_rgb = rgb

            time.sleep(2)

        except KeyboardInterrupt:
            print("\nEncerrado.")
            control_server.shutdown()
            control_server.server_close()
            break
        except Exception as exc:
            print(f"Erro: {exc}")
            time.sleep(3)


if __name__ == "__main__":
    main()