diff --git a/cmd/skyeye/main.go b/cmd/skyeye/main.go index 77730eaa..8095e852 100644 --- a/cmd/skyeye/main.go +++ b/cmd/skyeye/main.go @@ -31,57 +31,59 @@ import ( "github.com/dharmab/skyeye/pkg/coalitions" "github.com/dharmab/skyeye/pkg/encyclopedia" "github.com/dharmab/skyeye/pkg/locations" + "github.com/dharmab/skyeye/pkg/simpleradio" "github.com/dharmab/skyeye/pkg/synthesizer/voices" "github.com/ggerganov/whisper.cpp/bindings/go/pkg/whisper" ) // Used for CLI configuration values. var ( - configFile string - logLevel string - logFormat string - enableTranscriptionLogging bool - acmiFile string - telemetryAddress string - telemetryConnectionTimeout time.Duration - telemetryPassword string - srsAddress string - srsConnectionTimeout time.Duration - srsExternalAWACSModePassword string - srsFrequencies []string - enableGRPC bool - grpcAddress string - grpcAPIKey string - controllerCallsign string - controllerCallsigns []string - coalitionName string - telemetryUpdateInterval time.Duration - recognizerName string - whisperModelPath string - recognizerLockPath string - openAIAPIKey string - voiceName string - useSystemVoice bool - mute bool - voiceSpeed float64 - voiceVolume float64 - voicePauseLength time.Duration - voiceLockPath string - enableAutomaticPicture bool - automaticPictureInterval time.Duration - enableThreatMonitoring bool - threatMonitoringInterval time.Duration - threatMonitoringRequiresSRS bool - mandatoryThreatRadiusNM float64 - threatBRAABearingSpreadDeg float64 - threatBRAARangeSpreadNM float64 - enableTracing bool - discordWebhookID string - discordWebhookToken string - exitAfter time.Duration - enableTerrainDetection bool - locationsFile string - aircraftFile string + configFile string + logLevel string + logFormat string + enableTranscriptionLogging bool + acmiFile string + telemetryAddress string + telemetryConnectionTimeout time.Duration + telemetryPassword string + srsAddress string + srsConnectionTimeout time.Duration + srsExternalAWACSModePassword string + srsFrequencies []string + srsSplitTransmissionGracePeriod time.Duration + enableGRPC bool + grpcAddress string + grpcAPIKey string + controllerCallsign string + controllerCallsigns []string + coalitionName string + telemetryUpdateInterval time.Duration + recognizerName string + whisperModelPath string + recognizerLockPath string + openAIAPIKey string + voiceName string + useSystemVoice bool + mute bool + voiceSpeed float64 + voiceVolume float64 + voicePauseLength time.Duration + voiceLockPath string + enableAutomaticPicture bool + automaticPictureInterval time.Duration + enableThreatMonitoring bool + threatMonitoringInterval time.Duration + threatMonitoringRequiresSRS bool + mandatoryThreatRadiusNM float64 + threatBRAABearingSpreadDeg float64 + threatBRAARangeSpreadNM float64 + enableTracing bool + discordWebhookID string + discordWebhookToken string + exitAfter time.Duration + enableTerrainDetection bool + locationsFile string + aircraftFile string ) const ( @@ -114,6 +116,7 @@ func init() { skyeye.Flags().DurationVar(&srsConnectionTimeout, "srs-connection-timeout", 10*time.Second, "Connection timeout for SRS client") skyeye.Flags().StringVar(&srsExternalAWACSModePassword, "srs-eam-password", "", "SRS external AWACS mode password") skyeye.Flags().StringSliceVar(&srsFrequencies, "srs-frequencies", []string{"251.0AM", "133.0AM", "30.0FM"}, "List of SRS frequencies to use") + skyeye.Flags().DurationVar(&srsSplitTransmissionGracePeriod, "srs-split-transmission-grace-period", simpleradio.DefaultSplitTransmissionGracePeriod, "How long to wait for more audio before treating a transmission as finished. Increase this if long transmissions are being split in two on a lossy network") // DCS-gRPC skyeye.Flags().BoolVar(&enableGRPC, "enable-grpc", false, "Enable DCS-gRPC features") @@ -434,49 +437,50 @@ func run(_ *cobra.Command, _ []string) { customAircraft := loadAircraft() config := conf.Configuration{ - ACMIFile: acmiFile, - TelemetryAddress: telemetryAddress, - TelemetryConnectionTimeout: telemetryConnectionTimeout, - TelemetryClientName: callsign, - TelemetryPassword: telemetryPassword, - SRSAddress: srsAddress, - SRSConnectionTimeout: srsConnectionTimeout, - SRSClientName: fmt.Sprintf("GCI %s [BOT]", callsign), - SRSExternalAWACSModePassword: srsExternalAWACSModePassword, - SRSFrequencies: parsedSRSFrequencies, - EnableTranscriptionLogging: enableTranscriptionLogging, - Callsign: callsign, - Coalition: coalition, - RadarSweepInterval: telemetryUpdateInterval, - Recognizer: conf.Recognizer(recognizerName), - RecognizerLock: recognizerLock, - WhisperModel: whisperModel, - OpenAIAPIKey: openAIAPIKey, - Voice: voice, - UseSystemVoice: useSystemVoice, - VoiceLock: voiceLock, - Mute: mute, - VoiceSpeed: voiceSpeed, - Volume: volume, - VoicePauseLength: voicePauseLength, - EnableAutomaticPicture: enableAutomaticPicture, - PictureBroadcastInterval: automaticPictureInterval, - EnableThreatMonitoring: enableThreatMonitoring, - ThreatMonitoringInterval: threatMonitoringInterval, - ThreatMonitoringRequiresSRS: threatMonitoringRequiresSRS, - MandatoryThreatRadius: unit.Length(mandatoryThreatRadiusNM) * unit.NauticalMile, - ThreatBRAABearingSpread: unit.Angle(threatBRAABearingSpreadDeg) * unit.Degree, - ThreatBRAARangeSpread: unit.Length(threatBRAARangeSpreadNM) * unit.NauticalMile, - EnableTracing: enableTracing, - DiscordWebhookID: discordWebhookID, - DiscorbWebhookToken: discordWebhookToken, - ExitAfter: exitAfter, - EnableGRPC: enableGRPC, - GRPCAddress: grpcAddress, - GRPCAPIKey: grpcAPIKey, - EnableTerrainDetection: enableTerrainDetection, - Locations: locs, - CustomAircraft: customAircraft, + ACMIFile: acmiFile, + TelemetryAddress: telemetryAddress, + TelemetryConnectionTimeout: telemetryConnectionTimeout, + TelemetryClientName: callsign, + TelemetryPassword: telemetryPassword, + SRSAddress: srsAddress, + SRSConnectionTimeout: srsConnectionTimeout, + SRSClientName: fmt.Sprintf("GCI %s [BOT]", callsign), + SRSExternalAWACSModePassword: srsExternalAWACSModePassword, + SRSFrequencies: parsedSRSFrequencies, + SRSSplitTransmissionGracePeriod: srsSplitTransmissionGracePeriod, + EnableTranscriptionLogging: enableTranscriptionLogging, + Callsign: callsign, + Coalition: coalition, + RadarSweepInterval: telemetryUpdateInterval, + Recognizer: conf.Recognizer(recognizerName), + RecognizerLock: recognizerLock, + WhisperModel: whisperModel, + OpenAIAPIKey: openAIAPIKey, + Voice: voice, + UseSystemVoice: useSystemVoice, + VoiceLock: voiceLock, + Mute: mute, + VoiceSpeed: voiceSpeed, + Volume: volume, + VoicePauseLength: voicePauseLength, + EnableAutomaticPicture: enableAutomaticPicture, + PictureBroadcastInterval: automaticPictureInterval, + EnableThreatMonitoring: enableThreatMonitoring, + ThreatMonitoringInterval: threatMonitoringInterval, + ThreatMonitoringRequiresSRS: threatMonitoringRequiresSRS, + MandatoryThreatRadius: unit.Length(mandatoryThreatRadiusNM) * unit.NauticalMile, + ThreatBRAABearingSpread: unit.Angle(threatBRAABearingSpreadDeg) * unit.Degree, + ThreatBRAARangeSpread: unit.Length(threatBRAARangeSpreadNM) * unit.NauticalMile, + EnableTracing: enableTracing, + DiscordWebhookID: discordWebhookID, + DiscorbWebhookToken: discordWebhookToken, + ExitAfter: exitAfter, + EnableGRPC: enableGRPC, + GRPCAddress: grpcAddress, + GRPCAPIKey: grpcAPIKey, + EnableTerrainDetection: enableTerrainDetection, + Locations: locs, + CustomAircraft: customAircraft, } log.Info().Msg("starting application") diff --git a/config.yaml b/config.yaml index ac536f50..9ac135b4 100644 --- a/config.yaml +++ b/config.yaml @@ -92,6 +92,15 @@ # on the aux radio. Meanwhile, the F-16 can only tune 225.000-399.975 on COM1 and # 108.000-151.975 on COM2. #srs-frequencies: 251.0AM,133.0AM,30.0FM +# +# How long the GCI waits for more audio before deciding a transmission has +# finished. SRS sends one packet every 40ms, so the default tolerates a fair +# amount of packet loss before a single transmission is split into two. +# +# Raise this if players report the GCI mishearing or ignoring the end of long +# transmissions on a lossy network. The cost is that the GCI waits slightly +# longer before replying. +#srs-split-transmission-grace-period: 300ms # DCS-gRPC (optional) # Enable DCS-gRPC features (requires https://github.com/DCS-gRPC/rust-server) diff --git a/internal/application/app.go b/internal/application/app.go index 3eaf387c..b07868df 100644 --- a/internal/application/app.go +++ b/internal/application/app.go @@ -128,13 +128,14 @@ func NewApplication(config conf.Configuration) (*Application, error) { Int("modulationID", int(srs.ModulationAM)). Msg("constructing SRS client") srsClient, err := simpleradio.NewClient(srs.ClientConfiguration{ - Address: config.SRSAddress, - ConnectionTimeout: config.SRSConnectionTimeout, - ClientName: config.SRSClientName, - ExternalAWACSModePassword: config.SRSExternalAWACSModePassword, - Coalition: config.Coalition, - Radios: radios, - Mute: config.Mute, + Address: config.SRSAddress, + ConnectionTimeout: config.SRSConnectionTimeout, + ClientName: config.SRSClientName, + ExternalAWACSModePassword: config.SRSExternalAWACSModePassword, + Coalition: config.Coalition, + Radios: radios, + Mute: config.Mute, + SplitTransmissionGracePeriod: config.SRSSplitTransmissionGracePeriod, }) if err != nil { return nil, fmt.Errorf("failed to construct application: %w", err) diff --git a/internal/conf/configuration.go b/internal/conf/configuration.go index 37f3b248..9b7c593b 100644 --- a/internal/conf/configuration.go +++ b/internal/conf/configuration.go @@ -44,6 +44,10 @@ type Configuration struct { SRSExternalAWACSModePassword string // SRSFrequencies that the bot simultaneously receives and transmits on SRSFrequencies []simpleradio.RadioFrequency + // SRSSplitTransmissionGracePeriod is how long to wait for another voice packet before treating + // a transmission as finished. Raising it makes the bot tolerate more packet loss before + // splitting one transmission into two, at the cost of responding slightly later. + SRSSplitTransmissionGracePeriod time.Duration // EnableGRPC controls whether DCS-gRPC features are enabled EnableGRPC bool // GRPCAddress is the network address of the DCS-gRPC server (including port) diff --git a/pkg/simpleradio/client.go b/pkg/simpleradio/client.go index 69134141..9a9bc8aa 100644 --- a/pkg/simpleradio/client.go +++ b/pkg/simpleradio/client.go @@ -80,7 +80,7 @@ func NewClient(config types.ClientConfiguration) (*Client, error) { receivers := make(map[types.Radio]*receiver, len(config.Radios)) for _, radio := range config.Radios { - receivers[radio] = &receiver{} + receivers[radio] = newReceiver(config.SplitTransmissionGracePeriod) } client := &Client{ @@ -208,10 +208,16 @@ func (c *Client) Run(ctx context.Context, wg *sync.WaitGroup) error { return nil } -func (c *Client) getPeerName(guid types.GUID) (string, bool) { +// getPeer looks up a peer's client info by GUID. +func (c *Client) getPeer(guid types.GUID) (types.ClientInfo, bool) { c.clientsLock.RLock() defer c.clientsLock.RUnlock() info, ok := c.clients[guid] + return info, ok +} + +func (c *Client) getPeerName(guid types.GUID) (string, bool) { + info, ok := c.getPeer(guid) if ok { return info.Name, true } diff --git a/pkg/simpleradio/receive.go b/pkg/simpleradio/receive.go index d6d65fb2..9d73d1d8 100644 --- a/pkg/simpleradio/receive.go +++ b/pkg/simpleradio/receive.go @@ -10,19 +10,69 @@ import ( "github.com/rs/zerolog/log" ) +// DefaultSplitTransmissionGracePeriod is used when the client configuration does not set one. +const DefaultSplitTransmissionGracePeriod = 300 * time.Millisecond + +// minRxDuration is the minimum duration of a transmission to be considered for speech recognition. This reduces +// thrashing due to transmissions too short to contain any useful content. +const minRxDuration = 1 * time.Second // 1s is whisper.cpp's minimum duration, it errors for any samples shorter than this. + +// senderTTL is how long a transmitter's packet high-water mark is remembered after its last packet. +// It only needs to outlast network delays by a wide margin. +const senderTTL = 5 * time.Minute + +// sender is the most recent packet seen from one transmitting client. +type sender struct { + // packetNumber is the highest packet number seen from this transmitter. + packetNumber uint64 + // at is when that packet was received. + at time.Time +} + // receiver buffers incoming transmissions on a single radio frequency. +// +// The channel owns the transmission window, not the caller: the window opens when a packet arrives +// on an idle channel and closes when no packet has arrived for splitTransmissionGracePeriod. Each +// receiver's window is independent of every other receiver's, the way a GCI monitoring several +// channels hears each one separately. +// +// Within a window the first transmitter wins and anyone stepping on them is dropped, mirroring +// radio capture effect. That also keeps the buffer single-origin, which the Opus decoder requires. type receiver struct { // lock protects the receiver's state. lock sync.RWMutex // buffer of received voice packets. buffer []voice.Packet - // origin is the GUID of a client we are currently listening to. We can only listen to one client at a time, and whoever started broadcasting first wins. + // origin is the GUID of the client that won the current transmission window. origin types.GUID - // deadline is extended every time another voice packet is received. When we pass the deadline, the transmission is considered over. + // startedAt is when the current transmission window opened. + startedAt time.Time + // deadline is extended every time another voice packet is received. Once it passes, the + // transmission is considered over. deadline time.Time - // packetNumber is the number of the last received voice packet. We only record a packet if its packet number is larger than the last received packet's, and skip any that were dropped or delivered out of order. - // If we were more ambitious we would reassemble the packets and use Opus's forward error correction to recover from lost packets... too bad! - packetNumber uint64 + // splitTransmissionGracePeriod is how long to wait for another packet before treating a + // transmission as finished. + splitTransmissionGracePeriod time.Duration + // senders records the highest packet number seen from each transmitter. It deliberately + // outlives individual transmissions: SRS packet numbers increment for the life of a client, so + // remembering them is what stops a packet delayed past the end of its own transmission from + // opening a new one. + // + // If we were more ambitious we would reassemble the packets and use Opus's forward error + // correction to recover from lost packets... too bad! + senders map[types.GUID]sender +} + +// newReceiver returns a receiver listening on one frequency. A grace period of zero selects +// DefaultSplitTransmissionGracePeriod. +func newReceiver(gracePeriod time.Duration) *receiver { + if gracePeriod <= 0 { + gracePeriod = DefaultSplitTransmissionGracePeriod + } + return &receiver{ + splitTransmissionGracePeriod: gracePeriod, + senders: make(map[types.GUID]sender), + } } // Receive returns a channel that receives transmissions over the radio. Each transmission is F32LE PCM audio data. @@ -30,41 +80,89 @@ func (c *Client) Receive() <-chan Transmission { return c.rxChan } -// receive checks if the given packet is part of a new transmission or matches a transmission in progress. -// If either case is true, the packet is buffered into the receiver. +// receive buffers the given packet if it continues the transmission in progress or starts a new +// one. +// +// Any completed transmission must be taken with completedTransmission before calling this, +// otherwise a new transmission would be appended onto the previous one's buffer. func (r *receiver) receive(packet *voice.Packet) { - // Accept the packet if it is either: - // - the first packet of a new transmission - isNewTransmission := r.origin == "" && r.packetNumber == 0 - // - a newer packet from the same origin - isNewerPacket := packet.PacketID > r.packetNumber - isSameOrigin := r.origin == types.GUID(packet.OriginGUID) - shouldAcceptPacket := isNewTransmission || (isNewerPacket && isSameOrigin) - if !shouldAcceptPacket { + now := time.Now() + origin := types.GUID(packet.OriginGUID) + + r.lock.Lock() + defer r.lock.Unlock() + + // Skip duplicates, packets delivered out of order, and packets delayed past the end of the + // transmission they belong to. + // + // A packet number that goes backwards only means a stale packet if it turns up while that + // transmitter is still being heard. Long afterwards it more likely means the transmitter + // restarted its numbering, and refusing it would silence that client for good. + if s, ok := r.senders[origin]; ok && packet.PacketID <= s.packetNumber { + if now.Sub(s.at) <= r.stragglerWindow() { + return + } + log.Debug(). + Str("origin", string(origin)). + Uint64("packetNumber", packet.PacketID). + Uint64("previousPacketNumber", s.packetNumber). + Msg("transmitter appears to have restarted its packet numbering") + } + + isWindowOpen := len(r.buffer) > 0 && now.Before(r.deadline) + if isWindowOpen && origin != r.origin { + // Someone stepped on the client we are already listening to. Whoever started first wins. return } - if isNewTransmission { - log.Info().Str("origin", string(packet.OriginGUID)).Msg("receiving transmission") + if !isWindowOpen { + log.Info().Str("origin", string(origin)).Msg("receiving transmission") + r.origin = origin + r.startedAt = now } - r.lock.Lock() - defer r.lock.Unlock() + r.senders[origin] = sender{packetNumber: packet.PacketID, at: now} if len(r.buffer) < maxRxPackets { r.buffer = append(r.buffer, *packet) } - r.origin = types.GUID(packet.OriginGUID) - r.deadline = time.Now().Add(maxRxGap) - r.packetNumber = packet.PacketID + r.deadline = now.Add(r.splitTransmissionGracePeriod) +} + +// completedTransmission returns the buffered transmission if its window has closed, the duration of +// audio it contains, and the wall clock span over which it arrived. It returns nil packets if no +// transmission is ready. Taking a transmission closes the window. +// +// The audio duration and the span differ when packets are lost: the audio duration is what +// whisper.cpp will see, while the span is how long the speaker was actually talking. +func (r *receiver) completedTransmission() (packets []voice.Packet, audio, span time.Duration) { + now := time.Now() + + r.lock.Lock() + defer r.lock.Unlock() + + if len(r.buffer) == 0 || !now.After(r.deadline) { + return nil, 0, 0 + } + + packets = r.buffer + audio = time.Duration(len(packets)) * frameLength + lastPacketAt := r.deadline.Add(-r.splitTransmissionGracePeriod) + span = lastPacketAt.Sub(r.startedAt) + frameLength + + r.buffer = nil + r.origin = "" + r.startedAt = time.Time{} + r.deadline = time.Time{} + r.forgetStaleSenders(now) + + return packets, audio, span } // hasTransmission checks if the receiver has a complete transmission buffered. func (r *receiver) hasTransmission() bool { r.lock.RLock() defer r.lock.RUnlock() - hasPackets := len(r.buffer) > 0 - isComplete := time.Now().After(r.deadline) - return hasPackets && isComplete + return len(r.buffer) > 0 && time.Now().After(r.deadline) } // isReceivingTransmission checks if the receiver is currently buffering an in-progress transmission. @@ -74,18 +172,33 @@ func (r *receiver) isReceivingTransmission() bool { return r.deadline.After(time.Now()) } -// reset clears the receiver's buffer. +// deadlineAt returns the time at which the current transmission window closes. It is the zero time +// if no transmission is in progress. +func (r *receiver) deadlineAt() time.Time { + r.lock.RLock() + defer r.lock.RUnlock() + return r.deadline +} + +// reset discards any transmission in progress. It is called when reconnecting to the SRS server, +// where any partial transmission is no longer useful and every transmitter may have a new GUID. func (r *receiver) reset() { r.lock.Lock() defer r.lock.Unlock() - r.buffer = make([]voice.Packet, 0) + r.buffer = nil r.origin = "" + r.startedAt = time.Time{} r.deadline = time.Time{} - r.packetNumber = 0 + clear(r.senders) } -// maxRxGap is a duration after which the receiver will assume the end of a transmission if no packets are received. -const maxRxGap = 300 * time.Millisecond +// stragglerWindow is how long after a transmitter's last packet a lower-numbered packet is still +// treated as one that was delayed rather than one that was renumbered. A packet later than this is +// long past being useful audio anyway: measured against a live server, packets arrive 40ms apart +// with a worst case of 52ms. +func (r *receiver) stragglerWindow() time.Duration { + return 2 * r.splitTransmissionGracePeriod +} // MaxTransmissionDuration is the longest acceptable received transmission // length. Longer transmissions are truncated to this. @@ -94,74 +207,115 @@ const MaxTransmissionDuration = 30 * time.Second // maxRxPackets is the longest acceptable received transmission length in packets. const maxRxPackets = int(MaxTransmissionDuration / frameLength) -// minRxDuration is the minimum duration of a transmission to be considered for speech recognition. This reduces -// thrashing due to transmissions too short to contain any useful content. -const minRxDuration = 1 * time.Second // 1s is whisper.cpp's minimum duration, it errors for any samples shorter than this. +// forgetStaleSenders drops the high-water marks of transmitters we have not heard from in a while, +// bounding the map on a long-running server. The caller must hold the lock. +func (r *receiver) forgetStaleSenders(now time.Time) { + for guid, s := range r.senders { + if now.Sub(s.at) > senderTTL { + delete(r.senders, guid) + } + } +} // receiveVoice listens for incoming UDP voice packets, decodes them into VoicePacket structs, and routes them to the out channel for audio decoding. func (c *Client) receiveVoice(ctx context.Context, in <-chan []byte, out chan<- []voice.Packet) { - // t is a ticker which triggers the check for the end of a transmission. - t := time.NewTicker(frameLength) + // ticker triggers the check for the end of a transmission. + ticker := time.NewTicker(frameLength) for { select { case b := <-in: - packet, err := voice.Decode(b) - if err != nil { - log.Debug().Err(err).Msg("failed to decode voice packet") - continue - } + // Publish any transmission that has already ended before buffering this packet, so + // that a finished transmission never blocks the next one from starting. + c.publishTransmissions(out) + c.handlePacket(b) + case <-ticker.C: + // Handle everything already queued before checking deadlines. Otherwise a loop that + // stalled could declare a transmission over while the packets that would have extended + // it are still waiting to be read. + c.drainPackets(in) + c.publishTransmissions(out) + case <-ctx.Done(): + log.Info().Msg("stopping SRS audio receiver due to context cancellation") + return + } + } +} - logger := log.With().Str("GUID", string(packet.OriginGUID)).Logger() - - if c.secureCoalitionRadios.Load() { - client, ok := c.clients[types.GUID(packet.OriginGUID)] - if !ok { - logger.Warn().Msg("ignoring voice packet from unknown client") - continue - } - if client.Coalition != c.clientInfo.Coalition { - logger.Trace().Msg("ignoring voice packet from different coalition") - continue - } - } +// drainPackets handles every packet currently queued, without blocking. +func (c *Client) drainPackets(in <-chan []byte) { + for { + select { + case b := <-in: + c.handlePacket(b) + default: + return + } + } +} + +// handlePacket decodes a voice packet and buffers it into each receiver tuned to a frequency the +// packet was transmitted on. +func (c *Client) handlePacket(b []byte) { + packet, err := voice.Decode(b) + if err != nil { + log.Debug().Err(err).Msg("failed to decode voice packet") + return + } + + logger := log.With().Str("GUID", string(packet.OriginGUID)).Logger() + + if c.secureCoalitionRadios.Load() { + peer, ok := c.getPeer(types.GUID(packet.OriginGUID)) + if !ok { + logger.Warn().Msg("ignoring voice packet from unknown client") + return + } + if peer.Coalition != c.clientInfo.Coalition { + logger.Trace().Msg("ignoring voice packet from different coalition") + return + } + } - for radio, receiver := range c.receivers { - for _, frequency := range packet.Frequencies { - testRadio := types.Radio{ - Frequency: frequency.Frequency, - Modulation: types.Modulation(frequency.Modulation), - IsEncrypted: frequency.Encryption != 0, - } - if testRadio.IsSameFrequency(radio) { - receiver.receive(packet) - } - } + for radio, receiver := range c.receivers { + for _, frequency := range packet.Frequencies { + testRadio := types.Radio{ + Frequency: frequency.Frequency, + Modulation: types.Modulation(frequency.Modulation), + IsEncrypted: frequency.Encryption != 0, } - case <-t.C: - // Check if everyone has stopped talking. - if len(in) == 0 { - for _, receiver := range c.receivers { - if receiver.hasTransmission() { - duration := time.Duration(len(receiver.buffer)) * frameLength - logger := log.With().Stringer("duration", duration).Logger() - if duration > minRxDuration { - if len(receiver.buffer) >= maxRxPackets { - logger.Warn().Stringer("max", MaxTransmissionDuration).Msg("transmission truncated to maximum duration") - } - logger.Info().Msg("received transmission") - audio := make([]voice.Packet, len(receiver.buffer)) - copy(audio, receiver.buffer) - out <- audio - } else { - logger.Info().Msg("discarding transmission below minimum size") - } - receiver.reset() - } - } + if testRadio.IsSameFrequency(radio) { + receiver.receive(packet) + // A packet can list the same frequency more than once. The per-transmitter packet + // numbering would skip the repeat anyway; stopping here saves the second lookup. + break } - case <-ctx.Done(): - log.Info().Msg("stopping SRS audio receiver due to context cancellation") - return + } + } +} + +// publishTransmissions publishes each transmission whose window has closed. Every receiver's window +// is evaluated on its own, so a busy channel cannot hold another channel's transmission open. +func (c *Client) publishTransmissions(out chan<- []voice.Packet) { + for _, receiver := range c.receivers { + packets, audio, span := receiver.completedTransmission() + if packets == nil { + continue + } + logger := log.With(). + Stringer("duration", audio). + Stringer("span", span). + Int("packets", len(packets)). + Logger() + if audio >= minRxDuration { + if len(packets) >= maxRxPackets { + logger.Warn().Stringer("max", MaxTransmissionDuration).Msg("transmission truncated to maximum duration") + } + logger.Info().Msg("received transmission") + out <- packets + } else { + // A span much longer than the duration means packets were lost, rather than the + // speaker simply saying something short. + logger.Info().Msg("discarding transmission below minimum size") } } } diff --git a/pkg/simpleradio/receive_test.go b/pkg/simpleradio/receive_test.go new file mode 100644 index 00000000..1fd1d6d0 --- /dev/null +++ b/pkg/simpleradio/receive_test.go @@ -0,0 +1,443 @@ +package simpleradio + +import ( + "context" + "strings" + "testing" + "testing/synctest" + "time" + + "github.com/dharmab/skyeye/pkg/coalitions" + "github.com/dharmab/skyeye/pkg/simpleradio/types" + "github.com/dharmab/skyeye/pkg/simpleradio/voice" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testGracePeriod = 300 * time.Millisecond + +// testGUID pads a readable name out to the length of a real SRS GUID. +func testGUID(name string) types.GUID { + return types.GUID(name + strings.Repeat("0", types.GUIDLength-len(name))) +} + +func testFrequency(radio types.Radio) voice.Frequency { + return voice.Frequency{Frequency: radio.Frequency, Modulation: byte(radio.Modulation)} +} + +// testPacket builds a voice packet carrying a stand-in for an Opus frame. The receiver never looks +// at the audio, only at the identity and frequency metadata. +func testPacket(origin types.GUID, packetID uint64, frequencies ...voice.Frequency) *voice.Packet { + if len(frequencies) == 0 { + frequencies = []voice.Frequency{testFrequency(testRadio)} + } + packet := voice.NewPacket( + []byte{0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8}, + frequencies, + 100000002, + packetID, + 0, + []byte(origin), + []byte(origin), + ) + return &packet +} + +// transmit feeds a whole transmission into the receiver, one 40ms frame at a time. +func transmit(r *receiver, origin types.GUID, firstPacketID uint64, frames int) { + for i := range frames { + r.receive(testPacket(origin, firstPacketID+uint64(i))) + time.Sleep(frameLength) + } +} + +func TestReceiverOpensWindowOnFirstPacket(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + r := newReceiver(testGracePeriod) + assert.False(t, r.isReceivingTransmission()) + assert.True(t, r.deadlineAt().IsZero()) + + r.receive(testPacket(testGUID("alpha"), 1)) + + assert.True(t, r.isReceivingTransmission(), "window should be open") + assert.False(t, r.hasTransmission(), "transmission is not finished yet") + assert.Equal(t, time.Now().Add(testGracePeriod), r.deadlineAt()) + }) +} + +func TestReceiverAccumulatesTransmission(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + r := newReceiver(testGracePeriod) + transmit(r, testGUID("alpha"), 1, 25) + + time.Sleep(testGracePeriod) + packets, audio, span := r.completedTransmission() + + require.Len(t, packets, 25) + assert.Equal(t, time.Second, audio) + assert.Equal(t, time.Second, span) + assert.False(t, r.isReceivingTransmission(), "window should be closed after taking it") + }) +} + +// The grace period is the whole boundary between one transmission and the next, so pin both sides +// of it exactly. +func TestReceiverGracePeriodBoundary(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + r := newReceiver(testGracePeriod) + r.receive(testPacket(testGUID("alpha"), 1)) + + time.Sleep(testGracePeriod - time.Nanosecond) + assert.False(t, r.hasTransmission(), "still within the grace period") + + time.Sleep(2 * time.Nanosecond) + assert.True(t, r.hasTransmission(), "grace period has elapsed") + }) +} + +func TestReceiverSkipsDuplicateAndOutOfOrderPackets(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + alpha := testGUID("alpha") + r := newReceiver(testGracePeriod) + + r.receive(testPacket(alpha, 5)) + r.receive(testPacket(alpha, 5)) // duplicate + r.receive(testPacket(alpha, 4)) // delivered out of order + r.receive(testPacket(alpha, 6)) + + time.Sleep(testGracePeriod + time.Nanosecond) + packets, _, _ := r.completedTransmission() + + require.Len(t, packets, 2) + assert.Equal(t, uint64(5), packets[0].PacketID) + assert.Equal(t, uint64(6), packets[1].PacketID) + }) +} + +// Radio capture effect: whoever gets the channel first keeps it for the whole transmission. +func TestReceiverFirstTransmitterWinsTheWindow(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + alpha, bravo := testGUID("alpha"), testGUID("bravo") + r := newReceiver(testGracePeriod) + + r.receive(testPacket(alpha, 1)) + time.Sleep(frameLength) + r.receive(testPacket(bravo, 100)) // steps on alpha + time.Sleep(frameLength) + r.receive(testPacket(alpha, 2)) + + time.Sleep(testGracePeriod + time.Nanosecond) + packets, _, _ := r.completedTransmission() + + require.Len(t, packets, 2) + for _, packet := range packets { + assert.Equal(t, alpha, types.GUID(packet.OriginGUID)) + } + }) +} + +// Regression: a finished but not yet published transmission used to lock out the next caller, +// because "in progress" meant "state has not been reset" rather than "the deadline has not passed". +func TestReceiverNewTransmitterOpensWindowAfterGracePeriod(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + alpha, bravo := testGUID("alpha"), testGUID("bravo") + r := newReceiver(testGracePeriod) + + transmit(r, alpha, 1, 3) + time.Sleep(testGracePeriod + time.Nanosecond) + first, _, _ := r.completedTransmission() + require.Len(t, first, 3) + + // bravo has a lower packet number than alpha ever reached, which must not matter. + r.receive(testPacket(bravo, 1)) + assert.True(t, r.isReceivingTransmission(), "bravo should own a fresh window") + + time.Sleep(testGracePeriod + time.Nanosecond) + second, _, _ := r.completedTransmission() + require.Len(t, second, 1) + assert.Equal(t, bravo, types.GUID(second[0].OriginGUID)) + }) +} + +// Regression: SRS packet numbers climb for the life of a client, so a packet delayed past the end +// of its own transmission used to be mistaken for the start of a new one. It would buffer a stale +// frame, arm a fresh deadline, and lock out the next real caller for that whole window. +func TestReceiverIgnoresStragglerFromPublishedTransmission(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + alpha, bravo := testGUID("alpha"), testGUID("bravo") + r := newReceiver(testGracePeriod) + + transmit(r, alpha, 1, 5) + time.Sleep(testGracePeriod + time.Nanosecond) + published, _, _ := r.completedTransmission() + require.Len(t, published, 5) + + // A packet from the middle of the transmission we just published finally turns up. + r.receive(testPacket(alpha, 3)) + assert.False(t, r.isReceivingTransmission(), "straggler must not open a window") + + // The channel is still free for the next caller. + r.receive(testPacket(bravo, 1)) + assert.True(t, r.isReceivingTransmission()) + assert.Equal(t, bravo, r.origin) + }) +} + +// Remembering packet numbers across transmissions must not silence a client whose numbering +// restarts, which would be far worse than the stale packet the memory exists to reject. +func TestReceiverAcceptsRenumberedTransmitter(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + alpha := testGUID("alpha") + r := newReceiver(testGracePeriod) + + transmit(r, alpha, 5000, 5) + time.Sleep(testGracePeriod + time.Nanosecond) + published, _, _ := r.completedTransmission() + require.Len(t, published, 5) + + // Well past the point where a delayed packet could still be in flight, the same client + // starts numbering from scratch. + time.Sleep(time.Minute) + r.receive(testPacket(alpha, 1)) + + assert.True(t, r.isReceivingTransmission(), "a restarted client must still be heard") + assert.Equal(t, alpha, r.origin) + }) +} + +// Lost packets shorten the audio without shortening the time the speaker was talking. The audio +// duration is what whisper.cpp sees and gates the minimum length; the span is what makes a +// discarded transmission diagnosable. +func TestReceiverReportsAudioAndSpanSeparately(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + alpha := testGUID("alpha") + r := newReceiver(testGracePeriod) + + // 25 frames' worth of talking, but one packet in every five is lost in transit. The first + // and last frames arrive, so the span still covers the whole utterance. + for i := range 25 { + if i%5 != 2 { + r.receive(testPacket(alpha, uint64(i+1))) + } + time.Sleep(frameLength) + } + + time.Sleep(testGracePeriod) + packets, audio, span := r.completedTransmission() + + require.Len(t, packets, 20) + assert.Equal(t, 800*time.Millisecond, audio, "only the frames that arrived are decodable") + assert.Equal(t, time.Second, span, "the speaker still talked for a second") + assert.Less(t, audio, minRxDuration, "which is why a real transmission can be discarded") + }) +} + +func TestReceiverResetClearsState(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + alpha := testGUID("alpha") + r := newReceiver(testGracePeriod) + transmit(r, alpha, 5, 3) + + r.reset() + + assert.False(t, r.isReceivingTransmission()) + assert.False(t, r.hasTransmission()) + assert.Empty(t, r.buffer) + assert.Empty(t, r.senders, "a reconnect may hand every transmitter a new GUID") + + // After a reset the receiver accepts a packet it would previously have skipped. + r.receive(testPacket(alpha, 1)) + assert.True(t, r.isReceivingTransmission()) + }) +} + +func TestNewReceiverDefaultsGracePeriod(t *testing.T) { + t.Parallel() + assert.Equal(t, DefaultSplitTransmissionGracePeriod, newReceiver(0).splitTransmissionGracePeriod) + assert.Equal(t, time.Second, newReceiver(time.Second).splitTransmissionGracePeriod) +} + +// newTestClient builds a Client without touching the network. NewClient dials TCP and UDP, which +// would block outside the synctest bubble. +func newTestClient(gracePeriod time.Duration, radios ...types.Radio) *Client { + receivers := make(map[types.Radio]*receiver, len(radios)) + for _, radio := range radios { + receivers[radio] = newReceiver(gracePeriod) + } + return &Client{ + clientInfo: types.ClientInfo{Coalition: coalitions.Blue}, + clients: make(map[types.GUID]types.ClientInfo), + receivers: receivers, + } +} + +// Regression: publishing a finished transmission used to be gated on the shared inbound queue being +// empty, which let traffic on any frequency - including ones no radio is tuned to - hold another +// channel's transmission indefinitely. +func TestPublishTransmissionsIgnoresInboundQueueDepth(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + c := newTestClient(testGracePeriod, testRadio) + out := make(chan []voice.Packet, 4) + + for i := range 25 { + c.handlePacket(testPacket(testGUID("alpha"), uint64(i+1)).Encode()) + time.Sleep(frameLength) + } + time.Sleep(testGracePeriod + time.Nanosecond) + + // A backlog on the inbound queue must not influence a receiver's own deadline. + in := make(chan []byte, 64) + for i := range 32 { + in <- testPacket(testGUID("noise"), uint64(i+1), voice.Frequency{Frequency: 500_000_000}).Encode() + } + require.NotEmpty(t, in) + + c.publishTransmissions(out) + + require.Len(t, out, 1) + assert.Len(t, <-out, 25) + }) +} + +func TestReceiveVoicePublishesTransmission(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + c := newTestClient(testGracePeriod, testRadio) + in := make(chan []byte, 64) + out := make(chan []voice.Packet, 4) + go c.receiveVoice(ctx, in, out) + + for i := range 25 { + in <- testPacket(testGUID("alpha"), uint64(i+1)).Encode() + time.Sleep(frameLength) + } + time.Sleep(testGracePeriod + frameLength) + synctest.Wait() + + require.Len(t, out, 1) + assert.Len(t, <-out, 25) + }) +} + +func TestReceiveVoiceDiscardsTransmissionBelowMinimum(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + c := newTestClient(testGracePeriod, testRadio) + in := make(chan []byte, 64) + out := make(chan []voice.Packet, 4) + go c.receiveVoice(ctx, in, out) + + for i := range 10 { + in <- testPacket(testGUID("alpha"), uint64(i+1)).Encode() + time.Sleep(frameLength) + } + time.Sleep(testGracePeriod + frameLength) + synctest.Wait() + + assert.Empty(t, out, "400ms of audio is below whisper.cpp's one second minimum") + }) +} + +// Each channel is monitored on its own, the way a GCI listening to several frequencies hears each +// one separately. Traffic on one must not hold another's transmission open. +func TestReceiveVoiceChannelsAreIndependent(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + c := newTestClient(testGracePeriod, testRadio, testOtherRadio) + in := make(chan []byte, 128) + out := make(chan []voice.Packet, 4) + go c.receiveVoice(ctx, in, out) + + // alpha finishes talking on one channel while bravo keeps talking on the other. + for i := range 25 { + in <- testPacket(testGUID("alpha"), uint64(i+1)).Encode() + in <- testPacket(testGUID("bravo"), uint64(i+1), testFrequency(testOtherRadio)).Encode() + time.Sleep(frameLength) + } + for i := range 25 { + in <- testPacket(testGUID("bravo"), uint64(i+26), testFrequency(testOtherRadio)).Encode() + time.Sleep(frameLength) + } + synctest.Wait() + + require.Len(t, out, 1, "alpha's transmission should publish while bravo is still talking") + published := <-out + require.Len(t, published, 25) + assert.Equal(t, testGUID("alpha"), types.GUID(published[0].OriginGUID)) + }) +} + +// Regression: a packet listing the same frequency twice used to buffer its first frame twice, +// putting a 40ms stutter at the front of every transmission. +func TestReceiveVoiceBuffersRepeatedFrequencyOnce(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + c := newTestClient(testGracePeriod, testRadio) + out := make(chan []voice.Packet, 4) + + duplicated := []voice.Frequency{testFrequency(testRadio), testFrequency(testRadio)} + for i := range 25 { + c.handlePacket(testPacket(testGUID("alpha"), uint64(i+1), duplicated...).Encode()) + time.Sleep(frameLength) + } + time.Sleep(testGracePeriod + time.Nanosecond) + c.publishTransmissions(out) + + require.Len(t, out, 1) + assert.Len(t, <-out, 25) + }) +} + +func TestReceiveVoiceHonoursConfiguredGracePeriod(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + gracePeriod := time.Second + c := newTestClient(gracePeriod, testRadio) + in := make(chan []byte, 64) + out := make(chan []voice.Packet, 4) + go c.receiveVoice(ctx, in, out) + + for i := range 25 { + in <- testPacket(testGUID("alpha"), uint64(i+1)).Encode() + time.Sleep(frameLength) + } + + // A gap that would have ended the transmission under the default grace period. + time.Sleep(DefaultSplitTransmissionGracePeriod + frameLength) + synctest.Wait() + require.Empty(t, out, "the configured grace period has not elapsed yet") + + for i := range 25 { + in <- testPacket(testGUID("alpha"), uint64(i+26)).Encode() + time.Sleep(frameLength) + } + time.Sleep(gracePeriod + frameLength) + synctest.Wait() + + require.Len(t, out, 1) + assert.Len(t, <-out, 50, "both halves belong to one transmission") + }) +} diff --git a/pkg/simpleradio/sync_test.go b/pkg/simpleradio/sync_test.go index 47e0e36a..dd0c81ad 100644 --- a/pkg/simpleradio/sync_test.go +++ b/pkg/simpleradio/sync_test.go @@ -131,3 +131,25 @@ func TestRemoveClient(t *testing.T) { // Removing a peer that was never tracked is harmless. c.removeClient(testPeer("peer000000000000000009", "Ghost", coalitions.Blue, testRadio)) } + +func TestGetPeer(t *testing.T) { + t.Parallel() + c := newSyncTestClient() + peer := testPeer("peer000000000000000001", "Eagle 1", coalitions.Blue, testRadio) + c.syncClient(peer) + + info, ok := c.getPeer(peer.GUID) + require.True(t, ok) + assert.Equal(t, "Eagle 1", info.Name) + + name, ok := c.getPeerName(peer.GUID) + require.True(t, ok) + assert.Equal(t, "Eagle 1", name) + + _, ok = c.getPeer("unknown00000000000000") + assert.False(t, ok) + + name, ok = c.getPeerName("unknown00000000000000") + assert.False(t, ok) + assert.Empty(t, name) +} diff --git a/pkg/simpleradio/transmit.go b/pkg/simpleradio/transmit.go index 9ea86635..dc122769 100644 --- a/pkg/simpleradio/transmit.go +++ b/pkg/simpleradio/transmit.go @@ -47,8 +47,8 @@ func (c *Client) waitForClearChannel() { for _, receiver := range c.receivers { if receiver.isReceivingTransmission() { isReceiving = true - if receiver.deadline.After(deadline) { - deadline = receiver.deadline + if rxDeadline := receiver.deadlineAt(); rxDeadline.After(deadline) { + deadline = rxDeadline } } } diff --git a/pkg/simpleradio/transmit_test.go b/pkg/simpleradio/transmit_test.go new file mode 100644 index 00000000..b7ec3bfa --- /dev/null +++ b/pkg/simpleradio/transmit_test.go @@ -0,0 +1,78 @@ +package simpleradio + +import ( + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/assert" +) + +// The bot waits for an incoming transmission to finish before talking, so it does not step on a +// player mid-sentence. +func TestWaitForClearChannel(t *testing.T) { + t.Parallel() + + t.Run("returns immediately when nobody is talking", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + c := newTestClient(testGracePeriod, testRadio, testOtherRadio) + + start := time.Now() + c.waitForClearChannel() + + assert.Equal(t, start, time.Now(), "no reason to wait") + }) + }) + + t.Run("waits past the deadline of an in-progress transmission", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + c := newTestClient(testGracePeriod, testRadio) + c.receivers[testRadio].receive(testPacket(testGUID("alpha"), 1)) + + start := time.Now() + c.waitForClearChannel() + + // The grace period must elapse before the channel is clear, plus the courtesy pause. + assert.GreaterOrEqual(t, time.Since(start), testGracePeriod) + }) + }) + + t.Run("waits for the busiest channel", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + c := newTestClient(testGracePeriod, testRadio, testOtherRadio) + + c.receivers[testRadio].receive(testPacket(testGUID("alpha"), 1)) + time.Sleep(200 * time.Millisecond) + c.receivers[testOtherRadio].receive( + testPacket(testGUID("bravo"), 1, testFrequency(testOtherRadio)), + ) + latest := c.receivers[testOtherRadio].deadlineAt() + + c.waitForClearChannel() + + assert.False(t, time.Now().Before(latest), "must outlast every channel's deadline") + for _, receiver := range c.receivers { + assert.False(t, receiver.isReceivingTransmission()) + } + }) + }) +} + +func TestReceiverDeadlineAt(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + r := newReceiver(testGracePeriod) + assert.True(t, r.deadlineAt().IsZero(), "no transmission, no deadline") + + r.receive(testPacket(testGUID("alpha"), 1)) + assert.Equal(t, time.Now().Add(testGracePeriod), r.deadlineAt()) + + time.Sleep(testGracePeriod + time.Nanosecond) + packets, _, _ := r.completedTransmission() + assert.Len(t, packets, 1) + assert.True(t, r.deadlineAt().IsZero(), "taking the transmission clears the deadline") + }) +} diff --git a/pkg/simpleradio/types/configuration.go b/pkg/simpleradio/types/configuration.go index 6399b52d..6fed35a5 100644 --- a/pkg/simpleradio/types/configuration.go +++ b/pkg/simpleradio/types/configuration.go @@ -26,4 +26,7 @@ type ClientConfiguration struct { AllowRecording bool // Mute is true if the client should not transmit. Mute bool + // SplitTransmissionGracePeriod is how long the client waits for another voice packet before + // treating a transmission as finished. Zero selects a sensible default. + SplitTransmissionGracePeriod time.Duration }