#include #include // std::string #include // std::transform #include // std::tolower #include "FACT.h" #include "Dim.h" #include "Event.h" #include "Shell.h" #include "StateMachineDim.h" #include "Connection.h" #include "LocalControl.h" #include "Configuration.h" #include "Timers.h" #include "Console.h" #include "Converter.h" #include "tools.h" #include "HeadersTNGWeather.h" #include namespace ba = boost::asio; namespace bs = boost::system; namespace dummy = ba::placeholders; using namespace std; using namespace TNGWeather; class ConnectionWeather : public Connection { uint16_t fInterval; bool fIsVerbose; string fSite; virtual void UpdateWeather(const Time &, const DimWeather &) { } virtual void UpdateDust(const Time &, const float &) { } string fRdfData; float fDust; protected: boost::array fArray; Time fLastReport; Time fLastReception; void HandleRead(const boost::system::error_code& err, size_t bytes_received) { // Do not schedule a new read if the connection failed. if (bytes_received==0 || err) { if (err==ba::error::eof) Warn("Connection closed by remote host."); // 107: Transport endpoint is not connected (bs::error_code(107, bs::system_category)) // 125: Operation canceled if (err && err!=ba::error::eof && // Connection closed by remote host err!=ba::error::basic_errors::not_connected && // Connection closed by remote host err!=ba::error::basic_errors::operation_aborted) // Connection closed by us { ostringstream str; str << "Reading from " << URL() << ": " << err.message() << " (" << err << ")";// << endl; Error(str); } PostClose(err!=ba::error::basic_errors::operation_aborted); fRdfData = ""; return; } fRdfData += string(fArray.data(), bytes_received); const size_t end = fRdfData.find("\r\n\r\n"); if (end==string::npos) { Out() << "Received data corrupted [1]." << endl; Out() << fRdfData << endl; return; } string data(fRdfData); data.erase(0, end+4); size_t pos = 0; while (1) { const size_t chunk = data.find("\r\n", pos); if (chunk==0 || chunk==string::npos) { StartReadReport(); return; } size_t len = 0; stringstream val(data.substr(pos, chunk-pos)); val >> hex >> len; data.erase(pos, chunk-pos+2); if (len==0) break; pos += len+2; // Count trailing \r\n of chunk } fLastReception = Time(); fRdfData = ""; PostClose(false); if (fIsVerbose) { Out() << "------------------------------------------------------" << endl; Out() << data << endl; Out() << "------------------------------------------------------" << endl; } const Soprano::Parser* p = Soprano::PluginManager::instance()->discoverParserForSerialization( Soprano::SerializationRdfXml ); Soprano::StatementIterator it = p->parseString(QString(data.c_str()), QUrl(""), Soprano::SerializationRdfXml ); DimWeather w; Time time(Time::none); try { while (it.next()) { const string pre = (*it).predicate().toString().toStdString(); const string obj = (*it).object().toString().toStdString(); const size_t slash = pre.find_last_of('/'); if (slash==string::npos) continue; const string id = pre.substr(slash+1); if (obj=="N/A") continue; if (id=="dimmSeeing") w.fSeeing = stof(obj); if (id=="dustTotal") w.fDustTotal = stof(obj); if (id=="deltaM1") w.fDeltaM1 = stof(obj); if (id=="airPressure") w.fAirPressure = stof(obj); if (id=="dewPoint") w.fDewPoint = stof(obj); if (id=="windDirection") w.fWindDirection = stof(obj); if (id=="windSpeed") w.fWindSpeed = stof(obj)*3.6; if (id=="hum") w.fHumidity = stof(obj); if (id=="tempGround") w.fTempGround = stof(obj); if (id=="temp2M") w.fTemp2M = stof(obj); if (id=="temp5M") w.fTemp5M = stof(obj); if (id=="temp10M") w.fTemp10M = stof(obj); if (id=="date") time.SetFromStr(obj, "%Y-%m-%dT%H:%M:%S"); /* Out() << "S: " << (*it).subject().toString().toStdString() << endl; Out() << "P: " << (*it).predicate().toString().toStdString() << endl; Out() << "O: " << (*it).object().toString().toStdString() << endl; Out() << "C: " << (*it).context().toString().toStdString() << endl; */ } if (!time.IsValid()) throw runtime_error("time invalid"); if (time!=fLastReport && fIsVerbose) { Out() << endl; Out() << "Date: " << time << endl; Out() << "Seeing: " << w.fSeeing << endl; Out() << "DustTotal: " << w.fDustTotal << endl; Out() << "DeltaM1: " << w.fDeltaM1 << endl; Out() << "AirPressure: " << w.fAirPressure << endl; Out() << "DewPoint: " << w.fDewPoint << endl; Out() << "WindDirection: " << w.fWindDirection << endl; Out() << "WindSpeed: " << w.fWindSpeed << endl; Out() << "Humidity: " << w.fHumidity << endl; Out() << "TempGround: " << w.fTempGround << endl; Out() << "Temp2M: " << w.fTemp2M << endl; Out() << "Temp5M: " << w.fTemp5M << endl; Out() << "Temp10M: " << w.fTemp10M << endl; Out() << endl; } fLastReport = time; UpdateWeather(time, w); if (fDust==w.fDustTotal) return; UpdateDust(time, w.fDustTotal); fDust = w.fDustTotal; ostringstream out; out << setprecision(3) << "Dust: " << fDust << "ug/m^3 [" << time << "]"; Message(out); } catch (const exception &e) { Out() << "Corrupted data received: " << e.what() << endl; fLastReport = Time(Time::none); return; } } void StartReadReport() { async_read_some(ba::buffer(fArray), boost::bind(&ConnectionWeather::HandleRead, this, dummy::error, dummy::bytes_transferred)); } boost::asio::deadline_timer fKeepAlive; void PostRequest() { const string cmd = "GET "+fSite+" HTTP/1.1\r\n" "Accept: */*\r\n" "Content-Type: application/octet-stream\r\n" "User-Agent: FACT\r\n" "Host: www.fact-project.org\r\n" "Pragma: no-cache\r\n" "Cache-Control: no-cache\r\n" "Expires: 0\r\n" "Connection: Keep-Alive\r\n" "Cache-Control: max-age=0\r\n" "\r\n"; PostMessage(cmd); } void Request() { PostRequest(); fKeepAlive.expires_from_now(boost::posix_time::seconds(fInterval)); fKeepAlive.async_wait(boost::bind(&ConnectionWeather::HandleRequest, this, dummy::error)); } void HandleRequest(const bs::error_code &error) { // 125: Operation canceled (bs::error_code(125, bs::system_category)) if (error && error!=ba::error::basic_errors::operation_aborted) { ostringstream str; str << "Write timeout of " << URL() << ": " << error.message() << " (" << error << ")";// << endl; Error(str); PostClose(false); return; } if (!is_open()) { // For example: Here we could schedule a new accept if we // would not want to allow two connections at the same time. PostClose(true); return; } // Check whether the deadline has passed. We compare the deadline // against the current time since a new asynchronous operation // may have moved the deadline before this actor had a chance // to run. if (fKeepAlive.expires_at() > ba::deadline_timer::traits_type::now()) return; Request(); } private: // This is called when a connection was established void ConnectionEstablished() { Request(); StartReadReport(); } public: static const uint16_t kMaxAddr; public: ConnectionWeather(ba::io_service& ioservice, MessageImp &imp) : Connection(ioservice, imp()), fIsVerbose(true), fDust(-1), fLastReport(Time::none), fLastReception(Time::none), fKeepAlive(ioservice) { SetLogStream(&imp); } void SetVerbose(bool b) { fIsVerbose = b; Connection::SetVerbose(b); } void SetInterval(uint16_t i) { fInterval = i; } void SetSite(const string &site) { fSite = site; } int GetState() const { if (fLastReport.IsValid() && fLastReport+boost::posix_time::seconds(fInterval*2)>Time()) return 3; if (fLastReception.IsValid() && fLastReception+boost::posix_time::seconds(fInterval*2)>Time()) return 2; return 1; } }; const uint16_t ConnectionWeather::kMaxAddr = 0xfff; // ------------------------------------------------------------------------ #include "DimDescriptionService.h" class ConnectionDimWeather : public ConnectionWeather { private: DimDescribedService fDimWeather; DimDescribedService fDimAtmosphere; virtual void UpdateWeather(const Time &t, const DimWeather &data) { fDimWeather.setData(&data, sizeof(DimWeather)); fDimWeather.Update(t); } virtual void UpdateDust(const Time &t, const float &dust) { fDimAtmosphere.setData(&dust, sizeof(float)); fDimAtmosphere.Update(t); } public: ConnectionDimWeather(ba::io_service& ioservice, MessageImp &imp) : ConnectionWeather(ioservice, imp), fDimWeather("TNG_WEATHER/DATA", "F:1;F:1;F:1;F:1;F:1;F:1;F:1;F:1;F:1;F:1;F:1;F:1", "|T_10M[deg C]:Temperature 10m above ground" "|T_5M[deg C]:Temperature 5m above ground" "|T_2M[deg C]:Temperature 2m above ground" "|T_0[deg C]:Temperature at ground" "|T_dew[deg C]:Dew point" "|H[%]:Humidity" "|P[mbar]:Air pressure" "|v[km/h]:Wind speed" "|d[deg]:Wind direction (N-E)" "|DeltaM1" "|Dust[ug/m^3]:Dust (total)" "|Seeing[W/m^2]:Seeing"), fDimAtmosphere("TNG_WEATHER/DUST", "F:1", "|Dust[ug/m^3]:Dust (total)") { } }; // ------------------------------------------------------------------------ template class StateMachineWeather : public T, public ba::io_service, public ba::io_service::work { private: S fWeather; bool CheckEventSize(size_t has, const char *name, size_t size) { if (has==size) return true; ostringstream msg; msg << name << " - Received event has " << has << " bytes, but expected " << size << "."; T::Fatal(msg); return false; } int SetVerbosity(const EventImp &evt) { if (!CheckEventSize(evt.GetSize(), "SetVerbosity", 1)) return T::kSM_FatalError; fWeather.SetVerbose(evt.GetBool()); return T::GetCurrentState(); } /* int Disconnect() { // Close all connections fWeather.PostClose(false); return T::GetCurrentState(); } int Reconnect(const EventImp &evt) { // Close all connections to supress the warning in SetEndpoint fWeather.PostClose(false); // Now wait until all connection have been closed and // all pending handlers have been processed poll(); if (evt.GetBool()) fWeather.SetEndpoint(evt.GetString()); // Now we can reopen the connection fWeather.PostClose(true); return T::GetCurrentState(); } */ int Execute() { // Dispatch (execute) at most one handler from the queue. In contrary // to run_one(), it doesn't wait until a handler is available // which can be dispatched, so poll_one() might return with 0 // handlers dispatched. The handlers are always dispatched/executed // synchronously, i.e. within the call to poll_one() poll_one(); return fWeather.GetState(); } public: StateMachineWeather(ostream &out=cout) : T(out, "TNG_WEATHER"), ba::io_service::work(static_cast(*this)), fWeather(*this, *this) { // ba::io_service::work is a kind of keep_alive for the loop. // It prevents the io_service to go to stopped state, which // would prevent any consecutive calls to run() // or poll() to do nothing. reset() could also revoke to the // previous state but this might introduce some overhead of // deletion and creation of threads and more. // State names T::AddStateName(State::kDisconnected, "NoConnection", "No connection to web-server could be established recently"); T::AddStateName(State::kConnected, "Invalid", "Connection to webserver can be established, but received data is not recent or invalid"); T::AddStateName(State::kReceiving, "Valid", "Connection to webserver can be established, receint data received"); // Verbosity commands T::AddEvent("SET_VERBOSE", "B") (bind(&StateMachineWeather::SetVerbosity, this, placeholders::_1)) ("set verbosity state" "|verbosity[bool]:disable or enable verbosity for received data (yes/no), except dynamic data"); /* // Conenction commands AddEvent("DISCONNECT") (bind(&StateMachineWeather::Disconnect, this)) ("disconnect from ethernet"); AddEvent("RECONNECT", "O") (bind(&StateMachineWeather::Reconnect, this, placeholders::_1)) ("(Re)connect ethernet connection to FTM, a new address can be given" "|[host][string]:new ethernet address in the form "); */ } int EvalOptions(Configuration &conf) { fWeather.SetVerbose(!conf.Get("quiet")); fWeather.SetInterval(conf.Get("interval")); fWeather.SetDebugTx(conf.Get("debug-tx")); fWeather.SetSite(conf.Get("url")); fWeather.SetEndpoint(conf.Get("addr")); fWeather.StartConnect(); return -1; } }; // ------------------------------------------------------------------------ #include "Main.h" template int RunShell(Configuration &conf) { return Main::execute>(conf); } void SetupConfiguration(Configuration &conf) { po::options_description control("TNG weather control options"); control.add_options() ("no-dim,d", po_switch(), "Disable dim services") ("addr,a", var("tngweb.tng.iac.es:80"), "Network address of Cosy") ("url,u", var("/weather/rss/"), "File name and path to load") ("quiet,q", po_bool(true), "Disable printing contents of all received messages (except dynamic data) in clear text.") ("interval,i", var(300), "Interval between two updates on the server in seconds") ("debug-tx", po_bool(), "Enable debugging of ethernet transmission.") ; conf.AddOptions(control); } /* Extract usage clause(s) [if any] for SYNOPSIS. Translators: "Usage" and "or" here are patterns (regular expressions) which are used to match the usage synopsis in program output. An example from cp (GNU coreutils) which contains both strings: Usage: cp [OPTION]... [-T] SOURCE DEST or: cp [OPTION]... SOURCE... DIRECTORY or: cp [OPTION]... -t DIRECTORY SOURCE... */ void PrintUsage() { cout << "The tngweather is an interface to the TNG weather data.\n" "\n" "The default is that the program is started without user intercation. " "All actions are supposed to arrive as DimCommands. Using the -c " "option, a local shell can be initialized. With h or help a short " "help message about the usuage can be brought to the screen.\n" "\n" "Usage: tngweather [-c type] [OPTIONS]\n" " or: tngweather [OPTIONS]\n"; cout << endl; } void PrintHelp() { // Main::PrintHelp>(); /* Additional help text which is printed after the configuration options goes here */ /* cout << "bla bla bla" << endl << endl; cout << endl; cout << "Environment:" << endl; cout << "environment" << endl; cout << endl; cout << "Examples:" << endl; cout << "test exam" << endl; cout << endl; cout << "Files:" << endl; cout << "files" << endl; cout << endl; */ } int main(int argc, const char* argv[]) { Configuration conf(argv[0]); conf.SetPrintUsage(PrintUsage); Main::SetupConfiguration(conf); SetupConfiguration(conf); if (!conf.DoParse(argc, argv, PrintHelp)) return 127; //try { // No console access at all if (!conf.Has("console")) { if (conf.Get("no-dim")) return RunShell(conf); else return RunShell(conf); } // Cosole access w/ and w/o Dim if (conf.Get("no-dim")) { if (conf.Get("console")==0) return RunShell(conf); else return RunShell(conf); } else { if (conf.Get("console")==0) return RunShell(conf); else return RunShell(conf); } } /*catch (std::exception& e) { cerr << "Exception: " << e.what() << endl; return -1; }*/ return 0; }