ltadeu6/music-light
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: ltadeu6:db2fda70bde70cc10c73d57d0f99d5646d8639ed
Branches Tags
ltadeu6:master
..
compare: ltadeu6:0d61eb63118a445837d0309e9b47fc5034d213e5
Branches Tags
ltadeu6:master

No commits in common. "db2fda70bde70cc10c73d57d0f99d5646d8639ed" and "0d61eb63118a445837d0309e9b47fc5034d213e5" have entirely different histories.
db2fda70bd ... 0d61eb6311

4 changed files with 9 additions and 417 deletions
Download patch file Download diff file Expand all files Collapse all files

2
AGENTS.md
View file

@ -33,12 +33,10 @@ Then run:
- The service loads optional environment overrides from `~/.config/music-light/env`.
- Use `make status-service`, `make logs-service`, `make restart-service`, and `make uninstall-service`.
- Keep it as a user service because `playerctl` needs access to the logged-in desktop user's MPRIS session.
- The running script exposes an HTTP control API for Home Assistant: `GET /status`, `POST /pause`, `POST /resume`, `POST /toggle`.
## Configuration
Edit constants at the top of `luz-musica.py`:
- `HA_URL`, `HA_TOKEN`, `LIGHT_ENTITY`
- control API: `CONTROL_HOST`, `CONTROL_PORT`, `CONTROL_TOKEN`
- tuning: `BRIGHTNESS_PCT`, `SATURATION_BOOST`, `MIN_SATURATION`, `MIN_VALUE`, `REQUEST_TIMEOUT`
## Expected behavior / troubleshooting

81
README.md
View file

@ -43,12 +43,6 @@ You can also tune:
- `MIN_VALUE`
- `REQUEST_TIMEOUT`
Network control settings:
- `CONTROL_HOST`: bind address for the control API, default `0.0.0.0`
- `CONTROL_PORT`: bind port for the control API, default `8765`
- `CONTROL_TOKEN`: optional shared token required for control requests
## Run
```bash
@ -102,84 +96,9 @@ make uninstall-service
This is installed as a user service because `playerctl` needs access to the desktop user's MPRIS session.
## Home Assistant Control
The running script exposes a small HTTP API that Home Assistant can call over the local network:
- `GET /status`
- `POST /pause`
- `POST /resume`
- `POST /toggle`
When paused, the systemd service stays running but the script stops changing the light. When resumed, it forces an update from the current track.
Example requests from another machine on the same network:
```bash
curl http://ubuntu-host:8765/status
curl -X POST http://ubuntu-host:8765/pause
curl -X POST http://ubuntu-host:8765/resume
curl -X POST http://ubuntu-host:8765/toggle
```
If you set `CONTROL_TOKEN`, include it as a header:
```bash
curl -X POST -H "X-Music-Light-Token: your-token" http://ubuntu-host:8765/toggle
```
Home Assistant `configuration.yaml` example:
```yaml
rest_command:
music_light_pause:
url: "http://ubuntu-host:8765/pause"
method: post
music_light_resume:
url: "http://ubuntu-host:8765/resume"
method: post
music_light_toggle:
url: "http://ubuntu-host:8765/toggle"
method: post
```
With `CONTROL_TOKEN`:
```yaml
rest_command:
music_light_toggle:
url: "http://ubuntu-host:8765/toggle"
method: post
headers:
X-Music-Light-Token: "your-token"
```
Create Home Assistant scripts so they can be shown as dashboard buttons:
```yaml
script:
pause_music_light:
alias: Pause Music Light
sequence:
- service: rest_command.music_light_pause
resume_music_light:
alias: Resume Music Light
sequence:
- service: rest_command.music_light_resume
toggle_music_light:
alias: Toggle Music Light
sequence:
- service: rest_command.music_light_toggle
```
Then add `script.pause_music_light`, `script.resume_music_light`, or `script.toggle_music_light` to a Home Assistant dashboard as button cards.
## Troubleshooting
- If nothing is playing, or `playerctl` cannot read metadata, the loop sleeps and retries.
- If `mpris:artUrl` is missing, the script logs it and waits for the next track change.
- If Home Assistant is unreachable or returns an error, the script logs the exception and retries.
- If the service cannot see the media player, make sure it is running under the same logged-in desktop user.
- If Home Assistant cannot reach the control API, check the Ubuntu host firewall and confirm `CONTROL_HOST=0.0.0.0`.

340
luz-musica.py
View file

@ -2,14 +2,10 @@
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
@ -35,9 +31,6 @@ 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
@ -48,284 +41,6 @@ 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:
@ -625,24 +340,21 @@ def should_use_purple_fallback(pixels: list[tuple[int, int, int]]) -> bool:
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:
# fallback simples
return PURPLE_FALLBACK_RGB
if should_use_purple_fallback(pixels):
return PURPLE_FALLBACK_RGB
buckets = build_buckets(img)
# 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]),
@ -650,26 +362,6 @@ def dominant_color(img: Image.Image) -> tuple[int, int, int]:
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)
@ -707,17 +399,11 @@ def set_light_color(rgb: tuple[int, int, int]) -> None:
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)
@ -729,7 +415,7 @@ def main() -> None:
time.sleep(2)
continue
if CONTROL_STATE.consume_force_update() or track_id != last_track:
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.")
@ -742,15 +428,9 @@ def main() -> None:
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}")
print(f"Aplicando cor: {rgb}")
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
@ -758,8 +438,6 @@ def main() -> None:
except KeyboardInterrupt:
print("\nEncerrado.")
control_server.shutdown()
control_server.server_close()
break
except Exception as exc:
print(f"Erro: {exc}")

3
music-light.env.example
View file

@ -3,6 +3,3 @@
# HA_TOKEN=replace-with-your-home-assistant-token
# LIGHT_ENTITY=light.luzsala_luz_sala
# PLAYERCTL_BIN=/usr/bin/playerctl
# CONTROL_HOST=0.0.0.0
# CONTROL_PORT=8765
# CONTROL_TOKEN=optional-shared-token