soapysdroutput: fix SoapyUHD TX signal path and MCR pinning

Fix multiple issues preventing SoapySDR output from producing RF:

- handleInputMessages() never called in DSP engine thread (Qt signal
  lost without event loop) - scheduled properly
- setGain() moved to start() post-activation (pre-activation gain
  silently fails on SoapyUHD)
- fullScale threshold corrected for CS16 format detection
- Timed first write pattern matching gr4-lora SoapySink

MCR pinning for SoapyUHD: inject auto_tick_rate=0 in
DeviceSoapySDR device-open path to prevent UHD from re-deriving the
master clock rate on set_tx_rate(), which breaks the decimator chain
and produces no RF. Also reads SDRANGEL_USRP_MASTER_CLOCK_RATE_HZ env
var for optional MCR override.

TX diagnostic counters: SoapySDROutputThread tracks packets,
underflows, errors. Exposed via REST as streamSettingsArgs
key-value pairs.

devices/soapysdr/devicesoapysdr.cpp

@@ -76,6 +76,33 @@ SoapySDR::Device *DeviceSoapySDR::openopenSoapySDRFromSequence(uint32_t sequence
                 kwargs[deviceEnum.m_idKey.toStdString()] = deviceEnum.m_idValue.toStdString();
             }
 
+            // Optional master_clock_rate override via environment.
+            // Device-agnostic: applies to any SoapySDR device that honors the
+            // master_clock_rate device-arg (SoapyUHD for B200/B210, etc.).
+            // Empty/unset = SoapyUHD picks MCR via auto_tick_rate (default).
+            // Used by headless harnesses to pin MCR for clean-decim TX/RX rates.
+            if (const char *mcr_env = std::getenv("SDRANGEL_USRP_MASTER_CLOCK_RATE_HZ"))
+            {
+                if (mcr_env[0] != '\0' && kwargs.find("master_clock_rate") == kwargs.end())
+                {
+                    kwargs["master_clock_rate"] = mcr_env;
+                    qDebug("DeviceSoapySDR::openopenSoapySDRFromSequence:"
+                           " SDRANGEL_USRP_MASTER_CLOCK_RATE_HZ=%s", mcr_env);
+                }
+            }
+
+            // Disable auto_tick_rate for SoapyUHD so that the MCR pinned above
+            // (or via device args) is preserved across setSampleRate() calls.
+            // Without this, UHD re-derives the MCR on setSampleRate/set_tx_rate
+            // even when MCR was already set, breaking the rate decimator chain.
+            if (deviceEnum.m_driverName == "uhd"
+                && kwargs.find("auto_tick_rate") == kwargs.end())
+            {
+                kwargs["auto_tick_rate"] = "0";
+                qDebug("DeviceSoapySDR::openopenSoapySDRFromSequence:"
+                       " forced auto_tick_rate=0 for SoapyUHD");
+            }
+
             SoapySDR::Kwargs::const_iterator it = kwargs.begin();
 
             for (; it != kwargs.end(); ++it) {

plugins/samplesink/soapysdroutput/soapysdroutput.cpp

@@ -39,6 +39,8 @@
 MESSAGE_CLASS_DEFINITION(SoapySDROutput::MsgConfigureSoapySDROutput, Message)
 MESSAGE_CLASS_DEFINITION(SoapySDROutput::MsgStartStop, Message)
 MESSAGE_CLASS_DEFINITION(SoapySDROutput::MsgReportGainChange, Message)
+MESSAGE_CLASS_DEFINITION(SoapySDROutput::MsgGetStreamInfo, Message)
+MESSAGE_CLASS_DEFINITION(SoapySDROutput::MsgReportStreamInfo, Message)
 
 SoapySDROutput::SoapySDROutput(DeviceAPI *deviceAPI) :
     m_deviceAPI(deviceAPI),
@@ -526,7 +528,16 @@ bool SoapySDROutput::start()
     else // first allocation
     {
         qDebug("SoapySDROutput::start: allocate thread and take ownership");
-        soapySDROutputThread = new SoapySDROutputThread(m_deviceShared.m_device, requestedChannel+1);
+        unsigned int nbChannels = requestedChannel + 1;
+        // Dual-channel TX with chan-1 zero-fill. Required on B210/B220
+        // to activate second DUC chain in FPGA, preventing USB corruption
+        // and enabling the TX PA (ATR switch needs both chains).
+        if (m_deviceShared.m_device->getNumChannels(SOAPY_SDR_TX) >= 2
+            && m_deviceAPI->getSourceBuddies().empty()) {
+            nbChannels = 2;
+            qWarning("SoapySDROutput::start: forced dual-channel TX");
+        }
+        soapySDROutputThread = new SoapySDROutputThread(m_deviceShared.m_device, nbChannels);
         m_thread = soapySDROutputThread; // take ownership
         needsStart = true;
     }
@@ -539,6 +550,16 @@ bool SoapySDROutput::start()
         qDebug("SoapySDROutput::start: (re)start buddy thread");
         soapySDROutputThread->setSampleRate(m_settings.m_devSampleRate);
         soapySDROutputThread->startWork();
+        // Re-apply gain now that stream is active — setGain before activation
+        // silently fails on SoapyUHD/B210.
+        if (m_deviceShared.m_device) {
+            try {
+                m_deviceShared.m_device->setGain(
+                    SOAPY_SDR_TX, requestedChannel, m_settings.m_globalGain);
+                qCritical("SoapySDROutput::start: setGain ch%d %d (post-activation)",
+                          requestedChannel, m_settings.m_globalGain);
+            } catch (...) {}
+        }
     }
 
     qDebug("SoapySDROutput::start: started");
