All major TX parameters are now hot-swappable during transmission: - DSP params (drive, stereo, pilot, RDS levels, limiter) via atomic.Pointer[LiveParams], loaded once per chunk (~50ms) - RDS text (PS, RT) via atomic.Value in encoder, applied at RDS group boundaries (~88ms) - TX frequency via driver.Tune(), applied between chunks Zero locks in DSP path. HTTP handler writes atomics, run loop reads them. Validation happens before store. New: SoapyDriver.Tune() for live frequency changes. New: LiveConfigUpdate/LivePatch types for type-safe patching. New: 4 engine tests for live DSP/freq/RDS updates + validation.tags/v0.9.0
| @@ -216,3 +216,17 @@ func (b *txBridge) TXStats() map[string]any { | |||
| "lastError": s.LastError, "uptimeSeconds": s.UptimeSeconds, | |||
| } | |||
| } | |||
| 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, | |||
| }) | |||
| } | |||
| @@ -67,6 +67,9 @@ type Engine struct { | |||
| totalSamples atomic.Uint64 | |||
| underruns atomic.Uint64 | |||
| lastError atomic.Value // string | |||
| // Live config: pending frequency change, applied between chunks | |||
| pendingFreq atomic.Pointer[float64] | |||
| } | |||
| func NewEngine(cfg cfgpkg.Config, driver platform.SoapyDriver) *Engine { | |||
| @@ -115,6 +118,86 @@ func (e *Engine) SetChunkDuration(d time.Duration) { | |||
| e.chunkDuration = d | |||
| } | |||
| // LiveConfigUpdate carries hot-reloadable parameters from the control API. | |||
| // nil pointers mean "no change". Validated before applying. | |||
| type LiveConfigUpdate struct { | |||
| FrequencyMHz *float64 | |||
| OutputDrive *float64 | |||
| StereoEnabled *bool | |||
| PilotLevel *float64 | |||
| RDSInjection *float64 | |||
| RDSEnabled *bool | |||
| LimiterEnabled *bool | |||
| LimiterCeiling *float64 | |||
| PS *string | |||
| RadioText *string | |||
| } | |||
| // UpdateConfig applies live parameter changes without restarting the engine. | |||
| // DSP params take effect at the next chunk boundary (~50ms max). | |||
| // Frequency changes are applied between chunks via driver.Tune(). | |||
| // RDS text updates are applied at the next RDS group boundary (~88ms). | |||
| func (e *Engine) UpdateConfig(u LiveConfigUpdate) error { | |||
| // --- Validate --- | |||
| if u.FrequencyMHz != nil { | |||
| if *u.FrequencyMHz < 65 || *u.FrequencyMHz > 110 { | |||
| return fmt.Errorf("frequencyMHz out of range (65-110)") | |||
| } | |||
| } | |||
| if u.OutputDrive != nil { | |||
| if *u.OutputDrive < 0 || *u.OutputDrive > 3 { | |||
| return fmt.Errorf("outputDrive out of range (0-3)") | |||
| } | |||
| } | |||
| if u.PilotLevel != nil { | |||
| if *u.PilotLevel < 0 || *u.PilotLevel > 0.2 { | |||
| return fmt.Errorf("pilotLevel out of range (0-0.2)") | |||
| } | |||
| } | |||
| if u.RDSInjection != nil { | |||
| if *u.RDSInjection < 0 || *u.RDSInjection > 0.15 { | |||
| return fmt.Errorf("rdsInjection out of range (0-0.15)") | |||
| } | |||
| } | |||
| if u.LimiterCeiling != nil { | |||
| if *u.LimiterCeiling < 0 || *u.LimiterCeiling > 2 { | |||
| return fmt.Errorf("limiterCeiling out of range (0-2)") | |||
| } | |||
| } | |||
| // --- Frequency: store for run loop to apply via driver.Tune() --- | |||
| if u.FrequencyMHz != nil { | |||
| freqHz := *u.FrequencyMHz * 1e6 | |||
| e.pendingFreq.Store(&freqHz) | |||
| } | |||
| // --- RDS text: forward to encoder atomics --- | |||
| if u.PS != nil || u.RadioText != nil { | |||
| if enc := e.generator.RDSEncoder(); enc != nil { | |||
| ps, rt := "", "" | |||
| if u.PS != nil { ps = *u.PS } | |||
| if u.RadioText != nil { rt = *u.RadioText } | |||
| enc.UpdateText(ps, rt) | |||
| } | |||
| } | |||
| // --- DSP params: build new LiveParams from current + patch --- | |||
| // Read current, apply deltas, store new | |||
| current := e.generator.CurrentLiveParams() | |||
| next := current // copy | |||
| if u.OutputDrive != nil { next.OutputDrive = *u.OutputDrive } | |||
| if u.StereoEnabled != nil { next.StereoEnabled = *u.StereoEnabled } | |||
| if u.PilotLevel != nil { next.PilotLevel = *u.PilotLevel } | |||
| if u.RDSInjection != nil { next.RDSInjection = *u.RDSInjection } | |||
| if u.RDSEnabled != nil { next.RDSEnabled = *u.RDSEnabled } | |||
| if u.LimiterEnabled != nil { next.LimiterEnabled = *u.LimiterEnabled } | |||
| if u.LimiterCeiling != nil { next.LimiterCeiling = *u.LimiterCeiling } | |||
| e.generator.UpdateLive(next) | |||
| return nil | |||
| } | |||
| func (e *Engine) Start(ctx context.Context) error { | |||
| e.mu.Lock() | |||
| if e.state != EngineIdle { | |||
| @@ -192,6 +275,16 @@ func (e *Engine) run(ctx context.Context) { | |||
| if ctx.Err() != nil { | |||
| return | |||
| } | |||
| // Apply pending frequency change between chunks | |||
| if pf := e.pendingFreq.Swap(nil); pf != nil { | |||
| if err := e.driver.Tune(ctx, *pf); err != nil { | |||
| e.lastError.Store(fmt.Sprintf("tune: %v", err)) | |||
| } else { | |||
| log.Printf("engine: tuned to %.3f MHz", *pf/1e6) | |||
| } | |||
| } | |||
| frame := e.generator.GenerateFrame(e.chunkDuration) | |||
| if e.upsampler != nil { | |||
| frame = e.upsampler.Process(frame) | |||
| @@ -131,3 +131,121 @@ func TestEngineSameRate(t *testing.T) { | |||
| t.Fatal("expected same-rate mode (upsampler == nil)") | |||
| } | |||
| } | |||
| func TestEngineLiveUpdateDSP(t *testing.T) { | |||
| cfg := cfgpkg.Default() | |||
| driver := platform.NewSimulatedDriver(nil) | |||
| eng := NewEngine(cfg, driver) | |||
| eng.SetChunkDuration(10 * time.Millisecond) | |||
| ctx := context.Background() | |||
| if err := eng.Start(ctx); err != nil { | |||
| t.Fatalf("start: %v", err) | |||
| } | |||
| defer eng.Stop(ctx) | |||
| time.Sleep(50 * time.Millisecond) | |||
| // Update DSP params while running | |||
| drive := 1.5 | |||
| stereo := false | |||
| err := eng.UpdateConfig(LiveConfigUpdate{ | |||
| OutputDrive: &drive, | |||
| StereoEnabled: &stereo, | |||
| }) | |||
| if err != nil { | |||
| t.Fatalf("UpdateConfig: %v", err) | |||
| } | |||
| // Engine should still be running after update | |||
| time.Sleep(50 * time.Millisecond) | |||
| stats := eng.Stats() | |||
| if stats.State != "running" { | |||
| t.Fatalf("expected running after update, got %s", stats.State) | |||
| } | |||
| if stats.Underruns > 0 { | |||
| t.Fatalf("unexpected underruns after update: %d", stats.Underruns) | |||
| } | |||
| } | |||
| func TestEngineLiveUpdateFrequency(t *testing.T) { | |||
| cfg := cfgpkg.Default() | |||
| driver := platform.NewSimulatedDriver(nil) | |||
| eng := NewEngine(cfg, driver) | |||
| eng.SetChunkDuration(10 * time.Millisecond) | |||
| ctx := context.Background() | |||
| if err := eng.Start(ctx); err != nil { | |||
| t.Fatalf("start: %v", err) | |||
| } | |||
| defer eng.Stop(ctx) | |||
| time.Sleep(50 * time.Millisecond) | |||
| // Tune frequency | |||
| freq := 99.5 | |||
| err := eng.UpdateConfig(LiveConfigUpdate{FrequencyMHz: &freq}) | |||
| if err != nil { | |||
| t.Fatalf("UpdateConfig freq: %v", err) | |||
| } | |||
| // Let it process for a bit so the pending freq gets applied | |||
| time.Sleep(50 * time.Millisecond) | |||
| stats := eng.Stats() | |||
| if stats.State != "running" { | |||
| t.Fatalf("expected running after tune, got %s", stats.State) | |||
| } | |||
| } | |||
| func TestEngineLiveUpdateRDS(t *testing.T) { | |||
| cfg := cfgpkg.Default() | |||
| driver := platform.NewSimulatedDriver(nil) | |||
| eng := NewEngine(cfg, driver) | |||
| eng.SetChunkDuration(10 * time.Millisecond) | |||
| ctx := context.Background() | |||
| if err := eng.Start(ctx); err != nil { | |||
| t.Fatalf("start: %v", err) | |||
| } | |||
| defer eng.Stop(ctx) | |||
| time.Sleep(50 * time.Millisecond) | |||
| // Update RDS text | |||
| ps := "NEWPS" | |||
| rt := "Now playing: test track" | |||
| err := eng.UpdateConfig(LiveConfigUpdate{PS: &ps, RadioText: &rt}) | |||
| if err != nil { | |||
| t.Fatalf("UpdateConfig RDS: %v", err) | |||
| } | |||
| time.Sleep(50 * time.Millisecond) | |||
| stats := eng.Stats() | |||
| if stats.Underruns > 0 { | |||
| t.Fatalf("underruns after RDS update: %d", stats.Underruns) | |||
| } | |||
| } | |||
| func TestEngineLiveUpdateValidation(t *testing.T) { | |||
| cfg := cfgpkg.Default() | |||
| driver := platform.NewSimulatedDriver(nil) | |||
| eng := NewEngine(cfg, driver) | |||
| // Out of range frequency | |||
| badFreq := 200.0 | |||
| if err := eng.UpdateConfig(LiveConfigUpdate{FrequencyMHz: &badFreq}); err == nil { | |||
| t.Fatal("expected validation error for bad frequency") | |||
| } | |||
| // Out of range drive | |||
| badDrive := 10.0 | |||
| if err := eng.UpdateConfig(LiveConfigUpdate{OutputDrive: &badDrive}); err == nil { | |||
| t.Fatal("expected validation error for bad drive") | |||
| } | |||
| // Valid update should succeed | |||
| goodDrive := 1.0 | |||
| if err := eng.UpdateConfig(LiveConfigUpdate{OutputDrive: &goodDrive}); err != nil { | |||
| t.Fatalf("expected valid update to succeed: %v", err) | |||
| } | |||
| } | |||
| @@ -10,11 +10,28 @@ import ( | |||
| "github.com/jan/fm-rds-tx/internal/platform" | |||
| ) | |||
| // TXController is an optional interface the Server uses to start/stop TX. | |||
| // TXController is an optional interface the Server uses to start/stop TX | |||
| // and apply live config changes. | |||
| type TXController interface { | |||
| StartTX() error | |||
| StopTX() error | |||
| TXStats() map[string]any | |||
| UpdateConfig(patch LivePatch) error | |||
| } | |||
| // LivePatch mirrors the patchable fields from ConfigPatch for the engine. | |||
| // nil = no change. | |||
| type LivePatch struct { | |||
| FrequencyMHz *float64 | |||
| OutputDrive *float64 | |||
| StereoEnabled *bool | |||
| PilotLevel *float64 | |||
| RDSInjection *float64 | |||
| RDSEnabled *bool | |||
| LimiterEnabled *bool | |||
| LimiterCeiling *float64 | |||
| PS *string | |||
| RadioText *string | |||
| } | |||
| type Server struct { | |||
| @@ -27,6 +44,10 @@ type Server struct { | |||
| type ConfigPatch struct { | |||
| FrequencyMHz *float64 `json:"frequencyMHz,omitempty"` | |||
| OutputDrive *float64 `json:"outputDrive,omitempty"` | |||
| StereoEnabled *bool `json:"stereoEnabled,omitempty"` | |||
| PilotLevel *float64 `json:"pilotLevel,omitempty"` | |||
| RDSInjection *float64 `json:"rdsInjection,omitempty"` | |||
| RDSEnabled *bool `json:"rdsEnabled,omitempty"` | |||
| ToneLeftHz *float64 `json:"toneLeftHz,omitempty"` | |||
| ToneRightHz *float64 `json:"toneRightHz,omitempty"` | |||
| ToneAmplitude *float64 `json:"toneAmplitude,omitempty"` | |||
| @@ -162,58 +183,60 @@ func (s *Server) handleConfig(w http.ResponseWriter, r *http.Request) { | |||
| w.Header().Set("Content-Type", "application/json") | |||
| _ = json.NewEncoder(w).Encode(cfg) | |||
| case http.MethodPost: | |||
| // TODO: config changes only update the control server's copy. | |||
| // The running Engine/Generator holds its own snapshot and won't | |||
| // pick up these changes until restarted. Wire up a hot-reload | |||
| // path or document this limitation clearly for operators. | |||
| var patch ConfigPatch | |||
| if err := json.NewDecoder(r.Body).Decode(&patch); err != nil { | |||
| http.Error(w, err.Error(), http.StatusBadRequest) | |||
| return | |||
| } | |||
| // Update the server's config snapshot (for GET /config and /status) | |||
| s.mu.Lock() | |||
| next := s.cfg | |||
| if patch.FrequencyMHz != nil { | |||
| next.FM.FrequencyMHz = *patch.FrequencyMHz | |||
| } | |||
| if patch.OutputDrive != nil { | |||
| next.FM.OutputDrive = *patch.OutputDrive | |||
| } | |||
| if patch.ToneLeftHz != nil { | |||
| next.Audio.ToneLeftHz = *patch.ToneLeftHz | |||
| } | |||
| if patch.ToneRightHz != nil { | |||
| next.Audio.ToneRightHz = *patch.ToneRightHz | |||
| } | |||
| if patch.ToneAmplitude != nil { | |||
| next.Audio.ToneAmplitude = *patch.ToneAmplitude | |||
| } | |||
| if patch.PS != nil { | |||
| next.RDS.PS = *patch.PS | |||
| } | |||
| if patch.RadioText != nil { | |||
| next.RDS.RadioText = *patch.RadioText | |||
| } | |||
| if patch.PreEmphasisTauUS != nil { | |||
| next.FM.PreEmphasisTauUS = *patch.PreEmphasisTauUS | |||
| } | |||
| if patch.LimiterEnabled != nil { | |||
| next.FM.LimiterEnabled = *patch.LimiterEnabled | |||
| } | |||
| if patch.LimiterCeiling != nil { | |||
| next.FM.LimiterCeiling = *patch.LimiterCeiling | |||
| } | |||
| if patch.FrequencyMHz != nil { next.FM.FrequencyMHz = *patch.FrequencyMHz } | |||
| if patch.OutputDrive != nil { next.FM.OutputDrive = *patch.OutputDrive } | |||
| if patch.ToneLeftHz != nil { next.Audio.ToneLeftHz = *patch.ToneLeftHz } | |||
| if patch.ToneRightHz != nil { next.Audio.ToneRightHz = *patch.ToneRightHz } | |||
| if patch.ToneAmplitude != nil { next.Audio.ToneAmplitude = *patch.ToneAmplitude } | |||
| if patch.PS != nil { next.RDS.PS = *patch.PS } | |||
| if patch.RadioText != nil { next.RDS.RadioText = *patch.RadioText } | |||
| if patch.PreEmphasisTauUS != nil { next.FM.PreEmphasisTauUS = *patch.PreEmphasisTauUS } | |||
| if patch.StereoEnabled != nil { next.FM.StereoEnabled = *patch.StereoEnabled } | |||
| if patch.LimiterEnabled != nil { next.FM.LimiterEnabled = *patch.LimiterEnabled } | |||
| if patch.LimiterCeiling != nil { next.FM.LimiterCeiling = *patch.LimiterCeiling } | |||
| if patch.RDSEnabled != nil { next.RDS.Enabled = *patch.RDSEnabled } | |||
| if patch.PilotLevel != nil { next.FM.PilotLevel = *patch.PilotLevel } | |||
| if patch.RDSInjection != nil { next.FM.RDSInjection = *patch.RDSInjection } | |||
| if err := next.Validate(); err != nil { | |||
| s.mu.Unlock() | |||
| http.Error(w, err.Error(), http.StatusBadRequest) | |||
| return | |||
| } | |||
| s.cfg = next | |||
| tx := s.tx | |||
| s.mu.Unlock() | |||
| // Forward live-patchable params to running engine (if active) | |||
| if tx != nil { | |||
| lp := LivePatch{ | |||
| FrequencyMHz: patch.FrequencyMHz, | |||
| OutputDrive: patch.OutputDrive, | |||
| StereoEnabled: patch.StereoEnabled, | |||
| PilotLevel: patch.PilotLevel, | |||
| RDSInjection: patch.RDSInjection, | |||
| RDSEnabled: patch.RDSEnabled, | |||
| LimiterEnabled: patch.LimiterEnabled, | |||
| LimiterCeiling: patch.LimiterCeiling, | |||
| PS: patch.PS, | |||
| RadioText: patch.RadioText, | |||
| } | |||
| if err := tx.UpdateConfig(lp); err != nil { | |||
| http.Error(w, err.Error(), http.StatusBadRequest) | |||
| return | |||
| } | |||
| } | |||
| w.Header().Set("Content-Type", "application/json") | |||
| _ = json.NewEncoder(w).Encode(map[string]any{"ok": true}) | |||
| _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "live": tx != nil}) | |||
| default: | |||
| http.Error(w, "method not allowed", http.StatusMethodNotAllowed) | |||
| } | |||
| @@ -5,6 +5,7 @@ import ( | |||
| "encoding/binary" | |||
| "fmt" | |||
| "path/filepath" | |||
| "sync/atomic" | |||
| "time" | |||
| "github.com/jan/fm-rds-tx/internal/audio" | |||
| @@ -20,6 +21,18 @@ type frameSource interface { | |||
| NextFrame() audio.Frame | |||
| } | |||
| // LiveParams carries DSP parameters that can be hot-swapped at runtime. | |||
| // Loaded once per chunk via atomic pointer — zero per-sample overhead. | |||
| type LiveParams struct { | |||
| OutputDrive float64 | |||
| StereoEnabled bool | |||
| PilotLevel float64 | |||
| RDSInjection float64 | |||
| RDSEnabled bool | |||
| LimiterEnabled bool | |||
| LimiterCeiling float64 | |||
| } | |||
| // PreEmphasizedSource wraps an audio source and applies pre-emphasis. | |||
| // The source is expected to already output at composite rate (resampled | |||
| // upstream). Pre-emphasis is applied per-sample at that rate. | |||
| @@ -71,17 +84,38 @@ type Generator struct { | |||
| frameSeq uint64 | |||
| // Pre-allocated frame buffer — reused every GenerateFrame call. | |||
| // Safe because driver.Write() is blocking: it returns only after | |||
| // the hardware has consumed the data. Do NOT hold references to | |||
| // frame.Samples beyond Write's return. | |||
| frameBuf *output.CompositeFrame | |||
| bufCap int | |||
| // Live-updatable DSP parameters — written by control API, read per chunk. | |||
| liveParams atomic.Pointer[LiveParams] | |||
| } | |||
| func NewGenerator(cfg cfgpkg.Config) *Generator { | |||
| return &Generator{cfg: cfg} | |||
| } | |||
| // UpdateLive hot-swaps DSP parameters. Thread-safe — called from control API, | |||
| // applied at the next chunk boundary by the DSP goroutine. | |||
| func (g *Generator) UpdateLive(p LiveParams) { | |||
| g.liveParams.Store(&p) | |||
| } | |||
| // CurrentLiveParams returns the current live parameter snapshot. | |||
| // Used by Engine.UpdateConfig to do read-modify-write on the params. | |||
| func (g *Generator) CurrentLiveParams() LiveParams { | |||
| if lp := g.liveParams.Load(); lp != nil { | |||
| return *lp | |||
| } | |||
| return LiveParams{OutputDrive: 1.0, LimiterCeiling: 1.0} | |||
| } | |||
| // RDSEncoder returns the live RDS encoder, or nil if RDS is disabled. | |||
| // Used by the Engine to forward text updates. | |||
| func (g *Generator) RDSEncoder() *rds.Encoder { | |||
| return g.rdsEnc | |||
| } | |||
| func (g *Generator) init() { | |||
| if g.initialized { | |||
| return | |||
| @@ -113,6 +147,18 @@ func (g *Generator) init() { | |||
| g.fmMod = dsp.NewFMModulator(g.sampleRate) | |||
| if g.cfg.FM.MaxDeviationHz > 0 { g.fmMod.MaxDeviation = g.cfg.FM.MaxDeviationHz } | |||
| } | |||
| // Seed initial live params from config | |||
| g.liveParams.Store(&LiveParams{ | |||
| OutputDrive: g.cfg.FM.OutputDrive, | |||
| StereoEnabled: g.cfg.FM.StereoEnabled, | |||
| PilotLevel: g.cfg.FM.PilotLevel, | |||
| RDSInjection: g.cfg.FM.RDSInjection, | |||
| RDSEnabled: g.cfg.RDS.Enabled, | |||
| LimiterEnabled: g.cfg.FM.LimiterEnabled, | |||
| LimiterCeiling: ceiling, | |||
| }) | |||
| g.initialized = true | |||
| } | |||
| @@ -146,27 +192,38 @@ func (g *Generator) GenerateFrame(duration time.Duration) *output.CompositeFrame | |||
| g.frameSeq++ | |||
| frame.Sequence = g.frameSeq | |||
| ceiling := g.cfg.FM.LimiterCeiling | |||
| // Load live params once per chunk — single atomic read, zero per-sample cost | |||
| lp := g.liveParams.Load() | |||
| if lp == nil { | |||
| // Fallback: should never happen after init(), but be safe | |||
| lp = &LiveParams{OutputDrive: 1.0, LimiterCeiling: 1.0} | |||
| } | |||
| // Apply live combiner gains | |||
| g.combiner.PilotGain = lp.PilotLevel | |||
| g.combiner.RDSGain = lp.RDSInjection | |||
| ceiling := lp.LimiterCeiling | |||
| if ceiling <= 0 { ceiling = 1.0 } | |||
| for i := 0; i < samples; i++ { | |||
| in := g.source.NextFrame() | |||
| comps := g.stereoEncoder.Encode(in) | |||
| if !g.cfg.FM.StereoEnabled { | |||
| if !lp.StereoEnabled { | |||
| comps.Stereo = 0; comps.Pilot = 0 | |||
| } | |||
| rdsValue := 0.0 | |||
| if g.rdsEnc != nil { | |||
| if g.rdsEnc != nil && lp.RDSEnabled { | |||
| rdsCarrier := g.stereoEncoder.RDSCarrier() | |||
| rdsValue = g.rdsEnc.NextSampleWithCarrier(rdsCarrier) | |||
| } | |||
| composite := g.combiner.Combine(comps.Mono, comps.Stereo, comps.Pilot, rdsValue) | |||
| composite *= g.cfg.FM.OutputDrive | |||
| composite *= lp.OutputDrive | |||
| if g.limiter != nil { | |||
| if lp.LimiterEnabled && g.limiter != nil { | |||
| composite = g.limiter.Process(composite) | |||
| composite = dsp.HardClip(composite, ceiling) | |||
| } | |||
| @@ -368,6 +368,20 @@ func (d *PlutoDriver) Stop(_ context.Context) error { | |||
| func (d *PlutoDriver) Flush(_ context.Context) error { return nil } | |||
| func (d *PlutoDriver) Tune(_ context.Context, freqHz float64) error { | |||
| d.mu.Lock() | |||
| defer d.mu.Unlock() | |||
| if d.phyDev == 0 { | |||
| return fmt.Errorf("pluto: not configured") | |||
| } | |||
| phyChanLO := d.findChannel(d.phyDev, "altvoltage1", true) | |||
| if phyChanLO == 0 { | |||
| return fmt.Errorf("pluto: TX LO channel not found") | |||
| } | |||
| d.writeChanAttrLL(phyChanLO, "frequency", int64(freqHz)) | |||
| return nil | |||
| } | |||
| func (d *PlutoDriver) Close(_ context.Context) error { | |||
| d.mu.Lock() | |||
| defer d.mu.Unlock() | |||
| @@ -68,6 +68,8 @@ type SoapyDriver interface { | |||
| Flush(ctx context.Context) error | |||
| Close(ctx context.Context) error | |||
| Stats() RuntimeStats | |||
| // Tune changes the TX center frequency while streaming. Thread-safe. | |||
| Tune(ctx context.Context, freqHz float64) error | |||
| } | |||
| // ----------------------------------------------------------------------- | |||
| @@ -220,3 +222,10 @@ func (sd *SimulatedDriver) Stats() RuntimeStats { | |||
| EffectiveRate: sd.cfg.SampleRateHz, | |||
| } | |||
| } | |||
| func (sd *SimulatedDriver) Tune(_ context.Context, freqHz float64) error { | |||
| sd.mu.Lock() | |||
| sd.cfg.CenterFreqHz = freqHz | |||
| sd.mu.Unlock() | |||
| return nil | |||
| } | |||
| @@ -209,6 +209,15 @@ func (d *nativeDriver) Stop(_ context.Context) error { | |||
| func (d *nativeDriver) Flush(_ context.Context) error { return nil } | |||
| func (d *nativeDriver) Tune(_ context.Context, freqHz float64) error { | |||
| d.mu.Lock() | |||
| defer d.mu.Unlock() | |||
| if d.dev == 0 || d.lib == nil { | |||
| return fmt.Errorf("soapy: not configured") | |||
| } | |||
| return d.lib.setFrequency(d.dev, dirTX, 0, freqHz) | |||
| } | |||
| func (d *nativeDriver) Close(_ context.Context) error { | |||
| d.mu.Lock() | |||
| defer d.mu.Unlock() | |||
| @@ -1,6 +1,9 @@ | |||
| package rds | |||
| import "math" | |||
| import ( | |||
| "math" | |||
| "sync/atomic" | |||
| ) | |||
| // RDS encoder — port of PiFmRds, adapted for arbitrary sample rates. | |||
| // At 228 kHz: uses exact {0,+1,0,-1} carrier (identical to PiFmRds). | |||
| @@ -88,6 +91,11 @@ type Encoder struct { | |||
| carrierStep float64 | |||
| SampleRate float64 | |||
| // Live-updatable text — written by control API, read at group boundaries. | |||
| // Zero-contention: atomic swap, checked once per RDS group (~88ms at 228kHz). | |||
| livePS atomic.Value // string | |||
| liveRT atomic.Value // string | |||
| } | |||
| func NewEncoder(cfg RDSConfig) (*Encoder, error) { | |||
| @@ -141,6 +149,18 @@ func (e *Encoder) Reset() { | |||
| for i := range e.ring { e.ring[i] = 0 } | |||
| } | |||
| // UpdateText hot-swaps PS and/or RT. Thread-safe — called from HTTP handlers, | |||
| // applied at the next RDS group boundary by the DSP goroutine. | |||
| // Pass empty string to leave a field unchanged. | |||
| func (e *Encoder) UpdateText(ps, rt string) { | |||
| if ps != "" { | |||
| e.livePS.Store(normalizePS(ps)) | |||
| } | |||
| if rt != "" { | |||
| e.liveRT.Store(normalizeRT(rt)) | |||
| } | |||
| } | |||
| // NextSample returns the next RDS subcarrier sample at the configured rate. | |||
| // Uses the internal free-running 57 kHz carrier. Prefer NextSampleWithCarrier | |||
| // for phase-locked operation in a stereo MPX chain. | |||
| @@ -157,6 +177,16 @@ func (e *Encoder) NextSample() float64 { | |||
| func (e *Encoder) NextSampleWithCarrier(carrier float64) float64 { | |||
| if e.sampleCount >= e.spb { | |||
| if e.bitPos >= bitsPerGroup { | |||
| // Apply live text updates at group boundaries (~88ms at 228kHz). | |||
| // This is the only place we read the atomics — zero per-sample overhead. | |||
| if ps, ok := e.livePS.Load().(string); ok && ps != "" { | |||
| e.scheduler.cfg.PS = ps | |||
| } | |||
| if rt, ok := e.liveRT.Load().(string); ok && rt != "" { | |||
| e.scheduler.cfg.RT = rt | |||
| e.scheduler.rtIdx = 0 // restart RT transmission for new text | |||
| e.scheduler.rtABFlag = !e.scheduler.rtABFlag // toggle A/B per RDS spec | |||
| } | |||
| e.getRDSGroup() | |||
| e.bitPos = 0 | |||
| } | |||