source: ntrip/branches/BNC_2.12/src/bnccaster.cpp @ 8398

Last change on this file since 8398 was 8398, checked in by stuerze, 21 months ago

some changes to allow RINEX file generation with correct version 3 file names, 10 Hz observation data processing and resampling (RINEX files, feed engine) and harmonization of all avalable sampling cases (RINEX files, feed engine, reqc edit)

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