diff --git a/cmd/fmrtx/main.go b/cmd/fmrtx/main.go index 3f234d6..cfa71f8 100644 --- a/cmd/fmrtx/main.go +++ b/cmd/fmrtx/main.go @@ -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, + }) +} diff --git a/internal/app/engine.go b/internal/app/engine.go index ae8e176..7384d9a 100644 --- a/internal/app/engine.go +++ b/internal/app/engine.go @@ -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) diff --git a/internal/app/engine_test.go b/internal/app/engine_test.go index b7c7738..95a111b 100644 --- a/internal/app/engine_test.go +++ b/internal/app/engine_test.go @@ -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) + } +} diff --git a/internal/control/control.go b/internal/control/control.go index 2dc4c72..25a5a5e 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -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) } diff --git a/internal/offline/generator.go b/internal/offline/generator.go index 25eebb7..e0513e4 100644 --- a/internal/offline/generator.go +++ b/internal/offline/generator.go @@ -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) } diff --git a/internal/platform/plutosdr/pluto_windows.go b/internal/platform/plutosdr/pluto_windows.go index 7bf5b57..fa57e3f 100644 --- a/internal/platform/plutosdr/pluto_windows.go +++ b/internal/platform/plutosdr/pluto_windows.go @@ -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() diff --git a/internal/platform/soapy.go b/internal/platform/soapy.go index 81e94b6..8fe4845 100644 --- a/internal/platform/soapy.go +++ b/internal/platform/soapy.go @@ -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 +} diff --git a/internal/platform/soapysdr/native.go b/internal/platform/soapysdr/native.go index de60027..4c9d824 100644 --- a/internal/platform/soapysdr/native.go +++ b/internal/platform/soapysdr/native.go @@ -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() diff --git a/internal/rds/encoder.go b/internal/rds/encoder.go index e87b8b0..8abcaf5 100644 --- a/internal/rds/encoder.go +++ b/internal/rds/encoder.go @@ -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 }