OpsLog/internal/cat/icomaudio.go

package cat

// icomaudio.go — the NETWORK AUDIO stream (UDP 50003) for the Icom LAN protocol.
// It is the third stream alongside control (50001) and CI-V (50002): once the
// control login + conninfo (with rxenable=1) authorize audio, the rig streams RX
// audio here as data packets. This file dials/handshakes/keeps-alive that socket
// exactly like the CI-V stream (icomnet.go) — those parts are byte-for-byte the
// PROVEN transport — and hands each received audio payload to a sink callback
// (the app decodes it via an audio.Codec and plays it through the RX monitor).
//
// Reuses icomnet.go's helpers (icnCtrl, icnHandshake, icnPingReply, icnRecv,
// icnLocalID, icnLE) and the same seq/retransmit discipline.
//
// ⚠️ PAYLOAD OFFSET PENDING ON-RIG VERIFICATION. The stream framing (handshake,
// ping, idle, retransmit, common 16-byte header) is identical to CI-V and proven.
// The AUDIO data packet's inner layout — where the PCM starts and the datalen
// field — is reconstructed from wfview's audio_packet (ident@0x10, datalen@0x12,
// sendseq@0x14, audio@0x16) but NOT yet confirmed against a real 50003 capture.
// audioPump logs the first few raw packets (icaDumpFirst) so the offset can be
// confirmed/corrected on the first on-rig test without a packet capture, the same
// way the CI-V/scope framing was iterated. Nothing here can destabilize CAT: the
// audio stream is opt-in and entirely separate from control/CI-V.

