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

Last change on this file since 5527 was 5527, checked in by mervart, 10 years ago
File size: 15.8 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#include "RTCM/GPSDecoder.h"
54#ifdef RTROVER_INTERFACE
55# include "RTRover/bncrtrover.h"
56#endif
57
58using namespace std;
59
60// Constructor
61////////////////////////////////////////////////////////////////////////////
62bncCaster::bncCaster(const QString& outFileName, int port) {
63
64 bncSettings settings;
65
66 connect(this, SIGNAL(newMessage(QByteArray,bool)),
67 BNC_CORE, SLOT(slotMessage(const QByteArray,bool)));
68
69 if ( !outFileName.isEmpty() ) {
70 QString lName = outFileName;
71 expandEnvVar(lName);
72 _outFile = new QFile(lName);
73 if ( Qt::CheckState(settings.value("rnxAppend").toInt()) == Qt::Checked) {
74 _outFile->open(QIODevice::WriteOnly | QIODevice::Append);
75 }
76 else {
77 _outFile->open(QIODevice::WriteOnly);
78 }
79 _out = new QTextStream(_outFile);
80 _out->setRealNumberNotation(QTextStream::FixedNotation);
81 }
82 else {
83 _outFile = 0;
84 _out = 0;
85 }
86
87 _port = port;
88
89 if (_port != 0) {
90 _server = new QTcpServer;
91 if ( !_server->listen(QHostAddress::Any, _port) ) {
92 emit newMessage("bncCaster: Cannot listen on sync port", true);
93 }
94 connect(_server, SIGNAL(newConnection()), this, SLOT(slotNewConnection()));
95 _sockets = new QList<QTcpSocket*>;
96 }
97 else {
98 _server = 0;
99 _sockets = 0;
100 }
101
102 int uPort = settings.value("outUPort").toInt();
103 if (uPort != 0) {
104 _uServer = new QTcpServer;
105 if ( !_uServer->listen(QHostAddress::Any, uPort) ) {
106 emit newMessage("bncCaster: Cannot listen on usync port", true);
107 }
108 connect(_uServer, SIGNAL(newConnection()), this, SLOT(slotNewUConnection()));
109 _uSockets = new QList<QTcpSocket*>;
110 }
111 else {
112 _uServer = 0;
113 _uSockets = 0;
114 }
115
116 int nmeaPort = settings.value("nmeaPort").toInt();
117 if (nmeaPort != 0) {
118 _nmeaServer = new QTcpServer;
119 if ( !_nmeaServer->listen(QHostAddress::Any, nmeaPort) ) {
120 emit newMessage("bncCaster: Cannot listen on port", true);
121 }
122 connect(_nmeaServer, SIGNAL(newConnection()), this, SLOT(slotNewNMEAConnection()));
123 _nmeaSockets = new QList<QTcpSocket*>;
124 }
125 else {
126 _nmeaServer = 0;
127 _nmeaSockets = 0;
128 }
129
130 _epochs = new QMultiMap<long, t_obs>;
131
132 _samplingRate = settings.value("binSampl").toInt();
133 _waitTime = settings.value("waitTime").toInt();
134 _lastDumpSec = 0;
135 _confInterval = -1;
136#ifdef RTROVER_INTERFACE
137 if (!settings.value("rtroverMode").toString().isEmpty()) {
138 _bncRtrover = new t_bncRtrover();
139 _bncRtrover->start();
140 }
141 else {
142 _bncRtrover = 0;
143 }
144#endif
145}
146
147// Destructor
148////////////////////////////////////////////////////////////////////////////
149bncCaster::~bncCaster() {
150
151 QMutexLocker locker(&_mutex);
152
153 QListIterator<bncGetThread*> it(_threads);
154 while(it.hasNext()){
155 bncGetThread* thread = it.next();
156 thread->terminate();
157 }
158 delete _out;
159 delete _outFile;
160 delete _server;
161 delete _sockets;
162 delete _uServer;
163 delete _uSockets;
164 delete _nmeaServer;
165 delete _nmeaSockets;
166 delete _epochs;
167#ifdef RTROVER_INTERFACE
168 if (_bncRtrover) {
169 _bncRtrover->quit();
170 _bncRtrover->deleteLater();
171 }
172#endif
173}
174
175// New Observations
176////////////////////////////////////////////////////////////////////////////
177void bncCaster::slotNewObs(const QByteArray staID, QList<t_obs> obsList) {
178
179 QMutexLocker locker(&_mutex);
180
181 unsigned index = 0;
182 QMutableListIterator<t_obs> it(obsList);
183 while (it.hasNext()) {
184 ++index;
185 t_obs& obs = it.next();
186
187 long iSec = long(floor(obs.GPSWeeks+0.5));
188 long newTime = obs.GPSWeek * 7*24*3600 + iSec;
189
190 // Rename the Station
191 // ------------------
192 strncpy(obs.StatID, staID.constData(),sizeof(obs.StatID));
193 obs.StatID[sizeof(obs.StatID)-1] = '\0';
194
195 // Output into the socket
196 // ----------------------
197 if (_uSockets) {
198
199 ostringstream oStr;
200 oStr.setf(ios::showpoint | ios::fixed);
201 oStr << obs.StatID << " "
202 << setw(4) << obs.GPSWeek << " "
203 << setw(14) << setprecision(7) << obs.GPSWeeks << " "
204 << bncRinex::asciiSatLine(obs) << endl;
205
206 string hlpStr = oStr.str();
207
208 QMutableListIterator<QTcpSocket*> is(*_uSockets);
209 while (is.hasNext()) {
210 QTcpSocket* sock = is.next();
211 if (sock->state() == QAbstractSocket::ConnectedState) {
212 int numBytes = hlpStr.length();
213 if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
214 delete sock;
215 is.remove();
216 }
217 }
218 else if (sock->state() != QAbstractSocket::ConnectingState) {
219 delete sock;
220 is.remove();
221 }
222 }
223 }
224
225 // First time, set the _lastDumpSec immediately
226 // --------------------------------------------
227 if (_lastDumpSec == 0) {
228 _lastDumpSec = newTime - 1;
229 }
230
231 // An old observation - throw it away
232 // ----------------------------------
233 if (newTime <= _lastDumpSec) {
234 if (index == 1) {
235 bncSettings settings;
236 if ( !settings.value("outFile").toString().isEmpty() ||
237 !settings.value("outPort").toString().isEmpty() ) {
238
239 QTime enomtime = QTime(0,0,0).addSecs(iSec);
240
241 emit( newMessage(QString("%1: Old epoch %2 (%3) thrown away")
242 .arg(staID.data()).arg(iSec)
243 .arg(enomtime.toString("HH:mm:ss"))
244 .toAscii(), true) );
245 }
246 }
247 continue;
248 }
249
250 // Save the observation
251 // --------------------
252 _epochs->insert(newTime, obs);
253
254 // Dump Epochs
255 // -----------
256 if (newTime - _waitTime > _lastDumpSec) {
257 dumpEpochs(_lastDumpSec + 1, newTime - _waitTime);
258 _lastDumpSec = newTime - _waitTime;
259 }
260 }
261}
262
263// New Connection
264////////////////////////////////////////////////////////////////////////////
265void bncCaster::slotNewConnection() {
266 _sockets->push_back( _server->nextPendingConnection() );
267 emit( newMessage(QString("New client connection on sync port: # %1")
268 .arg(_sockets->size()).toAscii(), true) );
269}
270
271void bncCaster::slotNewUConnection() {
272 _uSockets->push_back( _uServer->nextPendingConnection() );
273 emit( newMessage(QString("New client connection on usync port: # %1")
274 .arg(_uSockets->size()).toAscii(), true) );
275}
276
277void bncCaster::slotNewNMEAConnection() {
278 _nmeaSockets->push_back( _nmeaServer->nextPendingConnection() );
279 emit( newMessage(QString("New PPP client on port: # %1")
280 .arg(_nmeaSockets->size()).toAscii(), true) );
281}
282
283// Add New Thread
284////////////////////////////////////////////////////////////////////////////
285void bncCaster::addGetThread(bncGetThread* getThread, bool noNewThread) {
286
287 qRegisterMetaType<t_obs>("t_obs");
288 qRegisterMetaType< QList<t_obs> >("QList<t_obs>");
289 qRegisterMetaType<gpsephemeris>("gpsephemeris");
290 qRegisterMetaType<glonassephemeris>("glonassephemeris");
291 qRegisterMetaType<galileoephemeris>("galileoephemeris");
292
293 connect(getThread, SIGNAL(newObs(QByteArray, QList<t_obs>)),
294 this, SLOT(slotNewObs(QByteArray, QList<t_obs>)));
295
296#ifdef RTROVER_INTERFACE
297 if (_bncRtrover) {
298 connect(getThread, SIGNAL(newObs(QByteArray, QList<t_obs>)),
299 _bncRtrover, SLOT(slotNewObs(QByteArray, QList<t_obs>)));
300 }
301#endif
302
303 connect(getThread, SIGNAL(getThreadFinished(QByteArray)),
304 this, SLOT(slotGetThreadFinished(QByteArray)));
305
306 connect(getThread, SIGNAL(newNMEAstr(QByteArray)),
307 this, SLOT(slotNewNMEAstr(QByteArray)));
308
309 _staIDs.push_back(getThread->staID());
310 _threads.push_back(getThread);
311
312 if (noNewThread) {
313 getThread->run();
314 }
315 else {
316 getThread->start();
317 }
318}
319
320// Get Thread destroyed
321////////////////////////////////////////////////////////////////////////////
322void bncCaster::slotGetThreadFinished(QByteArray staID) {
323 QMutexLocker locker(&_mutex);
324
325 QListIterator<bncGetThread*> it(_threads);
326 while (it.hasNext()) {
327 bncGetThread* thread = it.next();
328 if (thread->staID() == staID) {
329 _threads.removeOne(thread);
330 }
331 }
332
333 _staIDs.removeAll(staID);
334 emit( newMessage(
335 QString("Decoding %1 stream(s)").arg(_staIDs.size()).toAscii(), true) );
336 if (_staIDs.size() == 0) {
337 emit(newMessage("bncCaster: Last get thread terminated", true));
338 emit getThreadsFinished();
339 }
340}
341
342// Dump Complete Epochs
343////////////////////////////////////////////////////////////////////////////
344void bncCaster::dumpEpochs(long minTime, long maxTime) {
345
346 for (long sec = minTime; sec <= maxTime; sec++) {
347
348 if ( (_out || _sockets) &&
349 (_samplingRate == 0 || sec % _samplingRate == 0) ) {
350
351 QList<t_obs> allObs = _epochs->values(sec);
352
353 QListIterator<t_obs> it(allObs);
354 bool firstObs = true;
355 while (it.hasNext()) {
356 const t_obs& obs = it.next();
357
358 ostringstream oStr;
359 oStr.setf(ios::showpoint | ios::fixed);
360 if (firstObs) {
361 firstObs = false;
362 oStr << "> " << obs.GPSWeek << ' '
363 << setprecision(7) << obs.GPSWeeks << endl;;
364 }
365 oStr << obs.StatID << ' ' << bncRinex::asciiSatLine(obs) << endl;
366 if (!it.hasNext()) {
367 oStr << endl;
368 }
369 string hlpStr = oStr.str();
370
371 // Output into the File
372 // --------------------
373 if (_out) {
374 *_out << hlpStr.c_str();
375 _out->flush();
376 }
377
378 // Output into the socket
379 // ----------------------
380 if (_sockets) {
381 QMutableListIterator<QTcpSocket*> is(*_sockets);
382 while (is.hasNext()) {
383 QTcpSocket* sock = is.next();
384 if (sock->state() == QAbstractSocket::ConnectedState) {
385 int numBytes = hlpStr.length();
386 if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
387 delete sock;
388 is.remove();
389 }
390 }
391 else if (sock->state() != QAbstractSocket::ConnectingState) {
392 delete sock;
393 is.remove();
394 }
395 }
396 }
397 }
398 }
399 _epochs->remove(sec);
400 }
401}
402
403// Reread configuration (private slot)
404////////////////////////////////////////////////////////////////////////////
405void bncCaster::slotReadMountPoints() {
406
407 bncSettings settings;
408 settings.reRead();
409
410 readMountPoints();
411}
412
413// Read Mountpoints
414////////////////////////////////////////////////////////////////////////////
415void bncCaster::readMountPoints() {
416
417 bncSettings settings;
418
419 // Reread several options
420 // ----------------------
421 _samplingRate = settings.value("binSampl").toInt();
422 _waitTime = settings.value("waitTime").toInt();
423 if (_waitTime < 1) {
424 _waitTime = 1;
425 }
426
427 // Add new mountpoints
428 // -------------------
429 int iMount = -1;
430 QListIterator<QString> it(settings.value("mountPoints").toStringList());
431 while (it.hasNext()) {
432 ++iMount;
433 QStringList hlp = it.next().split(" ");
434 if (hlp.size() <= 1) continue;
435 QUrl url(hlp[0]);
436
437 // Does it already exist?
438 // ----------------------
439 bool existFlg = false;
440 QListIterator<bncGetThread*> iTh(_threads);
441 while (iTh.hasNext()) {
442 bncGetThread* thread = iTh.next();
443 if (thread->mountPoint() == url) {
444 existFlg = true;
445 break;
446 }
447 }
448
449 // New bncGetThread
450 // ----------------
451 if (!existFlg) {
452 QByteArray format = hlp[1].toAscii();
453 QByteArray latitude = hlp[2].toAscii();
454 QByteArray longitude = hlp[3].toAscii();
455 QByteArray nmea = hlp[4].toAscii();
456 QByteArray ntripVersion = hlp[5].toAscii();
457
458 bncGetThread* getThread = new bncGetThread(url, format, latitude,
459 longitude, nmea, ntripVersion);
460 addGetThread(getThread);
461 }
462 }
463
464 // Remove mountpoints
465 // ------------------
466 QListIterator<bncGetThread*> iTh(_threads);
467 while (iTh.hasNext()) {
468 bncGetThread* thread = iTh.next();
469
470 bool existFlg = false;
471 QListIterator<QString> it(settings.value("mountPoints").toStringList());
472 while (it.hasNext()) {
473 QStringList hlp = it.next().split(" ");
474 if (hlp.size() <= 1) continue;
475 QUrl url(hlp[0]);
476
477 if (thread->mountPoint() == url) {
478 existFlg = true;
479 break;
480 }
481 }
482
483 if (!existFlg) {
484 disconnect(thread, 0, 0, 0);
485 _staIDs.removeAll(thread->staID());
486 _threads.removeAll(thread);
487 thread->terminate();
488 }
489 }
490
491 emit mountPointsRead(_threads);
492 emit( newMessage(QString("Configuration read: "
493 + BNC_CORE->confFileName()
494 + ", %1 stream(s)")
495 .arg(_threads.count()).toAscii(), true) );
496
497 // (Re-) Start the configuration timer
498 // -----------------------------------
499 int ms = 0;
500
501 if (_confInterval != -1) {
502 ms = 1000 * _confInterval;
503 }
504 else {
505 QTime currTime = currentDateAndTimeGPS().time();
506 QTime nextShotTime;
507
508 if (settings.value("onTheFlyInterval").toString() == "1 min") {
509 _confInterval = 60;
510 nextShotTime = QTime(currTime.hour(), currTime.minute()+1, 0);
511 }
512 else if (settings.value("onTheFlyInterval").toString() == "5 min") {
513 _confInterval = 300;
514 nextShotTime = QTime(currTime.hour(), currTime.minute()+5, 0);
515 }
516 else if (settings.value("onTheFlyInterval").toString() == "1 hour") {
517 _confInterval = 3600;
518 nextShotTime = QTime(currTime.hour()+1, 0, 0);
519 }
520 else {
521 _confInterval = 86400;
522 nextShotTime = QTime(23, 59, 59, 999);
523 }
524
525 ms = currTime.msecsTo(nextShotTime);
526 if (ms < 30000) {
527 ms = 30000;
528 }
529 }
530
531 QTimer::singleShot(ms, this, SLOT(slotReadMountPoints()));
532}
533
534//
535////////////////////////////////////////////////////////////////////////////
536int bncCaster::myWrite(QTcpSocket* sock, const char* buf, int bufLen) {
537 sock->write(buf, bufLen);
538 for (int ii = 1; ii <= 10; ii++) {
539 if (sock->waitForBytesWritten(10)) { // wait 10 ms
540 return bufLen;
541 }
542 }
543 return -1;
544}
545
546//
547////////////////////////////////////////////////////////////////////////////
548void bncCaster::slotNewNMEAstr(QByteArray str) {
549 if (_nmeaSockets) {
550 QMutableListIterator<QTcpSocket*> is(*_nmeaSockets);
551 while (is.hasNext()) {
552 QTcpSocket* sock = is.next();
553 if (sock->state() == QAbstractSocket::ConnectedState) {
554 sock->write(str);
555 }
556 else if (sock->state() != QAbstractSocket::ConnectingState) {
557 delete sock;
558 is.remove();
559 }
560 }
561 }
562}
Note: See TracBrowser for help on using the repository browser.