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

Last change on this file since 4765 was 4765, checked in by mervart, 12 years ago
File size: 15.4 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 "bncapp.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 (bncApp*) qApp, 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 _bncRtrover = new t_bncRtrover();
138 _bncRtrover->start();
139#endif
140}
141
142// Destructor
143////////////////////////////////////////////////////////////////////////////
144bncCaster::~bncCaster() {
145
146 QMutexLocker locker(&_mutex);
147
148 QListIterator<bncGetThread*> it(_threads);
149 while(it.hasNext()){
150 bncGetThread* thread = it.next();
151 thread->terminate();
152 }
153 delete _out;
154 delete _outFile;
155 delete _server;
156 delete _sockets;
157 delete _uServer;
158 delete _uSockets;
159 delete _nmeaServer;
160 delete _nmeaSockets;
161 delete _epochs;
162#ifdef RTROVER_INTERFACE
163 _bncRtrover->quit();
164 _bncRtrover->deleteLater();
165#endif
166}
167
168// New Observations
169////////////////////////////////////////////////////////////////////////////
170void bncCaster::newObs(const QByteArray staID, bool firstObs, t_obs obs) {
171
172 QMutexLocker locker(&_mutex);
173
174 long iSec = long(floor(obs.GPSWeeks+0.5));
175 long newTime = obs.GPSWeek * 7*24*3600 + iSec;
176
177 // Rename the Station
178 // ------------------
179 strncpy(obs.StatID, staID.constData(),sizeof(obs.StatID));
180 obs.StatID[sizeof(obs.StatID)-1] = '\0';
181
182 // Output into the socket
183 // ----------------------
184 if (_uSockets) {
185
186 ostringstream oStr;
187 oStr.setf(ios::showpoint | ios::fixed);
188 oStr << obs.StatID << " "
189 << obs.GPSWeek << " "
190 << setprecision(7) << obs.GPSWeeks << " "
191 << bncRinex::asciiSatLine(obs) << 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 _lastDumpSec immediately
213 // --------------------------------------------
214 if (_lastDumpSec == 0) {
215 _lastDumpSec = newTime - 1;
216 }
217
218 // An old observation - throw it away
219 // ----------------------------------
220 if (newTime <= _lastDumpSec) {
221 if (firstObs) {
222 bncSettings settings;
223 if ( !settings.value("outFile").toString().isEmpty() ||
224 !settings.value("outPort").toString().isEmpty() ) {
225
226 QTime enomtime = QTime(0,0,0).addSecs(iSec);
227
228 emit( newMessage(QString("%1: Old epoch %2 (%3) thrown away")
229 .arg(staID.data()).arg(iSec)
230 .arg(enomtime.toString("HH:mm:ss"))
231 .toAscii(), true) );
232 }
233 }
234 return;
235 }
236
237 // Save the observation
238 // --------------------
239 _epochs->insert(newTime, obs);
240
241 // Dump Epochs
242 // -----------
243 if (newTime - _waitTime > _lastDumpSec) {
244 dumpEpochs(_lastDumpSec + 1, newTime - _waitTime);
245 _lastDumpSec = newTime - _waitTime;
246 }
247}
248
249// New Connection
250////////////////////////////////////////////////////////////////////////////
251void bncCaster::slotNewConnection() {
252 _sockets->push_back( _server->nextPendingConnection() );
253 emit( newMessage(QString("New client connection on sync port: # %1")
254 .arg(_sockets->size()).toAscii(), true) );
255}
256
257void bncCaster::slotNewUConnection() {
258 _uSockets->push_back( _uServer->nextPendingConnection() );
259 emit( newMessage(QString("New client connection on usync port: # %1")
260 .arg(_uSockets->size()).toAscii(), true) );
261}
262
263void bncCaster::slotNewNMEAConnection() {
264 _nmeaSockets->push_back( _nmeaServer->nextPendingConnection() );
265 emit( newMessage(QString("New PPP client on port: # %1")
266 .arg(_nmeaSockets->size()).toAscii(), true) );
267}
268
269// Add New Thread
270////////////////////////////////////////////////////////////////////////////
271void bncCaster::addGetThread(bncGetThread* getThread, bool noNewThread) {
272
273 qRegisterMetaType<t_obs>("t_obs");
274 qRegisterMetaType<gpsephemeris>("gpsephemeris");
275 qRegisterMetaType<glonassephemeris>("glonassephemeris");
276 qRegisterMetaType<galileoephemeris>("galileoephemeris");
277
278 connect(getThread, SIGNAL(newObs(QByteArray, bool, t_obs)),
279 this, SLOT(newObs(QByteArray, bool, t_obs)));
280
281#ifdef RTROVER_INTERFACE
282 connect(getThread, SIGNAL(newObs(QByteArray, bool, t_obs)),
283 _bncRtrover, SLOT(slotNewObs(QByteArray, bool, t_obs)));
284#endif
285
286 connect(getThread, SIGNAL(getThreadFinished(QByteArray)),
287 this, SLOT(slotGetThreadFinished(QByteArray)));
288
289 connect(getThread, SIGNAL(newNMEAstr(QByteArray)),
290 this, SLOT(slotNewNMEAstr(QByteArray)));
291
292 _staIDs.push_back(getThread->staID());
293 _threads.push_back(getThread);
294
295 if (noNewThread) {
296 getThread->run();
297 }
298 else {
299 getThread->start();
300 }
301}
302
303// Get Thread destroyed
304////////////////////////////////////////////////////////////////////////////
305void bncCaster::slotGetThreadFinished(QByteArray staID) {
306 QMutexLocker locker(&_mutex);
307
308 QListIterator<bncGetThread*> it(_threads);
309 while (it.hasNext()) {
310 bncGetThread* thread = it.next();
311 if (thread->staID() == staID) {
312 _threads.removeOne(thread);
313 }
314 }
315
316 _staIDs.removeAll(staID);
317 emit( newMessage(
318 QString("Decoding %1 stream(s)").arg(_staIDs.size()).toAscii(), true) );
319 if (_staIDs.size() == 0) {
320 emit(newMessage("bncCaster: Last get thread terminated", true));
321 emit getThreadsFinished();
322 }
323}
324
325// Dump Complete Epochs
326////////////////////////////////////////////////////////////////////////////
327void bncCaster::dumpEpochs(long minTime, long maxTime) {
328
329 for (long sec = minTime; sec <= maxTime; sec++) {
330
331 QList<t_obs> allObs = _epochs->values(sec);
332
333 QListIterator<t_obs> it(allObs);
334 while (it.hasNext()) {
335 const t_obs& obs = it.next();
336
337 if (_samplingRate == 0 || sec % _samplingRate == 0) {
338
339 if (_out || _sockets) {
340 ostringstream oStr;
341 oStr.setf(ios::showpoint | ios::fixed);
342 oStr << obs.StatID << " "
343 << obs.GPSWeek << " "
344 << setprecision(7) << obs.GPSWeeks << " "
345 << bncRinex::asciiSatLine(obs) << endl;
346 if (!it.hasNext()) {
347 oStr << endl;
348 }
349 string hlpStr = oStr.str();
350
351 // Output into the File
352 // --------------------
353 if (_out) {
354 *_out << hlpStr.c_str();
355 _out->flush();
356 }
357
358 // Output into the socket
359 // ----------------------
360 if (_sockets) {
361 QMutableListIterator<QTcpSocket*> is(*_sockets);
362 while (is.hasNext()) {
363 QTcpSocket* sock = is.next();
364 if (sock->state() == QAbstractSocket::ConnectedState) {
365 int numBytes = hlpStr.length();
366 if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
367 delete sock;
368 is.remove();
369 }
370 }
371 else if (sock->state() != QAbstractSocket::ConnectingState) {
372 delete sock;
373 is.remove();
374 }
375 }
376 }
377 }
378 }
379
380 _epochs->remove(sec);
381 }
382 }
383}
384
385// Reread configuration (private slot)
386////////////////////////////////////////////////////////////////////////////
387void bncCaster::slotReadMountPoints() {
388
389 bncSettings settings;
390 settings.reRead();
391
392 readMountPoints();
393}
394
395// Read Mountpoints
396////////////////////////////////////////////////////////////////////////////
397void bncCaster::readMountPoints() {
398
399 bncSettings settings;
400
401 // Reread several options
402 // ----------------------
403 _samplingRate = settings.value("binSampl").toInt();
404 _waitTime = settings.value("waitTime").toInt();
405 if (_waitTime < 1) {
406 _waitTime = 1;
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() <= 1) 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].toAscii();
435 QByteArray latitude = hlp[2].toAscii();
436 QByteArray longitude = hlp[3].toAscii();
437 QByteArray nmea = hlp[4].toAscii();
438 QByteArray ntripVersion = hlp[5].toAscii();
439
440 bncGetThread* getThread = new bncGetThread(url, format, latitude,
441 longitude, nmea, ntripVersion);
442 addGetThread(getThread);
443 }
444 }
445
446 // Remove mountpoints
447 // ------------------
448 QListIterator<bncGetThread*> iTh(_threads);
449 while (iTh.hasNext()) {
450 bncGetThread* thread = iTh.next();
451
452 bool existFlg = false;
453 QListIterator<QString> it(settings.value("mountPoints").toStringList());
454 while (it.hasNext()) {
455 QStringList hlp = it.next().split(" ");
456 if (hlp.size() <= 1) continue;
457 QUrl url(hlp[0]);
458
459 if (thread->mountPoint() == url) {
460 existFlg = true;
461 break;
462 }
463 }
464
465 if (!existFlg) {
466 disconnect(thread, 0, 0, 0);
467 _staIDs.removeAll(thread->staID());
468 _threads.removeAll(thread);
469 thread->terminate();
470 }
471 }
472
473 emit mountPointsRead(_threads);
474 emit( newMessage(QString("Configuration read: "
475 + ((bncApp*) qApp)->confFileName()
476 + ", %1 stream(s)")
477 .arg(_threads.count()).toAscii(), true) );
478
479 // (Re-) Start the configuration timer
480 // -----------------------------------
481 int ms = 0;
482
483 if (_confInterval != -1) {
484 ms = 1000 * _confInterval;
485 }
486 else {
487 QTime currTime = currentDateAndTimeGPS().time();
488 QTime nextShotTime;
489
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::slotNewNMEAstr(QByteArray str) {
531 if (_nmeaSockets) {
532 QMutableListIterator<QTcpSocket*> is(*_nmeaSockets);
533 while (is.hasNext()) {
534 QTcpSocket* sock = is.next();
535 if (sock->state() == QAbstractSocket::ConnectedState) {
536 sock->write(str);
537 }
538 else if (sock->state() != QAbstractSocket::ConnectingState) {
539 delete sock;
540 is.remove();
541 }
542 }
543 }
544}
Note: See TracBrowser for help on using the repository browser.