OpsLog/internal/cat/tci_audio.go

package cat

// TCI audio — receiving the radio's audio over the same WebSocket that carries
// the commands, so a SunSDR needs no virtual audio cable.
//
// TCI mixes two kinds of frame on one socket: TEXT frames are the commands
// ("trx:0,true;"), BINARY frames are streams. A binary frame is a fixed header
// followed by float32 samples:
//
//	uint32 receiver     which receiver the stream belongs to
//	uint32 sampleRate   Hz
//	uint32 format       0 = float32
//	uint32 codec        0 = uncompressed
//	uint32 crc          unused in practice
//	uint32 length       samples in the payload
//	uint32 type         which stream this is (see tciStream*)
//	uint32 reserved[9]
//	float32 payload[…]  stereo, interleaved
//
// The stream is asked for with "audio_samplerate:" then "audio_start:<rx>;",
// and stopped with "audio_stop:<rx>;".
//
// NOTHING HERE IS CONFIRMED ON A RADIO YET. The layout above is read from the
// TCI documentation, and the stream-type numbers in particular are the sort of
// detail a document gets right and a memory of it does not — so every header is
// logged for the first few seconds of a session, and the numbers the radio
// actually sends will settle it. Same discipline as the Yaesu meters and the
// Flex spot feed: measure on the real thing, then write the constant down.

