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

Last change on this file since 293 was 280, checked in by mervart, 18 years ago

* empty log message *

File size: 7.6 KB
Line 
1// Part of BNC, a utility for retrieving decoding and
2// converting GNSS data streams from NTRIP broadcasters,
3// written by Leos Mervart.
4//
5// Copyright (C) 2006
6// German Federal Agency for Cartography and Geodesy (BKG)
7// http://www.bkg.bund.de
8// Czech Technical University Prague, Department of Advanced Geodesy
9// http://www.fsv.cvut.cz
10//
11// Email: euref-ip@bkg.bund.de
12//
13// This program is free software; you can redistribute it and/or
14// modify it under the terms of the GNU General Public License
15// as published by the Free Software Foundation, version 2.
16//
17// This program is distributed in the hope that it will be useful,
18// but WITHOUT ANY WARRANTY; without even the implied warranty of
19// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20// GNU General Public License for more details.
21//
22// You should have received a copy of the GNU General Public License
23// along with this program; if not, write to the Free Software
24// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
25
26/* -------------------------------------------------------------------------
27 * BKG NTRIP Client
28 * -------------------------------------------------------------------------
29 *
30 * Class: bncCaster
31 *
32 * Purpose: buffers and disseminates the data
33 *
34 * Author: L. Mervart
35 *
36 * Created: 24-Dec-2005
37 *
38 * Changes:
39 *
40 * -----------------------------------------------------------------------*/
41
42#include <math.h>
43
44#include "bnccaster.h"
45#include "bncgetthread.h"
46#include "bncutils.h"
47#include "RTCM/GPSDecoder.h"
48
49// Constructor
50////////////////////////////////////////////////////////////////////////////
51bncCaster::bncCaster(const QString& outFileName, int port) {
52
53 QSettings settings;
54
55 if ( !outFileName.isEmpty() ) {
56 QString lName = outFileName;
57 expandEnvVar(lName);
58 _outFile = new QFile(lName);
59 if ( Qt::CheckState(settings.value("rnxAppend").toInt()) == Qt::Checked) {
60 _outFile->open(QIODevice::WriteOnly | QIODevice::Append);
61 }
62 else {
63 _outFile->open(QIODevice::WriteOnly);
64 }
65 _out = new QTextStream(_outFile);
66 _out->setRealNumberNotation(QTextStream::FixedNotation);
67 _out->setRealNumberPrecision(5);
68 }
69 else {
70 _outFile = 0;
71 _out = 0;
72 }
73
74 _port = port;
75
76 if (_port != 0) {
77 _server = new QTcpServer;
78 _server->listen(QHostAddress::Any, _port);
79 connect(_server, SIGNAL(newConnection()), this, SLOT(slotNewConnection()));
80 _sockets = new QList<QTcpSocket*>;
81 }
82 else {
83 _server = 0;
84 _sockets = 0;
85 }
86
87 _epochs = new QMultiMap<long, Observation*>;
88
89 _lastDumpSec = 0;
90
91 _samplingRate = settings.value("rnxSampl").toInt();
92 _waitTime = settings.value("waitTime").toInt();
93 if (_waitTime < 1) {
94 _waitTime = 1;
95 }
96}
97
98// Destructor
99////////////////////////////////////////////////////////////////////////////
100bncCaster::~bncCaster() {
101 QListIterator<bncGetThread*> it(_threads);
102 while(it.hasNext()){
103 bncGetThread* thread = it.next();
104 thread->terminate();
105 thread->wait();
106 delete thread;
107 }
108 delete _out;
109 delete _outFile;
110 delete _server;
111 delete _sockets;
112 delete _epochs;
113}
114
115// New Observations
116////////////////////////////////////////////////////////////////////////////
117void bncCaster::newObs(const QByteArray& staID, const QUrl& mountPoint,
118 bool firstObs, Observation* obs) {
119
120 QMutexLocker locker(&_mutex);
121
122 long iSec = long(floor(obs->GPSWeeks+0.5));
123 long newTime = obs->GPSWeek * 7*24*3600 + iSec;
124
125 // Rename the Station
126 // ------------------
127 strncpy(obs->StatID, staID.constData(),sizeof(obs->StatID));
128
129 // Prepare RINEX Output
130 // --------------------
131 if (_rinexWriters.find(obs->StatID) == _rinexWriters.end()) {
132 _rinexWriters.insert(obs->StatID, new bncRinex(obs->StatID, mountPoint));
133 }
134 bncRinex* rnx = _rinexWriters.find(obs->StatID).value();
135 if (_samplingRate == 0 || iSec % _samplingRate == 0) {
136 rnx->deepCopy(obs);
137 }
138 rnx->dumpEpoch(newTime);
139
140 // First time, set the _lastDumpSec immediately
141 // --------------------------------------------
142 if (_lastDumpSec == 0) {
143 _lastDumpSec = newTime - 1;
144 }
145
146 // An old observation - throw it away
147 // ----------------------------------
148 if (newTime <= _lastDumpSec) {
149 if (firstObs) {
150 QSettings settings;
151 if ( !settings.value("outFile").toString().isEmpty() ||
152 !settings.value("outPort").toString().isEmpty() ) {
153 emit( newMessage(QString("Station %1: old epoch %2 thrown away")
154 .arg(staID.data()).arg(iSec).toAscii()) );
155 }
156 }
157 delete obs;
158 return;
159 }
160
161 // Save the observation
162 // --------------------
163 _epochs->insert(newTime, obs);
164
165 // Dump older epochs
166 // -----------------
167 dumpEpochs(_lastDumpSec + 1, newTime - _waitTime);
168
169 if (_lastDumpSec < newTime - _waitTime) {
170 _lastDumpSec = newTime - _waitTime;
171 }
172}
173
174// New Connection
175////////////////////////////////////////////////////////////////////////////
176void bncCaster::slotNewConnection() {
177 _sockets->push_back( _server->nextPendingConnection() );
178}
179
180// Add New Thread
181////////////////////////////////////////////////////////////////////////////
182void bncCaster::addGetThread(bncGetThread* getThread) {
183 connect(getThread, SIGNAL(error(const QByteArray&)),
184 this, SLOT(slotGetThreadError(const QByteArray&)));
185
186 _staIDs.push_back(getThread->staID());
187 _threads.push_back(getThread);
188}
189
190// Error in get thread
191////////////////////////////////////////////////////////////////////////////
192void bncCaster::slotGetThreadError(const QByteArray& staID) {
193 QMutexLocker locker(&_mutex);
194 _staIDs.removeAll(staID);
195 emit( newMessage(
196 QString("Mountpoint size %1").arg(_staIDs.size()).toAscii()) );
197 if (_staIDs.size() == 0) {
198 emit(newMessage("bncCaster:: last get thread terminated"));
199 emit getThreadErrors();
200 }
201}
202
203// Dump Complete Epochs
204////////////////////////////////////////////////////////////////////////////
205void bncCaster::dumpEpochs(long minTime, long maxTime) {
206
207 const char begEpoch = 'A';
208 const char begObs = 'B';
209 const char endEpoch = 'C';
210
211 for (long sec = minTime; sec <= maxTime; sec++) {
212
213 bool first = true;
214 QList<Observation*> allObs = _epochs->values(sec);
215 QListIterator<Observation*> it(allObs);
216 while (it.hasNext()) {
217 Observation* obs = it.next();
218
219 if (_samplingRate == 0 || sec % _samplingRate == 0) {
220
221 // Output into the file
222 // --------------------
223 if (_out) {
224 if (first) {
225 *_out << begEpoch << endl;;
226 }
227 *_out << obs->StatID << " "
228 << obs->SVPRN << " "
229 << obs->GPSWeek << " "
230 << obs->GPSWeeks << " "
231 << obs->C1 << " "
232 << obs->P1 << " "
233 << obs->P2 << " "
234 << obs->L1 << " "
235 << obs->L2 << " "
236 << obs->SNR1 << " "
237 << obs->SNR2 << endl;
238 if (!it.hasNext()) {
239 *_out << endEpoch << endl;
240 }
241 }
242
243 // Output into the socket
244 // ----------------------
245 if (_sockets) {
246 int numBytes = sizeof(*obs);
247 QListIterator<QTcpSocket*> is(*_sockets);
248 while (is.hasNext()) {
249 QTcpSocket* sock = is.next();
250 if (first) {
251 sock->write(&begEpoch, 1);
252 }
253 sock->write(&begObs, 1);
254 sock->write((char*) obs, numBytes);
255 if (!it.hasNext()) {
256 sock->write(&endEpoch, 1);
257 }
258 }
259 }
260 }
261
262 delete obs;
263 _epochs->remove(sec);
264 first = false;
265 }
266 }
267}
Note: See TracBrowser for help on using the repository browser.