fm-rds-tx/cmd/fmrtx/main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	apppkg "github.com/jan/fm-rds-tx/internal/app"
	"github.com/jan/fm-rds-tx/internal/audio"
	cfgpkg "github.com/jan/fm-rds-tx/internal/config"
	ctrlpkg "github.com/jan/fm-rds-tx/internal/control"
	drypkg "github.com/jan/fm-rds-tx/internal/dryrun"
	"github.com/jan/fm-rds-tx/internal/platform"
	"github.com/jan/fm-rds-tx/internal/platform/plutosdr"
	"github.com/jan/fm-rds-tx/internal/platform/soapysdr"
)

func main() {
	configPath := flag.String("config", "", "path to JSON config")
	printConfig := flag.Bool("print-config", false, "print effective config and exit")
	dryRun := flag.Bool("dry-run", false, "run no-hardware dry-run output")
	dryOutput := flag.String("dry-output", "-", "dry-run output path or - for stdout")
	simulate := flag.Bool("simulate-tx", false, "run simulated Soapy/backend transmit path")
	simulateOutput := flag.String("simulate-output", "", "simulated transmit output file")
	simulateDuration := flag.Duration("simulate-duration", 500*time.Millisecond, "simulated transmit duration")
	txMode := flag.Bool("tx", false, "start real TX mode (requires hardware + build tags)")
	txAutoStart := flag.Bool("tx-auto-start", false, "auto-start TX on launch")
	listDevices := flag.Bool("list-devices", false, "enumerate SoapySDR devices and exit")
	audioStdin := flag.Bool("audio-stdin", false, "read S16LE stereo PCM audio from stdin")
	audioRate := flag.Int("audio-rate", 44100, "sample rate of stdin audio input (Hz)")
	audioHTTP := flag.Bool("audio-http", false, "enable HTTP audio ingest via /audio/stream")
	flag.Parse()

	// --- list-devices (SoapySDR) ---
	if *listDevices {
		devices, err := soapysdr.Enumerate()
		if err != nil {
			log.Fatalf("enumerate: %v", err)
		}
		if len(devices) == 0 {
			fmt.Println("no SoapySDR devices found")
			return
		}
		for i, dev := range devices {
			fmt.Printf("device %d:\n", i)
			for k, v := range dev {
				fmt.Printf("  %s = %s\n", k, v)
			}
		}
		return
	}

	cfg, err := cfgpkg.Load(*configPath)
	if err != nil {
		log.Fatalf("load config: %v", err)
	}

	// --- print-config ---
	if *printConfig {
		preemph := "off"
		if cfg.FM.PreEmphasisTauUS > 0 {
			preemph = fmt.Sprintf("%.0fµs", cfg.FM.PreEmphasisTauUS)
		}
		fmt.Printf("backend=%s freq=%.1fMHz stereo=%t rds=%t preemph=%s limiter=%t fmmod=%t deviation=±%.0fHz compositeRate=%dHz deviceRate=%.0fHz listen=%s pluto=%t soapy=%t\n",
			cfg.Backend.Kind, cfg.FM.FrequencyMHz, cfg.FM.StereoEnabled, cfg.RDS.Enabled,
			preemph, cfg.FM.LimiterEnabled, cfg.FM.FMModulationEnabled, cfg.FM.MaxDeviationHz,
			cfg.FM.CompositeRateHz, cfg.EffectiveDeviceRate(), cfg.Control.ListenAddress,
			plutosdr.Available(), soapysdr.Available())
		return
	}

	// --- dry-run ---
	if *dryRun {
		frame := drypkg.Generate(cfg)
		if err := drypkg.WriteJSON(*dryOutput, frame); err != nil {
			log.Fatalf("dry-run: %v", err)
		}
		if *dryOutput != "" && *dryOutput != "-" {
			fmt.Fprintf(os.Stderr, "dry run frame written to %s\n", *dryOutput)
		}
		return
	}

	// --- simulate ---
	if *simulate {
		summary, err := apppkg.RunSimulatedTransmit(cfg, *simulateOutput, *simulateDuration)
		if err != nil {
			log.Fatalf("simulate-tx: %v", err)
		}
		fmt.Println(summary)
		return
	}

	// --- TX mode ---
	if *txMode {
		driver := selectDriver(cfg)
		if driver == nil {
			log.Fatal("no hardware driver available — build with -tags pluto (or -tags soapy)")
		}
		runTXMode(cfg, driver, *txAutoStart, *audioStdin, *audioRate, *audioHTTP)
		return
	}

	// --- default: HTTP only ---
	srv := ctrlpkg.NewServer(cfg)
	log.Printf("fm-rds-tx listening on %s (TX default: off, use --tx for hardware)", cfg.Control.ListenAddress)
	log.Fatal(http.ListenAndServe(cfg.Control.ListenAddress, srv.Handler()))
}

