sdrangel/plugins/samplesource/aaroniartsainput/aaroniartsainputworker.cpp

423 wiersze
14 KiB
C++

///////////////////////////////////////////////////////////////////////////////////
// Copyright (C) 2023 Edouard Griffiths, F4EXB //
// //
// 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 <boost/endian/conversion.hpp>
#include <QJsonObject>
#include <QJsonArray>
#include <QJsonValue>
#include "util/messagequeue.h"
#include "dsp/dspcommands.h"
#include "aaroniartsainputsettings.h"
#include "aaroniartsainputworker.h"
MESSAGE_CLASS_DEFINITION(AaroniaRTSAInputWorker::MsgReportSampleRateAndFrequency, Message)
AaroniaRTSAInputWorker::AaroniaRTSAInputWorker(SampleSinkFifo* sampleFifo) :
QObject(),
m_timer(this),
m_samplesBuf(),
m_sampleFifo(sampleFifo),
m_centerFrequency(0),
m_sampleRate(1),
m_inputMessageQueue(nullptr),
m_status(AaroniaRTSAInputSettings::ConnectionIdle),
mReply(nullptr),
m_convertBuffer(64e6)
{
// Initialize network managers
m_networkAccessManager = new QNetworkAccessManager(this);
m_networkAccessManagerConfig = new QNetworkAccessManager(this);
QObject::connect(
m_networkAccessManagerConfig,
&QNetworkAccessManager::finished,
this,
&AaroniaRTSAInputWorker::handleConfigReply
);
// Request 16bit raw samples
// m_serverAddress = "localhost:55123";
// QUrl url(tr("http://%1/stream?format=float32").arg(m_serverAddress));
// QNetworkRequest req(url);
// mReply = m_networkAccessManager->get(req);
// // Connect Qt slots to network events
// connect(mReply, SIGNAL(error(QNetworkReply::NetworkError)), this, SLOT(onError(QNetworkReply::NetworkError)));
// connect(mReply, SIGNAL(finished()), this, SLOT(onFinished()));
// connect(mReply, SIGNAL(readyRead()), this, SLOT(onReadyRead()));
mPrevTime = 0;
mPacketSamples = 0;
}
AaroniaRTSAInputWorker::~AaroniaRTSAInputWorker()
{
if (mReply)
{
// disconnect previous sugnals
disconnect(mReply, SIGNAL(error(QNetworkReply::NetworkError)), this, SLOT(onError(QNetworkReply::NetworkError)));
disconnect(mReply, SIGNAL(finished()), this, SLOT(onFinished()));
disconnect(mReply, SIGNAL(readyRead()), this, SLOT(onReadyRead()));
mReply->abort();
mReply->deleteLater();
}
m_networkAccessManager->deleteLater();
QObject::disconnect(
m_networkAccessManagerConfig,
&QNetworkAccessManager::finished,
this,
&AaroniaRTSAInputWorker::handleConfigReply
);
m_networkAccessManagerConfig->deleteLater();
}
void AaroniaRTSAInputWorker::onSocketError(QAbstractSocket::SocketError error)
{
(void) error;
m_status = AaroniaRTSAInputSettings::ConnectionError;
emit updateStatus(m_status);
}
void AaroniaRTSAInputWorker::sendCenterFrequencyAndSampleRate()
{
if (m_iqDemodName.size() == 0) {
return;
}
qDebug("AaroniaRTSAInputWorker::sendCenterFrequencyAndSampleRate: %llu samplerate: %d", m_centerFrequency, m_sampleRate);
QJsonObject object {
{"receiverName", m_iqDemodName},
{"simpleconfig", QJsonObject({
{"main", QJsonObject({
{"centerfreq", QJsonValue((qint64) m_centerFrequency)},
{"samplerate", QJsonValue(m_sampleRate)},
{"spanfreq", QJsonValue(m_sampleRate)},
})}
})}
};
QJsonDocument document;
document.setObject(object);
QUrl url(tr("http://%1/remoteconfig").arg(m_serverAddress));
QNetworkRequest request(url);
request.setHeader(QNetworkRequest::ContentTypeHeader, "application/json");
m_networkAccessManagerConfig->put(request, document.toJson());
}
void AaroniaRTSAInputWorker::getConfig()
{
QUrl url(tr("http://%1/remoteconfig").arg(m_serverAddress));
QNetworkRequest request(url);
m_networkAccessManagerConfig->get(request);
}
void AaroniaRTSAInputWorker::onCenterFrequencyChanged(quint64 centerFrequency)
{
if (m_centerFrequency == centerFrequency) {
return;
}
m_centerFrequency = centerFrequency;
sendCenterFrequencyAndSampleRate();
}
void AaroniaRTSAInputWorker::onSampleRateChanged(int sampleRate)
{
if (m_sampleRate == sampleRate) {
return;
}
m_sampleRate = sampleRate;
sendCenterFrequencyAndSampleRate();
}
void AaroniaRTSAInputWorker::onServerAddressChanged(QString serverAddress)
{
m_status = AaroniaRTSAInputSettings::ConnectionDisconnected;
updateStatus(m_status);
if (mReply)
{
// disconnect previous sugnals
disconnect(mReply, SIGNAL(error(QNetworkReply::NetworkError)), this, SLOT(onError(QNetworkReply::NetworkError)));
disconnect(mReply, SIGNAL(finished()), this, SLOT(onFinished()));
disconnect(mReply, SIGNAL(readyRead()), this, SLOT(onReadyRead()));
mReply->abort();
mReply->deleteLater();
}
QUrl url(tr("http://%1/stream?format=float32").arg(serverAddress));
QNetworkRequest req(url);
mReply = m_networkAccessManager->get(req);
// Connect Qt slots to network events
connect(mReply, SIGNAL(error(QNetworkReply::NetworkError)), this, SLOT(onError(QNetworkReply::NetworkError)));
connect(mReply, SIGNAL(finished()), this, SLOT(onFinished()));
connect(mReply, SIGNAL(readyRead()), this, SLOT(onReadyRead()));
mPrevTime = 0;
mPacketSamples = 0;
m_serverAddress = serverAddress;
getConfig();
}
void AaroniaRTSAInputWorker::tick()
{
}
/**************************CPY ********************************* */
void AaroniaRTSAInputWorker::onError(QNetworkReply::NetworkError code)
{
(void) code;
qWarning() << "AaroniaRTSAInputWorker::onError: network Error: " << mReply->errorString();
m_status = 3;
emit updateStatus(3);
}
void AaroniaRTSAInputWorker::onFinished()
{
qDebug() << "AaroniaRTSAInputWorker::onFinished(: finished: " << mReply->errorString();
mBuffer.append(mReply->readAll());
mReply->deleteLater();
mReply = nullptr;
}
// bytes received from the socket
void AaroniaRTSAInputWorker::onReadyRead()
{
if (m_status != AaroniaRTSAInputSettings::ConnectionOK)
{
m_status = AaroniaRTSAInputSettings::ConnectionOK;
emit updateStatus(m_status);
}
// read as many bytes as possible into input buffer
qint64 n = mReply->bytesAvailable();
qint64 bs = mBuffer.size();
mBuffer.resize(bs + n);
qint64 done = mReply->read(mBuffer.data() + bs, n);
mBuffer.resize(bs + done);
// intialize parsing
int offset = 0;
int avail = mBuffer.size();
// consume all input data if possible
while (offset < avail)
{
// any samples so far (not looking for meta data)
if (mPacketSamples)
{
// enough samples
if (offset + mPacketSamples * 2 * sizeof(float) <= (unsigned long) avail)
{
// do something with the IQ data
const float *sp = (const float *)(mBuffer.constData() + offset);
SampleVector::iterator it = m_convertBuffer.begin();
m_decimatorsFloatIQ.decimate1(&it, sp, 2*mPacketSamples);
/*m_samplesBuf.clear();
for (int i = 0; i < mPacketSamples*2; i+=2)
{
m_samplesBuf.push_back(Sample(
sp[i] << (SDR_RX_SAMP_SZ - 8),
sp[i+1] << (SDR_RX_SAMP_SZ - 8)
));
}*/
//m_sampleFifo->write(m_samplesBuf.begin(), m_samplesBuf.end());
m_sampleFifo->write(m_convertBuffer.begin(), it);
//qDebug() << "IQ " << sp[0] << ", " << sp[1];
//m_sampleFifo->write()
// consume all samples from the input buffer
offset += mPacketSamples * 2 * sizeof(float);
mPacketSamples = 0;
}
else
{
break;
}
}
else
{
// is there a complete JSON metadata object in the buffer
int split = mBuffer.indexOf('\x1e', offset);
if (split != -1)
{
// Extract it
QByteArray data = mBuffer.mid(offset, split - offset);
offset = split + 1;
// Parse the JSON data
QJsonParseError error;
QJsonDocument jdoc = QJsonDocument::fromJson(data, &error);
if (error.error == QJsonParseError::NoError)
{
// Extract fields of interest
// double startTime = jdoc["startTime"].toDouble(), endTime = jdoc["endTime"].toDouble();
int samples = jdoc["samples"].toInt();
// Dump packet loss
// if (startTime != mPrevTime)
// {
// qDebug() << "AaroniaRTSAInputWorker::onReadyRead: packet loss: "
// << QDateTime::fromMSecsSinceEpoch(startTime * 1000).toString()
// << " D " << endTime - startTime
// << " O " << startTime - mPrevTime
// << " S " << samples
// << " L " << QDateTime::currentMSecsSinceEpoch() / 1000.0 - startTime;
// if (m_status != AaroniaRTSAInputSettings::ConnectionUnstable)
// {
// m_status = AaroniaRTSAInputSettings::ConnectionUnstable;
// emit updateStatus(m_status);
// }
// }
// Switch to data phase
// mPrevTime = endTime;
mPacketSamples = samples;
// qDebug() << jdoc.toJson();
quint64 endFreq = jdoc["endFrequency"].toDouble();
quint64 startFreq = jdoc["startFrequency"].toDouble();
int bw = jdoc["sampleFrequency"].toInt();
quint64 midFreq = (endFreq + startFreq) / 2;
if ((bw != m_sampleRate) || (midFreq != m_centerFrequency))
{
if (m_inputMessageQueue)
{
MsgReportSampleRateAndFrequency *msg = MsgReportSampleRateAndFrequency::create(bw, midFreq);
m_inputMessageQueue->push(msg);
}
}
m_sampleRate = bw;
m_centerFrequency = midFreq;
}
else
{
QTextStream qerr(stderr);
qerr << "Json Parse Error: " + error.errorString();
}
}
else
{
break;
}
}
}
// Remove consumed data from the buffer
mBuffer.remove(0, offset);
}
void AaroniaRTSAInputWorker::handleConfigReply(QNetworkReply* reply)
{
if (reply->operation() == QNetworkAccessManager::GetOperation) // return from GET to /remoteconfig
{
parseConfig(reply->readAll());
}
else if (reply->operation() == QNetworkAccessManager::PutOperation) // return from PUT to /remoteconfig
{
int httpStatusCode = reply->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt();
if ((httpStatusCode / 100) == 2) {
qDebug("AaroniaRTSAInputWorker::handleConfigReply: remoteconfig OK (%d)", httpStatusCode);
} else {
qWarning("AaroniaRTSAInputWorker::handleConfigReply: remoteconfig ended with error (%d)", httpStatusCode);
}
}
reply->deleteLater();
}
void AaroniaRTSAInputWorker::parseConfig(QByteArray bytes)
{
QJsonDocument document = QJsonDocument::fromJson(bytes);
m_iqDemodName = "";
if (document.isObject())
{
QJsonObject documentObject = document.object();
if (documentObject.contains(QStringLiteral("config")))
{
QJsonObject config = documentObject.value(QStringLiteral("config")).toObject();
if (config.contains(QStringLiteral("items")))
{
QJsonArray configItems = config.value(QStringLiteral("items")).toArray();
for (const auto& configIem : configItems)
{
QJsonObject configIemObject = configIem.toObject();
if (configIemObject.contains(QStringLiteral("name")))
{
QString nameItem = configIemObject.value(QStringLiteral("name")).toString();
if (nameItem.startsWith("Block_IQDemodulator"))
{
m_iqDemodName = nameItem;
break;
}
}
}
}
else
{
qDebug() << "AaroniaRTSAInputWorker::parseConfig: config has no items: " << config;
}
}
else
{
qDebug() << "AaroniaRTSAInputWorker::parseConfig: document has no config object: " << documentObject;
}
}
else
{
qDebug() << "AaroniaRTSAInputWorker::parseConfig: document is not an object: " << document;
}
if (m_iqDemodName == "") {
qWarning("AaroniaRTSAInputWorker.parseConfig: could not find IQ demdulator");
} else {
qDebug("AaroniaRTSAInputWorker::parseConfig: IQ demdulator name: %s", qPrintable(m_iqDemodName));
}
}