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

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

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