268 lines
11 KiB
Python
268 lines
11 KiB
Python
"""
|
|
Applet: SUB/WAVE Now Playing
|
|
Summary: Shows what's playing on a SUB/WAVE station
|
|
Description: Configure with your SUB/WAVE (github.com/perminder-klair/subwave) station's base URL to display the current track, artist, show, host, album art, time/temp, and a progress bar.
|
|
Author: poprhythm
|
|
"""
|
|
|
|
load("render.star", "render")
|
|
load("schema.star", "schema")
|
|
load("http.star", "http")
|
|
load("cache.star", "cache")
|
|
load("time.star", "time")
|
|
load("animation.star", "animation")
|
|
|
|
DEFAULT_STATION_URL = ""
|
|
NOWPLAYING_TTL = 20
|
|
COVER_TTL = 300
|
|
|
|
COLOR_TITLE = "#22d3ee" # cyan
|
|
COLOR_ARTIST = "#22d3ee" # cyan
|
|
COLOR_SEPARATOR = "#444444" # thin divider between artist and show/host
|
|
COLOR_SHOW = "#e0e0e0" # near-white
|
|
COLOR_HOST = "#ec4899" # magenta/pink
|
|
COLOR_META = "#888888" # muted gray, for the time/temp line
|
|
COLOR_BAR_FILLED = "#c026d3" # magenta/purple
|
|
COLOR_BAR_EMPTY = "#333333"
|
|
|
|
def main(config):
|
|
station_url = config.get("station_url", DEFAULT_STATION_URL)
|
|
if station_url:
|
|
station_url = station_url.rstrip("/")
|
|
|
|
if not station_url:
|
|
return render.Root(child = _message_screen(["Set station URL", "in app settings"]))
|
|
|
|
now = _fetch_now_playing(station_url)
|
|
if now == None:
|
|
return render.Root(child = _message_screen(["Can't reach", "station"]))
|
|
|
|
if not now["online"]:
|
|
return render.Root(child = _message_screen([now["station"], "offline"]))
|
|
|
|
art = _fetch_cover(station_url, now["subsonic_id"])
|
|
progress = _track_progress(now["subsonic_id"], now["duration"])
|
|
|
|
top_lines = [
|
|
render.Marquee(width = 32, child = render.Text(now["title"], color = COLOR_TITLE)),
|
|
render.Marquee(width = 32, child = render.Text(now["artist"], color = COLOR_ARTIST)),
|
|
]
|
|
if now["show"]:
|
|
top_lines.append(render.Box(width = 32, height = 1, color = COLOR_SEPARATOR))
|
|
show_host_row = [render.Text(now["show"], font = "tom-thumb", color = COLOR_SHOW)]
|
|
if now["host"]:
|
|
show_host_row.append(render.Text(" " + now["host"], font = "tom-thumb", color = COLOR_HOST))
|
|
top_lines.append(render.Marquee(width = 32, child = render.Row(children = show_host_row)))
|
|
if now["time_temp"]:
|
|
top_lines.append(render.Marquee(width = 32, child = render.Text(now["time_temp"], font = "tom-thumb", color = COLOR_META)))
|
|
|
|
children = [render.Column(children = top_lines)]
|
|
if progress != None:
|
|
children.append(_progress_bar(progress["fraction"], width = 32))
|
|
|
|
text_column = render.Box(
|
|
width = 32,
|
|
height = 32,
|
|
child = render.Column(
|
|
main_align = "space_between",
|
|
cross_align = "start",
|
|
expanded = True,
|
|
children = children,
|
|
),
|
|
)
|
|
|
|
if art:
|
|
content = render.Row(children = [_art_widget(art, now["subsonic_id"]), text_column])
|
|
else:
|
|
content = render.Column(
|
|
main_align = "center",
|
|
cross_align = "center",
|
|
expanded = True,
|
|
children = [
|
|
render.Marquee(width = 64, child = render.Text(now["title"], color = COLOR_TITLE)),
|
|
render.Marquee(width = 64, child = render.Text(now["artist"], color = COLOR_ARTIST)),
|
|
],
|
|
)
|
|
|
|
return render.Root(child = content)
|
|
|
|
# Album art gets a slow "Ken Burns" breathing effect: eases from one corner
|
|
# out to the full frame, then back into that same corner, on a continuous
|
|
# loop -- not a one-shot reveal. A one-shot version (start in a corner, end
|
|
# on the full frame, then hold) looked jumpy: since the app re-renders on
|
|
# NOWPLAYING_TTL and each fresh render restarts its own animation from tick
|
|
# 0, whatever the *previous* pushed image happened to be showing at that
|
|
# moment gets abruptly replaced by a fresh "zoomed into a corner" frame. The
|
|
# back-and-forth loop doesn't eliminate that (a server-side re-render always
|
|
# restarts at tick 0 regardless), but ending exactly where it started means
|
|
# the animation also loops seamlessly on-device between refreshes, which in
|
|
# practice reads as far less jarring than a one-way reveal that snaps back.
|
|
#
|
|
# Art is rendered oversized (45x45, vs. the 32x32 display slot) since
|
|
# animation.Scale above 1.0 has no visible effect on a widget whose declared
|
|
# size already matches its container (confirmed elsewhere this session) --
|
|
# starting at native 1.0 scale of the *oversized* image and easing down to
|
|
# ~32/45 achieves "zoom out to reveal everything" while only ever using
|
|
# Scale <= 1.0. Combining Scale + Translate in one Transformation, and
|
|
# combining that with the app's 4 already-scrolling Marquee lines, was
|
|
# verified not to trigger any *additional* glitching beyond the already-
|
|
# accepted minor Marquee quirk (see README). Duration 1200 ticks x the
|
|
# default 50ms/tick = 60s round-trip (30s out, 30s back); verified
|
|
# empirically that a slower zoom compresses *smaller*, not larger, since more
|
|
# adjacent ticks become pixel-identical at this resolution -- ~97-264KB
|
|
# depending on the art's own complexity, well under the 460KB device limit.
|
|
ART_SIZE = 45
|
|
ART_SLOT = 32
|
|
ART_MAX_OFFSET = (ART_SIZE - ART_SLOT) / 2.0 # 6.5px: centered -> true corner
|
|
ART_END_SCALE = float(ART_SLOT) / float(ART_SIZE) # ~0.71: full art fits in the slot
|
|
ART_ZOOM_DURATION = 1200 # 60s at the default 50ms/tick
|
|
|
|
CORNER_OFFSETS = [
|
|
(-ART_MAX_OFFSET, -ART_MAX_OFFSET),
|
|
(ART_MAX_OFFSET, -ART_MAX_OFFSET),
|
|
(-ART_MAX_OFFSET, ART_MAX_OFFSET),
|
|
(ART_MAX_OFFSET, ART_MAX_OFFSET),
|
|
]
|
|
|
|
def _corner_for_track(subsonic_id):
|
|
# Pseudo-random but stable per track, so re-rendering the same
|
|
# still-playing track (every NOWPLAYING_TTL seconds) doesn't jarringly
|
|
# jump to a different corner -- only a genuinely new track does.
|
|
total = 0
|
|
for ch in subsonic_id.elems():
|
|
total += ord(ch)
|
|
return CORNER_OFFSETS[total % len(CORNER_OFFSETS)]
|
|
|
|
def _art_widget(art_bytes, subsonic_id):
|
|
corner_x, corner_y = _corner_for_track(subsonic_id)
|
|
big_art = render.Image(src = art_bytes, width = ART_SIZE, height = ART_SIZE)
|
|
zoom = animation.Transformation(
|
|
child = big_art,
|
|
duration = ART_ZOOM_DURATION,
|
|
keyframes = [
|
|
animation.Keyframe(
|
|
percentage = 0.0,
|
|
transforms = [animation.Scale(1.0, 1.0), animation.Translate(corner_x, corner_y)],
|
|
),
|
|
animation.Keyframe(
|
|
percentage = 0.5,
|
|
transforms = [animation.Scale(ART_END_SCALE, ART_END_SCALE), animation.Translate(0, 0)],
|
|
curve = "ease_in_out",
|
|
),
|
|
animation.Keyframe(
|
|
percentage = 1.0,
|
|
transforms = [animation.Scale(1.0, 1.0), animation.Translate(corner_x, corner_y)],
|
|
curve = "ease_in_out",
|
|
),
|
|
],
|
|
)
|
|
return render.Box(width = ART_SLOT, height = ART_SLOT, child = zoom)
|
|
|
|
def _message_screen(lines):
|
|
return render.Column(
|
|
main_align = "center",
|
|
cross_align = "center",
|
|
expanded = True,
|
|
children = [render.Text(line, font = "tom-thumb") for line in lines],
|
|
)
|
|
|
|
def _progress_bar(fraction, width = 32, height = 2):
|
|
filled = int(width * fraction + 0.5)
|
|
filled = max(0, min(filled, width))
|
|
children = []
|
|
if filled > 0:
|
|
children.append(render.Box(width = filled, height = height, color = COLOR_BAR_FILLED))
|
|
if filled < width:
|
|
children.append(render.Box(width = width - filled, height = height, color = COLOR_BAR_EMPTY))
|
|
return render.Row(children = children)
|
|
|
|
def _fetch_now_playing(station_url):
|
|
rep = http.get(station_url + "/api/now-playing", ttl_seconds = NOWPLAYING_TTL)
|
|
if rep.status_code != 200:
|
|
return None
|
|
|
|
data = rep.json()
|
|
# `.get(key, {})` only falls back to the default when the key is
|
|
# *absent* -- SUB/WAVE sends these as explicit JSON null (e.g.
|
|
# activeShow between scheduled shows), which decodes to Starlark None
|
|
# and passes right through .get(), so `or {}` is needed too.
|
|
np = data.get("nowPlaying", {}) or {}
|
|
dj = data.get("dj", {}) or {}
|
|
context = data.get("context", {}) or {}
|
|
active_show = context.get("activeShow", {}) or {}
|
|
|
|
return {
|
|
"title": np.get("title", "Unknown title"),
|
|
"artist": np.get("artist", "Unknown artist"),
|
|
"subsonic_id": np.get("subsonic_id", ""),
|
|
"duration": np.get("duration", 0),
|
|
"station": dj.get("station", "SUB/WAVE"),
|
|
"show": active_show.get("name", ""),
|
|
"host": dj.get("name", ""),
|
|
"time_temp": _format_time_temp(context.get("clock", {}) or {}, context.get("weather", {}) or {}),
|
|
"online": bool(data.get("streamOnline", False)),
|
|
}
|
|
|
|
def _format_time_temp(clock, weather):
|
|
parts = []
|
|
display_time = clock.get("display", "")
|
|
if display_time:
|
|
parts.append(display_time)
|
|
|
|
temp = weather.get("temp", None)
|
|
unit = weather.get("tempUnit", "")
|
|
if temp != None:
|
|
parts.append("%d°%s" % (int(temp), unit))
|
|
|
|
return " · ".join(parts)
|
|
|
|
def _fetch_cover(station_url, subsonic_id):
|
|
if not subsonic_id:
|
|
return None
|
|
rep = http.get(station_url + "/api/cover/" + subsonic_id, ttl_seconds = COVER_TTL)
|
|
if rep.status_code != 200:
|
|
return None
|
|
return rep.body()
|
|
|
|
# The now-playing API reports track duration but not elapsed/remaining time,
|
|
# so we track playback position ourselves: the first time we see a given
|
|
# subsonic_id, we stamp the current wall-clock time in the cache; on later
|
|
# renders (while the cache entry is still alive) we diff against that stamp
|
|
# to get elapsed seconds. This only works on a long-lived server process
|
|
# with a persistent cache (confirmed true for tronbyt/server's in-memory
|
|
# cache) -- a one-shot `pixlet render` CLI call always starts a fresh
|
|
# process with an empty cache, so progress will always show 0% locally.
|
|
def _track_progress(subsonic_id, duration):
|
|
if not subsonic_id or not duration:
|
|
return None
|
|
|
|
key = "subwave_track_start_" + subsonic_id
|
|
now_unix = time.now().unix
|
|
cached = cache.get(key)
|
|
|
|
if cached == None:
|
|
cache.set(key, str(now_unix), ttl_seconds = int(duration) + 30)
|
|
start_unix = now_unix
|
|
else:
|
|
start_unix = int(cached)
|
|
|
|
elapsed = max(0, now_unix - start_unix)
|
|
remaining = max(0, duration - elapsed)
|
|
fraction = min(1.0, float(elapsed) / float(duration))
|
|
|
|
return {"elapsed": elapsed, "remaining": remaining, "fraction": fraction}
|
|
|
|
def get_schema():
|
|
return schema.Schema(
|
|
version = "1",
|
|
fields = [
|
|
schema.Text(
|
|
id = "station_url",
|
|
name = "Station URL",
|
|
desc = "Base URL of your SUB/WAVE station, e.g. https://radio.example.com",
|
|
icon = "server",
|
|
default = DEFAULT_STATION_URL,
|
|
),
|
|
],
|
|
)
|