sdrangel/sdrbase/dsp/dspdevicesinkengine.cpp

620 wiersze
19 KiB
C++

///////////////////////////////////////////////////////////////////////////////////
// Copyright (C) 2016 F4EXB //
// written by Edouard Griffiths //
// //
// This program is free software; you can redistribute it and/or modify //
// it under the terms of the GNU General Public License as published by //
// the Free Software Foundation as version 3 of the License, or //
// (at your option) any later version. //
// //
// This program is distributed in the hope that it will be useful, //
// but WITHOUT ANY WARRANTY; without even the implied warranty of //
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the //
// GNU General Public License V3 for more details. //
// //
// You should have received a copy of the GNU General Public License //
// along with this program. If not, see <http://www.gnu.org/licenses/>. //
///////////////////////////////////////////////////////////////////////////////////
#include <stdio.h>
#include <QDebug>
#include <QThread>
#include "dspdevicesinkengine.h"
#include "dsp/basebandsamplesource.h"
#include "dsp/basebandsamplesink.h"
#include "dsp/devicesamplesink.h"
#include "dsp/dspcommands.h"
#include "samplesourcefifo.h"
#include "threadedbasebandsamplesource.h"
DSPDeviceSinkEngine::DSPDeviceSinkEngine(uint32_t uid, QObject* parent) :
QThread(parent),
m_uid(uid),
m_state(StNotStarted),
m_deviceSampleSink(nullptr),
m_sampleSinkSequence(0),
m_basebandSampleSources(),
m_spectrumSink(nullptr),
m_sampleRate(0),
m_centerFrequency(0),
m_multipleSourcesDivisionFactor(1)
{
connect(&m_inputMessageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleInputMessages()), Qt::QueuedConnection);
connect(&m_syncMessenger, SIGNAL(messageSent()), this, SLOT(handleSynchronousMessages()), Qt::QueuedConnection);
moveToThread(this);
}
DSPDeviceSinkEngine::~DSPDeviceSinkEngine()
{
stop();
wait();
}
void DSPDeviceSinkEngine::run()
{
qDebug() << "DSPDeviceSinkEngine::run";
m_state = StIdle;
exec();
}
void DSPDeviceSinkEngine::start()
{
qDebug() << "DSPDeviceSinkEngine::start";
QThread::start();
}
void DSPDeviceSinkEngine::stop()
{
qDebug() << "DSPDeviceSinkEngine::stop";
gotoIdle();
m_state = StNotStarted;
QThread::exit();
// DSPExit cmd;
// m_syncMessenger.sendWait(cmd);
}
bool DSPDeviceSinkEngine::initGeneration()
{
qDebug() << "DSPDeviceSinkEngine::initGeneration";
DSPGenerationInit cmd;
return m_syncMessenger.sendWait(cmd) == StReady;
}
bool DSPDeviceSinkEngine::startGeneration()
{
qDebug() << "DSPDeviceSinkEngine::startGeneration";
DSPGenerationStart cmd;
return m_syncMessenger.sendWait(cmd) == StRunning;
}
void DSPDeviceSinkEngine::stopGeneration()
{
qDebug() << "DSPDeviceSinkEngine::stopGeneration";
DSPGenerationStop cmd;
m_syncMessenger.storeMessage(cmd);
handleSynchronousMessages();
}
void DSPDeviceSinkEngine::setSink(DeviceSampleSink* sink)
{
qDebug() << "DSPDeviceSinkEngine::setSink";
DSPSetSink cmd(sink);
m_syncMessenger.sendWait(cmd);
}
void DSPDeviceSinkEngine::setSinkSequence(int sequence)
{
qDebug("DSPDeviceSinkEngine::setSinkSequence: seq: %d", sequence);
m_sampleSinkSequence = sequence;
}
void DSPDeviceSinkEngine::addThreadedSource(ThreadedBasebandSampleSource* source)
{
qDebug() << "DSPDeviceSinkEngine::addThreadedSource: " << source->objectName().toStdString().c_str();
DSPAddThreadedBasebandSampleSource cmd(source);
m_syncMessenger.sendWait(cmd);
}
void DSPDeviceSinkEngine::removeThreadedSource(ThreadedBasebandSampleSource* source)
{
qDebug() << "DSPDeviceSinkEngine::removeThreadedSource: " << source->objectName().toStdString().c_str();
DSPRemoveThreadedBasebandSampleSource cmd(source);
m_syncMessenger.sendWait(cmd);
}
void DSPDeviceSinkEngine::addSpectrumSink(BasebandSampleSink* spectrumSink)
{
qDebug() << "DSPDeviceSinkEngine::addSpectrumSink: " << spectrumSink->objectName().toStdString().c_str();
DSPAddSpectrumSink cmd(spectrumSink);
m_syncMessenger.sendWait(cmd);
}
void DSPDeviceSinkEngine::removeSpectrumSink(BasebandSampleSink* spectrumSink)
{
qDebug() << "DSPDeviceSinkEngine::removeSpectrumSink: " << spectrumSink->objectName().toStdString().c_str();
DSPRemoveSpectrumSink cmd(spectrumSink);
m_syncMessenger.sendWait(cmd);
}
QString DSPDeviceSinkEngine::errorMessage()
{
qDebug() << "DSPDeviceSinkEngine::errorMessage";
DSPGetErrorMessage cmd;
m_syncMessenger.sendWait(cmd);
return cmd.getErrorMessage();
}
QString DSPDeviceSinkEngine::sinkDeviceDescription()
{
qDebug() << "DSPDeviceSinkEngine::sinkDeviceDescription";
DSPGetSinkDeviceDescription cmd;
m_syncMessenger.sendWait(cmd);
return cmd.getDeviceDescription();
}
void DSPDeviceSinkEngine::work(int nbWriteSamples)
{
// multiple channel sources handling
if ((m_threadedBasebandSampleSources.size() + m_basebandSampleSources.size()) > 1)
{
// qDebug("DSPDeviceSinkEngine::work: multiple channel sources handling: %u", m_multipleSourcesDivisionFactor);
SampleVector::iterator writeBegin;
SampleSourceFifo* sampleFifo = m_deviceSampleSink->getSampleFifo();
sampleFifo->getWriteIterator(writeBegin);
SampleVector::iterator writeAt = writeBegin;
std::vector<SampleVector::iterator> sampleSourceIterators;
for (ThreadedBasebandSampleSources::iterator it = m_threadedBasebandSampleSources.begin(); it != m_threadedBasebandSampleSources.end(); ++it)
{
sampleSourceIterators.push_back(SampleVector::iterator());
(*it)->getSampleSourceFifo().readAdvance(sampleSourceIterators.back(), nbWriteSamples);
sampleSourceIterators.back() -= nbWriteSamples;
}
for (BasebandSampleSources::iterator it = m_basebandSampleSources.begin(); it != m_basebandSampleSources.end(); ++it)
{
sampleSourceIterators.push_back(SampleVector::iterator());
(*it)->getSampleSourceFifo().readAdvance(sampleSourceIterators.back(), nbWriteSamples);
sampleSourceIterators.back() -= nbWriteSamples;
}
for (int is = 0; is < nbWriteSamples; is++)
{
// pull data from sources FIFOs and merge them in the device sample FIFO
for (std::vector<SampleVector::iterator>::iterator it = sampleSourceIterators.begin(); it != sampleSourceIterators.end(); ++it)
{
Sample s = (**it);
s /= m_multipleSourcesDivisionFactor;
++(*it);
if (it == sampleSourceIterators.begin()) {
(*writeAt) = s;
} else {
(*writeAt) += s;
}
}
sampleFifo->bumpIndex(writeAt);
}
}
}
// notStarted -> idle -> init -> running -+
// ^ |
// +-----------------------+
DSPDeviceSinkEngine::State DSPDeviceSinkEngine::gotoIdle()
{
qDebug() << "DSPDeviceSinkEngine::gotoIdle";
switch(m_state) {
case StNotStarted:
return StNotStarted;
case StIdle:
case StError:
return StIdle;
case StReady:
case StRunning:
break;
}
if(m_deviceSampleSink == 0)
{
return StIdle;
}
// stop everything
for(BasebandSampleSources::const_iterator it = m_basebandSampleSources.begin(); it != m_basebandSampleSources.end(); it++)
{
qDebug() << "DSPDeviceSinkEngine::gotoIdle: stopping " << (*it)->objectName().toStdString().c_str();
(*it)->stop();
}
for(ThreadedBasebandSampleSources::const_iterator it = m_threadedBasebandSampleSources.begin(); it != m_threadedBasebandSampleSources.end(); it++)
{
qDebug() << "DSPDeviceSinkEngine::gotoIdle: stopping ThreadedSampleSource(" << (*it)->getSampleSourceObjectName().toStdString().c_str() << ")";
(*it)->stop();
}
if (m_spectrumSink)
{
disconnect(m_deviceSampleSink->getSampleFifo(), SIGNAL(dataRead(int)), this, SLOT(handleForwardToSpectrumSink(int)));
m_spectrumSink->stop();
}
m_deviceSampleSink->stop();
m_deviceDescription.clear();
m_sampleRate = 0;
return StIdle;
}
DSPDeviceSinkEngine::State DSPDeviceSinkEngine::gotoInit()
{
switch(m_state) {
case StNotStarted:
return StNotStarted;
case StRunning: // FIXME: assumes it goes first through idle state. Could we get back to init from running directly?
return StRunning;
case StReady:
return StReady;
case StIdle:
case StError:
break;
}
if (m_deviceSampleSink == 0)
{
return gotoError("DSPDeviceSinkEngine::gotoInit: No sample source configured");
}
// init: pass sample rate and center frequency to all sample rate and/or center frequency dependent sinks and wait for completion
m_deviceDescription = m_deviceSampleSink->getDeviceDescription();
m_centerFrequency = m_deviceSampleSink->getCenterFrequency();
m_sampleRate = m_deviceSampleSink->getSampleRate();
qDebug() << "DSPDeviceSinkEngine::gotoInit: "
<< " m_deviceDescription: " << m_deviceDescription.toStdString().c_str()
<< " sampleRate: " << m_sampleRate
<< " centerFrequency: " << m_centerFrequency;
DSPSignalNotification notif(m_sampleRate, m_centerFrequency);
for (BasebandSampleSources::const_iterator it = m_basebandSampleSources.begin(); it != m_basebandSampleSources.end(); ++it)
{
qDebug() << "DSPDeviceSinkEngine::gotoInit: initializing " << (*it)->objectName().toStdString().c_str();
(*it)->handleMessage(notif);
}
for (ThreadedBasebandSampleSources::const_iterator it = m_threadedBasebandSampleSources.begin(); it != m_threadedBasebandSampleSources.end(); ++it)
{
qDebug() << "DSPDeviceSinkEngine::gotoInit: initializing ThreadedSampleSource(" << (*it)->getSampleSourceObjectName().toStdString().c_str() << ")";
(*it)->handleSourceMessage(notif);
}
if (m_spectrumSink) {
m_spectrumSink->handleMessage(notif);
}
// pass data to listeners
if (m_deviceSampleSink->getMessageQueueToGUI())
{
DSPSignalNotification* rep = new DSPSignalNotification(notif); // make a copy for the output queue
m_deviceSampleSink->getMessageQueueToGUI()->push(rep);
}
return StReady;
}
DSPDeviceSinkEngine::State DSPDeviceSinkEngine::gotoRunning()
{
qDebug() << "DSPDeviceSinkEngine::gotoRunning";
switch(m_state)
{
case StNotStarted:
return StNotStarted;
case StIdle:
return StIdle;
case StRunning:
return StRunning;
case StReady:
case StError:
break;
}
if(m_deviceSampleSink == 0) {
return gotoError("DSPDeviceSinkEngine::gotoRunning: No sample source configured");
}
qDebug() << "DSPDeviceSinkEngine::gotoRunning: " << m_deviceDescription.toStdString().c_str() << " started";
// Start everything
if(!m_deviceSampleSink->start())
{
return gotoError("DSPDeviceSinkEngine::gotoRunning: Could not start sample sink");
}
for(BasebandSampleSources::const_iterator it = m_basebandSampleSources.begin(); it != m_basebandSampleSources.end(); it++)
{
qDebug() << "DSPDeviceSinkEngine::gotoRunning: starting " << (*it)->objectName().toStdString().c_str();
(*it)->start();
}
for (ThreadedBasebandSampleSources::const_iterator it = m_threadedBasebandSampleSources.begin(); it != m_threadedBasebandSampleSources.end(); ++it)
{
qDebug() << "DSPDeviceSinkEngine::gotoRunning: starting ThreadedSampleSource(" << (*it)->getSampleSourceObjectName().toStdString().c_str() << ")";
(*it)->start();
}
if (m_spectrumSink)
{
connect(m_deviceSampleSink->getSampleFifo(), SIGNAL(dataRead(int)), this, SLOT(handleForwardToSpectrumSink(int)));
m_spectrumSink->start();
}
qDebug() << "DSPDeviceSinkEngine::gotoRunning: input message queue pending: " << m_inputMessageQueue.size();
return StRunning;
}
DSPDeviceSinkEngine::State DSPDeviceSinkEngine::gotoError(const QString& errorMessage)
{
qDebug() << "DSPDeviceSinkEngine::gotoError";
m_errorMessage = errorMessage;
m_deviceDescription.clear();
m_state = StError;
return StError;
}
void DSPDeviceSinkEngine::handleSetSink(DeviceSampleSink* sink)
{
gotoIdle();
m_deviceSampleSink = sink;
if(m_deviceSampleSink != 0) {
qDebug("DSPDeviceSinkEngine::handleSetSink: set %s", qPrintable(sink->getDeviceDescription()));
} else {
qDebug("DSPDeviceSinkEngine::handleSetSource: set none");
}
}
void DSPDeviceSinkEngine::handleData(int nbSamples)
{
if(m_state == StRunning)
{
work(nbSamples);
}
}
void DSPDeviceSinkEngine::handleSynchronousMessages()
{
Message *message = m_syncMessenger.getMessage();
qDebug() << "DSPDeviceSinkEngine::handleSynchronousMessages: " << message->getIdentifier();
if (DSPGenerationInit::match(*message))
{
m_state = gotoIdle();
if(m_state == StIdle) {
m_state = gotoInit(); // State goes ready if init is performed
}
}
else if (DSPGenerationStart::match(*message))
{
if(m_state == StReady) {
m_state = gotoRunning();
}
}
else if (DSPGenerationStop::match(*message))
{
m_state = gotoIdle();
}
else if (DSPGetSinkDeviceDescription::match(*message))
{
((DSPGetSinkDeviceDescription*) message)->setDeviceDescription(m_deviceDescription);
}
else if (DSPGetErrorMessage::match(*message))
{
((DSPGetErrorMessage*) message)->setErrorMessage(m_errorMessage);
}
else if (DSPSetSink::match(*message)) {
handleSetSink(((DSPSetSink*) message)->getSampleSink());
}
else if (DSPAddSpectrumSink::match(*message))
{
m_spectrumSink = ((DSPAddSpectrumSink*) message)->getSampleSink();
}
else if (DSPRemoveSpectrumSink::match(*message))
{
BasebandSampleSink* spectrumSink = ((DSPRemoveSpectrumSink*) message)->getSampleSink();
if(m_state == StRunning) {
spectrumSink->stop();
}
m_spectrumSink = 0;
}
else if (DSPAddBasebandSampleSource::match(*message))
{
BasebandSampleSource* source = ((DSPAddBasebandSampleSource*) message)->getSampleSource();
m_basebandSampleSources.push_back(source);
DSPSignalNotification notif(m_sampleRate, m_centerFrequency);
source->handleMessage(notif);
checkNumberOfBasebandSources();
if (m_state == StRunning)
{
source->start();
}
}
else if (DSPRemoveBasebandSampleSource::match(*message))
{
BasebandSampleSource* source = ((DSPRemoveBasebandSampleSource*) message)->getSampleSource();
if(m_state == StRunning) {
source->stop();
}
m_basebandSampleSources.remove(source);
checkNumberOfBasebandSources();
}
else if (DSPAddThreadedBasebandSampleSource::match(*message))
{
ThreadedBasebandSampleSource *threadedSource = ((DSPAddThreadedBasebandSampleSource*) message)->getThreadedSampleSource();
m_threadedBasebandSampleSources.push_back(threadedSource);
DSPSignalNotification notif(m_sampleRate, m_centerFrequency);
threadedSource->handleSourceMessage(notif);
checkNumberOfBasebandSources();
if (m_state == StRunning)
{
threadedSource->start();
}
}
else if (DSPRemoveThreadedBasebandSampleSource::match(*message))
{
ThreadedBasebandSampleSource* threadedSource = ((DSPRemoveThreadedBasebandSampleSource*) message)->getThreadedSampleSource();
if (m_state == StRunning) {
threadedSource->stop();
}
m_threadedBasebandSampleSources.remove(threadedSource);
checkNumberOfBasebandSources();
}
m_syncMessenger.done(m_state);
}
void DSPDeviceSinkEngine::handleInputMessages()
{
Message* message;
while ((message = m_inputMessageQueue.pop()) != 0)
{
qDebug("DSPDeviceSinkEngine::handleInputMessages: message: %s", message->getIdentifier());
if (DSPSignalNotification::match(*message))
{
DSPSignalNotification *notif = (DSPSignalNotification *) message;
// update DSP values
m_sampleRate = notif->getSampleRate();
m_centerFrequency = notif->getCenterFrequency();
qDebug() << "DSPDeviceSinkEngine::handleInputMessages: DSPSignalNotification:"
<< " m_sampleRate: " << m_sampleRate
<< " m_centerFrequency: " << m_centerFrequency;
// forward source changes to sources with immediate execution
for(BasebandSampleSources::const_iterator it = m_basebandSampleSources.begin(); it != m_basebandSampleSources.end(); it++)
{
qDebug() << "DSPDeviceSinkEngine::handleInputMessages: forward message to " << (*it)->objectName().toStdString().c_str();
(*it)->handleMessage(*message);
}
for (ThreadedBasebandSampleSources::const_iterator it = m_threadedBasebandSampleSources.begin(); it != m_threadedBasebandSampleSources.end(); ++it)
{
qDebug() << "DSPDeviceSinkEngine::handleSourceMessages: forward message to ThreadedSampleSource(" << (*it)->getSampleSourceObjectName().toStdString().c_str() << ")";
(*it)->handleSourceMessage(*message);
}
// forward changes to listeners on DSP output queue
MessageQueue *guiMessageQueue = m_deviceSampleSink->getMessageQueueToGUI();
qDebug("DSPDeviceSinkEngine::handleInputMessages: DSPSignalNotification: guiMessageQueue: %p", guiMessageQueue);
if (guiMessageQueue)
{
DSPSignalNotification* rep = new DSPSignalNotification(*notif); // make a copy for the output queue
guiMessageQueue->push(rep);
}
delete message;
}
}
}
void DSPDeviceSinkEngine::handleForwardToSpectrumSink(int nbSamples)
{
if (m_spectrumSink)
{
SampleSourceFifo* sampleFifo = m_deviceSampleSink->getSampleFifo();
SampleVector::iterator readUntil;
sampleFifo->getReadIterator(readUntil);
m_spectrumSink->feed(readUntil - nbSamples, readUntil, false);
}
}
void DSPDeviceSinkEngine::checkNumberOfBasebandSources()
{
SampleSourceFifo* sampleFifo = m_deviceSampleSink->getSampleFifo();
// single channel source handling
if ((m_threadedBasebandSampleSources.size() + m_basebandSampleSources.size()) == 1)
{
qDebug("DSPDeviceSinkEngine::checkNumberOfBasebandSources: single channel mode");
disconnect(sampleFifo, SIGNAL(dataWrite(int)), this, SLOT(handleData(int)));
if (m_threadedBasebandSampleSources.size() == 1) {
m_threadedBasebandSampleSources.back()->setDeviceSampleSourceFifo(sampleFifo);
} else if (m_basebandSampleSources.size() == 1) {
m_basebandSampleSources.back()->setDeviceSampleSourceFifo(sampleFifo);
}
m_multipleSourcesDivisionFactor = 1; // for consistency but it is not used in this case
}
// null or multiple channel sources handling
else
{
int nbSources = 0;
for (ThreadedBasebandSampleSources::iterator it = m_threadedBasebandSampleSources.begin(); it != m_threadedBasebandSampleSources.end(); ++it)
{
(*it)->setDeviceSampleSourceFifo(0);
nbSources++;
}
for (BasebandSampleSources::iterator it = m_basebandSampleSources.begin(); it != m_basebandSampleSources.end(); ++it)
{
(*it)->setDeviceSampleSourceFifo(0);
nbSources++;
}
if (nbSources == 0) {
m_multipleSourcesDivisionFactor = 1;
} else if (nbSources < 3) {
m_multipleSourcesDivisionFactor = nbSources;
} else {
m_multipleSourcesDivisionFactor = 1<<nbSources;
}
if (nbSources > 1) {
connect(sampleFifo, SIGNAL(dataWrite(int)), this, SLOT(handleData(int)), Qt::QueuedConnection);
}
qDebug("DSPDeviceSinkEngine::checkNumberOfBasebandSources: handle %d channel(s)", nbSources);
}
}