music-light/luz-musica.py

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

import colorsys
import subprocess
import time
from io import BytesIO
from urllib.parse import parse_qs, urlparse

import requests
from PIL import Image

HA_URL = "http://192.168.1.105:8123"
HA_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJkZjM1MmRhNjg1ZWM0MGJiYTc4YTZlNWY3ZmQ0YTA1YSIsImlhdCI6MTc3NjI4MTI0MiwiZXhwIjoyMDkxNjQxMjQyfQ.rR0Kfwk5Smr06_9KJSDn3YPXiRlokEZZFiAa_naNs78"
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


def run_playerctl(field: str, player: str | None = None) -> str:
    command = ["playerctl"]
    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", "-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", "--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", "--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 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

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

    return filtered


def dominant_color(img: Image.Image) -> tuple[int, int, int]:
    pixels = good_pixels(img)
    if not pixels:
        # fallback simples
        small = img.resize((50, 50))
        colors = small.getcolors(50 * 50)
        if not colors:
            return (255, 255, 255)
        colors.sort(reverse=True)
        return colors[0][1]

    # 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 x: x[1])[0]
    return boost_color(best)


def boost_color(rgb: tuple[int, int, int]) -> tuple[int, int, int]:
    r, g, b = rgb
    rf, gf, bf = r / 255.0, g / 255.0, b / 255.0
    h, s, v = colorsys.rgb_to_hsv(rf, gf, bf)

    s = min(1.0, s * SATURATION_BOOST)
    v = max(v, 0.45)

    nr, ng, nb = colorsys.hsv_to_rgb(h, s, v)
    return (int(nr * 255), int(ng * 255), int(nb * 255))


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.")
    last_track = ""

    while True:
        try:
            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 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)
                rgb = dominant_color(img)
                print(f"Aplicando cor: {rgb}")
                set_light_color(rgb)
                last_track = track_id

            time.sleep(2)

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


if __name__ == "__main__":
    main()