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

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