// selectDriver picks the best available driver based on config and build tags.
func selectDriver(cfg cfgpkg.Config) platform.SoapyDriver {
	kind := cfg.Backend.Kind

	// Explicit PlutoSDR
	if kind == "pluto" || kind == "plutosdr" {
		if plutosdr.Available() {
			return plutosdr.NewPlutoDriver()
		}
		log.Printf("warning: backend=%s but pluto driver not available (%s)", kind, plutosdr.AvailableError())
	}

	// Explicit SoapySDR
	if kind == "soapy" || kind == "soapysdr" {
		if soapysdr.Available() {
			return soapysdr.NewNativeDriver()
		}
		log.Printf("warning: backend=%s but soapy driver not available", kind)
	}

	// Auto-detect: prefer PlutoSDR, fall back to SoapySDR
	if plutosdr.Available() {
		log.Println("auto-selected: pluto-iio driver")
		return plutosdr.NewPlutoDriver()
	}
	if soapysdr.Available() {
		log.Println("auto-selected: soapy-native driver")
		return soapysdr.NewNativeDriver()
	}

	return nil
}

func runTXMode(cfg cfgpkg.Config, driver platform.SoapyDriver, autoStart bool, audioStdin bool, audioRate int, audioHTTP bool) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Configure driver
	// OutputDrive controls composite signal level, NOT hardware gain.
	// Hardware TX gain is always 0 dB (max power). Use external attenuator for power control.
	soapyCfg := platform.SoapyConfig{
		Driver:       cfg.Backend.Driver,
		Device:       cfg.Backend.Device,
		CenterFreqHz: cfg.FM.FrequencyMHz * 1e6,
		GainDB:       0, // 0 dB = max TX power on PlutoSDR
		DeviceArgs:   map[string]string{},
	}
	if cfg.Backend.URI != "" {
		soapyCfg.DeviceArgs["uri"] = cfg.Backend.URI
	}
	for k, v := range cfg.Backend.DeviceArgs {
		soapyCfg.DeviceArgs[k] = v
	}
	soapyCfg.SampleRateHz = cfg.EffectiveDeviceRate()

	log.Printf("TX: configuring %s freq=%.3fMHz rate=%.0fHz gain=%.1fdB",
		driver.Name(), cfg.FM.FrequencyMHz, soapyCfg.SampleRateHz, soapyCfg.GainDB)

	if err := driver.Configure(ctx, soapyCfg); err != nil {
		log.Fatalf("configure: %v", err)
	}

	caps, err := driver.Capabilities(ctx)
	if err == nil {
		log.Printf("TX: device caps: gain=%.0f..%.0f dB, rate=%.0f..%.0f Hz",
			caps.GainMinDB, caps.GainMaxDB, caps.MinSampleRate, caps.MaxSampleRate)
	}

	// Engine
	engine := apppkg.NewEngine(cfg, driver)

	// Live audio stream source (optional)
	var streamSrc *audio.StreamSource
	if audioStdin || audioHTTP {
		// Buffer: 2 seconds at input rate — enough to absorb jitter
		bufferFrames := audioRate * 2
		if bufferFrames <= 0 {
			bufferFrames = 1
		}
		streamSrc = audio.NewStreamSource(bufferFrames, audioRate)
		engine.SetStreamSource(streamSrc)

		if audioStdin {
			go func() {
				log.Printf("audio: reading S16LE stereo PCM from stdin at %d Hz", audioRate)
				if err := audio.IngestReader(os.Stdin, streamSrc); err != nil {
					log.Printf("audio: stdin ingest ended: %v", err)
				} else {
					log.Println("audio: stdin EOF")
				}
			}()
		}
		if audioHTTP {
			log.Printf("audio: HTTP ingest enabled on /audio/stream (rate=%dHz, buffer=%d frames)", audioRate, streamSrc.Stats().Capacity)
		}
	}

	// Control plane
	srv := ctrlpkg.NewServer(cfg)
	srv.SetDriver(driver)
	srv.SetTXController(&txBridge{engine: engine})
	if streamSrc != nil {
		srv.SetStreamSource(streamSrc)
	}

	if autoStart {
		log.Println("TX: auto-start enabled")
		if err := engine.Start(ctx); err != nil {
			log.Fatalf("engine start: %v", err)
		}
		log.Printf("TX ACTIVE: freq=%.3fMHz rate=%.0fHz", cfg.FM.FrequencyMHz, cfg.EffectiveDeviceRate())
	} else {
		log.Println("TX ready (idle) — POST /tx/start to begin")
	}

	go func() {
		log.Printf("control plane on %s", cfg.Control.ListenAddress)
		if err := http.ListenAndServe(cfg.Control.ListenAddress, srv.Handler()); err != nil {
			log.Printf("http: %v", err)
		}
	}()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	sig := <-sigCh
	log.Printf("received %s, shutting down...", sig)

	_ = engine.Stop(ctx)
	_ = driver.Close(ctx)
	log.Println("shutdown complete")
}

