source: trunk/FACT++/src/tngweather.cc@ 13868

Last change on this file since 13868 was 13866, checked in by tbretz, 13 years ago
First draft of a TNG weather application.
File size: 18.9 KB
Line 
1#include <boost/bind.hpp>
2
3#include <string> // std::string
4#include <algorithm> // std::transform
5#include <cctype> // std::tolower
6
7#include "FACT.h"
8#include "Dim.h"
9#include "Event.h"
10#include "Shell.h"
11#include "StateMachineDim.h"
12#include "Connection.h"
13#include "LocalControl.h"
14#include "Configuration.h"
15#include "Timers.h"
16#include "Console.h"
17#include "Converter.h"
18
19#include "tools.h"
20
21#include <Soprano/Soprano>
22
23namespace ba = boost::asio;
24namespace bs = boost::system;
25namespace dummy = ba::placeholders;
26
27using namespace std;
28
29// ------------------------------------------------------------------------
30
31struct DimWeather
32{
33 DimWeather() { memset(this, 0, sizeof(DimWeather)); }
34
35 uint16_t fStatus;
36
37 float fTemp10M;
38 float fTemp5M;
39 float fTemp2M;
40 float fTempGround;
41 float fDewPoint;
42 float fHumidity;
43 float fAirPressure;
44 float fWindSpeed;
45 float fWindDirection;
46 float fDeltaM1;
47 float fDustTotal;
48 float fSeeing;
49
50} __attribute__((__packed__));
51
52
53// ------------------------------------------------------------------------
54
55class ConnectionWeather : public Connection
56{
57 uint16_t fInterval;
58
59 bool fIsVerbose;
60
61 string fSite;
62
63 virtual void UpdateWeather(const Time &, const DimWeather &)
64 {
65 }
66
67 string fRdfData;
68 uint32_t missing;
69
70protected:
71
72 boost::array<char, 4096> fArray;
73
74 Time fLastReport;
75 Time fLastReception;
76
77 void HandleRead(const boost::system::error_code& err, size_t bytes_received)
78 {
79 // Do not schedule a new read if the connection failed.
80 if (bytes_received==0 || err)
81 {
82 if (err==ba::error::eof)
83 Warn("Connection closed by remote host.");
84
85 // 107: Transport endpoint is not connected (bs::error_code(107, bs::system_category))
86 // 125: Operation canceled
87 if (err && err!=ba::error::eof && // Connection closed by remote host
88 err!=ba::error::basic_errors::not_connected && // Connection closed by remote host
89 err!=ba::error::basic_errors::operation_aborted) // Connection closed by us
90 {
91 ostringstream str;
92 str << "Reading from " << URL() << ": " << err.message() << " (" << err << ")";// << endl;
93 Error(str);
94 }
95 PostClose(err!=ba::error::basic_errors::operation_aborted);
96
97 fRdfData = "";
98 return;
99 }
100
101 fRdfData += string(fArray.data(), bytes_received);
102
103 const size_t end = fRdfData.find("\r\n\r\n");
104 if (end==string::npos)
105 {
106 Out() << "Received data corrupted [1]." << endl;
107 Out() << fRdfData << endl;
108 return;
109 }
110
111 string data(fRdfData);
112 data.erase(0, end+4);
113
114 size_t pos = 0;
115 while (1)
116 {
117 const size_t chunk = data.find("\r\n", pos);
118 if (chunk==0 || chunk==string::npos)
119 {
120 StartReadReport();
121 return;
122 }
123
124 size_t len = 0;
125 stringstream val(data.substr(pos, chunk-pos));
126 val >> hex >> len;
127
128 data.erase(pos, chunk-pos+2);
129 if (len==0)
130 break;
131
132 pos += len+2; // Count trailing \r\n of chunk
133 }
134
135
136 fLastReception = Time();
137 fRdfData = "";
138 PostClose(false);
139
140
141 const Soprano::Parser* p = Soprano::PluginManager::instance()->discoverParserForSerialization( Soprano::SerializationRdfXml );
142 Soprano::StatementIterator it = p->parseString(QString(data.c_str()), QUrl(""), Soprano::SerializationRdfXml );
143
144
145 DimWeather w;
146 Time time(Time::none);
147 try
148 {
149 while (it.next())
150 {
151 const string pre = (*it).predicate().toString().toStdString();
152 const string obj = (*it).object().toString().toStdString();
153
154 const size_t slash = pre.find_last_of('/');
155 if (slash==string::npos)
156 continue;
157
158 const string id = pre.substr(slash+1);
159
160 if (obj=="N/A")
161 continue;
162
163 if (id=="dimmSeeing")
164 w.fSeeing = stof(obj);
165 if (id=="dustTotal")
166 w.fDustTotal = stof(obj);
167 if (id=="deltaM1")
168 w.fDeltaM1 = stof(obj);
169 if (id=="airPressure")
170 w.fAirPressure = stof(obj);
171 if (id=="dewPoint")
172 w.fDewPoint = stof(obj);
173 if (id=="windDirection")
174 w.fWindDirection = stof(obj);
175 if (id=="windSpeed")
176 w.fWindSpeed = stof(obj);
177 if (id=="hum")
178 w.fHumidity = stof(obj);
179 if (id=="tempGround")
180 w.fTempGround = stof(obj);
181 if (id=="temp2M")
182 w.fTemp2M = stof(obj);
183 if (id=="temp5M")
184 w.fTemp5M = stof(obj);
185 if (id=="temp10M")
186 w.fTemp10M = stof(obj);
187 if (id=="date")
188 time.SetFromStr(obj, "%Y-%m-%dT%H:%M:%S");
189
190 /*
191 Out() << "S: " << (*it).subject().toString().toStdString() << endl;
192 Out() << "P: " << (*it).predicate().toString().toStdString() << endl;
193 Out() << "O: " << (*it).object().toString().toStdString() << endl;
194 Out() << "C: " << (*it).context().toString().toStdString() << endl;
195 */
196 }
197
198 if (!time.IsValid())
199 throw runtime_error("time invalid");
200
201 if (time!=fLastReport)
202 {
203 Out() << endl;
204 Out() << "Date: " << time << endl;
205 Out() << "Seeing: " << w.fSeeing << endl;
206 Out() << "DustTotal: " << w.fDustTotal << endl;
207 Out() << "DeltaM1: " << w.fDeltaM1 << endl;
208 Out() << "AirPressure: " << w.fAirPressure << endl;
209 Out() << "DewPoint: " << w.fDewPoint << endl;
210 Out() << "WindDirection: " << w.fWindDirection << endl;
211 Out() << "WindSpeed: " << w.fWindSpeed << endl;
212 Out() << "Humidity: " << w.fHumidity << endl;
213 Out() << "TempGround: " << w.fTempGround << endl;
214 Out() << "Temp2M: " << w.fTemp2M << endl;
215 Out() << "Temp5M: " << w.fTemp5M << endl;
216 Out() << "Temp10M: " << w.fTemp10M << endl;
217 Out() << endl;
218 }
219
220 fLastReport = time;
221
222 UpdateWeather(time, w);
223
224 }
225 catch (const exception &e)
226 {
227 Out() << "Corrupted data received: " << e.what() << endl;
228 fLastReport = Time(Time::none);
229 return;
230 }
231 }
232
233 void StartReadReport()
234 {
235 async_read_some(ba::buffer(fArray),
236 boost::bind(&ConnectionWeather::HandleRead, this,
237 dummy::error, dummy::bytes_transferred));
238 }
239
240 boost::asio::deadline_timer fKeepAlive;
241
242 void PostRequest()
243 {
244 const string cmd =
245 "GET "+fSite+" HTTP/1.1\r\n"
246 "Accept: */*\r\n"
247 "Content-Type: application/octet-stream\r\n"
248 "User-Agent: FACT\r\n"
249 "Host: www.fact-project.org\r\n"
250 "Pragma: no-cache\r\n"
251 "Cache-Control: no-cache\r\n"
252 "Expires: 0\r\n"
253 "Connection: Keep-Alive\r\n"
254 "Cache-Control: max-age=0\r\n"
255 "\r\n";
256
257 PostMessage(cmd);
258 }
259
260 void Request()
261 {
262 PostRequest();
263
264 fKeepAlive.expires_from_now(boost::posix_time::seconds(fInterval));
265 fKeepAlive.async_wait(boost::bind(&ConnectionWeather::HandleRequest,
266 this, dummy::error));
267 }
268
269 void HandleRequest(const bs::error_code &error)
270 {
271 // 125: Operation canceled (bs::error_code(125, bs::system_category))
272 if (error && error!=ba::error::basic_errors::operation_aborted)
273 {
274 ostringstream str;
275 str << "Write timeout of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
276 Error(str);
277
278 PostClose(false);
279 return;
280 }
281
282 if (!is_open())
283 {
284 // For example: Here we could schedule a new accept if we
285 // would not want to allow two connections at the same time.
286 PostClose(true);
287 return;
288 }
289
290 // Check whether the deadline has passed. We compare the deadline
291 // against the current time since a new asynchronous operation
292 // may have moved the deadline before this actor had a chance
293 // to run.
294 if (fKeepAlive.expires_at() > ba::deadline_timer::traits_type::now())
295 return;
296
297 Request();
298 }
299
300
301private:
302 // This is called when a connection was established
303 void ConnectionEstablished()
304 {
305 Request();
306 StartReadReport();
307 }
308
309public:
310
311 static const uint16_t kMaxAddr;
312
313public:
314 ConnectionWeather(ba::io_service& ioservice, MessageImp &imp) : Connection(ioservice, imp()),
315 fIsVerbose(true), fLastReport(Time::none), fLastReception(Time::none), fKeepAlive(ioservice)
316 {
317 SetLogStream(&imp);
318 }
319
320 void SetVerbose(bool b)
321 {
322 fIsVerbose = b;
323 Connection::SetVerbose(b);
324 }
325
326 void SetInterval(uint16_t i)
327 {
328 fInterval = i;
329 }
330
331 void SetSite(const string &site)
332 {
333 fSite = site;
334 }
335
336 int GetState() const
337 {
338 if (fLastReport.IsValid() && fLastReport+boost::posix_time::seconds(fInterval*2)>Time())
339 return 3;
340
341 if (fLastReception.IsValid() && fLastReception+boost::posix_time::seconds(fInterval*2)>Time())
342 return 2;
343
344 return 1;
345 }
346};
347
348const uint16_t ConnectionWeather::kMaxAddr = 0xfff;
349
350// ------------------------------------------------------------------------
351
352#include "DimDescriptionService.h"
353
354class ConnectionDimWeather : public ConnectionWeather
355{
356private:
357
358 DimDescribedService fDimWeather;
359
360 virtual void UpdateWeather(const Time &t, const DimWeather &data)
361 {
362 fDimWeather.setData(&data, sizeof(DimWeather));
363 fDimWeather.Update(t);
364 }
365
366public:
367 ConnectionDimWeather(ba::io_service& ioservice, MessageImp &imp) :
368 ConnectionWeather(ioservice, imp),
369 fDimWeather("TNG_WEATHER/DATA", "S:1;F:1;F:1;F:1;F:1;F:1;F:1;F:1",
370 "|stat:Status"
371 "|T_10M[deg C]:Temperature 10m above ground"
372 "|T_5M[deg C]:Temperature 5m above ground"
373 "|T_2M[deg C]:Temperature 2m above ground"
374 "|T_0[deg C]:Temperature at ground"
375 "|T_dew[deg C]:Dew point"
376 "|H[%]:Humidity"
377 "|P[mbar]:Air pressure"
378 "|v[m/s]:Wind speed"
379 "|d[deg]:Wind direction (N-E)"
380 "|DeltaM1"
381 "|Dust[ug/m^3]:Dust (total)"
382 "|Seeing[W/m^2]:Seeing")
383 {
384 }
385};
386
387// ------------------------------------------------------------------------
388
389template <class T, class S>
390class StateMachineWeather : public T, public ba::io_service, public ba::io_service::work
391{
392private:
393 S fWeather;
394
395 enum states_t
396 {
397 kStateDisconnected = 1,
398 kStateConnected,
399 kStateReceiving,
400 };
401
402 bool CheckEventSize(size_t has, const char *name, size_t size)
403 {
404 if (has==size)
405 return true;
406
407 ostringstream msg;
408 msg << name << " - Received event has " << has << " bytes, but expected " << size << ".";
409 T::Fatal(msg);
410 return false;
411 }
412
413 int SetVerbosity(const EventImp &evt)
414 {
415 if (!CheckEventSize(evt.GetSize(), "SetVerbosity", 1))
416 return T::kSM_FatalError;
417
418 fWeather.SetVerbose(evt.GetBool());
419
420 return T::GetCurrentState();
421 }
422/*
423 int Disconnect()
424 {
425 // Close all connections
426 fWeather.PostClose(false);
427
428 return T::GetCurrentState();
429 }
430
431 int Reconnect(const EventImp &evt)
432 {
433 // Close all connections to supress the warning in SetEndpoint
434 fWeather.PostClose(false);
435
436 // Now wait until all connection have been closed and
437 // all pending handlers have been processed
438 poll();
439
440 if (evt.GetBool())
441 fWeather.SetEndpoint(evt.GetString());
442
443 // Now we can reopen the connection
444 fWeather.PostClose(true);
445
446 return T::GetCurrentState();
447 }
448*/
449 int Execute()
450 {
451 // Dispatch (execute) at most one handler from the queue. In contrary
452 // to run_one(), it doesn't wait until a handler is available
453 // which can be dispatched, so poll_one() might return with 0
454 // handlers dispatched. The handlers are always dispatched/executed
455 // synchronously, i.e. within the call to poll_one()
456 poll_one();
457
458 return fWeather.GetState();
459 }
460
461
462public:
463 StateMachineWeather(ostream &out=cout) :
464 T(out, "TNG_WEATHER"), ba::io_service::work(static_cast<ba::io_service&>(*this)),
465 fWeather(*this, *this)
466 {
467 // ba::io_service::work is a kind of keep_alive for the loop.
468 // It prevents the io_service to go to stopped state, which
469 // would prevent any consecutive calls to run()
470 // or poll() to do nothing. reset() could also revoke to the
471 // previous state but this might introduce some overhead of
472 // deletion and creation of threads and more.
473
474 // State names
475 AddStateName(kStateDisconnected, "NoConnection",
476 "No connection to web-server could be established recently");
477
478 AddStateName(kStateConnected, "Invalid",
479 "Connection to webserver can be established, but received data is not recent or invalid");
480
481 AddStateName(kStateReceiving, "Valid",
482 "Connection to webserver can be established, receint data received");
483
484 // Verbosity commands
485 T::AddEvent("SET_VERBOSE", "B")
486 (bind(&StateMachineWeather::SetVerbosity, this, placeholders::_1))
487 ("set verbosity state"
488 "|verbosity[bool]:disable or enable verbosity for received data (yes/no), except dynamic data");
489/*
490 // Conenction commands
491 AddEvent("DISCONNECT")
492 (bind(&StateMachineWeather::Disconnect, this))
493 ("disconnect from ethernet");
494
495 AddEvent("RECONNECT", "O")
496 (bind(&StateMachineWeather::Reconnect, this, placeholders::_1))
497 ("(Re)connect ethernet connection to FTM, a new address can be given"
498 "|[host][string]:new ethernet address in the form <host:port>");
499*/
500 }
501
502 int EvalOptions(Configuration &conf)
503 {
504 fWeather.SetVerbose(!conf.Get<bool>("quiet"));
505 fWeather.SetInterval(conf.Get<uint16_t>("interval"));
506 fWeather.SetDebugTx(conf.Get<bool>("debug-tx"));
507 fWeather.SetSite(conf.Get<string>("url"));
508 fWeather.SetEndpoint(conf.Get<string>("addr"));
509 fWeather.StartConnect();
510
511 return -1;
512 }
513};
514
515// ------------------------------------------------------------------------
516
517#include "Main.h"
518
519
520template<class T, class S, class R>
521int RunShell(Configuration &conf)
522{
523 return Main::execute<T, StateMachineWeather<S, R>>(conf);
524}
525
526void SetupConfiguration(Configuration &conf)
527{
528 po::options_description control("TNG weather control options");
529 control.add_options()
530 ("no-dim,d", po_switch(), "Disable dim services")
531 ("addr,a", var<string>("tngweb.tng.iac.es:80"), "Network address of Cosy")
532 ("url,u", var<string>("/weather/rss/"), "File name and path to load")
533 ("quiet,q", po_bool(true), "Disable printing contents of all received messages (except dynamic data) in clear text.")
534 ("interval,i", var<uint16_t>(120), "Interval between two updates on the server in seconds")
535 ("debug-tx", po_bool(), "Enable debugging of ethernet transmission.")
536 ;
537
538 conf.AddOptions(control);
539}
540
541/*
542 Extract usage clause(s) [if any] for SYNOPSIS.
543 Translators: "Usage" and "or" here are patterns (regular expressions) which
544 are used to match the usage synopsis in program output. An example from cp
545 (GNU coreutils) which contains both strings:
546 Usage: cp [OPTION]... [-T] SOURCE DEST
547 or: cp [OPTION]... SOURCE... DIRECTORY
548 or: cp [OPTION]... -t DIRECTORY SOURCE...
549 */
550void PrintUsage()
551{
552 cout <<
553 "The tngweather is an interface to the TNG weather data.\n"
554 "\n"
555 "The default is that the program is started without user intercation. "
556 "All actions are supposed to arrive as DimCommands. Using the -c "
557 "option, a local shell can be initialized. With h or help a short "
558 "help message about the usuage can be brought to the screen.\n"
559 "\n"
560 "Usage: tngweather [-c type] [OPTIONS]\n"
561 " or: tngweather [OPTIONS]\n";
562 cout << endl;
563}
564
565void PrintHelp()
566{
567// Main::PrintHelp<StateMachineFTM<StateMachine, ConnectionFTM>>();
568
569 /* Additional help text which is printed after the configuration
570 options goes here */
571
572 /*
573 cout << "bla bla bla" << endl << endl;
574 cout << endl;
575 cout << "Environment:" << endl;
576 cout << "environment" << endl;
577 cout << endl;
578 cout << "Examples:" << endl;
579 cout << "test exam" << endl;
580 cout << endl;
581 cout << "Files:" << endl;
582 cout << "files" << endl;
583 cout << endl;
584 */
585}
586
587int main(int argc, const char* argv[])
588{
589 Configuration conf(argv[0]);
590 conf.SetPrintUsage(PrintUsage);
591 Main::SetupConfiguration(conf);
592 SetupConfiguration(conf);
593
594 if (!conf.DoParse(argc, argv, PrintHelp))
595 return -1;
596
597 //try
598 {
599 // No console access at all
600 if (!conf.Has("console"))
601 {
602 if (conf.Get<bool>("no-dim"))
603 return RunShell<LocalStream, StateMachine, ConnectionWeather>(conf);
604 else
605 return RunShell<LocalStream, StateMachineDim, ConnectionDimWeather>(conf);
606 }
607 // Cosole access w/ and w/o Dim
608 if (conf.Get<bool>("no-dim"))
609 {
610 if (conf.Get<int>("console")==0)
611 return RunShell<LocalShell, StateMachine, ConnectionWeather>(conf);
612 else
613 return RunShell<LocalConsole, StateMachine, ConnectionWeather>(conf);
614 }
615 else
616 {
617 if (conf.Get<int>("console")==0)
618 return RunShell<LocalShell, StateMachineDim, ConnectionDimWeather>(conf);
619 else
620 return RunShell<LocalConsole, StateMachineDim, ConnectionDimWeather>(conf);
621 }
622 }
623 /*catch (std::exception& e)
624 {
625 cerr << "Exception: " << e.what() << endl;
626 return -1;
627 }*/
628
629 return 0;
630}
Note: See TracBrowser for help on using the repository browser.