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

Last change on this file since 9018 was 9018, checked in by stuerze, 10 months ago

harmonization of MSM feed engine output

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