homelab-radio/sdr/speaker/speaker.py

#!/usr/bin/env python3
"""Play new SDR voice recordings on the speakers at the Pi's 3.5 mm jack, while switched on.

State (on / volume) is written by the radio web UI (web/speaker.py, LAN clients only) to
/var/lib/sdr-speaker/state.json. Only recordings that appear after the switch was turned on are played,
one after another; switching off stops the current one at once.
"""
import json
import re
import os
import subprocess
import time
import urllib.request

STATE = os.environ.get('SPEAKER_STATE', '/var/lib/sdr-speaker/state.json')
RECENT = os.environ.get('SDR_RECENT', 'http://127.0.0.1:8090/api/recent?limit=10')
CARD = os.environ.get('SPEAKER_CARD', 'Headphones')
DEVICE = f'plughw:CARD={CARD},DEV=0'
MAX_QUEUE = 8          # a burst of traffic must not keep the speakers busy for half an hour
SLEEP = os.environ.get('PI_SLEEP', '/run/pi-sleep.json')
NOW = os.environ.get('SPEAKER_NOW', '/run/sdr-speaker/now.json')   # what is playing, for the Pi's little screen


def publish(playing, queue, on, vol):
    data = {'on': on, 'volume': vol, 'queue': len(queue), 'playing': playing}
    try:
        tmp = NOW + '.tmp'
        with open(tmp, 'w', encoding='utf-8') as fh:
            json.dump(data, fh, ensure_ascii=False)
        os.replace(tmp, NOW)
    except OSError:
        pass


def state():
    try:
        with open(STATE, encoding='utf-8') as fh:
            st = json.load(fh)
    except (OSError, ValueError):
        st = {}
    on = bool(st.get('on'))
    try:   # calendar sleep window (/opt/pi/bin/sleep-window): silent, the switch itself stays as the user left it
        with open(SLEEP, encoding='utf-8') as fh:
            sl = json.load(fh)
        if sl.get('asleep') and time.time() - sl.get('at', 0) < 600:
            on = False
    except (OSError, ValueError):
        pass
    return on, int(st.get('volume', 70))


def recent():
    try:
        with urllib.request.urlopen(RECENT, timeout=5) as r:
            return json.loads(r.read())
    except Exception:  # noqa: BLE001
        return None          # unreachable (e.g. the radio page still starting): not the same as "no recordings"


def older_than(x, t):
    """Recording made before t (unix time)? Recordings from before the service started are never played."""
    try:
        from datetime import datetime
        return datetime.fromisoformat(x['ts']).timestamp() < t
    except (KeyError, ValueError, TypeError):
        return False


# The speakers take their power from port 4 of the powered VIA hub (per-port power switching): idle they hiss,
# so the port is switched off after IDLE_OFF s without playback and back on (with a warm-up) before the next one.
HUB, HUB_PORT = os.environ.get('SPEAKER_HUB', '1-1'), os.environ.get('SPEAKER_HUB_PORT', '').strip()
# only a single port number switches anything: an empty/odd value would make uhubctl act on EVERY port
HUB_PORT = HUB_PORT if re.fullmatch(r'[1-9](,[1-9])*', HUB_PORT) else ''   # e.g. "3" or "3,4"
IDLE_OFF, WARMUP = 30, 1.5
_power = {'on': None, 'since': 0.0}


def speaker_power(on):
    if not HUB_PORT or _power['on'] == on:   # no port configured: power switching is off
        return
    # the speakers only draw power: never switch a port with a USB device on it (the camera was moved onto the
    # speakers' port once, and every pause cut it off)
    busy = [p for p in HUB_PORT.split(',') if os.path.exists(f'/sys/bus/usb/devices/{HUB}.{p}')]
    if busy:
        if _power.get('busy') != busy:
            print(f"speaker power: port {','.join(busy)} has a USB device on it, not switching", flush=True)
            _power['busy'] = busy
        return
    r = subprocess.run(['uhubctl', '-l', HUB, '-p', HUB_PORT, '-a', 'on' if on else 'off'],
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
    if r.returncode == 0:
        _power.update(on=on, since=time.time())
        print(f"speaker power {'on' if on else 'off'}", flush=True)
        try:   # for the dashboard (sysinfo-api): /run/sdr-speaker is this unit's RuntimeDirectory
            with open('/run/sdr-speaker/speakers.json', 'w') as fh:
                json.dump({'on': on, 'since': _power['since'], 'port': HUB_PORT}, fh)
        except OSError:
            pass
        if on:
            time.sleep(WARMUP)   # let the amplifier wake up before the first word


def set_volume(v):
    subprocess.run(['amixer', '-q', '-c', CARD, 'sset', 'PCM', f'{v}%'], check=False,
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def main():
    seen, queue, proc, was_on, vol_now, playing = set(), [], None, False, None, None
    started, baseline = time.time(), False
    last_play = 0.0
    speaker_power(False)
    while True:
        on, vol = state()
        if on and vol != vol_now:
            set_volume(vol)
            vol_now = vol
        got = recent()
        if got is None:       # the list could not be read: never treat that as an empty history
            time.sleep(1.5)
            continue
        items = [x for x in got if x.get('class') == 'voice']
        if not baseline:      # first answer after start (boot): whatever is there already is old, never play it
            seen, baseline, was_on = {x['id'] for x in got}, True, on
            time.sleep(1.5)
            continue
        if on and not was_on:
            seen = {x['id'] for x in items}          # switching on never replays what is already there
            queue = []
        elif on:
            for x in reversed(items):                 # oldest first
                if x['id'] not in seen and not older_than(x, started - 30):
                    seen.add(x['id'])
                    queue.append(x)
            del queue[:-MAX_QUEUE]
        if not on:
            queue = []
            if proc and proc.poll() is None:
                proc.terminate()
        was_on = on
        if on and queue and (proc is None or proc.poll() is not None):
            speaker_power(True)
            x = queue.pop(0)
            proc = subprocess.Popen(['ffmpeg', '-nostdin', '-hide_banner', '-loglevel', 'error', '-i', x['url'],
                                     '-af', 'apad=pad_dur=0.4', '-f', 'alsa', DEVICE],
                                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            print(f"play {x.get('channel')} {x.get('ts')}", flush=True)
            playing = {k: x.get(k) for k in ('channel', 'group', 'ts', 'freq_mhz', 'duration_s')}
            playing['started'] = time.time()
        if proc is None or proc.poll() is not None:
            if proc is not None and playing is None:
                proc = None
            if proc is not None and proc.poll() is not None:
                playing = None
        busy = proc is not None and proc.poll() is None
        if busy:
            last_play = time.time()
        if _power['on'] and not busy and not queue and (not on or time.time() - last_play > IDLE_OFF):
            speaker_power(False)
        publish(playing if on else None, queue, on, vol)
        if len(seen) > 2000:
            seen = {x['id'] for x in items}
        time.sleep(1.5)


if __name__ == '__main__':
    main()