funknetz/station/station.go

// Package station is the headset station. It loads the config, keeps one
// reflector session, and exposes the state the window shows.
package station

import (
	"context"
	"net"
	"os"
	"sync"
	"time"

	"gitea.neitzel.de/konrad/funknetz/internal/opus"
)

// Options selects the config file and the clock used for heartbeats and retries.
type Options struct {
	ConfigPath string
	Clock      Clock
}

// State is what the window shows.
type State struct {
	Connection   string `json:"connection"`
	Detail       string `json:"detail"`
	Callsign     string `json:"callsign"`
	TalkGroup    int    `json:"talkGroup"`
	Protocol     string `json:"protocol"`
	Talker       string `json:"talker"`
	Transmitting bool   `json:"transmitting"`
	NeedsConfig  bool   `json:"needsConfig"`
	ConfigOpen   bool   `json:"configOpen"`
}

// Station is the seam tests and the window use.
type Station struct {
	cfg         loaded
	clock       Clock
	path        string
	loaded      bool
	configOpen  bool
	configErr   string
	openBlocked string
	form        ConfigForm
	epoch       uint64
	kick        bool
	conn        net.Conn
	cancelDial  context.CancelFunc
	apply       chan struct{}

	mu           sync.Mutex
	state        State
	wake         chan struct{}
	buttonHeld   bool
	spaceHeld    bool
	fatal        string
	retryReason  string
	protocol     string
	remoteTalker string
	roomBusy     bool
	audioHeard   bool
	lastAudio    time.Time
	flushWaiting bool
	flushAt      time.Time
	rxSeq        uint16
	secure       bool
	udpWait      bool
	ivPrefix     []byte
	udpKey       []byte
	rxCounter    uint32
	txCounter    uint32
	mic          []int16
	play         []int16
	clientID     uint16
	seq          uint16
	udp          *net.UDPConn
	udpPeer      *net.UDPAddr
	lastUDPSend  time.Time
	enc          *opus.Encoder
	dec          *opus.Decoder

	logFile   *os.File
	logPath   string
	logNotice string
	up        bool
	outage    bool
	sending   bool
	sendStart time.Time
	talkCall  string
	talkStart time.Time
}

// New loads the config at opt.ConfigPath. A missing or invalid file does not
// stop the station: the Config window opens and the file is left uncreated
// until Save. New returns an error when the path cannot be prepared, such as
// an unknown home directory.
func New(opt Options) (*Station, error) {
	clock := opt.Clock
	if clock == nil {
		clock = wallClock{}
	}
	out, err := loadConfig(opt.ConfigPath)
	if err != nil {
		return nil, err
	}
	detail := "Idle"
	if out.detail != "" {
		detail = out.detail
	}
	call, tg := out.cfg.callsign, out.cfg.talkGroup
	if !out.loaded && out.cfg.callsign == "" {
		call = out.form.Callsign
		tg = out.form.TalkGroup
	}
	errText := ""
	if !out.loaded {
		errText = out.detail
	}
	s := &Station{
		cfg:        out.cfg,
		clock:      clock,
		path:       opt.ConfigPath,
		loaded:     out.loaded,
		configOpen: !out.loaded,
		configErr:  errText,
		form:       out.form,
		wake:       make(chan struct{}, 1),
		apply:      make(chan struct{}, 1),
		logPath:    out.cfg.logPath,
		logNotice:  out.cfg.logNotice,
		fatal:      out.cfg.certFatal,
		state: State{
			Connection:  "Disconnected",
			Detail:      detail,
			Callsign:    call,
			TalkGroup:   tg,
			NeedsConfig: !out.loaded,
			ConfigOpen:  !out.loaded,
		},
	}
	if out.loaded && out.cfg.enableLogging {
		s.openLog()
	}
	return s, nil
}

// State returns the latest window state.
func (s *Station) State() State {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.state
}

// Config returns the Config window's form and whether it is open.
func (s *Station) Config() ConfigState {
	s.mu.Lock()
	defer s.mu.Unlock()
	return ConfigState{
		Open:        s.configOpen,
		NeedsConfig: !s.loaded,
		Error:       s.configErr,
		Form:        s.form,
	}
}

// Wait blocks until pred is true or ctx ends.
func (s *Station) Wait(ctx context.Context, pred func(State) bool) (State, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for !pred(s.state) {
		s.mu.Unlock()
		select {
		case <-ctx.Done():
			s.mu.Lock()
			return State{}, ctx.Err()
		case <-s.wake:
			s.mu.Lock()
		}
	}
	return s.state, nil
}

// HoldButton reports that the on-screen button is held.
func (s *Station) HoldButton() { s.setButton(true) }

// ReleaseButton reports that the on-screen button was released.
func (s *Station) ReleaseButton() { s.setButton(false) }

// HoldSpace reports that Space is held.
func (s *Station) HoldSpace() { s.setSpace(true) }

// ReleaseSpace reports that Space was released.
func (s *Station) ReleaseSpace() { s.setSpace(false) }

func (s *Station) setButton(held bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	was := s.transmitting()
	s.buttonHeld = held
	s.afterHold(was)
}

func (s *Station) setSpace(held bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	was := s.transmitting()
	s.spaceHeld = held
	s.afterHold(was)
}

func (s *Station) afterHold(was bool) {
	if was && !s.transmitting() {
		s.endTransmit()
	}
	if !s.buttonHeld && !s.spaceHeld {
		s.openBlocked = ""
	}
	s.publish()
}

// WriteMicrophone accepts 16 kHz mono samples. Samples are sent only while
// this station is allowed to talk; otherwise they are discarded.
func (s *Station) WriteMicrophone(pcm []int16) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.transmitting() || s.enc == nil || s.udp == nil {
		s.mic = nil
		return
	}
	s.mic = append(s.mic, pcm...)
	for len(s.mic) >= opus.FrameSamples {
		frame := append([]int16(nil), s.mic[:opus.FrameSamples]...)
		s.mic = append([]int16(nil), s.mic[opus.FrameSamples:]...)
		pkt, err := s.enc.Encode(frame)
		if err != nil {
			continue
		}
		// A lost datagram ends the session when the receive timer expires.
		_ = s.sendUDP(udpAudio, appendBlob(nil, pkt))
	}
}

// TakePlayback returns PCM queued for the speaker since the last take.
func (s *Station) TakePlayback() []int16 {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.transmitting() {
		s.play = nil
		return nil
	}
	out := s.play
	s.play = nil
	return out
}

// WaitPlayback waits until at least n samples are queued, then returns those samples.
func (s *Station) WaitPlayback(ctx context.Context, n int) ([]int16, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for len(s.play) < n {
		s.mu.Unlock()
		select {
		case <-ctx.Done():
			s.mu.Lock()
			return nil, ctx.Err()
		case <-s.wake:
			s.mu.Lock()
		}
	}
	out := append([]int16(nil), s.play[:n]...)
	s.play = append([]int16(nil), s.play[n:]...)
	return out, nil
}