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

Last change on this file since 1186 was 1186, checked in by weber, 15 years ago

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