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

Last change on this file since 1170 was 1170, checked in by mervart, 16 years ago

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