type txBridge struct{ engine *apppkg.Engine }

func (b *txBridge) StartTX() error { return b.engine.Start(context.Background()) }
func (b *txBridge) StopTX() error  { return b.engine.Stop(context.Background()) }
func (b *txBridge) TXStats() map[string]any {
	s := b.engine.Stats()
	return map[string]any{
		"state":               s.State,
		"chunksProduced":      s.ChunksProduced,
		"totalSamples":        s.TotalSamples,
		"underruns":           s.Underruns,
		"lateBuffers":         s.LateBuffers,
		"lastError":           s.LastError,
		"uptimeSeconds":       s.UptimeSeconds,
		"maxCycleMs":          s.MaxCycleMs,
		"maxGenerateMs":       s.MaxGenerateMs,
		"maxUpsampleMs":       s.MaxUpsampleMs,
		"maxWriteMs":          s.MaxWriteMs,
		"queue":               s.Queue,
		"runtimeIndicator":    s.RuntimeIndicator,
		"runtimeAlert":        s.RuntimeAlert,
		"degradedTransitions": s.DegradedTransitions,
		"mutedTransitions":    s.MutedTransitions,
		"faultedTransitions":  s.FaultedTransitions,
		"faultCount":          s.FaultCount,
		"faultHistory":        s.FaultHistory,
		"transitionHistory":   s.TransitionHistory,
		"lastFault":           s.LastFault,
	}
}
func (b *txBridge) UpdateConfig(lp ctrlpkg.LivePatch) error {
	return b.engine.UpdateConfig(apppkg.LiveConfigUpdate{
		FrequencyMHz:   lp.FrequencyMHz,
		OutputDrive:    lp.OutputDrive,
		StereoEnabled:  lp.StereoEnabled,
		PilotLevel:     lp.PilotLevel,
		RDSInjection:   lp.RDSInjection,
		RDSEnabled:     lp.RDSEnabled,
		LimiterEnabled: lp.LimiterEnabled,
		LimiterCeiling: lp.LimiterCeiling,
		PS:             lp.PS,
		RadioText:      lp.RadioText,
	})
}

func (b *txBridge) ResetFault() error {
	return b.engine.ResetFault()
}