--- /dev/null
+package main
+
+import (
+ "code.octet-stream.net/broadcaster/internal/protocol"
+ "encoding/json"
+ "flag"
+ "fmt"
+ "github.com/gopxl/beep/v2"
+ "github.com/gopxl/beep/v2/mp3"
+ "github.com/gopxl/beep/v2/speaker"
+ "github.com/gopxl/beep/v2/wav"
+ "golang.org/x/net/websocket"
+ "log"
+ "os"
+ "os/signal"
+ "path/filepath"
+ "strings"
+ "syscall"
+ "time"
+)
+
+const version = "v1.0.0"
+const sampleRate = 44100
+
+var config RadioConfig = NewRadioConfig()
+
+func main() {
+ configFlag := flag.String("c", "", "path to configuration file")
+ versionFlag := flag.Bool("v", false, "print version and exit")
+ flag.Parse()
+
+ if *versionFlag {
+ fmt.Println("Broadcaster Radio", version)
+ os.Exit(0)
+ }
+ if *configFlag == "" {
+ log.Fatal("must specify a configuration file with -c")
+ }
+
+ log.Println("Broadcaster Radio", version, "starting up")
+ config.LoadFromFile(*configFlag)
+ statusCollector.Config <- config
+
+ playbackSampleRate := beep.SampleRate(sampleRate)
+ speaker.Init(playbackSampleRate, playbackSampleRate.N(time.Second/10))
+
+ if config.PTTPin != -1 {
+ InitRaspberryPiPTT(config.PTTPin, config.GpioDevice)
+ }
+ if config.COSPin != -1 {
+ InitRaspberryPiCOS(config.COSPin, config.GpioDevice)
+ }
+
+ sig := make(chan os.Signal, 1)
+ signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
+ go func() {
+ sig := <-sig
+ log.Println("Radio shutting down due to signal", sig)
+ // Make sure we always stop PTT when program ends
+ ptt.DisengagePTT()
+ os.Exit(0)
+ }()
+
+ log.Println("Config checks out, radio coming online")
+ log.Println("Audio file cache:", config.CachePath)
+
+ fileSpecChan := make(chan []protocol.FileSpec)
+ go filesWorker(config.CachePath, fileSpecChan)
+
+ stop := make(chan bool)
+ playlistSpecChan := make(chan []protocol.PlaylistSpec)
+ go playlistWorker(playlistSpecChan, stop)
+
+ for {
+ runWebsocket(fileSpecChan, playlistSpecChan, stop)
+ log.Println("Websocket failed, retry in 30 seconds")
+ time.Sleep(time.Second * time.Duration(30))
+ }
+}
+
+func runWebsocket(fileSpecChan chan []protocol.FileSpec, playlistSpecChan chan []protocol.PlaylistSpec, stop chan bool) error {
+ log.Println("Establishing websocket connection to:", config.WebsocketURL())
+ ws, err := websocket.Dial(config.WebsocketURL(), "", config.ServerURL)
+ if err != nil {
+ return err
+ }
+
+ auth := protocol.AuthenticateMessage{
+ T: "authenticate",
+ Token: config.Token,
+ }
+ msg, _ := json.Marshal(auth)
+
+ if _, err := ws.Write(msg); err != nil {
+ log.Fatal(err)
+ }
+ statusCollector.Websocket <- ws
+
+ buf := make([]byte, 16384)
+ badRead := false
+ for {
+ n, err := ws.Read(buf)
+ if err != nil {
+ log.Println("Lost websocket to server")
+ return err
+ }
+ // Ignore any massively oversize messages
+ if n == len(buf) {
+ badRead = true
+ continue
+ } else if badRead {
+ badRead = false
+ continue
+ }
+
+ t, msg, err := protocol.ParseMessage(buf[:n])
+ if err != nil {
+ log.Println("Message parse error", err)
+ return err
+ }
+
+ if t == protocol.FilesType {
+ filesMsg := msg.(protocol.FilesMessage)
+ fileSpecChan <- filesMsg.Files
+ }
+
+ if t == protocol.PlaylistsType {
+ playlistsMsg := msg.(protocol.PlaylistsMessage)
+ playlistSpecChan <- playlistsMsg.Playlists
+ }
+
+ if t == protocol.StopType {
+ stop <- true
+ }
+ }
+}
+
+func filesWorker(cachePath string, ch chan []protocol.FileSpec) {
+ machine := NewFilesMachine(cachePath)
+ isDownloading := false
+ downloadResult := make(chan error)
+ var timer *time.Timer
+
+ for {
+ var timerCh <-chan time.Time = nil
+ if timer != nil {
+ timerCh = timer.C
+ }
+ doNext := false
+ select {
+ case specs := <-ch:
+ log.Println("Received new file specs", specs)
+ machine.UpdateSpecs(specs)
+ doNext = true
+ timer = nil
+ case err := <-downloadResult:
+ isDownloading = false
+ machine.RefreshMissing()
+ if err != nil {
+ log.Println(err)
+ if !machine.IsCacheComplete() {
+ timer = time.NewTimer(30 * time.Second)
+ }
+ } else {
+ if !machine.IsCacheComplete() {
+ timer = time.NewTimer(10 * time.Millisecond)
+ }
+ }
+ case <-timerCh:
+ doNext = true
+ timer = nil
+ }
+
+ if doNext && !isDownloading && !machine.IsCacheComplete() {
+ next := machine.NextFile()
+ isDownloading = true
+ go machine.DownloadSingle(next, downloadResult)
+ }
+ }
+}
+
+func playlistWorker(ch <-chan []protocol.PlaylistSpec, stop <-chan bool) {
+ var specs []protocol.PlaylistSpec
+ isPlaying := false
+ playbackFinished := make(chan error)
+ cancel := make(chan bool)
+ nextId := 0
+ var timer *time.Timer
+
+ for {
+ var timerCh <-chan time.Time = nil
+ if timer != nil {
+ timerCh = timer.C
+ }
+ doNext := false
+ select {
+ case specs = <-ch:
+ log.Println("Received new playlist specs", specs)
+ doNext = true
+ case <-playbackFinished:
+ isPlaying = false
+ doNext = true
+ cancel = make(chan bool)
+ case <-timerCh:
+ timer = nil
+ isPlaying = true
+ for _, v := range specs {
+ if v.Id == nextId {
+ go playPlaylist(v, playbackFinished, cancel)
+ }
+ }
+ case <-stop:
+ if isPlaying {
+ log.Println("Cancelling playlist in progress")
+ cancel <- true
+ }
+ }
+
+ if doNext && !isPlaying {
+ timer = nil
+ found := false
+ loc, err := time.LoadLocation(config.TimeZone)
+ if err != nil {
+ log.Fatal(err)
+ }
+ var soonestTime time.Time
+ for _, v := range specs {
+ t, err := time.ParseInLocation(protocol.StartTimeFormat, v.StartTime, loc)
+ if err != nil {
+ log.Println("Error parsing start time", err)
+ continue
+ }
+ if t.Before(time.Now()) {
+ continue
+ }
+ if !found || t.Before(soonestTime) {
+ soonestTime = t
+ found = true
+ nextId = v.Id
+ }
+ }
+ if found {
+ duration := soonestTime.Sub(time.Now())
+ log.Println("Next playlist will be id", nextId, "in", duration.Seconds(), "seconds")
+ timer = time.NewTimer(duration)
+ } else {
+ log.Println("No future playlists")
+ }
+ }
+ }
+}
+
+func playPlaylist(playlist protocol.PlaylistSpec, playbackFinished chan<- error, cancel <-chan bool) {
+ startTime := time.Now()
+ log.Println("Beginning playback of playlist", playlist.Name)
+entries:
+ for _, p := range playlist.Entries {
+ // delay
+ var duration time.Duration
+ if p.IsRelative {
+ duration = time.Second * time.Duration(p.DelaySeconds)
+ } else {
+ duration = time.Until(startTime.Add(time.Second * time.Duration(p.DelaySeconds)))
+ }
+ statusCollector.PlaylistBeginDelay <- BeginDelayStatus{
+ Playlist: playlist.Name,
+ Seconds: int(duration.Seconds()),
+ Filename: p.Filename,
+ }
+ select {
+ case <-time.After(duration):
+ case <-cancel:
+ break entries
+ }
+
+ statusCollector.PlaylistBeginWaitForChannel <- BeginWaitForChannelStatus{
+ Playlist: playlist.Name,
+ Filename: p.Filename,
+ }
+ cos.WaitForChannelClear()
+
+ // then play
+ statusCollector.PlaylistBeginPlayback <- BeginPlaybackStatus{
+ Playlist: playlist.Name,
+ Filename: p.Filename,
+ }
+ ptt.EngagePTT()
+ f, err := os.Open(filepath.Join(config.CachePath, p.Filename))
+ if err != nil {
+ log.Println("Couldn't open file for playlist", p.Filename)
+ continue
+ }
+ log.Println("Playing file", p.Filename)
+ l := strings.ToLower(p.Filename)
+ var streamer beep.StreamSeekCloser
+ var format beep.Format
+ if strings.HasSuffix(l, ".mp3") {
+ streamer, format, err = mp3.Decode(f)
+ } else if strings.HasSuffix(l, ".wav") {
+ streamer, format, err = wav.Decode(f)
+ } else {
+ log.Println("Unrecognised file extension (.wav and .mp3 supported), moving on")
+ }
+ if err != nil {
+ log.Println("Could not decode media file", err)
+ continue
+ }
+ defer streamer.Close()
+
+ done := make(chan bool)
+
+ if format.SampleRate != sampleRate {
+ resampled := beep.Resample(4, format.SampleRate, sampleRate, streamer)
+ speaker.Play(beep.Seq(resampled, beep.Callback(func() {
+ done <- true
+ })))
+ } else {
+ speaker.Play(beep.Seq(streamer, beep.Callback(func() {
+ done <- true
+ })))
+ }
+
+ select {
+ case <-done:
+ case <-cancel:
+ ptt.DisengagePTT()
+ break entries
+ }
+ ptt.DisengagePTT()
+ }
+ log.Println("Playlist finished", playlist.Name)
+ statusCollector.PlaylistBeginIdle <- true
+ playbackFinished <- nil
+}
projects / broadcaster / blobdiff