""" 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), 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 the full # frame in to a zoomed-in crop of the top-left corner, then back out, on a # continuous loop -- not a one-shot reveal. A one-shot version (start # zoomed in, 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-in # 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 has no visible effect on a widget whose declared size # already matches its container. Only ever scaling *up* from 1.0 (never # below) is deliberate, not cosmetic: this app originally scaled down to # ~32/45 to "zoom out" from a corner reached via animation.Translate, which # produced a real, on-device black border. Empirical testing (rendering the # actual Row layout at native canvas size, no forced output dimensions) # showed the outer Box auto-fits an oversized child to its own size at # animation.Scale(1.0, 1.0) -- so scaling *down* from there shrinks the # already-fitted image further, exposing background outside its bounds # (the black border). Scaling *up* only ever crops further into the image, # which is always safe. animation.Scale's pivot defaults to the top-left # corner (confirmed empirically; an explicit `origin` param had no effect # at all, so isn't used here) -- there's no reliable way to zoom into any # other corner, so this always zooms into the top-left. Duration 1200 # ticks x the default 50ms/tick = 60s round-trip (30s in, 30s out); # verified empirically that a slower zoom compresses *smaller*, not # larger, since more adjacent ticks become pixel-identical at this # resolution -- well under the 460KB device limit. ART_SIZE = 45 ART_SLOT = 32 ART_ZOOM_SCALE = 1.4 # how far in to zoom on the top-left corner ART_ZOOM_DURATION = 1200 # 60s at the default 50ms/tick def _art_widget(art_bytes): art = render.Image(src = art_bytes, width = ART_SIZE, height = ART_SIZE) zoom = animation.Transformation( child = art, duration = ART_ZOOM_DURATION, keyframes = [ animation.Keyframe( percentage = 0.0, transforms = [animation.Scale(ART_ZOOM_SCALE, ART_ZOOM_SCALE)], ), animation.Keyframe( percentage = 0.5, transforms = [animation.Scale(1.0, 1.0)], curve = "ease_in_out", ), animation.Keyframe( percentage = 1.0, transforms = [animation.Scale(ART_ZOOM_SCALE, ART_ZOOM_SCALE)], 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, ), ], )