source: ntrip/branches/BNC_2.13.beta/src/bnccaster.cpp@ 10113

Last change on this file since 10113 was 9958, checked in by stuerze, 2 years ago

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