1 files changed, 66 insertions, 4 deletions

diff --git a/apps/SoilMoistureIo.cpp b/apps/SoilMoistureIo.cpp
index 1280fc6..98a256b 100644
--- a/apps/SoilMoistureIo.cpp
+++ b/apps/SoilMoistureIo.cpp
@@ -2,6 +2,8 @@
 
 #include "json.hpp"
 #include <set>
+#include <boost/regex.hpp>
+#include <chrono>
 
 namespace trygvis {
 namespace soil_moisture {
@@ -28,7 +30,7 @@ CsvSampleOutputStream::CsvSampleOutputStream(ostream &stream, vector<string> fie
     stream << endl;
 }
 
-void CsvSampleOutputStream::write(it values) {
+void CsvSampleOutputStream::write(Sample values) {
     auto i = fields.begin();
     while (true) {
         auto field = *i;
@@ -54,8 +56,8 @@ JsonSampleOutputStream::JsonSampleOutputStream(ostream &stream, vector<string> f
         stream(stream), fields(fields) {
 }
 
-void JsonSampleOutputStream::write(it values) {
-    json doc;
+void JsonSampleOutputStream::write(Sample values) {
+    json doc({});
 
     for (auto &f: fields) {
         auto value = values.find(f);
@@ -72,7 +74,7 @@ SqlSampleOutputStream::SqlSampleOutputStream(ostream &stream, vector<string> fie
         stream(stream), fields(fields) {
 }
 
-void SqlSampleOutputStream::write(it values) {
+void SqlSampleOutputStream::write(Sample values) {
     auto i = fields.begin();
 
     stringstream fs, vs;
@@ -103,5 +105,65 @@ void SqlSampleOutputStream::write(it values) {
     stream << "INSERT INTO (" << fs << ") VALUES(" << vs << ")" << endl;
 }
 
+void CsvParser::process(mutable_buffers_1 buffer) {
+
+    size_t some = buffer_size(buffer);
+    auto data = boost::asio::buffer_cast<const uint8_t *>(buffer);
+
+    for (int i = 0; i < some; i++) {
+        uint8_t b = data[i];
+
+        if (b == packet_delimiter) {
+            process_line(line);
+            line = make_shared<vector<uint8_t>>();
+        } else {
+            line->push_back(b);
+        }
+    }
+
+}
+
+void CsvParser::process_line(shared_ptr<vector<uint8_t>> packet) {
+    auto timestamp = std::chrono::system_clock::now().time_since_epoch().count();
+    auto s = std::string((char *) packet->data(), packet->size());
+//    cerr << "packet: " << s << ", size=" << packet->size() << endl;
+
+    static const boost::regex e("([_a-zA-Z0-9]+)=([0-9]+)");
+
+    std::string::const_iterator start = s.begin();
+    std::string::const_iterator end = s.end();
+    boost::match_results<std::string::const_iterator> what;
+    boost::match_flag_type flags = boost::match_default;
+
+    Sample sample;
+
+    while (regex_search(start, end, what, e, flags)) {
+        auto key = static_cast<string>(what[1]);
+        auto value = static_cast<string>(what[2]);
+        start = what[0].second;
+
+//        static const string device_type = "serial";
+        map<string, string> values;
+        values[key] = value;
+//        values["hostname"] = hostname;
+//        values["device"] = device;
+//        values["device_type"] = device_type;
+//        values["timestamp"] = to_string(timestamp);
+//        values["sensor"] = sensor;
+//        values["value"] = value;
+
+//        cerr << key << " => " << value << endl;
+
+        sample[key] = value;
+
+        flags |= boost::match_prev_avail;
+        flags |= boost::match_not_bob;
+    }
+
+    if (sample.begin() != sample.end()) {
+        output->write(sample);
+    }
+}
+
 }
 }