source: ntrip/trunk/BNC/bnccaster.cpp@ 2647

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