@@ -729,6 +750,16 @@ bool SoapySDROutput::setDeviceCenterFrequency(SoapySDR::Device *dev, int request
                 m_deviceShared.m_deviceParams->getTxChannelMainTunableElementName(requestedChannel),
                 freq_hz);
         qDebug("SoapySDROutput::setDeviceCenterFrequency: setFrequency(%llu)", freq_hz);
+        // In dual-channel mode, also set channel 1 to match so both AD9361
+        // LOs are configured (chain B needs frequency for LO lock).
+        if (m_thread && m_thread->getNbChannels() > 1
+            && dev->getNumChannels(SOAPY_SDR_TX) > 1) {
+            dev->setFrequency(SOAPY_SDR_TX, 1,
+                m_deviceShared.m_deviceParams->getTxChannelMainTunableElementName(1),
+                freq_hz);
+            // Do NOT set ch1 gain — B210's global gain is shared.
+            // Setting ch1 gain overrides ch0's configured gain.
+        }
         return true;
     }
     catch (const std::exception &ex)
@@ -746,11 +777,14 @@ void SoapySDROutput::updateGains(SoapySDR::Device *dev, int requestedChannel, So
 
     try
     {
-        settings.m_globalGain = round(dev->getGain(SOAPY_SDR_TX, requestedChannel));
+        double hwGain = dev->getGain(SOAPY_SDR_TX, requestedChannel);
+        settings.m_globalGain = round(hwGain);
 
         for (const auto &name : settings.m_individualGains.keys()) {
             settings.m_individualGains[name] = dev->getGain(SOAPY_SDR_TX, requestedChannel, name.toStdString());
         }
+
+        qCritical("SoapySDROutput::updateGains: hwGain=%.1f -> m_gain=%d", hwGain, settings.m_globalGain);
     }
     catch (const std::exception &ex)
     {
@@ -812,6 +846,35 @@ bool SoapySDROutput::handleMessage(const Message& message)
 
         return true;
     }
+    else if (MsgGetStreamInfo::match(message))
+    {
+        if (m_deviceAPI->getSamplingDeviceGUIMessageQueue())
+        {
+            if (m_thread && m_running)
+            {
+                bool active;
+                quint64 packets;
+                quint32 underflows;
+                quint32 errors;
+                m_thread->getStreamStatus(active, packets, underflows, errors);
+                (void) packets;
+                MsgReportStreamInfo *report = MsgReportStreamInfo::create(
+                        true,
+                        active,
+                        underflows,
+                        errors
+                        );
+                m_deviceAPI->getSamplingDeviceGUIMessageQueue()->push(report);
+            }
+            else
+            {
+                MsgReportStreamInfo *report = MsgReportStreamInfo::create(false, false, 0, 0);
+                m_deviceAPI->getSamplingDeviceGUIMessageQueue()->push(report);
+            }
+        }
+
+        return true;
+    }
     else if (DeviceSoapySDRShared::MsgReportBuddyChange::match(message))
     {
         int requestedChannel = m_deviceAPI->getDeviceItemIndex();
@@ -876,6 +939,10 @@ bool SoapySDROutput::handleMessage(const Message& message)
 
 bool SoapySDROutput::applySettings(const SoapySDROutputSettings& settings, bool force)
 {
+    qCritical("SoapySDROutput::applySettings: E freq=%llu->%llu gain=%d->%d force=%d this=%p settings=%p",
+              m_settings.m_centerFrequency, settings.m_centerFrequency,
+              m_settings.m_globalGain, settings.m_globalGain, force,
+              this, &settings);
     bool forwardChangeOwnDSP = false;
     bool forwardChangeToBuddies  = false;
     bool globalGainChanged = false;
@@ -980,6 +1047,8 @@ bool SoapySDROutput::applySettings(const SoapySDROutputSettings& settings, bool
         forwardChangeToBuddies = true;
 
         if (dev) {
+            qCritical("SoapySDROutput::applySettings: set freq=%llu gain=%d",
+                      settings.m_centerFrequency, settings.m_globalGain);
             setDeviceCenterFrequency(dev, requestedChannel, settings.m_centerFrequency, settings.m_LOppmTenths);
         }
     }
@@ -1293,7 +1362,9 @@ bool SoapySDROutput::applySettings(const SoapySDROutputSettings& settings, bool
     if (globalGainChanged || individualGainsChanged)
     {
         if (dev) {
+            int savedGlobalGain = m_settings.m_globalGain;
             updateGains(dev, requestedChannel, m_settings);
+            m_settings.m_globalGain = savedGlobalGain;
         }
 
         if (getMessageQueueToGUI())
@@ -1683,6 +1754,31 @@ void SoapySDROutput::webapiFormatDeviceReport(SWGSDRangel::SWGDeviceReport& resp
         webapiFormatArgInfo(itArg, response.getSoapySdrOutputReport()->getStreamSettingsArgs()->back());
     }
 
+    // Append TX diagnostic counters
+    if (m_thread)
+    {
+        bool active;
+        quint64 packets;
+        quint32 underflows;
+        quint32 errors;
+        m_thread->getStreamStatus(active, packets, underflows, errors);
+        (void) active;
+
+        auto addCounter = [&](const std::string& key, const std::string& name, const std::string& value) {
+            SoapySDR::ArgInfo info;
+            info.key = key;
+            info.value = value;
+            info.type = SoapySDR::ArgInfo::STRING;
+            info.name = name;
+            response.getSoapySdrOutputReport()->getStreamSettingsArgs()->append(new SWGSDRangel::SWGArgInfo);
+            webapiFormatArgInfo(info, response.getSoapySdrOutputReport()->getStreamSettingsArgs()->back());
+        };
+
+        addCounter("packets", "TX packets sent", std::to_string(packets));
+        addCounter("underflows", "TX underflow events", std::to_string(underflows));
+        addCounter("errors", "TX fatal errors", std::to_string(errors));
+    }
+
     response.getSoapySdrOutputReport()->setFrequencySettingsArgs(new QList<SWGSDRangel::SWGArgInfo*>);
 
     for (const auto& itArg : channelSettings->m_frequencySettingsArgs)

plugins/samplesink/soapysdroutput/soapysdroutput.h

@@ -119,6 +119,64 @@ public:
         { }
     };
 
+    class MsgGetStreamInfo : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        static MsgGetStreamInfo* create() {
+            return new MsgGetStreamInfo();
+        }
+
+    private:
+        MsgGetStreamInfo() :
+            Message()
+        { }
+    };
+
+    class MsgReportStreamInfo : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        bool     getSuccess() const { return m_success; }
+        bool     getActive() const { return m_active; }
+        uint32_t getUnderflows() const { return m_underflows; }
+        uint32_t getErrors() const { return m_errors; }
+
+        static MsgReportStreamInfo* create(
+                bool     success,
+                bool     active,
+                uint32_t underflows,
+                uint32_t errors
+                )
+        {
+            return new MsgReportStreamInfo(
+                    success,
+                    active,
+                    underflows,
+                    errors
+                    );
+        }
+
+    private:
+        bool     m_success;
+        bool     m_active;
+        uint32_t m_underflows;
+        uint32_t m_errors;
+
+        MsgReportStreamInfo(
+                bool     success,
+                bool     active,
+                uint32_t underflows,
+                uint32_t errors
+                ) :
+            Message(),
+            m_success(success),
+            m_active(active),
+            m_underflows(underflows),
+            m_errors(errors)
+        { }
+    };
+
     SoapySDROutput(DeviceAPI *deviceAPI);
     virtual ~SoapySDROutput();
     virtual void destroy();

plugins/samplesink/soapysdroutput/soapysdroutputthread.cpp

@@ -30,7 +30,11 @@ SoapySDROutputThread::SoapySDROutputThread(SoapySDR::Device* dev, unsigned int n
     m_dev(dev),
     m_sampleRate(0),
     m_nbChannels(nbTxChannels),
-    m_interpolatorType(InterpolatorFloat)
+    m_interpolatorType(InterpolatorFloat),
+    m_packets(0),
+    m_underflows(0),
+    m_errors(0),
+    m_consecutiveErrors(0)
 {
     qDebug("SoapySDROutputThread::SoapySDROutputThread");
     m_channels = new Channel[nbTxChannels];
@@ -53,6 +57,12 @@ void SoapySDROutputThread::startWork()
         return;
     }
 
+    // Reset diagnostic counters for new session
+    m_packets = 0;
+    m_underflows = 0;
+    m_errors = 0;
+    m_consecutiveErrors = 0;
+
     m_startWaitMutex.lock();
     start();
 
@@ -91,24 +101,34 @@ void SoapySDROutputThread::run()
         for (const auto &it : channels) {
             m_dev->setSampleRate(SOAPY_SDR_TX, it, m_sampleRate);
         }
+        if (m_nbChannels > 1) {
+            // Do NOT set ch1 gain — B210's global TX gain is shared between channels.
+            // Setting ch1 gain overrides ch0's configured gain from REST API.
+            double txFreq = m_dev->getFrequency(SOAPY_SDR_TX, 0);
+            m_dev->setFrequency(SOAPY_SDR_TX, 1, txFreq);
+        }
 
         // Determine sample format to be used
         double fullScale(0.0);
         std::string format = m_dev->getNativeStreamFormat(SOAPY_SDR_TX, channels.front(), fullScale);
 
         qDebug("SoapySDROutputThread::run: format: %s fullScale: %f", format.c_str(), fullScale);
+        qCritical("SoapySDROutput: fmt=[%s] len=%zu fs=%.15f cb=%d ch=%zu",
+                  format.c_str(), format.size(), fullScale, m_interpolatorType, channels.size());
 
         if ((format == "CS8") && (fullScale == 128.0)) { // 8 bit signed - native
             m_interpolatorType = Interpolator8;
         } else if ((format == "CS16") && (fullScale == 2048.0)) { // 12 bit signed - native
             m_interpolatorType = Interpolator12;
-        } else if ((format == "CS16") && (fullScale == 32768.0)) { // 16 bit signed - native
+        } else if ((format == "CS16") && (fullScale >= 2049.0)) { // 16 bit signed - native
             m_interpolatorType = Interpolator16;
         } else { // for other types make a conversion to float
             m_interpolatorType = InterpolatorFloat;
             format = "CF32";
         }
 
+        qCritical("SoapySDROutput: final fmt=[%s] cb=%d", format.c_str(), m_interpolatorType);
+
         unsigned int elemSize = SoapySDR::formatToSize(format); // sample (I+Q) size in bytes
         SoapySDR::Stream *stream = m_dev->setupStream(SOAPY_SDR_TX, format, channels);
 
@@ -121,43 +141,49 @@ void SoapySDROutputThread::run()
             buffs[i] = buffMem[i].data();
         }
 
+        // Activate stream at thread start (untimed).  USRPOutputThread
+        // never uses has_time_spec — match that behavior.  HAS_TIME on
+        // first write stalls UHD waiting for the future timestamp which
+        // fills the TX ring buffer → writeStream timeouts → USB transport
+        // corruption → LIBUSB_ERROR_NOT_FOUND.
         m_dev->activateStream(stream);
-        int flags(0);
+        // Gain applied by SoapySDROutput::start() after thread is active.
+        // Do NOT set gain here — setGain before stream activation is a
+        // no-op on SoapyUHD/B210, and reading back returns 0.
+        int writeFlags(0);
         long long timeNs(0);
-        float blockTime = ((float) numElems) / (m_sampleRate <= 0 ? 1024000 : m_sampleRate);
-        long initialTtimeoutUs = 10000000 * blockTime; // 10 times the block time
-        long timeoutUs = initialTtimeoutUs < 250000 ? 250000 : initialTtimeoutUs; // 250ms minimum
+        long timeoutUs = 10000; // 10ms max block per writeStream
 
-        qDebug("SoapySDROutputThread::run: numElems: %u elemSize: %u initialTtimeoutUs: %ld  timeoutUs: %ld",
-                numElems, elemSize, initialTtimeoutUs, timeoutUs);
+        {
+            double actFreq = m_dev->getFrequency(SOAPY_SDR_TX, channels[0]);
+            double actGain = m_dev->getGain(SOAPY_SDR_TX, channels[0]);
+            double actSR = m_dev->getSampleRate(SOAPY_SDR_TX, channels[0]);
+            qCritical("SoapySDROutputThread::run: ch0 freq=%.0f gain=%.1f SR=%.0f",
+                      actFreq, actGain, actSR);
+            if (channels.size() > 1) {
+                double actFreq1 = m_dev->getFrequency(SOAPY_SDR_TX, channels[1]);
+                double actGain1 = m_dev->getGain(SOAPY_SDR_TX, channels[1]);
+                qCritical("SoapySDROutputThread::run: ch1 freq=%.0f gain=%.1f",
+                          actFreq1, actGain1);
+            }
+        }
+
+        qDebug("SoapySDROutputThread::run: numElems: %u elemSize: %u timeoutUs: %ld",
+                numElems, elemSize, timeoutUs);
         qDebug("SoapySDROutputThread::run: start running loop");
 
         while (m_running)
         {
-            int ret = m_dev->writeStream(stream, buffs.data(), numElems, flags, timeNs, timeoutUs);
-
-            if (ret == SOAPY_SDR_TIMEOUT)
-            {
-                qWarning("SoapySDROutputThread::run: timeout: flags: %d timeNs: %lld timeoutUs: %ld", flags, timeNs, timeoutUs);
-            }
-            else if (ret == SOAPY_SDR_OVERFLOW)
-            {
-                qWarning("SoapySDROutputThread::run: overflow: flags: %d timeNs: %lld timeoutUs: %ld", flags, timeNs, timeoutUs);
-            }
-            else if (ret < 0)
-            {
-                qCritical("SoapySDROutputThread::run: Unexpected write stream error: %s", SoapySDR::errToStr(ret));
-                break;
+            // Zero buffers before fill — prevents stale data
+            for (auto& buf : buffMem) {
+                std::fill(buf.begin(), buf.end(), 0);
             }
 
-            if (m_nbChannels > 1)
-            {
-                callbackMO(buffs, numElems); // size given in number of samples (1 item per sample)
-            }
-            else
-            {
-                switch (m_interpolatorType)
-                {
+            // Fill buffers from FIFO
+            if (m_nbChannels > 1) {
+                callbackMO(buffs, numElems);
+            } else {
+                switch (m_interpolatorType) {
                 case Interpolator8:
                     callbackSO8((qint8*) buffs[0], numElems);
                     break;
@@ -173,6 +199,45 @@ void SoapySDROutputThread::run()
                     break;
                 }
             }
+
+            // Idle-skip: when FIFO is empty, sleep and retry.
+            // Matches USRPOutputThread behavior.
+            bool hasNonZero = false;
+            for (unsigned int i = 0; i < m_nbChannels && !hasNonZero; i++) {
+                const char* p = buffMem[i].data();
+                const char* end = p + std::min<size_t>(16, buffMem[i].size());
+                while (p < end) {
+                    if (*p++ != 0) { hasNonZero = true; break; }
+                }
+            }
+
+            if (!hasNonZero) {
+                QThread::usleep(100);
+                continue;
+            }
+            int ret = m_dev->writeStream(stream, buffs.data(), numElems, writeFlags, timeNs, timeoutUs);
+
+            if (ret == SOAPY_SDR_TIMEOUT)
+            {
+                m_underflows++;
+                qWarning("SoapySDROutputThread::run: timeout: flags: %d timeNs: %lld timeoutUs: %ld", writeFlags, timeNs, timeoutUs);
+            }
+            else if (ret == SOAPY_SDR_OVERFLOW)
+            {
+                m_underflows++;
+                qWarning("SoapySDROutputThread::run: overflow: flags: %d timeNs: %lld timeoutUs: %ld", writeFlags, timeNs, timeoutUs);
+            }
+            else if (ret < 0)
+            {
+                m_errors++;
+                qCritical("SoapySDROutputThread::run: Unexpected write stream error: %s", SoapySDR::errToStr(ret));
+                break;
+            }
+            else if (ret > 0)
+            {
+                m_packets++;
+            }
+
         }
 
         qDebug("SoapySDROutputThread::run: stop running loop");
@@ -253,7 +318,7 @@ void SoapySDROutputThread::callbackMO(std::vector<void *>& buffs, qint32 samples
                 break;
             case InterpolatorFloat:
             default:
-                // TODO
+                callbackSOIF((float*) buffs[ichan], samplesPerChannel, ichan);
                 break;
             }
         }
@@ -522,3 +587,11 @@ void SoapySDROutputThread::callbackPartF(float* buf, SampleVector& data, unsigne
         }
     }
 }
+
+void SoapySDROutputThread::getStreamStatus(bool& active, quint64& packets, quint32& underflows, quint32& errors)
+{
+    active = m_packets > 0;
+    packets = m_packets;
+    underflows = m_underflows;
+    errors = m_errors;
+}

plugins/samplesink/soapysdroutput/soapysdroutputthread.h

@@ -42,6 +42,7 @@ public:
     void stopWork();
     bool isRunning() const { return m_running; }
     unsigned int getNbChannels() const { return m_nbChannels; }
+    void getStreamStatus(bool& active, quint64& packets, quint32& underflows, quint32& errors);
     void setLog2Interpolation(unsigned int channel, unsigned int log2_interp);
     unsigned int getLog2Interpolation(unsigned int channel) const;
     void setSampleRate(unsigned int sampleRate) { m_sampleRate = sampleRate; }
@@ -87,6 +88,12 @@ private:
     unsigned int m_nbChannels;
     InterpolatorType m_interpolatorType;
 
+    // Diagnostic counters
+    quint64 m_packets;              //!< Total packets (writeStream calls) sent
+    quint32 m_underflows;           //!< Underflow / timeout events
+    quint32 m_errors;               //!< Fatal write stream errors (SOAPY_SDR_ERR_*)
+    quint32 m_consecutiveErrors;    //!< Consecutive non-fatal errors before break
+
     void run();
     unsigned int getNbFifos();
     void callbackSO8(qint8* buf, qint32 len, unsigned int channel = 0);