import (
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

// icaAudioOffset is where the PCM payload begins inside an audio data packet.
// CONFIRMED on a real IC-7760 (2026-08-29): 16-byte common header, ident@0x10,
// send seq (BE) @0x12, payload length (BE uint32) @0x14 — 0x500 observed on
// every packet — and the PCM starts at 0x18. The 0x16 first guessed from
// wfview's struct swallowed two header bytes into the audio, one broken sample
// per packet: a 50 Hz click track under everything.
const icaAudioOffset = 0x18

// icaDumpFirst is how many initial audio packets to hex-dump to the debug log for
// offset verification. After the layout is confirmed on a real rig this can go to
// 0 (or the const above corrected).
const icaDumpFirst = 6

// icomAudio is the connected audio stream. RX only for now (Phase 4); TX (Phase
// 5) will add an encode+send path mirroring icomNet.Write.
type icomAudio struct {
	conn         *net.UDPConn
	aID, aRemote uint32

	sink func([]byte) // receives each raw audio payload (app decodes + plays)

	// Receive-side retransmit (audio is a heavy stream, like the scope): track the
	// rig's data-packet send seq and ask it to resend gaps, or the rig drops the
	// session. Same mechanism as icomNet. Owned solely by audioPump → no lock.
	rxHaveSeq bool
	rxLastSeq uint16
	rxMissing map[uint16]int

	dumped  int    // packets hex-dumped so far (≤ icaDumpFirst)
	pingSeq uint16 // client-ping counter — the rig wants OUR pings too (see icnPing)
	started time.Time

	// Live-TX state — see SendTXChunk.
	txMu    sync.Mutex
	txRem   []byte
	txOuter uint16
	txSend  uint16
	lastRx  atomic.Int64 // UnixNano of last packet (liveness)

	done      chan struct{}
	closeOnce sync.Once
}

func (a *icomAudio) markRx() { a.lastRx.Store(time.Now().UnixNano()) }

// Close tears the audio stream down (disconnect a few times; UDP is lossy).
func (a *icomAudio) Close() {
	a.closeOnce.Do(func() {
		close(a.done)
		for i := 0; i < 3; i++ {
			_, _ = a.conn.Write(icnCtrl(0x05, 0, a.aID, a.aRemote)) // disconnect
			time.Sleep(15 * time.Millisecond)
		}
		_ = a.conn.Close()
		debugLog.Printf("icom audio: stream closed")
	})
}

// dialIcomAudio opens the audio UDP stream to rig:50003, binding LOCAL :50003
// (mirroring the civ stream's local :50002). The control conninfo (rxenable=1,
// audioport=50003) must already have authorized it. sink receives each raw audio
// payload. cancel aborts a slow dial (Stop/Start).
func dialIcomAudio(host string, sink func([]byte), cancel <-chan struct{}) (*icomAudio, error) {
	araddr, err := net.ResolveUDPAddr("udp4", net.JoinHostPort(host, "50003"))
	if err != nil {
		return nil, err
	}
	conn, err := net.DialUDP("udp4", &net.UDPAddr{Port: 50003}, araddr)
	if err != nil {
		debugLog.Printf("icom audio: cannot bind local :50003 (Remote Utility running?): %v", err)
		return nil, err
	}
	aID := icnLocalID(conn)
	aRemote, err := icnHandshake(conn, aID, cancel)
	if err != nil {
		_ = conn.Close()
		debugLog.Printf("icom audio: handshake FAILED: %v", err)
		return nil, err
	}
	_ = conn.SetReadBuffer(1 << 20)
	a := &icomAudio{
		conn: conn, aID: aID, aRemote: aRemote,
		started:   time.Now(),
		sink:      sink,
		rxMissing: make(map[uint16]int),
		done:      make(chan struct{}),
	}
	a.markRx()
	debugLog.Printf("icom audio: stream up (rig id 0x%08X) — awaiting RX audio", aRemote)
	go a.audioPump()
	return a, nil
}

// audioPump drains the audio socket: replies to pings, sends idle keepalives,
// requests retransmits for lost packets, and hands each audio payload to sink.
func (a *icomAudio) audioPump() {
	buf := make([]byte, 8192)
	lastIdle := time.Now()
	lastPing := time.Now()
	lastReq := time.Now()
	for {
		select {
		case <-a.done:
			return
		default:
		}
		_ = a.conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
		if k, err := a.conn.Read(buf); err == nil && k >= 16 {
			a.markRx()
			switch typ := icnLE.Uint16(buf[4:]); {
			case typ == 0x07: // ping
				_, _ = a.conn.Write(icnPingReply(buf[:k], a.aID, a.aRemote))
			case typ == 0x01: // retransmit request from the rig (we send no tracked audio yet)
			case typ == 0x05: // rig-initiated disconnect
				debugLog.Printf("icom audio: rig sent DISCONNECT — audio stream dropped by the rig")
			case typ == 0x00 && k > icaAudioOffset: // audio data packet
				fresh := a.trackRxSeq(icnLE.Uint16(buf[6:]))
				if a.dumped < icaDumpFirst {
					a.dumped++
					debugLog.Printf("icom audio raw #%d: len=%d head=% X", a.dumped, k, buf[:min(icaAudioOffset+8, k)])
				}
				// Only a packet that ADVANCES the sequence reaches the sink. A
				// duplicate or a late retransmit used to be delivered as if it
				// were the next 20 ms of audio: the monitor's capped ring threw
				// the surplus away (the speakers stayed clean), but the recorder
				// keeps every sample it is given — the file grew longer than the
				// QSO and played back slowed and stuttering, each lost-then-
				// resent packet heard twice.
				if fresh && a.sink != nil {
					payload := append([]byte(nil), buf[icaAudioOffset:k]...)
					a.sink(payload)
				}
			}
		}
		if time.Since(lastIdle) > 100*time.Millisecond {
			_, _ = a.conn.Write(icnCtrl(0x00, 0, a.aID, a.aRemote))
			lastIdle = time.Now()
		}
		if time.Since(lastPing) > 500*time.Millisecond {
			a.pingSeq++
			_, _ = a.conn.Write(icnPing(a.pingSeq, a.aID, a.aRemote, uint32(time.Since(a.started).Milliseconds())))
			lastPing = time.Now()
		}
		if time.Since(lastReq) > 100*time.Millisecond {
			a.sendRetransmitReq()
			lastReq = time.Now()
		}
	}
}

// trackRxSeq / sendRetransmitReq mirror icomNet's receive-side retransmit exactly
// (audio is as loss-sensitive as the scope stream). Duplicated deliberately so
// the audio stream owns its own seq state with no shared locking.
//
// The return value says whether this packet moves the stream FORWARD — the
// only kind the sink may hear. A duplicate is the same 20 ms again; a late
// retransmit would play old audio in the middle of new. Both are accounted
// for here and dropped by the caller.
func (a *icomAudio) trackRxSeq(seq uint16) bool {
	if !a.rxHaveSeq {
		a.rxHaveSeq = true
		a.rxLastSeq = seq
		return true
	}
	switch d := int16(seq - a.rxLastSeq); {
	case d == 0:
		return false
	case d < 0:
		delete(a.rxMissing, seq)
		return false
	case d == 1:
		a.rxLastSeq = seq
		return true
	case int(d) <= icnMaxMissing:
		for f := a.rxLastSeq + 1; f != seq; f++ {
			a.rxMissing[f] = 0
		}
		a.rxLastSeq = seq
		return true
	default:
		a.rxMissing = make(map[uint16]int)
		a.rxLastSeq = seq
		return true
	}
}

func (a *icomAudio) sendRetransmitReq() {
	if len(a.rxMissing) == 0 {
		return
	}
	if len(a.rxMissing) > icnMaxMissing {
		a.rxMissing = make(map[uint16]int)
		return
	}
	var seqs []uint16
	for s, cnt := range a.rxMissing {
		if cnt >= 4 {
			delete(a.rxMissing, s)
			continue
		}
		a.rxMissing[s] = cnt + 1
		seqs = append(seqs, s)
	}
	switch {
	case len(seqs) == 0:
		return
	case len(seqs) == 1:
		_, _ = a.conn.Write(icnCtrl(0x01, seqs[0], a.aID, a.aRemote))
	default:
		b := make([]byte, 16+4*len(seqs))
		icnLE.PutUint32(b[0:], uint32(len(b)))
		icnLE.PutUint16(b[4:], 0x01)
		icnLE.PutUint32(b[8:], a.aID)
		icnLE.PutUint32(b[12:], a.aRemote)
		off := 16
		for _, s := range seqs {
			icnLE.PutUint16(b[off:], s)
			icnLE.PutUint16(b[off+2:], s)
			off += 4
		}
		_, _ = a.conn.Write(b)
	}
}

// PlayTX sends one already-decoded message out over the audio session, paced
// at the stream's own 20 ms / 320-sample cadence, and returns when the last
// packet has gone or stop is closed.
//
// The frame mirrors what the rig itself sends on this socket byte for byte —
// ident 0x81 0x01 (LPCM mono 16-bit), a big-endian send sequence at 0x12, the
// payload length at 0x14, PCM from 0x18 — with the IDs swapped for direction.
// PTT is the caller's business, exactly as on the TCI and sound-card paths.
func (a *icomAudio) PlayTX(pcm []byte, rate, ch, bits int, stop <-chan struct{}) error {
	mono := decodeToMono(pcm, ch, bits)
	if len(mono) == 0 {
		return fmt.Errorf("the message is empty")
	}
	if rate > 0 && rate != 16000 {
		mono = resampleLinear(mono, rate, 16000)
	}
	// Through the SAME framer and counters as the live microphone. Each of the
	// two used to number its own packets from 1, and a voice-keyer message sent
	// after a talk session re-used sequence numbers the rig had already seen —
	// it keyed for the full length of the message and modulated none of it.
	const frame = 320 // samples per packet: 20 ms at 16 kHz, the rig's own cadence
	tick := time.NewTicker(20 * time.Millisecond)
	defer tick.Stop()
	buf := make([]byte, frame*2)
	for pos := 0; pos < len(mono); pos += frame {
		select {
		case <-stop:
			return nil
		case <-a.done:
			return fmt.Errorf("the audio stream closed mid-message")
		case <-tick.C:
		}
		for i := 0; i < frame; i++ {
			var v float32
			if pos+i < len(mono) {
				v = mono[pos+i] * 32767
			}
			if v > 32767 {
				v = 32767
			} else if v < -32768 {
				v = -32768
			}
			icnLE.PutUint16(buf[i*2:], uint16(int16(v)))
		}
		if err := a.SendTXChunk(buf); err != nil {
			return err
		}
	}
	return nil
}

// Live TX — the microphone, not a recorded message. Chunks arrive at the
// microphone's own real-time pace (16 kHz mono 16-bit, whatever length the
// capture delivers) and are re-framed into the rig's 320-sample packets; the
// remainder waits for the next chunk. Counters and remainder live on the
// stream so a talk session survives across calls.
func (a *icomAudio) SendTXChunk(pcm []byte) error {
	a.txMu.Lock()
	defer a.txMu.Unlock()
	a.txRem = append(a.txRem, pcm...)
	const frameBytes = 640
	for len(a.txRem) >= frameBytes {
		select {
		case <-a.done:
			return fmt.Errorf("the audio stream closed")
		default:
		}
		pkt := make([]byte, 0x18+frameBytes)
		icnLE.PutUint32(pkt[0:], uint32(len(pkt)))
		a.txOuter++
		icnLE.PutUint16(pkt[6:], a.txOuter)
		icnLE.PutUint32(pkt[8:], a.aID)
		icnLE.PutUint32(pkt[12:], a.aRemote)
		pkt[0x10], pkt[0x11] = 0x81, 0x01
		icnBE.PutUint16(pkt[0x12:], a.txSend)
		a.txSend++
		icnBE.PutUint32(pkt[0x14:], frameBytes)
		copy(pkt[0x18:], a.txRem[:frameBytes])
		a.txRem = a.txRem[frameBytes:]
		if _, err := a.conn.Write(pkt); err != nil {
			return fmt.Errorf("sending mic audio to the rig: %w", err)
		}
	}
	return nil
}