db2fda70bd
show_skeleton is now managed by the Plasma wallpaper config directly, so the server no longer needs toggle-skeleton, set-shortcut endpoints, pynput hotkey listener, or the show_skeleton field in /current. Also fixes unclosed file handle in the /art handler. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
770 lines
24 KiB
Python
770 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import colorsys
|
||
import json
|
||
import math
|
||
import os
|
||
import re
|
||
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], rgb2: tuple[int, int, int]) -> None:
|
||
with self._lock:
|
||
self._art_url = art_url
|
||
self._rgb = rgb
|
||
self._rgb2 = rgb2
|
||
|
||
def set_lyrics(self, lyrics: list[tuple[float, str]]) -> None:
|
||
with self._lock:
|
||
self._lyrics = lyrics
|
||
self._verses = group_into_verses(lyrics)
|
||
|
||
def set_position(self, position: float) -> None:
|
||
with self._lock:
|
||
self._position = position
|
||
|
||
def current(self) -> dict[str, object]:
|
||
with self._lock:
|
||
verses = getattr(self, "_verses", [])
|
||
position = getattr(self, "_position", 0.0)
|
||
verse_lines, verse_active = find_active_verse(verses, position)
|
||
line = verse_lines[verse_active] if verse_active >= 0 and verse_lines else ""
|
||
return {
|
||
"art_url": getattr(self, "_art_url", None),
|
||
"rgb": list(getattr(self, "_rgb", [0, 200, 255])),
|
||
"rgb2": list(getattr(self, "_rgb2", [255, 0, 150])),
|
||
"lyric_line": line,
|
||
"verse_lines": verse_lines,
|
||
"verse_active": verse_active,
|
||
}
|
||
|
||
|
||
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:
|
||
if art_url.startswith("file://"):
|
||
with open(art_url[7:], "rb") as f:
|
||
body = f.read()
|
||
ctype = "image/png" if art_url.endswith(".png") else "image/jpeg"
|
||
else:
|
||
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.send_header("Access-Control-Allow-Origin", "*")
|
||
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
|
||
|
||
|
||
VERSE_GAP_SECS = 5.0 # gap between lines that splits a verse
|
||
INTER_VERSE_CAP = 8.0 # how long to keep last line of a verse visible
|
||
|
||
_LRC_TS = re.compile(r'\[(\d{1,2}):(\d{2}(?:\.\d+)?)\]')
|
||
|
||
|
||
def parse_lrc(text: str) -> list[tuple[float, str]]:
|
||
lines = []
|
||
for raw in text.splitlines():
|
||
m = _LRC_TS.match(raw)
|
||
if not m:
|
||
continue
|
||
ts = int(m.group(1)) * 60 + float(m.group(2))
|
||
line = raw[m.end():].strip()
|
||
lines.append((ts, line))
|
||
return lines
|
||
|
||
|
||
_TITLE_JUNK = re.compile(
|
||
r"[\s\-–]+("
|
||
r"\d{4}\s*remaster(ed)?|remaster(ed)?(\s+\d{4})?|"
|
||
r"deluxe(\s+edition)?|expanded(\s+edition)?|"
|
||
r"anniversary(\s+edition)?|bonus\s+track|"
|
||
r"live(\s+version)?|acoustic(\s+version)?|"
|
||
r"radio\s+edit|single\s+version|album\s+version"
|
||
r")[\s\)\]]*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def clean_title(title: str) -> str:
|
||
# strip parenthesised/bracketed suffixes first, then dash-separated junk
|
||
t = re.sub(r"\s*[\(\[][^\)\]]*[\)\]]$", "", title).strip()
|
||
t = _TITLE_JUNK.sub("", t).strip()
|
||
return t
|
||
|
||
|
||
def group_into_verses(
|
||
lyrics: list[tuple[float, str]],
|
||
) -> list[list[tuple[float, str]]]:
|
||
"""Split non-empty lyrics into verses on empty LRC lines or large timestamp gaps."""
|
||
verses: list[list[tuple[float, str]]] = []
|
||
current: list[tuple[float, str]] = []
|
||
for ts, text in lyrics:
|
||
if not text.strip():
|
||
if current:
|
||
verses.append(current)
|
||
current = []
|
||
continue
|
||
if current and (ts - current[-1][0]) > VERSE_GAP_SECS:
|
||
verses.append(current)
|
||
current = []
|
||
current.append((ts, text))
|
||
if current:
|
||
verses.append(current)
|
||
return verses
|
||
|
||
|
||
def find_active_verse(
|
||
verses: list[list[tuple[float, str]]],
|
||
position: float,
|
||
) -> tuple[list[str], int]:
|
||
"""Return (verse_text_lines, active_index) or ([], -1) when in a gap."""
|
||
for v_idx, verse in enumerate(verses):
|
||
for l_idx, (ts, _) in enumerate(verse):
|
||
if ts > position:
|
||
break
|
||
if l_idx + 1 < len(verse):
|
||
deadline = verse[l_idx + 1][0]
|
||
else:
|
||
nxt = verses[v_idx + 1][0][0] if v_idx + 1 < len(verses) else ts + INTER_VERSE_CAP
|
||
deadline = min(nxt, ts + INTER_VERSE_CAP)
|
||
if position <= deadline:
|
||
return [t for _, t in verse], l_idx
|
||
return [], -1
|
||
|
||
|
||
def fetch_lyrics(artist: str, title: str) -> list[tuple[float, str]]:
|
||
if not artist or not title:
|
||
return []
|
||
clean = clean_title(title)
|
||
for t in ([clean, title] if clean != title else [title]):
|
||
try:
|
||
resp = requests.get(
|
||
"https://lrclib.net/api/get",
|
||
params={"artist_name": artist, "track_name": t},
|
||
timeout=20,
|
||
)
|
||
if resp.status_code == 200:
|
||
synced = resp.json().get("syncedLyrics") or ""
|
||
if synced:
|
||
return parse_lrc(synced)
|
||
except Exception:
|
||
pass
|
||
return []
|
||
|
||
|
||
def fetch_lyrics_async(artist: str, title: str) -> None:
|
||
def _run() -> None:
|
||
lyrics = fetch_lyrics(artist, title)
|
||
CONTROL_STATE.set_lyrics(lyrics)
|
||
print(f"Letras: {len(lyrics)} linhas")
|
||
threading.Thread(target=_run, daemon=True).start()
|
||
|
||
|
||
def start_position_thread() -> threading.Thread:
|
||
def _run() -> None:
|
||
while True:
|
||
try:
|
||
player = choose_active_player()
|
||
if player:
|
||
result = subprocess.run(
|
||
[PLAYERCTL_BIN, "--player", player, "position"],
|
||
capture_output=True, text=True, timeout=1,
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
CONTROL_STATE.set_position(float(result.stdout.strip()))
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.5)
|
||
|
||
t = threading.Thread(target=_run, daemon=True)
|
||
t.start()
|
||
return t
|
||
|
||
|
||
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 build_buckets(img: Image.Image) -> dict[tuple[int, int, int], int]:
|
||
pixels = good_pixels(img)
|
||
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
|
||
return buckets
|
||
|
||
|
||
def dominant_color(img: Image.Image) -> tuple[int, int, int]:
|
||
pixels = good_pixels(img)
|
||
if not pixels:
|
||
return PURPLE_FALLBACK_RGB
|
||
|
||
if should_use_purple_fallback(pixels):
|
||
return PURPLE_FALLBACK_RGB
|
||
|
||
buckets = build_buckets(img)
|
||
best = max(
|
||
buckets.items(),
|
||
key=lambda item: item[1] * bucket_weight(item[0]),
|
||
)[0]
|
||
return boost_color(best)
|
||
|
||
|
||
def secondary_color(img: Image.Image, primary: tuple[int, int, int]) -> tuple[int, int, int]:
|
||
buckets = build_buckets(img)
|
||
primary_h = rgb_to_hsv(primary)[0]
|
||
MIN_HUE_DIST = 0.12 # at least ~43° apart
|
||
|
||
ranked = sorted(
|
||
buckets.items(),
|
||
key=lambda item: item[1] * bucket_weight(item[0]),
|
||
reverse=True,
|
||
)
|
||
for bucket, _ in ranked:
|
||
h = rgb_to_hsv(bucket)[0]
|
||
if hue_distance(h, primary_h) >= MIN_HUE_DIST:
|
||
return boost_color(bucket)
|
||
|
||
# fallback: complementary hue of primary
|
||
h, s, v = rgb_to_hsv(primary)
|
||
return hsv_to_rgb((h + 0.5) % 1.0, max(s, 0.5), max(v, 0.5))
|
||
|
||
|
||
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()
|
||
start_position_thread()
|
||
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)
|
||
rgb2 = secondary_color(img, rgb)
|
||
print(f"Cor base: {raw_rgb} -> paleta: {palette_rgb}")
|
||
print(f"Aplicando cor: {rgb} secundária: {rgb2}")
|
||
set_light_color(rgb)
|
||
CONTROL_STATE.set_current(art_url, rgb, rgb2)
|
||
artist = run_playerctl("artist", player)
|
||
title = run_playerctl("title", player)
|
||
CONTROL_STATE.set_lyrics([])
|
||
fetch_lyrics_async(artist, title)
|
||
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()
|