diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5e1078d --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +build/bin/ +frontend/wailsjs/ diff --git a/app.go b/app.go new file mode 100644 index 0000000..9a51c74 --- /dev/null +++ b/app.go @@ -0,0 +1,64 @@ +package main + +import ( + "context" + "fmt" + "os" + "path/filepath" + + "gitea.neitzel.de/konrad/funknetz/audio" + "gitea.neitzel.de/konrad/funknetz/station" +) + +// App is the window's view of the station. +type App struct { + cancel context.CancelFunc + station *station.Station +} + +func (a *App) startup(ctx context.Context) { + runCtx, cancel := context.WithCancel(ctx) + a.cancel = cancel + if err := audio.Start(runCtx, a.station.WriteMicrophone, a.station.TakePlayback); err != nil { + a.station.Fail("could not open the audio device") + return + } + go func() { _ = a.station.Run(runCtx) }() +} + +func (a *App) shutdown(context.Context) { + if a.cancel != nil { + a.cancel() + } +} + +// State returns the station state the page renders. +func (a *App) State() station.State { return a.station.State() } + +// HoldButton reports that the on-screen button is held. +func (a *App) HoldButton() { a.station.HoldButton() } + +// ReleaseButton reports that the on-screen button was released. +func (a *App) ReleaseButton() { a.station.ReleaseButton() } + +// HoldSpace reports that Space is held. +func (a *App) HoldSpace() { a.station.HoldSpace() } + +// ReleaseSpace reports that Space was released. +func (a *App) ReleaseSpace() { a.station.ReleaseSpace() } + +func configPath() (string, error) { + home, err := os.UserHomeDir() + if err != nil { + return "", err + } + return filepath.Join(home, ".funknetz.json"), nil +} + +func openStation() (*station.Station, error) { + path, err := configPath() + if err != nil { + return nil, fmt.Errorf("could not be read") + } + return station.New(station.Options{ConfigPath: path}) +} diff --git a/audio/device.go b/audio/device.go new file mode 100644 index 0000000..0345812 --- /dev/null +++ b/audio/device.go @@ -0,0 +1,164 @@ +// Package audio talks to the default input and output devices and presents +// 16 kHz mono PCM to the station. +package audio + +import ( + "context" + "encoding/binary" + "math" + + "github.com/gen2brain/malgo" +) + +const stationRate = 16000 + +// Start opens the default input and output and converts them to and from +// 16 kHz mono. It returns when the devices are running. They close when ctx ends. +func Start(ctx context.Context, capture func([]int16), playback func() []int16) error { + mctx, err := malgo.InitContext(nil, malgo.ContextConfig{}, nil) + if err != nil { + return err + } + cfg := malgo.DefaultDeviceConfig(malgo.Duplex) + cfg.SampleRate = stationRate + cfg.Capture.Format = malgo.FormatS16 + cfg.Capture.Channels = 1 + cfg.Playback.Format = malgo.FormatS16 + cfg.Playback.Channels = 1 + + var playRes, capRes converter + var playBuf []int16 + callbacks := malgo.DeviceCallbacks{ + Data: func(output, input []byte, framecount uint32) { + if len(input) > 0 { + channels := 1 + if framecount > 0 { + channels = len(input) / int(framecount) / 2 + } + if channels < 1 { + channels = 1 + } + pcm := capRes.Convert(downmix(bytesToS16(input), channels)) + if len(pcm) > 0 { + capture(pcm) + } + } + if len(output) == 0 { + return + } + channels := 1 + if framecount > 0 { + channels = len(output) / int(framecount) / 2 + } + if channels < 1 { + channels = 1 + } + playBuf = append(playBuf, playRes.Convert(playback())...) + need := int(framecount) + var frame []int16 + if len(playBuf) >= need { + frame = playBuf[:need] + rest := append([]int16(nil), playBuf[need:]...) + playBuf = rest + } else { + frame = playBuf + playBuf = nil + } + writeS16(output, frame, channels) + }, + } + dev, err := malgo.InitDevice(mctx.Context, cfg, callbacks) + if err != nil { + mctx.Free() + return err + } + rate := int(dev.SampleRate()) + if rate == 0 { + rate = stationRate + } + capRes = converter{from: rate, to: stationRate} + playRes = converter{from: stationRate, to: rate} + if err := dev.Start(); err != nil { + dev.Uninit() + mctx.Free() + return err + } + go func() { + <-ctx.Done() + dev.Stop() + dev.Uninit() + mctx.Free() + }() + return nil +} + +func bytesToS16(b []byte) []int16 { + n := len(b) / 2 + out := make([]int16, n) + for i := range out { + out[i] = int16(binary.NativeEndian.Uint16(b[i*2:])) + } + return out +} + +func writeS16(dst []byte, samples []int16, channels int) { + if channels < 1 { + return + } + frames := len(dst) / (channels * 2) + for i := 0; i < frames; i++ { + var s int16 + if i < len(samples) { + s = samples[i] + } + for ch := 0; ch < channels; ch++ { + off := (i*channels + ch) * 2 + binary.NativeEndian.PutUint16(dst[off:], uint16(s)) + } + } +} + +func downmix(samples []int16, channels int) []int16 { + if channels <= 1 { + return samples + } + n := len(samples) / channels + out := make([]int16, n) + for i := range out { + var sum int + for ch := 0; ch < channels; ch++ { + sum += int(samples[i*channels+ch]) + } + out[i] = int16(sum / channels) + } + return out +} + +type converter struct { + from int + to int + pos float64 + buf []int16 +} + +func (c *converter) Convert(in []int16) []int16 { + if c.from == 0 || c.to == 0 || c.from == c.to { + return in + } + c.buf = append(c.buf, in...) + step := float64(c.from) / float64(c.to) + var out []int16 + for c.pos+1 < float64(len(c.buf)) { + i := int(c.pos) + frac := c.pos - float64(i) + sample := (1-frac)*float64(c.buf[i]) + frac*float64(c.buf[i+1]) + out = append(out, int16(math.Round(sample))) + c.pos += step + } + drop := int(c.pos) + if drop > 0 && drop <= len(c.buf) { + c.buf = append([]int16(nil), c.buf[drop:]...) + c.pos -= float64(drop) + } + return out +} diff --git a/build/appicon.png b/build/appicon.png new file mode 100644 index 0000000..ecf4960 Binary files /dev/null and b/build/appicon.png differ diff --git a/build/darwin/Info.plist b/build/darwin/Info.plist new file mode 100644 index 0000000..d17a747 --- /dev/null +++ b/build/darwin/Info.plist @@ -0,0 +1,63 @@ + + + + CFBundlePackageType + APPL + CFBundleName + {{.Info.ProductName}} + CFBundleExecutable + {{.OutputFilename}} + CFBundleIdentifier + com.wails.{{.Name}} + CFBundleVersion + {{.Info.ProductVersion}} + CFBundleGetInfoString + {{.Info.Comments}} + CFBundleShortVersionString + {{.Info.ProductVersion}} + CFBundleIconFile + iconfile + LSMinimumSystemVersion + 10.13.0 + NSHighResolutionCapable + true + NSHumanReadableCopyright + {{.Info.Copyright}} + {{if .Info.FileAssociations}} + CFBundleDocumentTypes + + {{range .Info.FileAssociations}} + + CFBundleTypeExtensions + + {{.Ext}} + + CFBundleTypeName + {{.Name}} + CFBundleTypeRole + {{.Role}} + CFBundleTypeIconFile + {{.IconName}} + + {{end}} + + {{end}} + {{if .Info.Protocols}} + CFBundleURLTypes + + {{range .Info.Protocols}} + + CFBundleURLName + com.wails.{{.Scheme}} + CFBundleURLSchemes + + {{.Scheme}} + + CFBundleTypeRole + {{.Role}} + + {{end}} + + {{end}} + + diff --git a/frontend/dist/app.css b/frontend/dist/app.css new file mode 100644 index 0000000..fe8dd09 --- /dev/null +++ b/frontend/dist/app.css @@ -0,0 +1,36 @@ +body { + margin: 0; + background: #101418; + color: #f2f4f6; + font: 16px/1.4 system-ui, sans-serif; +} + +main { + padding: 24px; +} + +#connection { + margin: 0 0 16px; + font-size: 14px; +} + +#detail { + min-height: 1.4em; + font-size: 20px; +} + +button { + margin-top: 24px; + width: 100%; + height: 64px; + border: 0; + border-radius: 8px; + background: #2d6a4f; + color: white; + font: inherit; + font-size: 18px; +} + +button.down { + background: #9b2226; +} diff --git a/frontend/dist/app.js b/frontend/dist/app.js new file mode 100644 index 0000000..bf1bc26 --- /dev/null +++ b/frontend/dist/app.js @@ -0,0 +1,98 @@ +const button = document.getElementById("ptt"); + +function api() { + if (!window.go || !window.go.main || !window.go.main.App) { + return null; + } + return window.go.main.App; +} + +function line(state) { + if (state.detail && state.detail !== "Idle") { + return state.detail; + } + if (state.talker) { + return state.talker; + } + return "Idle"; +} + +function render(state) { + document.getElementById("connection").textContent = state.connection; + document.getElementById("callsign").textContent = state.callsign; + document.getElementById("talkgroup").textContent = String(state.talkGroup); + document.getElementById("detail").textContent = line(state); + button.classList.toggle("down", !!state.transmitting); +} + +function refresh() { + const app = api(); + if (!app) { + return; + } + app.State().then(render).catch(function () {}); +} + +button.addEventListener("pointerdown", function (event) { + event.preventDefault(); + button.setPointerCapture(event.pointerId); + const app = api(); + if (app) { + app.HoldButton(); + } +}); + +function releaseButton() { + const app = api(); + if (app) { + app.ReleaseButton(); + } +} + +button.addEventListener("pointerup", releaseButton); +button.addEventListener("pointercancel", releaseButton); + +let spaceHeld = false; + +window.addEventListener("keydown", function (event) { + if (event.code !== "Space") { + return; + } + event.preventDefault(); + if (event.repeat || spaceHeld) { + return; + } + spaceHeld = true; + const app = api(); + if (app) { + app.HoldSpace(); + } +}); + +window.addEventListener("keyup", function (event) { + if (event.code !== "Space") { + return; + } + event.preventDefault(); + if (!spaceHeld) { + return; + } + spaceHeld = false; + const app = api(); + if (app) { + app.ReleaseSpace(); + } +}); + +window.addEventListener("blur", function () { + spaceHeld = false; + const app = api(); + if (!app) { + return; + } + app.ReleaseButton(); + app.ReleaseSpace(); +}); + +setInterval(refresh, 100); +refresh(); diff --git a/frontend/dist/index.html b/frontend/dist/index.html new file mode 100644 index 0000000..0c0e279 --- /dev/null +++ b/frontend/dist/index.html @@ -0,0 +1,18 @@ + + + + + funknetz + + + +
+

