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

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