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

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

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