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

Last change on this file since 2013 was 2001, checked in by mervart, 14 years ago

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