import (
	"encoding/binary"
	"fmt"
	"math"
	"strings"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

// TCI stream types. RX audio is the one this file consumes; the others are
// named so a log line says what arrived rather than "type 3".
const (
	tciStreamIQ       = 0
	tciStreamRXAudio  = 1
	tciStreamTXAudio  = 2
	tciStreamTXChrono = 3
)

// tciHeaderWords is the header length in uint32 words (7 named + 9 reserved).
const tciHeaderWords = 16

// tciHeaderBytes is the same in bytes.
const tciHeaderBytes = tciHeaderWords * 4

// tciAudioProbeMax bounds the header logging. Enough frames to see the shape
// and the rate; few enough that an evening of listening does not fill the log.
const tciAudioProbeMax = 40

// TCIAudioStatus is what the panel polls while testing the stream.
type TCIAudioStatus struct {
	Running    bool  `json:"running"`
	SampleRate int   `json:"sample_rate"`
	Frames     int64 `json:"frames"`  // binary frames accepted
	Samples    int64 `json:"samples"` // audio samples decoded
	// PeakDB is the loudest sample of the last second, in dBFS: the one number
	// that says "audio is really arriving" rather than "a socket is open".
	PeakDB  float64 `json:"peak_db"`
	LastErr string  `json:"last_err,omitempty"`
}

// tciAudio is the receive-side state, kept on the backend so it lives exactly
// as long as the connection does.
type tciAudio struct {
	mu          sync.Mutex
	want        bool // the host asked for audio
	rx          int  // which receiver
	rate        int
	frames      int64
	samples     int64
	peak        float64
	peakAt      time.Time
	probeByType map[int]int
	// countByType counts EVERY frame per stream type, capped by nothing.
	// The probe above stops logging after forty frames of a type; these keep
	// counting, so a transmission that produced no transmit frames at all can
	// be reported as a fact rather than inferred from an absence of lines.
	countByType map[int]int64
	lastErr     string
	// widthLogged keeps the one-line note about the sample width to once a
	// session — it is a fact about the radio, not an event.
	widthLogged bool
	// txMark is the per-type frame count when transmission began, so the census
	// at the end reports the pass rather than the whole session.
	txMark map[int]int64

	// txFeed supplies the next frame of transmit audio when the radio asks for
	// one, or is nil when nothing is being sent. Set under this same lock, and
	// read on the reader goroutine — the radio's request and our answer are two
	// halves of one exchange and must not straddle a race.
	txFeed  func(samples int) []byte
	txSent  int64
	txShort int64 // requests the feed could not fill (it had run out)

	// What the radio SAID about its stream at connect (audio_stream_sample_type,
	// audio_stream_channels). Its own declaration, and it arrives before the
	// first frame — the frame arithmetic below stays as the check on it rather
	// than as the only source.
	declaredType  string
	declaredChans int

	// OnSamples receives decoded MONO samples (the two channels averaged) at
	// the negotiated rate. Mono because everything downstream — the QSO
	// recorder, the CW decoder — works on one channel, and a receiver's two
	// channels carry the same audio.
	OnSamples func(rate int, samples []float32)
}

// StartTCIAudio asks the radio to stream receiver rx's audio.
func (t *TCI) StartTCIAudio(rx, rate int) error {
	if rate <= 0 {
		rate = 48000
	}
	t.audio.mu.Lock()
	t.audio.want = true
	t.audio.rx = rx
	t.audio.rate = rate
	t.audio.frames, t.audio.samples, t.audio.peak = 0, 0, 0
	t.audio.lastErr = ""
	t.audio.mu.Unlock()

	// Sample rate first: the radio applies it to the stream it is about to
	// open, and asking afterwards restarts the stream on some firmware.
	if err := t.send(fmt.Sprintf("audio_samplerate:%d;", rate)); err != nil {
		return err
	}
	// Said rather than assumed. float32 and two channels are the documented
	// defaults and what this radio streams, but a default is a thing another
	// program can have changed — they share the radio, not just the protocol —
	// and a stream arriving in a format the decoder was not expecting is heard
	// as noise, not as a mistake.
	_ = t.send("audio_stream_sample_type:float32;")
	_ = t.send("audio_stream_channels:2;")
	return t.send(fmt.Sprintf("audio_start:%d;", rx))
}

// SetTCIAudioSink installs (or removes) the consumer of the decoded samples.
//
// One sink, not a list: today it is a test recording, tomorrow the QSO
// recorder, and two consumers of a live stream would need a policy about which
// one wins that nothing yet has an opinion about.
func (t *TCI) SetTCIAudioSink(fn func(rate int, samples []float32)) {
	t.audio.mu.Lock()
	t.audio.OnSamples = fn
	t.audio.mu.Unlock()
}

// StopTCIAudio closes the stream.
func (t *TCI) StopTCIAudio() error {
	t.audio.mu.Lock()
	t.audio.want = false
	rx := t.audio.rx
	t.audio.mu.Unlock()
	return t.send(fmt.Sprintf("audio_stop:%d;", rx))
}

// TCIAudioStatus reports what has arrived.
func (t *TCI) TCIAudioStatus() TCIAudioStatus {
	t.audio.mu.Lock()
	defer t.audio.mu.Unlock()
	st := TCIAudioStatus{
		Running:    t.audio.want,
		SampleRate: t.audio.rate,
		Frames:     t.audio.frames,
		Samples:    t.audio.samples,
		LastErr:    t.audio.lastErr,
	}
	// A peak older than a second is not a level, it is a memory. Reported as
	// silence rather than left standing, so a stream that has stopped arriving
	// looks stopped.
	if time.Since(t.audio.peakAt) < time.Second && t.audio.peak > 0 {
		st.PeakDB = 20 * math.Log10(t.audio.peak)
	} else {
		st.PeakDB = -99
	}
	return st
}

// handleBinary decodes one binary WebSocket frame.
//
// Called from the reader goroutine. Anything malformed is counted and dropped:
// a stream frame is not worth breaking the command connection over, and the
// command connection is what keeps the radio usable.
func (t *TCI) handleBinary(data []byte) {
	if len(data) < tciHeaderBytes {
		t.audioErr(fmt.Sprintf("binary frame of %d bytes is shorter than a header", len(data)))
		return
	}
	le := binary.LittleEndian
	receiver := int(le.Uint32(data[0:]))
	rate := int(le.Uint32(data[4:]))
	format := le.Uint32(data[8:])
	codec := le.Uint32(data[12:])
	length := int(le.Uint32(data[20:]))
	stype := int(le.Uint32(data[24:]))

	// Counted PER STREAM TYPE, not overall.
	//
	// A single counter was spent on the first forty receive-audio frames, which
	// arrive twenty-four times a second — so a transmit-chrono or transmit-audio
	// frame, the two this needs to see before the voice keyer can be written,
	// would never have been logged at all. They only appear once the operator
	// keys the radio, long after any global budget is gone.
	t.audio.mu.Lock()
	if t.audio.probeByType == nil {
		t.audio.probeByType = map[int]int{}
	}
	if t.audio.countByType == nil {
		t.audio.countByType = map[int]int64{}
	}
	t.audio.countByType[stype]++
	probe := t.audio.probeByType[stype]
	if probe < tciAudioProbeMax {
		t.audio.probeByType[stype]++
	}
	t.audio.mu.Unlock()
	if probe < tciAudioProbeMax {
		debugLog.Printf("TCI: binary frame — rx=%d rate=%d format=%d codec=%d length=%d type=%d payload=%d bytes",
			receiver, rate, format, codec, length, stype, len(data)-tciHeaderBytes)
	}

	if stype == tciStreamTXChrono {
		// The radio asking for the next frame of transmit audio. It is empty —
		// the whole message IS the request — and it carries the size it wants in
		// the header's length field, so the answer is written from what it says
		// rather than from what we assumed.
		t.serveChrono(rate, length)
		return
	}
	if stype != tciStreamRXAudio {
		// IQ and transmit audio. The latter is ours to send, not to receive:
		// counted above, and dropped.
		return
	}
	if codec != 0 {
		t.audioErr(fmt.Sprintf("stream is codec=%d, and nothing here decodes a compressed stream", codec))
		return
	}

	// The FORMAT number is decided by measurement, not by the number itself.
	//
	// A real SunSDR answered format=3, where the code expected 0 — and 0 was a
	// guess from reading the documentation, which is exactly the kind of detail
	// a memory of a document gets wrong. Rather than swap one magic number for
	// another, the sample width is derived from what arrived: the header says
	// how many samples the payload holds, so the bytes per sample follow from
	// dividing. That is true whatever number the format field carries, on this
	// firmware and the next.
	payload := data[tciHeaderBytes:]
	if len(payload) == 0 || length <= 0 {
		return
	}
	width := len(payload) / length
	var n int
	switch width {
	case 4:
		n = len(payload) / 4 // float32
	case 2:
		n = len(payload) / 2 // 16-bit PCM
	default:
		t.audioErr(fmt.Sprintf("frame carries %d bytes for %d samples (format=%d) — not a width this reads",
			len(payload), length, format))
		return
	}
	if n == 0 {
		return
	}
	// Under the lock like the rest of the counters: the reader is the only
	// writer today, but a fact about the radio that is read from another
	// goroutine has no business being the one field left unguarded.
	t.audio.mu.Lock()
	first := !t.audio.widthLogged
	t.audio.widthLogged = true
	t.audio.mu.Unlock()
	if first {
		debugLog.Printf("TCI: audio is %d bytes per sample at %d Hz (format field says %d)", width, rate, format)
	}
	// Stereo interleaved → mono. Both channels of a receiver carry the same
	// audio, and everything downstream works on one.
	// How many channels are interleaved. The radio says so at connect; two is
	// the fallback, which is what every SunSDR seen so far streams.
	t.audio.mu.Lock()
	chans := t.audio.declaredChans
	t.audio.mu.Unlock()
	if chans <= 0 {
		chans = 2
	}
	mono := make([]float32, 0, n/chans+1)
	var peak float64
	sample := func(i int) float32 {
		if width == 2 {
			// 16-bit PCM, scaled to the same -1…1 the rest of the audio path
			// works in, so a change of format cannot change what a level means.
			return float32(int16(le.Uint16(payload[i*2:]))) / 32768
		}
		return math.Float32frombits(le.Uint32(payload[i*4:]))
	}
	for i := 0; i+chans-1 < n; i += chans {
		var sum float32
		for c := 0; c < chans; c++ {
			sum += sample(i + c)
		}
		v := sum / float32(chans)
		if a := math.Abs(float64(v)); a > peak {
			peak = a
		}
		mono = append(mono, v)
	}

	t.audio.mu.Lock()
	t.audio.frames++
	t.audio.samples += int64(len(mono))
	if rate > 0 {
		t.audio.rate = rate
	}
	if peak > t.audio.peak || time.Since(t.audio.peakAt) > time.Second {
		t.audio.peak = peak
		t.audio.peakAt = time.Now()
	}
	cb := t.audio.OnSamples
	t.audio.mu.Unlock()

	if cb != nil {
		cb(rate, mono)
	}
}

// audioErr records a decoding complaint, once, so the panel can show it without
// the log filling with the same line at fifty frames a second.
func (t *TCI) audioErr(msg string) {
	t.audio.mu.Lock()
	first := t.audio.lastErr != msg
	t.audio.lastErr = msg
	t.audio.mu.Unlock()
	if first {
		debugLog.Printf("TCI: audio: %s", msg)
	}
}

// resumeAudio re-opens the stream after a reconnect, if the host had asked for
// it. A dropped WebSocket takes the audio with it, and an operator who switched
// recording on does not expect to switch it on again.
func (t *TCI) resumeAudio() {
	t.audio.mu.Lock()
	want, rx, rate := t.audio.want, t.audio.rx, t.audio.rate
	t.audio.mu.Unlock()
	if !want {
		return
	}
	if err := t.StartTCIAudio(rx, rate); err != nil {
		debugLog.Printf("TCI: re-opening the audio stream failed: %v", err)
	}
}

// wsMessageIsBinary keeps the type test in one place — the reader used to
// ignore the message type entirely and split every frame on ';', which would
// have fed audio bytes to the command parser the moment a stream was opened.
func wsMessageIsBinary(mt int) bool { return mt == websocket.BinaryMessage }

// noteTXTransition reports what the stream did across a transmission.
//
// The voice keyer needs two numbers the documentation does not give: the size
// and the cadence of the frames the radio expects while transmitting. They can
// only be read off a real transmission — and the first attempt came back with a
// log that said nothing at all, which is ambiguous: either no transmit frames
// arrived, or they arrived and went unlogged.
//
// So the boundaries are marked and every stream type is counted. A pass that
// produces "type 1: 240, and nothing else" is a RESULT — it says the radio
// sends no chrono unless something more is asked of it — where a log with no
// transmit lines in it was merely a silence.
func (t *TCI) noteTXTransition(on bool) {
	t.audio.mu.Lock()
	if t.audio.countByType == nil {
		t.audio.countByType = map[int]int64{}
	}
	if on {
		// Let the transmit types speak again on every pass: forty frames is a
		// budget spent long before the operator gets round to keying.
		if t.audio.probeByType != nil {
			delete(t.audio.probeByType, tciStreamTXAudio)
			delete(t.audio.probeByType, tciStreamTXChrono)
		}
		t.audio.txMark = map[int]int64{}
		for k, v := range t.audio.countByType {
			t.audio.txMark[k] = v
		}
		streaming := t.audio.want
		t.audio.mu.Unlock()
		debugLog.Printf("TCI: TRANSMIT started — watching for transmit-audio (type %d) and chrono (type %d) frames; receive stream is %s",
			tciStreamTXAudio, tciStreamTXChrono, map[bool]string{true: "open", false: "CLOSED (tick the TCI recording option, or the radio has no reason to stream)"}[streaming])
		return
	}
	names := map[int]string{
		tciStreamIQ:       "IQ",
		tciStreamRXAudio:  "receive audio",
		tciStreamTXAudio:  "transmit audio",
		tciStreamTXChrono: "transmit chrono",
	}
	var parts []string
	for _, k := range []int{tciStreamIQ, tciStreamRXAudio, tciStreamTXAudio, tciStreamTXChrono} {
		if n := t.audio.countByType[k] - t.audio.txMark[k]; n > 0 {
			parts = append(parts, fmt.Sprintf("%s (type %d): %d", names[k], k, n))
		}
	}
	t.audio.mu.Unlock()
	if len(parts) == 0 {
		debugLog.Printf("TCI: TRANSMIT ended — NO binary frames of any type arrived during it")
		return
	}
	debugLog.Printf("TCI: TRANSMIT ended — frames during the pass: %s", strings.Join(parts, ", "))
}