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

Last change on this file since 10518 was 10503, checked in by stuerze, 5 months ago

add an docoder-string ZERO2FILE

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