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

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

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