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 (
	"net"
	"sync"
	"sync/atomic"
	"time"
)

// icaAudioOffset is where the PCM payload begins inside an audio data packet
// (wfview audio_packet: 16-byte common header + ident@0x10 + datalen@0x12 +
// sendseq@0x14 → audio@0x16). Isolated as a const so a capture-confirmed change
// is a one-line edit.
const icaAudioOffset = 0x16

// 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)
	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,
		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()
	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
				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)])
				}
				if 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(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.
func (a *icomAudio) trackRxSeq(seq uint16) {
	if !a.rxHaveSeq {
		a.rxHaveSeq = true
		a.rxLastSeq = seq
		return
	}
	switch d := int16(seq - a.rxLastSeq); {
	case d == 0:
	case d < 0:
		delete(a.rxMissing, seq)
	case d == 1:
		a.rxLastSeq = seq
	case int(d) <= icnMaxMissing:
		for f := a.rxLastSeq + 1; f != seq; f++ {
			a.rxMissing[f] = 0
		}
		a.rxLastSeq = seq
	default:
		a.rxMissing = make(map[uint16]int)
		a.rxLastSeq = seq
	}
}

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