Index: /trunk/FACT++/src/tngweather.cc =================================================================== --- /trunk/FACT++/src/tngweather.cc (revision 13866) +++ /trunk/FACT++/src/tngweather.cc (revision 13866) @@ -0,0 +1,630 @@ +#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 + +namespace ba = boost::asio; +namespace bs = boost::system; +namespace dummy = ba::placeholders; + +using namespace std; + +// ------------------------------------------------------------------------ + +struct DimWeather +{ + DimWeather() { memset(this, 0, sizeof(DimWeather)); } + + uint16_t fStatus; + + float fTemp10M; + float fTemp5M; + float fTemp2M; + float fTempGround; + float fDewPoint; + float fHumidity; + float fAirPressure; + float fWindSpeed; + float fWindDirection; + float fDeltaM1; + float fDustTotal; + float fSeeing; + +} __attribute__((__packed__)); + + +// ------------------------------------------------------------------------ + +class ConnectionWeather : public Connection +{ + uint16_t fInterval; + + bool fIsVerbose; + + string fSite; + + virtual void UpdateWeather(const Time &, const DimWeather &) + { + } + + string fRdfData; + uint32_t missing; + +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); + + + 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); + 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) + { + 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); + + } + 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), 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; + + virtual void UpdateWeather(const Time &t, const DimWeather &data) + { + fDimWeather.setData(&data, sizeof(DimWeather)); + fDimWeather.Update(t); + } + +public: + ConnectionDimWeather(ba::io_service& ioservice, MessageImp &imp) : + ConnectionWeather(ioservice, imp), + fDimWeather("TNG_WEATHER/DATA", "S:1;F:1;F:1;F:1;F:1;F:1;F:1;F:1", + "|stat:Status" + "|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[m/s]:Wind speed" + "|d[deg]:Wind direction (N-E)" + "|DeltaM1" + "|Dust[ug/m^3]:Dust (total)" + "|Seeing[W/m^2]:Seeing") + { + } +}; + +// ------------------------------------------------------------------------ + +template +class StateMachineWeather : public T, public ba::io_service, public ba::io_service::work +{ +private: + S fWeather; + + enum states_t + { + kStateDisconnected = 1, + kStateConnected, + kStateReceiving, + }; + + 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 + AddStateName(kStateDisconnected, "NoConnection", + "No connection to web-server could be established recently"); + + AddStateName(kStateConnected, "Invalid", + "Connection to webserver can be established, but received data is not recent or invalid"); + + AddStateName(kStateReceiving, "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(120), "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 -1; + + //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; +}