subwave-nowplaying/subwave_nowplaying.star

"""
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")

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 footer
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)),
    ]
    # Note: show/host is plain Text/Row, not Marquee. Pixlet's Marquee runs
    # its own internal scroll-cycle animation regardless of whether the text
    # actually needs to scroll, and combining multiple Marquees with very
    # different cycle lengths in one Root desyncs them -- confirmed via an
    # isolated repro where a short Marquee'd line intermittently collapsed to
    # a single character on frames where the long title/artist marquees were
    # mid-cycle. Plain Text just clips if too wide, an acceptable tradeoff
    # for this short field.
    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.Row(children = show_host_row))

    text_column = render.Column(children = top_lines)

    if art:
        base = render.Row(children = [render.Image(src = art, width = 32, height = 32), text_column])
    else:
        base = 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)),
            ],
        )

    # Time/temp and the progress bar are rendered as a full-width (64px)
    # footer overlaid on top of the base layout via Stack, rather than
    # squeezed into the 32px-wide text column -- gives roughly 2x the
    # character budget for the time/temp line, at the cost of covering the
    # bottom ~9px of the album art with a translucent bar.
    footer_children = []
    if now["time_temp"]:
        footer_children.append(
            render.Box(
                width = 64,
                height = 7,
                color = "#000000cc",
                child = render.Row(
                    expanded = True,
                    main_align = "center",
                    cross_align = "center",
                    children = [render.Text(now["time_temp"], font = "tom-thumb", color = COLOR_META)],
                ),
            ),
        )
    if progress != None:
        footer_children.append(_progress_bar(progress["fraction"], width = 64))

    if not footer_children:
        return render.Root(child = base)

    footer = render.Box(
        width = 64,
        height = 32,
        child = render.Column(
            main_align = "end",
            cross_align = "start",
            expanded = True,
            children = [render.Column(children = footer_children)],
        ),
    )

    return render.Root(child = render.Stack(children = [base, footer]))

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()
    np = data.get("nowPlaying", {})
    dj = data.get("dj", {})
    context = data.get("context", {})
    active_show = context.get("activeShow", {})

    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", {}), context.get("weather", {})),
        "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,
            ),
        ],
    )