Disconnected

+

Callsign

+

Talk group

+

Idle

+ +
+ + + diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..e7a599b --- /dev/null +++ b/go.mod @@ -0,0 +1,38 @@ +module gitea.neitzel.de/konrad/funknetz + +go 1.27.0 + +require ( + github.com/gen2brain/malgo v0.11.24 + github.com/wailsapp/wails/v2 v2.11.0 +) + +require ( + github.com/bep/debounce v1.2.1 // indirect + github.com/go-ole/go-ole v1.3.0 // indirect + github.com/godbus/dbus/v5 v5.1.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/gorilla/websocket v1.5.3 // indirect + github.com/jchv/go-winloader v0.0.0-20210711035445-715c2860da7e // indirect + github.com/labstack/echo/v4 v4.13.3 // indirect + github.com/labstack/gommon v0.4.2 // indirect + github.com/leaanthony/go-ansi-parser v1.6.1 // indirect + github.com/leaanthony/gosod v1.0.4 // indirect + github.com/leaanthony/slicer v1.6.0 // indirect + github.com/leaanthony/u v1.1.1 // indirect + github.com/mattn/go-colorable v0.1.13 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/rivo/uniseg v0.4.7 // indirect + github.com/samber/lo v1.49.1 // indirect + github.com/tkrajina/go-reflector v0.5.8 // indirect + github.com/valyala/bytebufferpool v1.0.0 // indirect + github.com/valyala/fasttemplate v1.2.2 // indirect + github.com/wailsapp/go-webview2 v1.0.22 // indirect + github.com/wailsapp/mimetype v1.4.1 // indirect + golang.org/x/crypto v0.33.0 // indirect + golang.org/x/net v0.35.0 // indirect + golang.org/x/sys v0.30.0 // indirect + golang.org/x/text v0.22.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..2101af2 --- /dev/null +++ b/go.sum @@ -0,0 +1,83 @@ +github.com/bep/debounce v1.2.1 h1:v67fRdBA9UQu2NhLFXrSg0Brw7CexQekrBwDMM8bzeY= +github.com/bep/debounce v1.2.1/go.mod h1:H8yggRPQKLUhUoqrJC1bO2xNya7vanpDl7xR3ISbCJ0= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/gen2brain/malgo v0.11.24 h1:hHcIJVfzWcEDHFdPl5Dl/CUSOjzOleY0zzAV8Kx+imE= +github.com/gen2brain/malgo v0.11.24/go.mod h1:f9TtuN7DVrXMiV/yIceMeWpvanyVzJQMlBecJFVMxww= +github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE= +github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78= +github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk= +github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/jchv/go-winloader v0.0.0-20210711035445-715c2860da7e h1:Q3+PugElBCf4PFpxhErSzU3/PY5sFL5Z6rfv4AbGAck= +github.com/jchv/go-winloader v0.0.0-20210711035445-715c2860da7e/go.mod h1:alcuEEnZsY1WQsagKhZDsoPCRoOijYqhZvPwLG0kzVs= +github.com/labstack/echo/v4 v4.13.3 h1:pwhpCPrTl5qry5HRdM5FwdXnhXSLSY+WE+YQSeCaafY= +github.com/labstack/echo/v4 v4.13.3/go.mod h1:o90YNEeQWjDozo584l7AwhJMHN0bOC4tAfg+Xox9q5g= +github.com/labstack/gommon v0.4.2 h1:F8qTUNXgG1+6WQmqoUWnz8WiEU60mXVVw0P4ht1WRA0= +github.com/labstack/gommon v0.4.2/go.mod h1:QlUFxVM+SNXhDL/Z7YhocGIBYOiwB0mXm1+1bAPHPyU= +github.com/leaanthony/debme v1.2.1 h1:9Tgwf+kjcrbMQ4WnPcEIUcQuIZYqdWftzZkBr+i/oOc= +github.com/leaanthony/debme v1.2.1/go.mod h1:3V+sCm5tYAgQymvSOfYQ5Xx2JCr+OXiD9Jkw3otUjiA= +github.com/leaanthony/go-ansi-parser v1.6.1 h1:xd8bzARK3dErqkPFtoF9F3/HgN8UQk0ed1YDKpEz01A= +github.com/leaanthony/go-ansi-parser v1.6.1/go.mod h1:+vva/2y4alzVmmIEpk9QDhA7vLC5zKDTRwfZGOp3IWU= +github.com/leaanthony/gosod v1.0.4 h1:YLAbVyd591MRffDgxUOU1NwLhT9T1/YiwjKZpkNFeaI= +github.com/leaanthony/gosod v1.0.4/go.mod h1:GKuIL0zzPj3O1SdWQOdgURSuhkF+Urizzxh26t9f1cw= +github.com/leaanthony/slicer v1.6.0 h1:1RFP5uiPJvT93TAHi+ipd3NACobkW53yUiBqZheE/Js= +github.com/leaanthony/slicer v1.6.0/go.mod h1:o/Iz29g7LN0GqH3aMjWAe90381nyZlDNquK+mtH2Fj8= +github.com/leaanthony/u v1.1.1 h1:TUFjwDGlNX+WuwVEzDqQwC2lOv0P4uhTQw7CMFdiK7M= +github.com/leaanthony/u v1.1.1/go.mod h1:9+o6hejoRljvZ3BzdYlVL0JYCwtnAsVuN9pVTQcaRfI= +github.com/matryer/is v1.4.0/go.mod h1:8I/i5uYgLzgsgEloJE1U6xx5HkBQpAZvepWuujKwMRU= +github.com/matryer/is v1.4.1 h1:55ehd8zaGABKLXQUe2awZ99BD/PTc2ls+KV/dXphgEQ= +github.com/matryer/is v1.4.1/go.mod h1:8I/i5uYgLzgsgEloJE1U6xx5HkBQpAZvepWuujKwMRU= +github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= +github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= +github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= +github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= +github.com/samber/lo v1.49.1 h1:4BIFyVfuQSEpluc7Fua+j1NolZHiEHEpaSEKdsH0tew= +github.com/samber/lo v1.49.1/go.mod h1:dO6KHFzUKXgP8LDhU0oI8d2hekjXnGOu0DB8Jecxd6o= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/tkrajina/go-reflector v0.5.8 h1:yPADHrwmUbMq4RGEyaOUpz2H90sRsETNVpjzo3DLVQQ= +github.com/tkrajina/go-reflector v0.5.8/go.mod h1:ECbqLgccecY5kPmPmXg1MrHW585yMcDkVl6IvJe64T4= +github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= +github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= +github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo= +github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ= +github.com/wailsapp/go-webview2 v1.0.22 h1:YT61F5lj+GGaat5OB96Aa3b4QA+mybD0Ggq6NZijQ58= +github.com/wailsapp/go-webview2 v1.0.22/go.mod h1:qJmWAmAmaniuKGZPWwne+uor3AHMB5PFhqiK0Bbj8kc= +github.com/wailsapp/mimetype v1.4.1 h1:pQN9ycO7uo4vsUUuPeHEYoUkLVkaRntMnHJxVwYhwHs= +github.com/wailsapp/mimetype v1.4.1/go.mod h1:9aV5k31bBOv5z6u+QP8TltzvNGJPmNJD4XlAL3U+j3o= +github.com/wailsapp/wails/v2 v2.11.0 h1:seLacV8pqupq32IjS4Y7V8ucab0WZwtK6VvUVxSBtqQ= +github.com/wailsapp/wails/v2 v2.11.0/go.mod h1:jrf0ZaM6+GBc1wRmXsM8cIvzlg0karYin3erahI4+0k= +golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus= +golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M= +golang.org/x/net v0.0.0-20210505024714-0287a6fb4125/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= +golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8= +golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk= +golang.org/x/sys v0.0.0-20200810151505-1b9f1253b3ed/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM= +golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/opus/opus.go b/internal/opus/opus.go new file mode 100644 index 0000000..cbb33fb --- /dev/null +++ b/internal/opus/opus.go @@ -0,0 +1,123 @@ +// Package opus encodes and decodes the reflector's 16 kHz mono frames with libopus. +package opus + +/* +#cgo pkg-config: opus +#include + +static int configure_encoder(OpusEncoder *enc) { + int err; + err = opus_encoder_ctl(enc, OPUS_SET_BITRATE(20000)); + if (err < 0) return err; + err = opus_encoder_ctl(enc, OPUS_SET_VBR(1)); + if (err < 0) return err; + err = opus_encoder_ctl(enc, OPUS_SET_SIGNAL(OPUS_SIGNAL_VOICE)); + if (err < 0) return err; + err = opus_encoder_ctl(enc, OPUS_SET_MAX_BANDWIDTH(OPUS_BANDWIDTH_MEDIUMBAND)); + if (err < 0) return err; + err = opus_encoder_ctl(enc, OPUS_SET_DTX(0)); + return err; +} +*/ +import "C" + +import ( + "fmt" + "unsafe" +) + +const FrameSamples = 320 + +// Encoder produces one raw Opus packet per 20 ms frame. +type Encoder struct { + enc *C.OpusEncoder +} + +// Decoder turns one raw Opus packet into 320 mono samples. +type Decoder struct { + dec *C.OpusDecoder +} + +// NewEncoder returns a mono 16 kHz voice encoder at 20 kbit/s. +func NewEncoder() (*Encoder, error) { + var err C.int + enc := C.opus_encoder_create(16000, 1, C.OPUS_APPLICATION_AUDIO, &err) + if err < 0 || enc == nil { + return nil, fmt.Errorf("opus encoder %d", int(err)) + } + if c := C.configure_encoder(enc); c < 0 { + C.opus_encoder_destroy(enc) + return nil, fmt.Errorf("opus encoder setup %d", int(c)) + } + return &Encoder{enc: enc}, nil +} + +// Encode encodes exactly 320 samples. +func (e *Encoder) Encode(pcm []int16) ([]byte, error) { + if len(pcm) != FrameSamples { + return nil, fmt.Errorf("opus frame is %d samples", len(pcm)) + } + out := make([]byte, 4000) + n := C.opus_encode( + e.enc, + (*C.opus_int16)(unsafe.Pointer(&pcm[0])), + FrameSamples, + (*C.uchar)(unsafe.Pointer(&out[0])), + C.opus_int32(len(out)), + ) + if n < 0 { + return nil, fmt.Errorf("opus encode %d", int(n)) + } + return append([]byte(nil), out[:n]...), nil +} + +// Close releases the encoder. +func (e *Encoder) Close() { + if e == nil || e.enc == nil { + return + } + C.opus_encoder_destroy(e.enc) + e.enc = nil +} + +// NewDecoder returns a mono 16 kHz decoder. +func NewDecoder() (*Decoder, error) { + var err C.int + dec := C.opus_decoder_create(16000, 1, &err) + if err < 0 || dec == nil { + return nil, fmt.Errorf("opus decoder %d", int(err)) + } + return &Decoder{dec: dec}, nil +} + +// Decode decodes one packet into 320 samples. +func (d *Decoder) Decode(packet []byte) ([]int16, error) { + if len(packet) == 0 { + return nil, fmt.Errorf("empty opus packet") + } + pcm := make([]int16, FrameSamples) + n := C.opus_decode( + d.dec, + (*C.uchar)(unsafe.Pointer(&packet[0])), + C.opus_int32(len(packet)), + (*C.opus_int16)(unsafe.Pointer(&pcm[0])), + FrameSamples, + 0, + ) + if n < 0 { + return nil, fmt.Errorf("opus decode %d", int(n)) + } + if int(n) != FrameSamples { + return nil, fmt.Errorf("opus decode returned %d samples", int(n)) + } + return pcm, nil +} + +// Close releases the decoder. +func (d *Decoder) Close() { + if d == nil || d.dec == nil { + return + } + C.opus_decoder_destroy(d.dec) + d.dec = nil +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..48a6256 --- /dev/null +++ b/main.go @@ -0,0 +1,39 @@ +//go:build !bindings + +package main + +import ( + "embed" + "fmt" + "os" + + "github.com/wailsapp/wails/v2" + "github.com/wailsapp/wails/v2/pkg/options" + "github.com/wailsapp/wails/v2/pkg/options/assetserver" +) + +//go:embed all:frontend/dist +var assets embed.FS + +func main() { + st, err := openStation() + if err != nil { + fmt.Fprintln(os.Stderr, err.Error()) + os.Exit(1) + } + app := &App{station: st} + err = wails.Run(&options.App{ + Title: "funknetz", + Width: 420, + Height: 280, + BackgroundColour: &options.RGBA{R: 16, G: 20, B: 24, A: 255}, + AssetServer: &assetserver.Options{Assets: assets}, + OnStartup: app.startup, + OnShutdown: app.shutdown, + Bind: []any{app}, + }) + if err != nil { + fmt.Fprintln(os.Stderr, err.Error()) + os.Exit(1) + } +} diff --git a/main_bindings.go b/main_bindings.go new file mode 100644 index 0000000..0720908 --- /dev/null +++ b/main_bindings.go @@ -0,0 +1,12 @@ +//go:build bindings + +package main + +import ( + "github.com/wailsapp/wails/v2" + "github.com/wailsapp/wails/v2/pkg/options" +) + +func main() { + _ = wails.Run(&options.App{Bind: []any{&App{}}}) +} diff --git a/station/behavior_test.go b/station/behavior_test.go new file mode 100644 index 0000000..ff0ff3b --- /dev/null +++ b/station/behavior_test.go @@ -0,0 +1,520 @@ +package station_test + +import ( + "context" + "crypto/hmac" + "crypto/sha1" + "encoding/binary" + "io" + "net" + "strconv" + "testing" + "time" + + "gitea.neitzel.de/konrad/funknetz/internal/opus" + "gitea.neitzel.de/konrad/funknetz/station" +) + +type loggedIn struct { + st *station.Station + clock *fakeClock + refl *reflector + cancel context.CancelFunc + id uint16 +} + +func login(t *testing.T, callsign, key string, tg int, id uint16, codecs []string) loggedIn { + t.Helper() + refl := newReflector(t) + path := writeConfigFile(t, configJSON("127.0.0.1", refl.port, callsign, key, tg)) + clock := newFakeClock() + st, err := station.New(station.Options{ConfigPath: path, Clock: clock}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + t.Cleanup(cancel) + go func() { _ = st.Run(ctx) }() + refl.accept(t) + + typ, body := refl.readFrame(t) + if typ != 5 || len(body) != 4 || body[1] != 2 { + t.Fatalf("protocol frame type %d body %v", typ, body) + } + challenge := make([]byte, 20) + for i := range challenge { + challenge[i] = byte(i + 1) + } + refl.writeFrame(t, 10, appendBytes(nil, challenge)) + typ, body = refl.readFrame(t) + if typ != 11 { + t.Fatalf("auth type %d", typ) + } + call, rest := takeStr(t, body) + digest, _ := takeBytes(t, rest) + if call != "DL1ABC" && call != callsign { + t.Fatalf("callsign %q", call) + } + mac := hmac.New(sha1.New, []byte(key)) + mac.Write(challenge) + if !hmac.Equal(digest, mac.Sum(nil)) { + t.Fatal("auth digest does not match the stored key") + } + refl.writeFrame(t, 12, nil) + refl.writeFrame(t, 100, serverInfo(id, []string{"db0abc"}, codecs)) + typ, body = refl.readFrame(t) + if typ != 106 { + t.Fatalf("type %d, want select talk group", typ) + } + gotTG, _ := takeU32(t, body) + if int(gotTG) != tg { + t.Fatalf("talk group %d, want %d", gotTG, tg) + } + utyp, gotID, seq, ubody := refl.readUDP(t) + if utyp != 1 || gotID != id || seq != 0 || len(ubody) != 0 { + t.Fatalf("udp heartbeat type %d id %d seq %d len %d", utyp, gotID, seq, len(ubody)) + } + if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connected" }); err != nil { + t.Fatal(err) + } + return loggedIn{st: st, clock: clock, refl: refl, cancel: cancel, id: id} +} + +func TestQuietSessionSendsHeartbeats(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + + in.clock.BlockUntilWaiting(t) + in.clock.Advance(9 * time.Second) + if _, _, ok := tryFrame(in.refl); ok { + t.Fatal("TCP heartbeat arrived before 10 seconds") + } + in.clock.Advance(1 * time.Second) + in.clock.BlockUntilWaiting(t) + typ, body := in.refl.readFrame(t) + if typ != 1 || len(body) != 0 { + t.Fatalf("TCP heartbeat type %d len %d", typ, len(body)) + } + gen := in.clock.Generation() + in.refl.writeFrame(t, 1, nil) + waitClock(t, in.clock, gen) + + in.clock.Advance(5 * time.Second) + utyp, id, seq, ubody := in.refl.readUDP(t) + if utyp != 1 || id != 7 || seq != 1 || len(ubody) != 0 { + t.Fatalf("UDP heartbeat type %d id %d seq %d len %d", utyp, id, seq, len(ubody)) + } +} + +func TestDropRetriesUntilQuit(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + in.refl.tcp.Close() + if _, err := in.st.Wait(context.Background(), func(s station.State) bool { + return s.Connection == "Disconnected" && s.Detail == "Idle" + }); err != nil { + t.Fatal(err) + } + in.clock.BlockUntilWaiting(t) + in.clock.Advance(2 * time.Second) + in.refl.accept(t) + typ, body := in.refl.readFrame(t) + if typ != 5 || body[1] != 2 { + t.Fatalf("retry protocol type %d body %v", typ, body) + } + in.cancel() + in.clock.Advance(2 * time.Second) + in.refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond)) + if conn, err := in.refl.ln.Accept(); err == nil { + conn.Close() + t.Fatal("dialed after quit") + } +} + +func TestRefusedConnectionRetriesUntilQuit(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := ln.Addr().(*net.TCPAddr).Port + ln.Close() + + path := writeConfigFile(t, configJSON("127.0.0.1", port, "DL1ABC", "key", 4)) + clock := newFakeClock() + st, err := station.New(station.Options{ConfigPath: path, Clock: clock}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go func() { _ = st.Run(ctx) }() + + if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil { + t.Fatal(err) + } + st.HoldButton() + st.WriteMicrophone(make([]int16, opus.FrameSamples)) + + ln, err = net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + clock.BlockUntilWaiting(t) + clock.Advance(2 * time.Second) + ln.(*net.TCPListener).SetDeadline(time.Now().Add(2 * time.Second)) + conn, err := ln.Accept() + if err != nil { + t.Fatal(err) + } + conn.Close() + if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil { + t.Fatal(err) + } + cancel() + clock.Advance(2 * time.Second) + ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond)) + if conn, err := ln.Accept(); err == nil { + conn.Close() + t.Fatal("dialed after quit") + } +} + +func TestReflectorErrorStopsRetries(t *testing.T) { + for _, msg := range []string{"Access denied", "no such station"} { + t.Run(msg, func(t *testing.T) { + refl := newReflector(t) + path := writeConfigFile(t, configJSON("127.0.0.1", refl.port, "DL1ABC", "key", 20)) + clock := newFakeClock() + st, err := station.New(station.Options{ConfigPath: path, Clock: clock}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go func() { _ = st.Run(ctx) }() + refl.accept(t) + refl.readFrame(t) + refl.writeFrame(t, 13, appendStr(nil, msg)) + got, err := st.Wait(ctx, func(s station.State) bool { return s.Detail == msg }) + if err != nil { + t.Fatal(err) + } + if got.Connection != "Disconnected" { + t.Fatalf("Connection = %s, want Disconnected", got.Connection) + } + clock.Advance(2 * time.Second) + refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond)) + if conn, err := refl.ln.Accept(); err == nil { + conn.Close() + t.Fatal("retried after reflector error") + } + if got := st.State().Detail; got != msg { + t.Fatalf("Detail = %q, want it to stay %q", got, msg) + } + }) + } +} + +func TestMissingOpusStopsRetries(t *testing.T) { + refl := newReflector(t) + path := writeConfigFile(t, configJSON("127.0.0.1", refl.port, "DL1ABC", "key", 20)) + clock := newFakeClock() + st, err := station.New(station.Options{ConfigPath: path, Clock: clock}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go func() { _ = st.Run(ctx) }() + refl.accept(t) + refl.readFrame(t) + challenge := make([]byte, 20) + refl.writeFrame(t, 10, appendBytes(nil, challenge)) + refl.readFrame(t) + refl.writeFrame(t, 12, nil) + refl.writeFrame(t, 100, serverInfo(9, nil, []string{"GSM"})) + got, err := st.Wait(ctx, func(s station.State) bool { + return s.Detail == "The reflector did not offer OPUS." + }) + if err != nil { + t.Fatal(err) + } + if got.Connection != "Disconnected" { + t.Fatalf("Connection = %s", got.Connection) + } + clock.Advance(2 * time.Second) + refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond)) + if conn, err := refl.ln.Accept(); err == nil { + conn.Close() + t.Fatal("retried after missing OPUS") + } +} + +func TestTCPReceiveTimeoutDropsTheSession(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + in.clock.BlockUntilWaiting(t) + in.clock.Advance(15 * time.Second) + if _, err := in.st.Wait(context.Background(), func(s station.State) bool { + return s.Connection == "Disconnected" + }); err != nil { + t.Fatal(err) + } + in.clock.BlockUntilWaiting(t) + in.clock.Advance(2 * time.Second) + in.refl.accept(t) +} + +func TestUDPReceiveTimeoutDropsTheSession(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + for step := 0; step < 12; step++ { + in.clock.BlockUntilWaiting(t) + in.clock.Advance(5 * time.Second) + gen := in.clock.Generation() + if replyTCP(t, in.refl) { + waitClock(t, in.clock, gen) + } + discardUDP(t, in.refl) + if in.st.State().Connection == "Disconnected" { + if step < 11 { + t.Fatalf("dropped on step %d", step) + } + return + } + } + t.Fatal("session stayed up for 60 seconds without UDP") +} + +func TestRemoteAudioAndTalker(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + enc, err := opus.NewEncoder() + if err != nil { + t.Fatal(err) + } + defer enc.Close() + pkt, err := enc.Encode(make([]int16, opus.FrameSamples)) + if err != nil { + t.Fatal(err) + } + in.refl.writeUDP(t, 101, 1, 0, appendBytes(nil, pkt)) + in.refl.writeUDP(t, 101, 7, 1, appendBytes(nil, pkt)) + + ctx := t.Context() + pcm, err := in.st.WaitPlayback(ctx, opus.FrameSamples) + if err != nil { + t.Fatal(err) + } + if len(pcm) != opus.FrameSamples { + t.Fatalf("playback %d samples", len(pcm)) + } + if extra := in.st.TakePlayback(); len(extra) != 0 { + t.Fatalf("played %d extra samples from the wrong client id", len(extra)) + } + if got := in.st.State(); got.Talker != "" || got.Detail != "Idle" { + t.Fatalf("State = %+v, want Idle before the talker is named", got) + } + + in.refl.writeFrame(t, 104, appendStr(appendU32(nil, 99), "OTHER1")) + waitCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) + if _, err := in.st.Wait(waitCtx, func(s station.State) bool { return s.Talker == "OTHER1" }); err == nil { + cancel() + t.Fatal("showed a talker from another talk group") + } + cancel() + + in.refl.writeFrame(t, 104, appendStr(appendU32(nil, 20), "DB0XYZ")) + got, err := in.st.Wait(ctx, func(s station.State) bool { return s.Talker == "DB0XYZ" }) + if err != nil { + t.Fatal(err) + } + if got.Detail != "Idle" { + t.Fatalf("Detail = %q", got.Detail) + } + in.st.HoldButton() + in.st.WriteMicrophone(make([]int16, opus.FrameSamples)) + if _, _, ok := tryUDP(in.refl); ok { + t.Fatal("sent audio while the room was busy") + } + + in.refl.writeFrame(t, 105, appendStr(appendU32(nil, 20), "DB0XYZ")) + if _, err := in.st.Wait(ctx, func(s station.State) bool { + return s.Talker == "DL1ABC" && s.Transmitting + }); err != nil { + t.Fatal(err) + } + in.st.WriteMicrophone(make([]int16, opus.FrameSamples)) + utyp, id, _, body := in.refl.readUDP(t) + if utyp != 101 || id != 7 { + t.Fatalf("audio type %d id %d", utyp, id) + } + packet, _ := takeBytes(t, body) + dec, err := opus.NewDecoder() + if err != nil { + t.Fatal(err) + } + defer dec.Close() + decoded, err := dec.Decode(packet) + if err != nil { + t.Fatal(err) + } + if len(decoded) != opus.FrameSamples { + t.Fatalf("decoded %d samples", len(decoded)) + } +} + +func TestMicrophoneIsNotPlayed(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + in.st.HoldButton() + in.st.WriteMicrophone(make([]int16, opus.FrameSamples)) + in.refl.readUDP(t) + if pcm := in.st.TakePlayback(); len(pcm) != 0 { + t.Fatalf("playback %d samples from the microphone", len(pcm)) + } + in.st.ReleaseButton() + in.refl.readUDP(t) // flush, or padded frame then flush; drained below + discardUDP(t, in.refl) +} + +func TestPushToTalk(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + dec, err := opus.NewDecoder() + if err != nil { + t.Fatal(err) + } + defer dec.Close() + + in.st.WriteMicrophone(make([]int16, opus.FrameSamples*10)) + if _, _, ok := tryUDP(in.refl); ok { + t.Fatal("sent audio before push-to-talk") + } + + in.st.HoldButton() + in.st.HoldSpace() + if got := in.st.State(); got.Talker != "DL1ABC" || !got.Transmitting { + t.Fatalf("State = %+v, want this station as the talker", got) + } + in.st.WriteMicrophone(make([]int16, opus.FrameSamples)) + expectAudio(t, in.refl, dec, 7, 1) + in.st.ReleaseButton() + in.st.WriteMicrophone(make([]int16, opus.FrameSamples)) + expectAudio(t, in.refl, dec, 7, 2) + + in.st.WriteMicrophone(make([]int16, 100)) + in.st.ReleaseSpace() + expectAudio(t, in.refl, dec, 7, 3) + utyp, id, seq, body := in.refl.readUDP(t) + if utyp != 102 || id != 7 || seq != 4 || len(body) != 0 { + t.Fatalf("flush type %d id %d seq %d len %d", utyp, id, seq, len(body)) + } + if got := in.st.State(); got.Transmitting || got.Talker != "" || got.Detail != "Idle" { + t.Fatalf("State = %+v, want Idle after release", got) + } + in.st.WriteMicrophone(make([]int16, opus.FrameSamples)) + if _, _, ok := tryUDP(in.refl); ok { + t.Fatal("sent audio after release") + } + + enc, err := opus.NewEncoder() + if err != nil { + t.Fatal(err) + } + defer enc.Close() + pkt, err := enc.Encode(make([]int16, opus.FrameSamples)) + if err != nil { + t.Fatal(err) + } + in.st.HoldSpace() + in.refl.writeUDP(t, 101, 7, 9, appendBytes(nil, pkt)) + waitCtx, cancel := context.WithTimeout(t.Context(), 150*time.Millisecond) + defer cancel() + if _, err := in.st.WaitPlayback(waitCtx, opus.FrameSamples); err == nil { + t.Fatal("played remote audio while transmitting") + } +} + +func TestSequenceWraps(t *testing.T) { + in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"}) + in.st.HoldButton() + var prev uint16 = 0 + for i := 1; i <= 65536; i++ { + in.st.WriteMicrophone(make([]int16, opus.FrameSamples)) + _, _, seq, _ := in.refl.readUDP(t) + want := uint16(i) + if seq != want { + t.Fatalf("packet %d sequence %d, want %d", i, seq, want) + } + if i > 1 && seq != prev+1 { + t.Fatalf("sequence %d followed %d", seq, prev) + } + prev = seq + } +} + +func expectAudio(t *testing.T, refl *reflector, dec *opus.Decoder, id, seq uint16) { + t.Helper() + typ, gotID, gotSeq, body := refl.readUDP(t) + if typ != 101 || gotID != id || gotSeq != seq { + t.Fatalf("datagram type %d id %d seq %d, want audio id %d seq %d", typ, gotID, gotSeq, id, seq) + } + packet, rest := takeBytes(t, body) + if len(rest) != 0 { + t.Fatalf("audio has %d trailing bytes", len(rest)) + } + pcm, err := dec.Decode(packet) + if err != nil { + t.Fatal(err) + } + if len(pcm) != opus.FrameSamples { + t.Fatalf("decoded %d samples", len(pcm)) + } +} + +func tryFrame(r *reflector) (uint16, []byte, bool) { + _ = r.tcp.SetReadDeadline(time.Now().Add(30 * time.Millisecond)) + var hdr [4]byte + if _, err := io.ReadFull(r.tcp, hdr[:]); err != nil { + return 0, nil, false + } + n := binary.BigEndian.Uint32(hdr[:]) + buf := make([]byte, n) + if _, err := io.ReadFull(r.tcp, buf); err != nil { + return 0, nil, false + } + if len(buf) < 2 { + return 0, nil, false + } + return binary.BigEndian.Uint16(buf[:2]), buf[2:], true +} + +func tryUDP(r *reflector) ([]byte, *net.UDPAddr, bool) { + _ = r.udp.SetReadDeadline(time.Now().Add(40 * time.Millisecond)) + buf := make([]byte, 2048) + n, addr, err := r.udp.ReadFromUDP(buf) + if err != nil { + return nil, nil, false + } + r.peer = addr + return append([]byte(nil), buf[:n]...), addr, true +} + +func replyTCP(t *testing.T, r *reflector) bool { + t.Helper() + wrote := false + for { + typ, _, ok := tryFrame(r) + if !ok { + return wrote + } + if typ == 1 { + r.writeFrame(t, 1, nil) + wrote = true + } + } +} + +func discardUDP(t *testing.T, r *reflector) { + t.Helper() + for { + if _, _, ok := tryUDP(r); !ok { + return + } + } +} diff --git a/station/clock.go b/station/clock.go new file mode 100644 index 0000000..5128e88 --- /dev/null +++ b/station/clock.go @@ -0,0 +1,32 @@ +package station + +import "time" + +// Clock is the time source for heartbeats and the retry wait. +type Clock interface { + Now() time.Time + NewTimer(d time.Duration) Timer +} + +// Timer is a single deadline. Stop prevents the channel from firing. +type Timer interface { + C() <-chan time.Time + Stop() bool +} + +type wallClock struct{} + +func (wallClock) Now() time.Time { return time.Now() } + +func (wallClock) NewTimer(d time.Duration) Timer { + if d < 0 { + d = 0 + } + return wallTimer{t: time.NewTimer(d)} +} + +type wallTimer struct{ t *time.Timer } + +func (w wallTimer) C() <-chan time.Time { return w.t.C } + +func (w wallTimer) Stop() bool { return w.t.Stop() } diff --git a/station/config.go b/station/config.go new file mode 100644 index 0000000..33610e8 --- /dev/null +++ b/station/config.go @@ -0,0 +1,155 @@ +package station + +import ( + "fmt" + "math" + "os" + "regexp" + "strings" + + "encoding/json/jsontext" + jsonv2 "encoding/json/v2" +) + +type loaded struct { + host string + port int + callsign string + authKey string + talkGroup int +} + +type fileShape struct { + Host jsontext.Value `json:"host"` + Port jsontext.Value `json:"port"` + Callsign jsontext.Value `json:"callsign"` + AuthKey jsontext.Value `json:"authKey"` + TalkGroup jsontext.Value `json:"talkGroup"` +} + +type templateFile struct { + Host string `json:"host"` + Port int `json:"port"` + Callsign string `json:"callsign"` + AuthKey string `json:"authKey"` + TalkGroup int `json:"talkGroup"` +} + +var callsignPattern = regexp.MustCompile(`^[A-Z0-9][A-Z]{0,2}\d[A-Z0-9]{0,3}[A-Z](?:-[A-Z0-9]{1,3})?$`) + +func loadConfig(path string) (loaded, error) { + body, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + if err := writeTemplate(path); err != nil { + return loaded{}, err + } + return loaded{}, fmt.Errorf("created %s", path) + } + return loaded{}, fmt.Errorf("could not be read") + } + return parseConfig(body) +} + +func writeTemplate(path string) error { + body, err := jsonv2.Marshal(templateFile{}) + if err != nil { + return err + } + body = append(body, '\n') + f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if err != nil { + return err + } + _, werr := f.Write(body) + cerr := f.Close() + if werr != nil { + return werr + } + if cerr != nil { + return cerr + } + return os.Chmod(path, 0o600) +} + +func parseConfig(body []byte) (loaded, error) { + var shape fileShape + if err := jsonv2.Unmarshal(body, &shape); err != nil { + return loaded{}, fmt.Errorf("could not be read") + } + host, err := stringField("host", shape.Host) + if err != nil { + return loaded{}, err + } + port, err := intField("port", shape.Port, 1, 65535) + if err != nil { + return loaded{}, err + } + call, err := stringField("callsign", shape.Callsign) + if err != nil { + return loaded{}, err + } + key, err := authField(shape.AuthKey) + if err != nil { + return loaded{}, err + } + tg, err := intField("talkGroup", shape.TalkGroup, 1, int(math.MaxUint32)) + if err != nil { + return loaded{}, err + } + + host = strings.TrimSpace(host) + if host == "" { + return loaded{}, fmt.Errorf("host is empty") + } + call = strings.ToUpper(strings.TrimSpace(call)) + if call == "" { + return loaded{}, fmt.Errorf("callsign is empty") + } + if !callsignPattern.MatchString(call) { + return loaded{}, fmt.Errorf("callsign is invalid") + } + return loaded{host: host, port: port, callsign: call, authKey: key, talkGroup: tg}, nil +} + +func stringField(name string, v jsontext.Value) (string, error) { + if len(v) == 0 { + return "", nil + } + if v.Kind() != jsontext.KindString { + return "", fmt.Errorf("%s has the wrong type", name) + } + var s string + if err := jsonv2.Unmarshal(v, &s); err != nil { + return "", fmt.Errorf("%s has the wrong type", name) + } + return s, nil +} + +func authField(v jsontext.Value) (string, error) { + s, err := stringField("authKey", v) + if err != nil { + return "", err + } + if strings.TrimSpace(s) == "" { + return "", fmt.Errorf("authKey is empty") + } + return s, nil +} + +func intField(name string, v jsontext.Value, min, max int) (int, error) { + if len(v) == 0 { + return 0, fmt.Errorf("%s is invalid", name) + } + if v.Kind() != jsontext.KindNumber { + return 0, fmt.Errorf("%s has the wrong type", name) + } + var n int + if err := jsonv2.Unmarshal(v, &n); err != nil { + return 0, fmt.Errorf("%s is invalid", name) + } + if n < min || n > max { + return 0, fmt.Errorf("%s is invalid", name) + } + return n, nil +} diff --git a/station/config_test.go b/station/config_test.go new file mode 100644 index 0000000..0dbbba8 --- /dev/null +++ b/station/config_test.go @@ -0,0 +1,178 @@ +package station_test + +import ( + "os" + "path/filepath" + "strings" + "testing" + + jsonv2 "encoding/json/v2" + + "gitea.neitzel.de/konrad/funknetz/station" +) + +func TestMissingConfigIsCreatedAndRejected(t *testing.T) { + path := filepath.Join(t.TempDir(), ".funknetz.json") + + _, err := station.New(station.Options{ConfigPath: path}) + if err == nil { + t.Fatal("New() error = nil, want a rejection after creating the file") + } + if !strings.Contains(err.Error(), "created") { + t.Fatalf("New() error = %q, want it to say the file was created", err.Error()) + } + + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if got, want := info.Mode().Perm(), os.FileMode(0o600); got != want { + t.Fatalf("mode = %o, want %o", got, want) + } + + body, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var got map[string]any + if err := jsonv2.Unmarshal(body, &got); err != nil { + t.Fatalf("created file is not JSON: %v\n%s", err, body) + } + if got["host"] != "" || got["callsign"] != "" || got["authKey"] != "" { + t.Fatalf("string fields = %#v, want empty host, callsign, and authKey", got) + } + if got["port"] != float64(0) || got["talkGroup"] != float64(0) { + t.Fatalf("numeric fields = %#v, want port and talkGroup at zero", got) + } +} + +func TestExistingConfigIsNotRewritten(t *testing.T) { + path := filepath.Join(t.TempDir(), "funknetz.json") + original := []byte("{\"host\":\"127.0.0.1\",\"port\":5308,\"callsign\":\"DL1ABC\",\"authKey\":\"a b\",\"talkGroup\":20}\n") + if err := os.WriteFile(path, original, 0o640); err != nil { + t.Fatal(err) + } + + if _, err := station.New(station.Options{ConfigPath: path}); err != nil { + t.Fatal(err) + } + got, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(got) != string(original) { + t.Fatalf("file = %q, want %q", got, original) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if got, want := info.Mode().Perm(), os.FileMode(0o640); got != want { + t.Fatalf("mode = %o, want %o", got, want) + } +} + +func TestInvalidConfigNamesTheField(t *testing.T) { + tests := []struct { + name string + body string + want string + }{ + {name: "empty host", body: configJSON("", 5308, "DL1ABC", "key", 1), want: "host is empty"}, + {name: "blank host", body: configJSON(" ", 5308, "DL1ABC", "key", 1), want: "host is empty"}, + {name: "host type", body: `{"host":1,"port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":1}`, want: "host has the wrong type"}, + {name: "empty callsign", body: configJSON("127.0.0.1", 5308, "", "key", 1), want: "callsign is empty"}, + {name: "callsign type", body: `{"host":"127.0.0.1","port":5308,"callsign":1,"authKey":"key","talkGroup":1}`, want: "callsign has the wrong type"}, + {name: "empty auth key", body: configJSON("127.0.0.1", 5308, "DL1ABC", "", 1), want: "authKey is empty"}, + {name: "whitespace auth key", body: configJSON("127.0.0.1", 5308, "DL1ABC", " \t", 1), want: "authKey is empty"}, + {name: "auth key type", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":1,"talkGroup":1}`, want: "authKey has the wrong type"}, + {name: "port zero", body: configJSON("127.0.0.1", 0, "DL1ABC", "key", 1), want: "port is invalid"}, + {name: "port negative", body: configJSON("127.0.0.1", -1, "DL1ABC", "key", 1), want: "port is invalid"}, + {name: "port too high", body: configJSON("127.0.0.1", 65536, "DL1ABC", "key", 1), want: "port is invalid"}, + {name: "port type", body: `{"host":"127.0.0.1","port":"5308","callsign":"DL1ABC","authKey":"key","talkGroup":1}`, want: "port has the wrong type"}, + {name: "port fraction", body: `{"host":"127.0.0.1","port":5308.5,"callsign":"DL1ABC","authKey":"key","talkGroup":1}`, want: "port is invalid"}, + {name: "talk group zero", body: configJSON("127.0.0.1", 5308, "DL1ABC", "key", 0), want: "talkGroup is invalid"}, + {name: "talk group negative", body: configJSON("127.0.0.1", 5308, "DL1ABC", "key", -3), want: "talkGroup is invalid"}, + {name: "talk group type", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":"20"}`, want: "talkGroup has the wrong type"}, + {name: "bad callsign", body: configJSON("127.0.0.1", 5308, "HELLO", "key", 1), want: "callsign is invalid"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "funknetz.json") + if err := os.WriteFile(path, []byte(tt.body), 0o600); err != nil { + t.Fatal(err) + } + before, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + _, err = station.New(station.Options{ConfigPath: path}) + if err == nil { + t.Fatal("New() error = nil, want a field rejection") + } + if err.Error() != tt.want { + t.Fatalf("New() error = %q, want %q", err.Error(), tt.want) + } + after, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(after) != string(before) { + t.Fatalf("file = %q, want it left untouched", after) + } + }) + } +} + +func TestBrokenConfigSaysTheFileCouldNotBeRead(t *testing.T) { + path := filepath.Join(t.TempDir(), "funknetz.json") + if err := os.WriteFile(path, []byte("{"), 0o600); err != nil { + t.Fatal(err) + } + _, err := station.New(station.Options{ConfigPath: path}) + if err == nil { + t.Fatal("New() error = nil, want a read failure") + } + if err.Error() != "could not be read" { + t.Fatalf("New() error = %q, want %q", err.Error(), "could not be read") + } + if strings.Contains(err.Error(), "callsign") { + t.Fatalf("New() error = %q, want no field name", err.Error()) + } +} + +func TestCallsignIsTrimmedAndUppercased(t *testing.T) { + path := filepath.Join(t.TempDir(), "funknetz.json") + body := configJSON(" 127.0.0.1 ", 5308, " dl1abc-r ", "a b", 20) + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + st, err := station.New(station.Options{ConfigPath: path}) + if err != nil { + t.Fatal(err) + } + got := st.State() + if got.Callsign != "DL1ABC-R" { + t.Fatalf("Callsign = %q, want DL1ABC-R", got.Callsign) + } + if got.TalkGroup != 20 { + t.Fatalf("TalkGroup = %d, want 20", got.TalkGroup) + } + if got.Connection != "Disconnected" || got.Detail != "Idle" || got.Talker != "" || got.Transmitting { + t.Fatalf("State = %+v, want Disconnected, Idle, no talker", got) + } +} + +func configJSON(host string, port int, callsign, authKey string, talkGroup int) string { + body, err := jsonv2.Marshal(map[string]any{ + "host": host, + "port": port, + "callsign": callsign, + "authKey": authKey, + "talkGroup": talkGroup, + }) + if err != nil { + panic(err) + } + return string(body) +} diff --git a/station/protocol.go b/station/protocol.go new file mode 100644 index 0000000..2d795a3 --- /dev/null +++ b/station/protocol.go @@ -0,0 +1,158 @@ +package station + +import ( + "encoding/binary" + "fmt" + "io" +) + +const ( + tcpHeartbeat = 1 + tcpProtoVer = 5 + tcpAuthChallenge = 10 + tcpAuthResponse = 11 + tcpAuthOK = 12 + tcpError = 13 + tcpServerInfo = 100 + tcpTalkerStart = 104 + tcpTalkerStop = 105 + tcpSelectTG = 106 + + udpHeartbeat = 1 + udpAudio = 101 + udpFlush = 102 + + maxFrame = 32768 +) + +type frame struct { + typ uint16 + body []byte +} + +func writeFrame(w io.Writer, typ uint16, body []byte) error { + payload := make([]byte, 2+len(body)) + binary.BigEndian.PutUint16(payload[:2], typ) + copy(payload[2:], body) + var hdr [4]byte + binary.BigEndian.PutUint32(hdr[:], uint32(len(payload))) + if _, err := w.Write(hdr[:]); err != nil { + return err + } + _, err := w.Write(payload) + return err +} + +func readFrame(r io.Reader) (frame, error) { + var hdr [4]byte + if _, err := io.ReadFull(r, hdr[:]); err != nil { + return frame{}, err + } + n := binary.BigEndian.Uint32(hdr[:]) + if n < 2 || n > maxFrame { + return frame{}, fmt.Errorf("frame length %d", n) + } + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return frame{}, err + } + return frame{typ: binary.BigEndian.Uint16(buf[:2]), body: buf[2:]}, nil +} + +func appendU16(b []byte, v uint16) []byte { + var buf [2]byte + binary.BigEndian.PutUint16(buf[:], v) + return append(b, buf[:]...) +} + +func appendString(b []byte, s string) []byte { + b = appendU16(b, uint16(len(s))) + return append(b, s...) +} + +func appendBlob(b []byte, p []byte) []byte { + b = appendU16(b, uint16(len(p))) + return append(b, p...) +} + +func parseU16(b []byte) (uint16, []byte, error) { + if len(b) < 2 { + return 0, nil, io.ErrUnexpectedEOF + } + return binary.BigEndian.Uint16(b[:2]), b[2:], nil +} + +func parseU32(b []byte) (uint32, []byte, error) { + if len(b) < 4 { + return 0, nil, io.ErrUnexpectedEOF + } + return binary.BigEndian.Uint32(b[:4]), b[4:], nil +} + +func parseString(b []byte) (string, []byte, error) { + n, rest, err := parseU16(b) + if err != nil { + return "", nil, err + } + if len(rest) < int(n) { + return "", nil, io.ErrUnexpectedEOF + } + return string(rest[:n]), rest[n:], nil +} + +func parseBlob(b []byte) ([]byte, []byte, error) { + n, rest, err := parseU16(b) + if err != nil { + return nil, nil, err + } + if len(rest) < int(n) { + return nil, nil, io.ErrUnexpectedEOF + } + return rest[:n], rest[n:], nil +} + +func parseStringList(b []byte) ([]string, []byte, error) { + n, rest, err := parseU16(b) + if err != nil { + return nil, nil, err + } + out := make([]string, 0, n) + for range int(n) { + s, next, err := parseString(rest) + if err != nil { + return nil, nil, err + } + out = append(out, s) + rest = next + } + return out, rest, nil +} + +func parseServerInfo(b []byte) (uint16, []string, error) { + _, rest, err := parseU16(b) + if err != nil { + return 0, nil, err + } + id, rest, err := parseU16(rest) + if err != nil { + return 0, nil, err + } + _, rest, err = parseStringList(rest) + if err != nil { + return 0, nil, err + } + codecs, _, err := parseStringList(rest) + if err != nil { + return 0, nil, err + } + return id, codecs, nil +} + +func hasOPUS(codecs []string) bool { + for _, c := range codecs { + if c == "OPUS" { + return true + } + } + return false +} diff --git a/station/reflector_test.go b/station/reflector_test.go new file mode 100644 index 0000000..b7dafe0 --- /dev/null +++ b/station/reflector_test.go @@ -0,0 +1,325 @@ +package station_test + +import ( + "encoding/binary" + "io" + "net" + "os" + "sync" + "testing" + "time" + + "gitea.neitzel.de/konrad/funknetz/station" +) + +type fakeClock struct { + mu sync.Mutex + now time.Time + waiters []*fakeWaiter + gen int +} + +type fakeWaiter struct { + deadline time.Time + ch chan time.Time + done bool +} + +func newFakeClock() *fakeClock { + return &fakeClock{now: time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)} +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) NewTimer(d time.Duration) station.Timer { + c.mu.Lock() + defer c.mu.Unlock() + if d < 0 { + d = 0 + } + ch := make(chan time.Time, 1) + c.gen++ + w := &fakeWaiter{deadline: c.now.Add(d), ch: ch} + if d == 0 { + ch <- c.now + w.done = true + return &fakeTimer{c: c, w: w} + } + c.waiters = append(c.waiters, w) + return &fakeTimer{c: c, w: w} +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.now = c.now.Add(d) + kept := c.waiters[:0] + for _, w := range c.waiters { + if w.done { + continue + } + if !w.deadline.After(c.now) { + w.done = true + w.ch <- c.now + continue + } + kept = append(kept, w) + } + c.waiters = kept +} + +func (c *fakeClock) Generation() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.gen +} + +func waitClock(t *testing.T, c *fakeClock, after int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for c.Generation() <= after { + if time.Now().After(deadline) { + t.Fatal("station did not rearm the clock") + } + time.Sleep(time.Millisecond) + } +} + +func (c *fakeClock) BlockUntilWaiting(t *testing.T) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for { + c.mu.Lock() + n := 0 + for _, w := range c.waiters { + if !w.done { + n++ + } + } + c.mu.Unlock() + if n > 0 { + return + } + if time.Now().After(deadline) { + t.Fatal("station did not wait on the clock") + } + time.Sleep(time.Millisecond) + } +} + +type fakeTimer struct { + c *fakeClock + w *fakeWaiter +} + +func (t *fakeTimer) C() <-chan time.Time { return t.w.ch } + +func (t *fakeTimer) Stop() bool { + t.c.mu.Lock() + defer t.c.mu.Unlock() + if t.w.done { + return false + } + t.w.done = true + return true +} + +type reflector struct { + ln net.Listener + udp *net.UDPConn + tcp net.Conn + peer *net.UDPAddr + port int +} + +func newReflector(t *testing.T) *reflector { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := ln.Addr().(*net.TCPAddr).Port + udp, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}) + if err != nil { + ln.Close() + t.Fatal(err) + } + t.Cleanup(func() { + if ln != nil { + ln.Close() + } + udp.Close() + }) + return &reflector{ln: ln, udp: udp, port: port} +} + +func (r *reflector) accept(t *testing.T) { + t.Helper() + r.ln.(*net.TCPListener).SetDeadline(time.Now().Add(2 * time.Second)) + conn, err := r.ln.Accept() + if err != nil { + t.Fatal(err) + } + r.tcp = conn + t.Cleanup(func() { conn.Close() }) +} + +func (r *reflector) readFrame(t *testing.T) (uint16, []byte) { + t.Helper() + if err := r.tcp.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + var hdr [4]byte + if _, err := io.ReadFull(r.tcp, hdr[:]); err != nil { + t.Fatal(err) + } + n := binary.BigEndian.Uint32(hdr[:]) + if n > 32768 { + t.Fatalf("frame length %d", n) + } + buf := make([]byte, n) + if _, err := io.ReadFull(r.tcp, buf); err != nil { + t.Fatal(err) + } + if len(buf) < 2 { + t.Fatalf("short frame %d", len(buf)) + } + return binary.BigEndian.Uint16(buf[:2]), buf[2:] +} + +func (r *reflector) writeFrame(t *testing.T, typ uint16, body []byte) { + t.Helper() + payload := make([]byte, 2+len(body)) + binary.BigEndian.PutUint16(payload, typ) + copy(payload[2:], body) + var hdr [4]byte + binary.BigEndian.PutUint32(hdr[:], uint32(len(payload))) + if err := r.tcp.SetWriteDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + if _, err := r.tcp.Write(append(hdr[:], payload...)); err != nil { + t.Fatal(err) + } +} + +func (r *reflector) readUDP(t *testing.T) (typ, id, seq uint16, body []byte) { + t.Helper() + if err := r.udp.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + buf := make([]byte, 2048) + n, addr, err := r.udp.ReadFromUDP(buf) + if err != nil { + t.Fatal(err) + } + r.peer = addr + if n < 6 { + t.Fatalf("short datagram %d", n) + } + typ = binary.BigEndian.Uint16(buf[0:2]) + id = binary.BigEndian.Uint16(buf[2:4]) + seq = binary.BigEndian.Uint16(buf[4:6]) + body = append([]byte(nil), buf[6:n]...) + return typ, id, seq, body +} + +func (r *reflector) writeUDP(t *testing.T, typ, id, seq uint16, body []byte) { + t.Helper() + if r.peer == nil { + t.Fatal("no udp peer") + } + buf := make([]byte, 6+len(body)) + binary.BigEndian.PutUint16(buf[0:2], typ) + binary.BigEndian.PutUint16(buf[2:4], id) + binary.BigEndian.PutUint16(buf[4:6], seq) + copy(buf[6:], body) + if err := r.udp.SetWriteDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + if _, err := r.udp.WriteToUDP(buf, r.peer); err != nil { + t.Fatal(err) + } +} + +func appendU16(b []byte, v uint16) []byte { + var buf [2]byte + binary.BigEndian.PutUint16(buf[:], v) + return append(b, buf[:]...) +} + +func appendU32(b []byte, v uint32) []byte { + var buf [4]byte + binary.BigEndian.PutUint32(buf[:], v) + return append(b, buf[:]...) +} + +func appendStr(b []byte, s string) []byte { + b = appendU16(b, uint16(len(s))) + return append(b, s...) +} + +func appendBytes(b []byte, p []byte) []byte { + b = appendU16(b, uint16(len(p))) + return append(b, p...) +} + +func takeU16(t *testing.T, b []byte) (uint16, []byte) { + t.Helper() + if len(b) < 2 { + t.Fatalf("need 2 bytes, have %d", len(b)) + } + return binary.BigEndian.Uint16(b[:2]), b[2:] +} + +func takeU32(t *testing.T, b []byte) (uint32, []byte) { + t.Helper() + if len(b) < 4 { + t.Fatalf("need 4 bytes, have %d", len(b)) + } + return binary.BigEndian.Uint32(b[:4]), b[4:] +} + +func takeStr(t *testing.T, b []byte) (string, []byte) { + t.Helper() + n, rest := takeU16(t, b) + if len(rest) < int(n) { + t.Fatalf("string needs %d bytes, have %d", n, len(rest)) + } + return string(rest[:n]), rest[n:] +} + +func takeBytes(t *testing.T, b []byte) ([]byte, []byte) { + t.Helper() + n, rest := takeU16(t, b) + if len(rest) < int(n) { + t.Fatalf("bytes need %d, have %d", n, len(rest)) + } + return append([]byte(nil), rest[:n]...), rest[n:] +} + +func serverInfo(clientID uint16, nodes, codecs []string) []byte { + b := appendU16(nil, 0) + b = appendU16(b, clientID) + b = appendU16(b, uint16(len(nodes))) + for _, n := range nodes { + b = appendStr(b, n) + } + b = appendU16(b, uint16(len(codecs))) + for _, c := range codecs { + b = appendStr(b, c) + } + return b +} + +func writeConfigFile(t *testing.T, body string) string { + t.Helper() + path := t.TempDir() + "/funknetz.json" + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + return path +} diff --git a/station/session.go b/station/session.go new file mode 100644 index 0000000..dc8e457 --- /dev/null +++ b/station/session.go @@ -0,0 +1,522 @@ +package station + +import ( + "context" + "crypto/hmac" + "crypto/sha1" + "encoding/binary" + "errors" + "net" + "strconv" + "sync" + "time" + + "gitea.neitzel.de/konrad/funknetz/internal/opus" +) + +const noOpus = "The reflector did not offer OPUS." + +var errStopped = errors.New("session stopped") + +// Run keeps the reflector session until ctx is canceled. +func (s *Station) Run(ctx context.Context) error { + for { + if ctx.Err() != nil { + return nil + } + s.mu.Lock() + if s.fatal != "" { + s.mu.Unlock() + <-ctx.Done() + return nil + } + s.state.Connection = "Connecting" + s.publish() + host, port := s.cfg.host, s.cfg.port + s.mu.Unlock() + + conn, err := dialStation(ctx, host, port) + if err != nil { + if ctx.Err() != nil { + return nil + } + s.mu.Lock() + s.state.Connection = "Disconnected" + s.publish() + s.mu.Unlock() + if !s.wait(ctx, 2*time.Second) { + return nil + } + continue + } + + err = s.session(ctx, conn) + conn.Close() + s.mu.Lock() + s.closeUDP() + if s.fatal == "" { + s.state.Connection = "Disconnected" + s.remoteTalker = "" + s.roomBusy = false + } + s.mic = nil + s.play = nil + s.publish() + fatal := s.fatal != "" + s.mu.Unlock() + if ctx.Err() != nil { + return nil + } + if fatal { + <-ctx.Done() + return nil + } + if err != nil && errors.Is(err, errStopped) { + <-ctx.Done() + return nil + } + if !s.wait(ctx, 2*time.Second) { + return nil + } + } +} + +func dialStation(ctx context.Context, host string, port int) (net.Conn, error) { + var d net.Dialer + d.Timeout = 5 * time.Second + return d.DialContext(ctx, "tcp", net.JoinHostPort(host, strconv.Itoa(port))) +} + +func (s *Station) wait(ctx context.Context, d time.Duration) bool { + timer := s.clock.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C(): + return true + } +} + +func (s *Station) session(ctx context.Context, conn net.Conn) error { + done := make(chan struct{}) + var closeOnce sync.Once + closeConn := func() { closeOnce.Do(func() { conn.Close() }) } + defer close(done) + go func() { + select { + case <-ctx.Done(): + closeConn() + case <-done: + } + }() + + tcpCh := make(chan frame, 8) + tcpErr := make(chan error, 1) + go readTCP(conn, tcpCh, tcpErr, done) + + if err := writeFrame(conn, tcpProtoVer, []byte{0, 2, 0, 0}); err != nil { + return err + } + lastTCP := s.clock.Now() + tcpRx := lastTCP.Add(15 * time.Second) + + var udpCh chan []byte + var udpErr chan error + udpReady := false + var udpRx time.Time + + for { + s.mu.Lock() + lastUDP := s.lastUDPSend + s.mu.Unlock() + + next := lastTCP.Add(10 * time.Second) + if tcpRx.Before(next) { + next = tcpRx + } + if udpReady { + if u := lastUDP.Add(15 * time.Second); u.Before(next) { + next = u + } + if udpRx.Before(next) { + next = udpRx + } + } + wait := next.Sub(s.clock.Now()) + if wait < 0 { + wait = 0 + } + timer := s.clock.NewTimer(wait) + + var udpRecv <-chan []byte + var udpFail <-chan error + if udpCh != nil { + udpRecv = udpCh + udpFail = udpErr + } + + var reason string + var tcpf frame + var udpp []byte + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-tcpErr: + timer.Stop() + return errors.New("tcp closed") + case <-udpFail: + timer.Stop() + return errors.New("udp closed") + case tcpf = <-tcpCh: + timer.Stop() + reason = "tcp" + case udpp = <-udpRecv: + timer.Stop() + reason = "udp" + case <-timer.C(): + select { + case tcpf = <-tcpCh: + reason = "tcp" + default: + select { + case udpp = <-udpRecv: + reason = "udp" + default: + reason = "timer" + } + } + } + + now := s.clock.Now() + switch reason { + case "tcp": + tcpRx = now.Add(15 * time.Second) + if err := s.onTCP(conn, tcpf, done, &udpCh, &udpErr, &udpReady, &udpRx, &lastTCP); err != nil { + return err + } + case "udp": + s.mu.Lock() + if clientIDOf(udpp) == s.clientID { + udpRx = now.Add(60 * time.Second) + s.onDatagram(udpp) + } + s.mu.Unlock() + default: + if !now.Before(tcpRx) { + return errors.New("tcp receive timeout") + } + if udpReady && !now.Before(udpRx) { + return errors.New("udp receive timeout") + } + if !now.Before(lastTCP.Add(10 * time.Second)) { + if err := writeFrame(conn, tcpHeartbeat, nil); err != nil { + return err + } + lastTCP = s.clock.Now() + } + s.mu.Lock() + if udpReady && s.udp != nil && !s.clock.Now().Before(s.lastUDPSend.Add(15*time.Second)) { + if err := s.sendUDP(udpHeartbeat, nil); err != nil { + s.mu.Unlock() + return err + } + } + s.mu.Unlock() + } + } +} + +func readTCP(conn net.Conn, out chan<- frame, errc chan<- error, done <-chan struct{}) { + for { + f, err := readFrame(conn) + if err != nil { + select { + case errc <- err: + case <-done: + } + return + } + select { + case out <- f: + case <-done: + return + } + } +} + +func readUDP(conn *net.UDPConn, out chan<- []byte, errc chan<- error, done <-chan struct{}) { + buf := make([]byte, 2048) + for { + n, err := conn.Read(buf) + if err != nil { + select { + case errc <- err: + case <-done: + } + return + } + pkt := append([]byte(nil), buf[:n]...) + select { + case out <- pkt: + case <-done: + return + } + } +} + +func (s *Station) onTCP(conn net.Conn, f frame, done <-chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, udpRx *time.Time, lastTCP *time.Time) error { + switch f.typ { + case tcpHeartbeat, tcpAuthOK: + return nil + case tcpAuthChallenge: + chal, _, err := parseBlob(f.body) + if err != nil || len(chal) != 20 { + return errors.New("bad challenge") + } + mac := hmac.New(sha1.New, []byte(s.cfg.authKey)) + mac.Write(chal) + body := appendString(nil, s.cfg.callsign) + body = appendBlob(body, mac.Sum(nil)) + if err := writeFrame(conn, tcpAuthResponse, body); err != nil { + return err + } + *lastTCP = s.clock.Now() + return nil + case tcpError: + msg, _, err := parseString(f.body) + if err != nil { + msg = "could not be read" + } + s.stop(msg) + return errStopped + case tcpServerInfo: + return s.onServerInfo(conn, f.body, done, udpCh, udpErr, udpReady, udpRx, lastTCP) + case tcpTalkerStart: + s.onTalker(f.body, true) + return nil + case tcpTalkerStop: + s.onTalker(f.body, false) + return nil + default: + return nil + } +} + +func (s *Station) onServerInfo(conn net.Conn, body []byte, done <-chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, udpRx *time.Time, lastTCP *time.Time) error { + id, codecs, err := parseServerInfo(body) + if err != nil { + return err + } + if !hasOPUS(codecs) { + s.stop(noOpus) + return errStopped + } + enc, err := opus.NewEncoder() + if err != nil { + return err + } + dec, err := opus.NewDecoder() + if err != nil { + enc.Close() + return err + } + var tg [4]byte + binary.BigEndian.PutUint32(tg[:], uint32(s.cfg.talkGroup)) + if err := writeFrame(conn, tcpSelectTG, tg[:]); err != nil { + enc.Close() + dec.Close() + return err + } + *lastTCP = s.clock.Now() + + remote, ok := conn.RemoteAddr().(*net.TCPAddr) + if !ok { + enc.Close() + dec.Close() + return errors.New("tcp peer has no address") + } + udpConn, err := net.DialUDP("udp", nil, &net.UDPAddr{IP: remote.IP, Port: remote.Port}) + if err != nil { + enc.Close() + dec.Close() + return err + } + ch := make(chan []byte, 32) + errc := make(chan error, 1) + go readUDP(udpConn, ch, errc, done) + + s.mu.Lock() + s.clientID = id + s.seq = 0 + s.udp = udpConn + s.enc = enc + s.dec = dec + err = s.sendUDP(udpHeartbeat, nil) + if err != nil { + s.mu.Unlock() + enc.Close() + dec.Close() + udpConn.Close() + return err + } + s.state.Connection = "Connected" + s.publish() + s.mu.Unlock() + + *udpCh = ch + *udpErr = errc + *udpReady = true + *udpRx = s.clock.Now().Add(60 * time.Second) + return nil +} + +func (s *Station) stop(msg string) { + s.mu.Lock() + defer s.mu.Unlock() + s.fatal = msg + s.publish() +} + +// Fail stops retries and shows message until the operator quits. +func (s *Station) Fail(message string) { s.stop(message) } + +func (s *Station) onTalker(body []byte, start bool) { + tg, rest, err := parseU32(body) + if err != nil || int(tg) != s.cfg.talkGroup { + return + } + call, _, err := parseString(rest) + if err != nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if start { + if call == s.cfg.callsign { + return + } + was := s.transmitting() + s.remoteTalker = call + s.roomBusy = true + if was && !s.transmitting() { + s.endTransmit() + } + s.publish() + return + } + s.remoteTalker = "" + s.roomBusy = false + s.publish() +} + +func (s *Station) onDatagram(pkt []byte) { + if len(pkt) < 6 { + return + } + typ := binary.BigEndian.Uint16(pkt[:2]) + if typ != udpAudio || s.dec == nil { + return + } + if s.transmitting() { + return + } + packet, _, err := parseBlob(pkt[6:]) + if err != nil { + return + } + pcm, err := s.dec.Decode(packet) + if err != nil { + return + } + s.play = append(s.play, pcm...) + s.roomBusy = true + s.publish() +} + +func clientIDOf(pkt []byte) uint16 { + if len(pkt) < 6 { + return ^uint16(0) + } + return binary.BigEndian.Uint16(pkt[2:4]) +} + +func (s *Station) sendUDP(typ uint16, body []byte) error { + if s.udp == nil { + return errors.New("udp is down") + } + buf := make([]byte, 6+len(body)) + binary.BigEndian.PutUint16(buf[0:2], typ) + binary.BigEndian.PutUint16(buf[2:4], s.clientID) + binary.BigEndian.PutUint16(buf[4:6], s.seq) + s.seq++ + copy(buf[6:], body) + if _, err := s.udp.Write(buf); err != nil { + return err + } + s.lastUDPSend = s.clock.Now() + return nil +} + +func (s *Station) endTransmit() { + if s.enc != nil && s.udp != nil && len(s.mic) > 0 { + frame := make([]int16, opus.FrameSamples) + copy(frame, s.mic) + s.mic = nil + if pkt, err := s.enc.Encode(frame); err == nil { + _ = s.sendUDP(udpAudio, appendBlob(nil, pkt)) + } + } else { + s.mic = nil + } + if s.udp != nil { + _ = s.sendUDP(udpFlush, nil) + } +} + +func (s *Station) closeUDP() { + if s.udp != nil { + s.udp.Close() + s.udp = nil + } + if s.enc != nil { + s.enc.Close() + s.enc = nil + } + if s.dec != nil { + s.dec.Close() + s.dec = nil + } +} + +func (s *Station) transmitting() bool { + if s.fatal != "" || s.state.Connection != "Connected" || s.roomBusy { + return false + } + return s.buttonHeld || s.spaceHeld +} + +func (s *Station) publish() { + s.state.Callsign = s.cfg.callsign + s.state.TalkGroup = s.cfg.talkGroup + if s.fatal != "" { + s.state.Connection = "Disconnected" + s.state.Detail = s.fatal + s.state.Talker = "" + s.state.Transmitting = false + } else { + s.state.Detail = "Idle" + tx := s.transmitting() + s.state.Transmitting = tx + if tx { + s.state.Talker = s.cfg.callsign + } else { + s.state.Talker = s.remoteTalker + } + } + select { + case s.wake <- struct{}{}: + default: + } +} diff --git a/station/session_test.go b/station/session_test.go new file mode 100644 index 0000000..21f7e82 --- /dev/null +++ b/station/session_test.go @@ -0,0 +1,99 @@ +package station_test + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha1" + "testing" + "time" + + "gitea.neitzel.de/konrad/funknetz/station" +) + +func TestLoginAnnouncesProtocolAndJoinsTheTalkGroup(t *testing.T) { + refl := newReflector(t) + path := writeConfigFile(t, configJSON(" 127.0.0.1 ", refl.port, " dl1abc ", "auth key with spaces", 20)) + clock := newFakeClock() + st, err := station.New(station.Options{ConfigPath: path, Clock: clock}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go func() { _ = st.Run(ctx) }() + + got, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connecting" }) + if err != nil { + t.Fatal(err) + } + if got.Callsign != "DL1ABC" || got.TalkGroup != 20 || got.Detail != "Idle" { + t.Fatalf("State = %+v, want DL1ABC talk group 20 Idle", got) + } + + refl.accept(t) + typ, body := refl.readFrame(t) + if typ != 5 || len(body) != 4 || body[0] != 0 || body[1] != 2 || body[2] != 0 || body[3] != 0 { + t.Fatalf("first frame type %d body %v, want protocol 2.0", typ, body) + } + + challenge := make([]byte, 20) + for i := range challenge { + challenge[i] = byte(i + 1) + } + refl.writeFrame(t, 10, appendBytes(nil, challenge)) + + typ, body = refl.readFrame(t) + if typ != 11 { + t.Fatalf("auth response type %d, want 11", typ) + } + call, rest := takeStr(t, body) + digest, rest := takeBytes(t, rest) + if len(rest) != 0 { + t.Fatalf("auth response has %d trailing bytes", len(rest)) + } + if call != "DL1ABC" { + t.Fatalf("callsign = %q, want DL1ABC", call) + } + mac := hmac.New(sha1.New, []byte("auth key with spaces")) + mac.Write(challenge) + if !hmac.Equal(digest, mac.Sum(nil)) { + t.Fatalf("digest = %x, want hmac-sha1 of the challenge under the stored key", digest) + } + if bytes.Contains(body, []byte("auth key with spaces")) { + t.Fatal("auth response contains the auth key") + } + + refl.writeFrame(t, 12, nil) + refl.writeFrame(t, 100, serverInfo(0x1234, []string{"db0abc"}, []string{"GSM", "OPUS"})) + + typ, body = refl.readFrame(t) + if typ != 106 { + t.Fatalf("frame type %d, want select talk group", typ) + } + tg, rest := takeU32(t, body) + if tg != 20 || len(rest) != 0 { + t.Fatalf("select talk group = %d rest %d, want 20", tg, len(rest)) + } + + utyp, id, seq, ubody := refl.readUDP(t) + if utyp != 1 || id != 0x1234 || seq != 0 || len(ubody) != 0 { + t.Fatalf("udp heartbeat type %d id %d seq %d len %d, want type 1 id 0x1234 seq 0", utyp, id, seq, len(ubody)) + } + + got, err = st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connected" }) + if err != nil { + t.Fatal(err) + } + if got.Detail != "Idle" || got.Talker != "" { + t.Fatalf("State = %+v, want Connected Idle", got) + } + + // No other talk group and no station profile: the next TCP message is a heartbeat, not type 107 or 111. + clock.BlockUntilWaiting(t) + clock.Advance(10 * time.Second) + typ, _ = refl.readFrame(t) + if typ != 1 { + t.Fatalf("quiet TCP frame type %d, want heartbeat 1", typ) + } +} diff --git a/station/station.go b/station/station.go new file mode 100644 index 0000000..64f2bcf --- /dev/null +++ b/station/station.go @@ -0,0 +1,184 @@ +// 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" + "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"` + Talker string `json:"talker"` + Transmitting bool `json:"transmitting"` +} + +// Station is the seam tests and the window use. +type Station struct { + cfg loaded + clock Clock + + mu sync.Mutex + state State + wake chan struct{} + buttonHeld bool + spaceHeld bool + fatal string + remoteTalker string + roomBusy bool + mic []int16 + play []int16 + clientID uint16 + seq uint16 + udp *net.UDPConn + lastUDPSend time.Time + enc *opus.Encoder + dec *opus.Decoder +} + +// New loads the config at opt.ConfigPath. A missing file is created mode 0600 +// and New returns an error so the caller can exit before opening a window. +func New(opt Options) (*Station, error) { + cfg, err := loadConfig(opt.ConfigPath) + if err != nil { + return nil, err + } + clock := opt.Clock + if clock == nil { + clock = wallClock{} + } + s := &Station{ + cfg: cfg, + clock: clock, + wake: make(chan struct{}, 1), + state: State{ + Connection: "Disconnected", + Detail: "Idle", + Callsign: cfg.callsign, + TalkGroup: cfg.talkGroup, + }, + } + return s, nil +} + +// State returns the latest window state. +func (s *Station) State() State { + s.mu.Lock() + defer s.mu.Unlock() + return s.state +} + +// 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() + } + 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 + } + _ = 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() + 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 +} diff --git a/wails.json b/wails.json new file mode 100644 index 0000000..e0393db --- /dev/null +++ b/wails.json @@ -0,0 +1,20 @@ +{ + "name": "funknetz", + "outputfilename": "funknetz", + "frontend:dir": "frontend", + "frontend:install": "", + "frontend:build": "", + "wailsjsdir": "frontend", + "version": "2", + "author": { + "name": "funknetz", + "email": "" + }, + "info": { + "companyName": "funknetz", + "productName": "funknetz", + "productVersion": "1.0.0", + "copyright": "funknetz", + "comments": "Headset station for the FM-Funknetz" + } +}