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

Last change on this file since 2467 was 2316, checked in by mervart, 14 years ago

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