source: ntrip/trunk/BNC/src/bnccaster.cpp@ 10792

Last change on this file since 10792 was 10753, checked in by stuerze, 3 months ago

NtripSever functionallity added

File size: 18.5 KB
Line 
1// Part of BNC, a utility for retrieving decoding and
2// converting GNSS data streams from NTRIP broadcasters.
3//
4// Copyright (C) 2007
5// German Federal Agency for Cartography and Geodesy (BKG)
6// http://www.bkg.bund.de
7// Czech Technical University Prague, Department of Geodesy
8// http://www.fsv.cvut.cz
9//
10// Email: euref-ip@bkg.bund.de
11//
12// This program is free software; you can redistribute it and/or
13// modify it under the terms of the GNU General Public License
14// as published by the Free Software Foundation, version 2.
15//
16// This program is distributed in the hope that it will be useful,
17// but WITHOUT ANY WARRANTY; without even the implied warranty of
18// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19// GNU General Public License for more details.
20//
21// You should have received a copy of the GNU General Public License
22// along with this program; if not, write to the Free Software
23// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
24
25/* -------------------------------------------------------------------------
26 * BKG NTRIP Client
27 * -------------------------------------------------------------------------
28 *
29 * Class: bncCaster
30 *
31 * Purpose: buffers and disseminates the data
32 *
33 * Author: L. Mervart
34 *
35 * Created: 24-Dec-2005
36 *
37 * Changes:
38 *
39 * -----------------------------------------------------------------------*/
40
41#include <math.h>
42#ifndef WIN32
43#include <unistd.h>
44#else
45#include <windows.h>
46#endif
47#include <iostream>
48#include <iomanip>
49#include <sstream>
50
51#include "bnccaster.h"
52#include "bncrinex.h"
53#include "bnccore.h"
54#include "bncgetthread.h"
55#include "bncutils.h"
56#include "bncsettings.h"
57
58using namespace std;
59
60// Constructor
61////////////////////////////////////////////////////////////////////////////
62bncCaster::bncCaster() {
63
64 bncSettings settings;
65
66 connect(this, SIGNAL(newMessage(QByteArray,bool)),
67 BNC_CORE, SLOT(slotMessage(const QByteArray,bool)));
68
69 _outFile = 0;
70 _out = 0;
71 reopenOutFile();
72
73 // FeedEngine output port
74 // ----------------------
75 int port = settings.value("outPort").toInt();
76 if (port != 0) {
77 _server = new QTcpServer;
78 _server->setProxy(QNetworkProxy::NoProxy);
79 if ( !_server->listen(QHostAddress::LocalHost, port) ) {
80 QString message = "bncCaster: Cannot listen on sync port "
81 + QByteArray::number(port) + ": "
82 + _server->errorString();
83 emit newMessage(message.toLatin1(), true);
84 }
85 connect(_server, SIGNAL(newConnection()), this, SLOT(slotNewConnection()));
86 _sockets = new QList<QTcpSocket*>;
87 }
88 else {
89 _server = 0;
90 _sockets = 0;
91 }
92
93 int uPort = settings.value("outUPort").toInt();
94 if (uPort != 0) {
95 _uServer = new QTcpServer;
96 _uServer->setProxy(QNetworkProxy::NoProxy);
97 if ( !_uServer->listen(QHostAddress::LocalHost, uPort) ) {
98 QString message = "bncCaster: Cannot listen on usync port "
99 + QByteArray::number(uPort) + ": "
100 + _uServer->errorString();
101 emit newMessage(message.toLatin1(), true);
102 }
103 connect(_uServer, SIGNAL(newConnection()), this, SLOT(slotNewUConnection()));
104 _uSockets = new QList<QTcpSocket*>;
105 }
106 else {
107 _uServer = 0;
108 _uSockets = 0;
109 }
110
111 _outLockTime = settings.value("outLockTime",false).toBool();
112 _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
113 _outWait = settings.value("outWait").toDouble();
114 if (_outWait <= 0.0) {
115 _outWait = 0.01;
116 }
117 _confInterval = -1;
118
119 // Miscellaneous output port
120 // -------------------------
121 _miscMount = settings.value("miscMount").toString();
122 _miscPort = settings.value("miscPort").toInt();
123 if (!_miscMount.isEmpty() && _miscPort != 0) {
124 _miscServer = new QTcpServer;
125 _miscServer->setProxy(QNetworkProxy::NoProxy);
126 if ( !_miscServer->listen(QHostAddress::LocalHost, _miscPort) ) {
127 QString message = "bncCaster: Cannot listen on miscellaneous output port "
128 + QByteArray::number(_miscPort) + ": "
129 + _miscServer->errorString();
130 emit newMessage(message.toLatin1(), true);
131 }
132 connect(_miscServer, SIGNAL(newConnection()), this, SLOT(slotNewMiscConnection()));
133 _miscSockets = new QList<QTcpSocket*>;
134 }
135 else {
136 _miscServer = 0;
137 _miscSockets = 0;
138 }
139
140 // RawUploadCaster
141 _bncRawUploadCaster = new bncRawUploadCaster();
142
143}
144
145// Destructor
146////////////////////////////////////////////////////////////////////////////
147bncCaster::~bncCaster() {
148
149 QMutexLocker locker(&_mutex);
150
151 QListIterator<bncGetThread*> it(_threads);
152
153 while(it.hasNext()){
154 bncGetThread* thread = it.next();
155 disconnect(thread, 0, 0, 0);
156 _staIDs.removeAll(thread->staID());
157 _threads.removeAll(thread);
158 thread->terminate();
159 }
160
161 delete _out;
162 delete _outFile;
163 delete _server;
164 delete _sockets;
165 delete _uServer;
166 delete _uSockets;
167 delete _miscServer;
168 delete _miscSockets;
169 delete _bncRawUploadCaster;
170}
171
172// New Observations
173////////////////////////////////////////////////////////////////////////////
174void bncCaster::slotNewObs(const QByteArray staID, QList<t_satObs> obsList) {
175
176 QMutexLocker locker(&_mutex);
177
178 reopenOutFile();
179
180 unsigned index = 0;
181 QMutableListIterator<t_satObs> it(obsList);
182 while (it.hasNext()) {
183 ++index;
184 t_satObs& obs = it.next();
185
186 // Rename the Station
187 // ------------------
188 obs._staID = staID.data();
189
190 // Update/Set Slip Counters
191 // ------------------------
192 setSlipCounters(obs);
193
194 // Output into the socket
195 // ----------------------
196 if (_uSockets) {
197
198 ostringstream oStr;
199 oStr.setf(ios::showpoint | ios::fixed);
200 oStr << obs._staID << " "
201 << setw(4) << obs._time.gpsw() << " "
202 << setw(14) << setprecision(7) << obs._time.gpssec() << " "
203 << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
204
205 string hlpStr = oStr.str();
206
207 QMutableListIterator<QTcpSocket*> is(*_uSockets);
208 while (is.hasNext()) {
209 QTcpSocket* sock = is.next();
210 if (sock->state() == QAbstractSocket::ConnectedState) {
211 int numBytes = hlpStr.length();
212 if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
213 delete sock;
214 is.remove();
215 }
216 }
217 else if (sock->state() != QAbstractSocket::ConnectingState) {
218 delete sock;
219 is.remove();
220 }
221 }
222 }
223
224 // First time: set the _lastDumpTime
225 // ---------------------------------
226 if (!_lastDumpTime.valid()) {
227 _lastDumpTime = obs._time - 1.0;
228 }
229
230 // An old observation - throw it away
231 // ----------------------------------
232 if (obs._time <= _lastDumpTime) {
233 if (index == 1) {
234 bncSettings settings;
235 if ( !settings.value("outFile").toString().isEmpty() ||
236 !settings.value("outPort").toString().isEmpty() ) {
237 emit( newMessage(QString("%1: Old or erroneous observation epoch %2 thrown away from %3")
238 .arg(staID.data())
239 .arg(string(obs._time).c_str())
240 .arg(obs._prn.toString().c_str())
241 .toLatin1(), true) );
242 }
243 }
244 continue;
245 }
246
247 // Save the observation
248 // --------------------
249 _epochs[obs._time].append(obs);
250
251 // Dump Epochs
252 // -----------
253 if ((obs._time - _outWait) > _lastDumpTime) {
254 dumpEpochs(obs._time - _outWait);
255 _lastDumpTime = obs._time - _outWait;
256 }
257 }
258}
259
260// New Connection
261////////////////////////////////////////////////////////////////////////////
262void bncCaster::slotNewConnection() {
263 _sockets->push_back( _server->nextPendingConnection() );
264 emit( newMessage(QString("New client connection on sync port: # %1")
265 .arg(_sockets->size()).toLatin1(), true) );
266}
267
268void bncCaster::slotNewUConnection() {
269 _uSockets->push_back( _uServer->nextPendingConnection() );
270 emit( newMessage(QString("New client connection on usync port: # %1")
271 .arg(_uSockets->size()).toLatin1(), true) );
272}
273
274// Add New Thread
275////////////////////////////////////////////////////////////////////////////
276void bncCaster::addGetThread(bncGetThread* getThread, bool noNewThread) {
277
278 qRegisterMetaType<t_satObs>("t_satObs");
279 qRegisterMetaType< QList<t_satObs> >("QList<t_satObs>");
280
281 connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
282 this, SLOT(slotNewObs(QByteArray, QList<t_satObs>)));
283
284 connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
285 this, SIGNAL(newObs(QByteArray, QList<t_satObs>)));
286
287 connect(getThread, SIGNAL(newRawData(QByteArray, QByteArray)),
288 this, SLOT(slotNewRawData(QByteArray, QByteArray)));
289
290 connect(getThread, SIGNAL(newRawData(QByteArray, QByteArray)),
291 _bncRawUploadCaster, SLOT(slotNewRawData(QByteArray, QByteArray)));
292
293 connect(getThread, SIGNAL(getThreadFinished(QByteArray)),
294 this, SLOT(slotGetThreadFinished(QByteArray)));
295
296 _staIDs.push_back(getThread->staID());
297 _threads.push_back(getThread);
298
299 if (noNewThread) {
300 getThread->run();
301 }
302 else {
303 getThread->start();
304 }
305}
306
307// Get Thread destroyed
308////////////////////////////////////////////////////////////////////////////
309void bncCaster::slotGetThreadFinished(QByteArray staID) {
310 //QMutexLocker locker(&_mutex);
311
312 QListIterator<bncGetThread*> it(_threads);
313 while (it.hasNext()) {
314 bncGetThread* thread = it.next();
315 if (thread->staID() == staID) {
316 _threads.removeOne(thread);
317 }
318 }
319 _staIDs.removeAll(staID);
320 emit( newMessage(
321 QString("Decoding %1 stream(s)").arg(_staIDs.size()).toLatin1(), true) );
322 if (_staIDs.size() == 0) {
323 emit getThreadsFinished();
324 }
325
326}
327
328// Dump Complete Epochs
329////////////////////////////////////////////////////////////////////////////
330void bncCaster::dumpEpochs(const bncTime& maxTime) {
331
332 QMutableMapIterator<bncTime, QList<t_satObs> > itEpo(_epochs);
333 while (itEpo.hasNext()) {
334 itEpo.next();
335 const bncTime& epoTime = itEpo.key();
336 if (epoTime <= maxTime) {
337 const QList<t_satObs>& allObs = itEpo.value();
338 int sec = int(nint(epoTime.gpssec()*10));
339 if ( (_out || _sockets) && (sec % (_samplingRateMult10) == 0) ) {
340 QListIterator<t_satObs> it(allObs);
341 bool firstObs = true;
342 while (it.hasNext()) {
343 const t_satObs& obs = it.next();
344
345 ostringstream oStr;
346 oStr.setf(ios::showpoint | ios::fixed);
347 if (firstObs) {
348 firstObs = false;
349 oStr << "> " << obs._time.gpsw() << ' '
350 << setprecision(7) << obs._time.gpssec() << endl;
351 }
352 oStr << obs._staID << ' '
353 << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
354 if (!it.hasNext()) {
355 oStr << endl;
356 }
357 string hlpStr = oStr.str();
358
359 // Output into the File
360 // --------------------
361 if (_out) {
362 *_out << hlpStr.c_str();
363 _out->flush();
364 }
365
366 // Output into the socket
367 // ----------------------
368 if (_sockets) {
369 QMutableListIterator<QTcpSocket*> is(*_sockets);
370 while (is.hasNext()) {
371 QTcpSocket* sock = is.next();
372 if (sock->state() == QAbstractSocket::ConnectedState) {
373 int numBytes = hlpStr.length();
374 if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
375 delete sock;
376 is.remove();
377 }
378 }
379 else if (sock->state() != QAbstractSocket::ConnectingState) {
380 delete sock;
381 is.remove();
382 }
383 }
384 }
385 }
386 }
387 //_epochs.remove(epoTime);
388 itEpo.remove();
389 }
390 }
391}
392
393// Reread configuration (private slot)
394////////////////////////////////////////////////////////////////////////////
395void bncCaster::slotReadMountPoints() {
396
397 bncSettings settings;
398 settings.reRead();
399
400 readMountPoints();
401}
402
403// Read Mountpoints
404////////////////////////////////////////////////////////////////////////////
405void bncCaster::readMountPoints() {
406
407 bncSettings settings;
408
409 // Reread several options
410 // ----------------------
411 _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
412
413 _outWait = settings.value("outWait").toDouble();
414 if (_outWait <= 0.0) {
415 _outWait = 0.01;
416 }
417
418 // Add new mountpoints
419 // -------------------
420 int iMount = -1;
421 QListIterator<QString> it(settings.value("mountPoints").toStringList());
422 while (it.hasNext()) {
423 ++iMount;
424 QStringList hlp = it.next().split(" ");
425 if (hlp.size() < 7) continue;
426 QUrl url(hlp[0]);
427
428 // Does it already exist?
429 // ----------------------
430 bool existFlg = false;
431 QListIterator<bncGetThread*> iTh(_threads);
432 while (iTh.hasNext()) {
433 bncGetThread* thread = iTh.next();
434 if (thread->mountPoint() == url) {
435 existFlg = true;
436 break;
437 }
438 }
439
440 // New bncGetThread
441 // ----------------
442 if (!existFlg) {
443 QByteArray format = hlp[1].toLatin1();
444 QByteArray latitude = hlp[3].toLatin1();
445 QByteArray longitude = hlp[4].toLatin1();
446 QByteArray nmea = hlp[5].toLatin1();
447 QByteArray ntripVersion = hlp[6].toLatin1();
448
449 bncGetThread* getThread = new bncGetThread(url, format, latitude,
450 longitude, nmea, ntripVersion);
451 addGetThread(getThread);
452
453 }
454 }
455
456 // Remove mountpoints
457 // ------------------
458 QListIterator<bncGetThread*> iTh(_threads);
459 while (iTh.hasNext()) {
460 bncGetThread* thread = iTh.next();
461
462 bool existFlg = false;
463 QListIterator<QString> it(settings.value("mountPoints").toStringList());
464 while (it.hasNext()) {
465 QStringList hlp = it.next().split(" ");
466 if (hlp.size() <= 1) continue;
467 QUrl url(hlp[0]);
468
469 if (thread->mountPoint() == url) {
470 existFlg = true;
471 break;
472 }
473 }
474
475 if (!existFlg) {
476 disconnect(thread, 0, 0, 0);
477 _staIDs.removeAll(thread->staID());
478 _threads.removeAll(thread);
479 thread->terminate();
480 }
481 }
482
483 emit mountPointsRead(_threads);
484 emit(newMessage(QString("Configuration read: " + BNC_CORE->confFileName() + ", %1 stream(s)").arg(_threads.count()).toLatin1(), true));
485
486 // (Re-) Start the configuration timer
487 // -----------------------------------
488 if (settings.value("onTheFlyInterval").toString() == "no") {
489 return;
490 }
491
492 int ms = 0;
493 if (_confInterval != -1) {
494 ms = 1000 * _confInterval;
495 }
496 else {
497 QTime currTime = currentDateAndTimeGPS().time();
498 QTime nextShotTime;
499 if (settings.value("onTheFlyInterval").toString() == "1 min") {
500 _confInterval = 60;
501 nextShotTime = QTime(currTime.hour(), currTime.minute()+1, 0);
502 }
503 else if (settings.value("onTheFlyInterval").toString() == "5 min") {
504 _confInterval = 300;
505 nextShotTime = QTime(currTime.hour(), currTime.minute()+5, 0);
506 }
507 else if (settings.value("onTheFlyInterval").toString() == "1 hour") {
508 _confInterval = 3600;
509 nextShotTime = QTime(currTime.hour()+1, 0, 0);
510 }
511 else {
512 _confInterval = 86400;
513 nextShotTime = QTime(23, 59, 59, 999);
514 }
515
516 ms = currTime.msecsTo(nextShotTime);
517 if (ms < 30000) {
518 ms = 30000;
519 }
520 }
521
522 QTimer::singleShot(ms, this, SLOT(slotReadMountPoints()));
523}
524
525//
526////////////////////////////////////////////////////////////////////////////
527int bncCaster::myWrite(QTcpSocket* sock, const char* buf, int bufLen) {
528 sock->write(buf, bufLen);
529 for (int ii = 1; ii <= 10; ii++) {
530 if (sock->waitForBytesWritten(10)) { // wait 10 ms
531 return bufLen;
532 }
533 }
534 return -1;
535}
536
537//
538////////////////////////////////////////////////////////////////////////////
539void bncCaster::reopenOutFile() {
540
541 bncSettings settings;
542
543 QString outFileName = settings.value("outFile").toString();
544 if ( !outFileName.isEmpty() ) {
545 expandEnvVar(outFileName);
546 if (!_outFile || _outFile->fileName() != outFileName) {
547 delete _out;
548 delete _outFile;
549 _outFile = new QFile(outFileName);
550 if ( Qt::CheckState(settings.value("rnxAppend").toInt()) == Qt::Checked) {
551 _outFile->open(QIODevice::WriteOnly | QIODevice::Append);
552 }
553 else {
554 _outFile->open(QIODevice::WriteOnly);
555 }
556 _out = new QTextStream(_outFile);
557 _out->setRealNumberNotation(QTextStream::FixedNotation);
558 }
559 }
560 else {
561 delete _out; _out = 0;
562 delete _outFile; _outFile = 0;
563 }
564}
565
566// Output into the Miscellaneous socket
567////////////////////////////////////////////////////////////////////////////
568void bncCaster::slotNewRawData(QByteArray staID, QByteArray data) {
569 if (_miscSockets && (_miscMount == "ALL" || _miscMount == staID)) {
570 QMutableListIterator<QTcpSocket*> is(*_miscSockets);
571 while (is.hasNext()) {
572 QTcpSocket* sock = is.next();
573 if (sock->state() == QAbstractSocket::ConnectedState) {
574 sock->write(data);
575 }
576 else if (sock->state() != QAbstractSocket::ConnectingState) {
577 delete sock;
578 is.remove();
579 }
580 }
581 }
582}
583
584// New Connection
585////////////////////////////////////////////////////////////////////////////
586void bncCaster::slotNewMiscConnection() {
587 _miscSockets->push_back( _miscServer->nextPendingConnection() );
588 emit( newMessage(QString("New client connection on Miscellaneous Output Port: # %1")
589 .arg(_miscSockets->size()).toLatin1(), true) );
590}
591
592// Set/Update Slip Counters
593////////////////////////////////////////////////////////////////////////////
594void bncCaster::setSlipCounters(t_satObs& obs) {
595
596 double minLockTime = -1.0;
597 for (unsigned ii = 0; ii < obs._obs.size(); ii++) {
598 const t_frqObs* frqObs = obs._obs[ii];
599 if (frqObs->_lockTimeValid) {
600 if (minLockTime == -1.0 || minLockTime < frqObs->_lockTime) {
601 minLockTime = frqObs->_lockTime;
602 }
603 }
604 }
605
606 if (minLockTime == -1.0) {
607 return;
608 }
609
610 QMap<t_prn, double>& ltMap = _lockTimeMap[obs._staID];
611 QMap<t_prn, int>& jcMap = _jumpCounterMap[obs._staID];
612 QMap<t_prn, double>::const_iterator it = ltMap.find(obs._prn);
613 if (it == ltMap.end()) { // init satellite
614 ltMap[obs._prn] = minLockTime;
615 jcMap[obs._prn] = 0;
616 }
617 else {
618 if (minLockTime < ltMap[obs._prn]) {
619 jcMap[obs._prn] += 1;
620 if (jcMap[obs._prn] > 9999) {
621 jcMap[obs._prn] = 0;
622 }
623 }
624 ltMap[obs._prn] = minLockTime;
625 }
626 for (unsigned ii = 0; ii < obs._obs.size(); ii++) {
627 t_frqObs* frqObs = obs._obs[ii];
628 frqObs->_slipCounter = jcMap[obs._prn];
629 }
630}
Note: See TracBrowser for help on using the repository browser.