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

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