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

Last change on this file since 9124 was 9124, checked in by stuerze, 20 months ago

minor changes regarding latency check, ssr and long mountpoint names

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