video upload prototype

docs/video_file_request.md

@@ -79,3 +79,7 @@ payload: {
 
 
 
+```json
+{    "time":1567669674,    "cmd":"upload_video",    "rid":"001231554A20",    "data":{       "start":1590542800,       "end":1590542900,       "type":6    } }
+```
+

docs/video_upload_stream.md

@@ -33,7 +33,7 @@ sequenceDiagram
 
 ## 数据格式定义(DD)
 
-### STREAM_DATA_T: 大端在前
+### STREAM_DATA_T: 小端在前
 
 ```BNF
 STREAM_DATA_T := MAGIC_HEAD, {PAYLOAD}-, MAGIC_TAIL     # 总体定义

hi3518/main.cc

@@ -23,6 +23,7 @@
 #include <jsoncons_ext/jsonpatch/jsonpatch.hpp>
 #include "internal_types.h"
 #include "ptz.h"
+#include <queue>
 
 using namespace std;
 using namespace jsoncons;
@@ -83,7 +84,14 @@ char *host = hostArr;
 char *port = portArr;
 
 //
-OrderedList<long long> *gRecFilesList = nullptr;
+OrderedList<int64_t> *gRecFilesList = nullptr;
+typedef struct upload_item_t {
+    int64_t tss;
+    int64_t tse;
+    int type;
+    set<int64_t> slices;
+} upload_item_t;
+
 //
 string strMaQuePath = "/mnt/sd/Config/";
 
@@ -339,7 +347,7 @@ void frame_send_entry(void *args)
     }
 }
 
-void print_ts_files(OrderedList<long long> &_list)
+void print_ts_files(OrderedList<int64_t> &_list)
 {
     spdlog::info("print file ts {}:", _list.items().size());
     int cnt = 0;
@@ -355,7 +363,7 @@ void print_ts_files(OrderedList<long long> &_list)
 }
 
 
-void load_sd_files(OrderedList<long long> &_list)
+void load_sd_files(OrderedList<int64_t> &_list)
 {
     auto now = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
     try {
@@ -368,7 +376,7 @@ void load_sd_files(OrderedList<long long> &_list)
                 continue;
             }
 
-            long long ts = stoll(entry.path().stem().string());
+            int64_t ts = stoll(entry.path().stem().string());
             if(ts < consts::TS_2020) {
                 /// TODO: files records when offline with no valid time got
                 fs::remove(entry.path());
@@ -383,7 +391,7 @@ void load_sd_files(OrderedList<long long> &_list)
     }
 }
 
-void remove_ts_file(long long ts)
+void remove_ts_file(int64_t ts)
 {
     string fname = consts::recFilePath + to_string(ts) + ".h264";
     fs::remove(fname);
@@ -410,7 +418,7 @@ void record_video_entry(ev_module_config_t *pArgs)
         // reserve 0.5GB space
         // multipled by 1.5 because videos take less space in midnight (variable bitrate)
         ssize_t maxSlices = (ssize_t)((total - 512)/10*1.5);
-        gRecFilesList = new OrderedList<long long>(maxSlices, remove_ts_file);
+        gRecFilesList = new OrderedList<int64_t>(maxSlices, remove_ts_file);
         // load existing videos
         load_sd_files(*gRecFilesList);
 
@@ -430,8 +438,8 @@ void record_video_entry(ev_module_config_t *pArgs)
         spdlog::info("zmq sub successed (record)");
 
         zmq_msg_t msg;
-        long long preTs = 0;
-        long long nowTs = 0;
+        int64_t preTs = 0;
+        int64_t nowTs = 0;
         uint32_t cnt = 0;
         fstream * fp = nullptr;
         while (1) {
@@ -671,6 +679,158 @@ string apply_config(json &data)
     }
     return rc;
 }
+///
+#define MAGIC_HEAD0 ((uint8_t)0xBE)
+#define MAGIC_HEAD1 ((uint8_t)0xEF)
+#define MATIC_TAIL0 ((uint8_t)0xFF)
+#define MAGIC_TAIL1 ((uint8_t)0xFF)
+#define MAGIC_TAIL2 ((uint8_t)0xDE)
+#define MAGIC_TAIL3 ((uint8_t)0xAD)
+
+#define EV_UPLOAD_MAX_BUF_SIZE 1200
+
+string raw_send(int s, const void *buf, size_t len){
+    string rc;
+    ssize_t sent = 0;
+    while(len > 0) {
+        sent = ::send(s, buf, len, 0);
+        if(sent <= 0){
+            break;
+        }
+        len -= sent;
+    }
+    if(len > 0 || sent<=0){
+        rc = fmt::format("failed to send: {}", strerror(errno));
+    }
+    return rc;
+}
+
+///
+string upload_video(int64_t tss, int64_t tse, set<int64_t> slices, int8_t type){
+    string rc;
+
+    uint8_t * buf = (uint8_t *)malloc(EV_UPLOAD_MAX_BUF_SIZE);
+
+    int s = 0;
+    try{
+        auto up = gJsonConfig[consts::kMsgConfigUpload].as<string>();
+        auto uri = httplib::Uri::Parse(up);
+        if(uri.Host.empty() || uri.Port.empty()){
+            rc = fmt::format("invalid upload url configure: {}", up);
+        }else{
+            auto i = raw_connect(uri.Host, uri.Port, &s);
+            if(i < 0 || s == 0) {
+                rc = fmt::format("failed to connect to {}", up);
+            }else{
+                auto metaLen = 2 + 12 + 8 + 8 + 1 + 4 + (8 + 4)*slices.size();
+                auto headLen = 2 + metaLen + 2;
+                if(headLen > EV_UPLOAD_MAX_BUF_SIZE) {
+                    rc = fmt::format("too many files to upload, ignored: {}", slices.size());
+                }else{
+                    vector<uint32_t> fsize;
+                    for(auto &k:slices){
+                        struct stat st;
+                        if(stat((string(consts::recFilePath) + to_string(k) + ".h264").c_str(), &st) < 0){
+                            rc = fmt::format("failed to get file {}.h264: {}", k, strerror(errno));
+                            break;
+                        }else{
+                            fsize.push_back(st.st_size);
+                        }
+                    }
+                    if(!rc.empty()){
+                        throw StrException(rc);
+                    }
+
+                    int totalLen = 0;
+                    buf[0] = MAGIC_HEAD0;               // 1, 0 
+                    buf[1] = MAGIC_HEAD1;               // 1, 1
+                    totalLen += 2;
+                    *((uint16_t *)(buf+totalLen)) = 12;       // 2, 2
+                    totalLen += 2;
+                    memcpy(buf+totalLen, dev_sn, 12);         // 12, 4
+                    totalLen += 12;
+                    *((uint64_t *)(buf+totalLen)) = tss;      // 8, 16
+                    totalLen += 8;
+                    *((uint64_t *)(buf+totalLen)) = tse;      // 8, 24
+                    totalLen += 8;
+                    buf[totalLen]= type;                      // 1, 32
+                    totalLen += 1;
+                    *((uint32_t *)(buf+totalLen)) = slices.size(); // 4, 33
+                    totalLen += 4;
+                    int idx = 0;
+                    for(auto &k:slices){
+                        *((uint64_t *)(buf + totalLen)) = k;    // 8, 37
+                        totalLen += 8;
+                        *((uint32_t *)(buf + totalLen)) = fsize[idx]; // 4, 45
+                        totalLen += 4;
+                        idx++;
+                    }
+                    auto crc_ = crc16(buf, totalLen);
+                    memcpy(buf+totalLen, &crc_, 2);
+                    totalLen +=2;
+                    assert(totalLen = headLen);
+                    spdlog::info("header len: {}, crc: {}", totalLen, crc_);
+                    // send header
+                    rc = raw_send(s, buf, totalLen);
+                    if(!rc.empty()){
+                        rc = fmt::format("{}, {}:{}", rc, host, port);
+                    }else{
+                        // construct body
+                        idx = 0;
+                        for(auto &k:slices){
+                            int readsize = 1;
+                            uint32_t total_sent = 0;
+                            auto fpath = string(consts::recFilePath) + to_string(k) + ".h264";
+                            memcpy(buf, &fsize[idx], 4);
+                            FILE *fp = fopen(fpath.c_str(), "rb");
+                            if(!fp){
+                                rc = fmt::format("failed open file : {}", fpath);
+                                break;
+                            }else{
+                                spdlog::info("sending file {}: {}", fpath, fsize[idx]);
+                                for(int i = 0; readsize > 0; i++){
+                                    if(i == 0){
+                                        readsize = ::fread(buf+4, 1, EV_UPLOAD_MAX_BUF_SIZE - 4, fp) + 4;
+                                    }else{
+                                        readsize = ::fread(buf, 1, EV_UPLOAD_MAX_BUF_SIZE, fp);
+                                    }
+                                    
+                                    if((i == 0 && readsize > 4) ||(i !=0 && readsize > 0)) {
+                                        rc = raw_send(s, buf, readsize);
+                                        if(!rc.empty()){
+                                            break;
+                                        }
+                                    }else{
+                                        break;
+                                    }
+                                    spdlog::info("sent bytes: {}", readsize);
+                                    total_sent += readsize;
+                                }
+
+                                ::fclose(fp);
+
+                                if(readsize < 0){
+                                    rc = fmt::format("error read file: {}", strerror(errno));
+                                    spdlog::error(rc);
+                                    break;
+                                }else{
+                                    spdlog::info("sent file {}: total {}, sent {}", fpath, fsize[idx], total_sent - 4);
+                                }
+                            }
+                            idx++;
+                        }
+                    }
+                }
+            }
+        }
+    }catch(exception &e){
+        rc = fmt::format("failed to send to {}:{}: {}, {}:{}", host, port, e.what(), __FILE__, __LINE__);
+    }
+
+    free(buf);
+
+    return rc;
+}
 
 /// handles all incoming request from cloud services
 void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string topic)
@@ -734,14 +894,13 @@ void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string t
                 MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, -1, "no sd card", cmd, rid, data);
             }
             else {
-                auto tss = data["start"].as<long long>();
-                auto tse = data["end"].as<long long>();
-                // guess unit in ms or s
-                if(tss/100000000000 == 0) {
-                    tss = tss * 1000;
-                    tse = tse * 1000;
-                }
-
+                auto tss = data["start"].as<int64_t>();
+                auto tse = data["end"].as<int64_t>();
+                // // guess unit in ms or s
+                // if(tss/100000000000 == 0) {
+                //     tss = tss * 1000;
+                //     tse = tse * 1000;
+                // }
 
                 bool isTsValid = false;
                 if(tss <= tse && tss > consts::TS_2020 && tse < consts::TS_MAX) {
@@ -749,18 +908,19 @@ void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string t
                     isTsValid = true;
                 }
                 else {
-                    tss = tss/1000;
-                    tse = tse/1000;
-                    if(tss <= tse && tss > consts::TS_2020/1000 && tse < consts::TS_MAX/1000 ) {
+                    // in seconds
+                    tss = tss*1000;
+                    tse = tse*1000;
+                    if(tss <= tse && tss > consts::TS_2020 && tse < consts::TS_MAX ) {
                         // in seconds
                         isTsValid = true;
                     }
                 }
                 if(!isTsValid) {
-                    MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_INVALID_PARAM, "time stamps errror. start <= end and both in millisecon or second format ", cmd, rid, data);
+                    MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_INVALID_PARAM, "time stamps error. start <= end && start > 1577836800(000) and both in millisecon or second format ", cmd, rid, data);
                 }
                 else {
-                    long long offsetE = 0, offsetS = 0;
+                    int64_t offsetE = 0, offsetS = 0;
                     auto res = gRecFilesList->findByRange(tss, tse, offsetS, offsetE);
                     if(res.empty()) {
                         MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, -2, "no video found", cmd, rid, data);
@@ -773,6 +933,12 @@ void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string t
                             printf("%lld, ", k);
                         }
                         printf("\n");
+                        /// TODO: upload videos
+                        auto rc = upload_video(tss, tse, res, 6);
+                        if(!rc.empty()){
+                            spdlog::error(rc);
+                            MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_EXCEPTION, rc, cmd, rid, data);
+                        }
                     }
                 }
             }

hi3518/motiondetect.hpp

@@ -22,8 +22,10 @@ namespace md{
     const string kKeyM= "m";
     const string kKeyN= "n";
     const string kKeyP= "p";
+    const int kMaxCntMotionInPre = 5; // about 20s
+
     /// helpers
-    auto event_inspection(auto e, std::map<std::string,int> &m){
+    auto event_inspection(auto e, std::map<const std::string, int> &m){
         if(std::is_same<decltype(e), people>::value){
             spdlog::info("type is people: {}",  ++m[kKeyP]);
         }else if(std::is_same<decltype(e), motion>::value){
@@ -37,9 +39,9 @@ namespace md{
         return false;
     }
 
-    auto make_guard_fn(const std::map<const std::string, int> &konst){
-        return [&konst](auto e){
-            static std::map<const std::string, int>guards = {{"m", 0}, {"p", 0}, {"n", 0}};
+    auto make_guard_fn(const std::map<const std::string, int> &konst, std::map<const std::string, int> &guards ){
+        return [&konst, &guards](auto e){
+            //static std::map<const std::string, int>guards = {{"m", 0}, {"p", 0}, {"n", 0}};
             bool trans = false;
             bool reset = true;
             if(event_inspection(e, guards)||(guards[kKeyM] != 0 && guards[kKeyP] != 0)){
@@ -63,30 +65,34 @@ namespace md{
         };
     }
 
-    typedef bool (*fn_guard_t)(auto e);
     struct fsm {
-        const fn_guard_t guard_in2post = make_guard_fn(kNumIn2Post);
-        const bool guard_post2none = make_guard_fn(kNumPost2None);
-        const void action_none2in = [this](auto e){
-            this->timeStart = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
+        auto operator()() noexcept{
+            const auto guard_in2post = make_guard_fn(kNumIn2Post, guards);
+            const auto guard_post2none = make_guard_fn(kNumPost2None, guards);
+            const auto action_none2in = [this](auto e){
+                hasMotion = false;
+                timeStart = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
                 if(std::is_same<decltype(e), people>::value){
-                spdlog::info("none -p-> in, {}", this->timeStart/1000);
+                    spdlog::info("none -p-> in, {}", timeStart/1000);
                 }else if(std::is_same<decltype(e), mpboth>::value){
-                spdlog::info("none -b-> in, {}", this->timeStart/1000);
-                this->hasMotion = true;
+                    spdlog::info("none -b-> in, {}", timeStart/1000);
+                    hasMotion = true;
                 }
+
+                /// TODO: force event generation when max time duration reached
+
             };
-        const void action_pre2in = [this](auto e){
-            if(this->timeStart==0)
-                this->timeStart = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
+            const auto action_pre2in = [this](auto e){
+                if(timeStart==0)
+                    timeStart = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
                 if(std::is_same<decltype(e), people>::value){
-                spdlog::info("pre -p-> in: time {}", this->timeStart/1000);
+                    spdlog::info("pre -p-> in: time {}", timeStart/1000);
                 }else if(std::is_same<decltype(e), mpboth>::value){
-                spdlog::info("pre -b-> in: time {}", this->timeStart/1000);
+                    spdlog::info("pre -b-> in: time {}", timeStart/1000);
                 }
-            this->hasMotion = true;
+                hasMotion = true;
             };
-        const void action_post2none = [this](auto e){
+            const auto action_post2none = [this](auto e){
                 auto timeEnd = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
                 if(std::is_same<decltype(e), people>::value){
                     spdlog::info("post -p-> none, {}", timeEnd/1000);
@@ -95,50 +101,20 @@ namespace md{
                 }else if(std::is_same<decltype(e), none>::value){
                     spdlog::info("post -n-> none, {}", timeEnd/1000);
                 }
+
+                /// TODO: force event generation when max time duration reached
+
+                spdlog::info("event start:{}, end:{}, hasMotion: {}", timeStart, timeEnd, hasMotion);
+                /// TODO: generate event for uploading
             };
-        const void action_pre2pre = [this](auto e){
-            static int cntMotionOnly = 0;
+            const auto action_pre2pre = [this](auto e){
                 cntMotionOnly++;
                 if(cntMotionOnly >= kMaxCntMotionInPre) {
                     cntMotionOnly = 0;
-                this->timeStart = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
-            }
-        };
-
-        inline auto operator()() noexcept{
-            using namespace sml;
-=======
-        }else if(std::is_same<decltype(e), none>::value){
-            spdlog::info("type is none: {}",  ++m[kKeyN]);
+                    timeStart = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
                 }
-    }
-
-    auto make_guard_fn(const std::map<std::string,int> &konst){
-        return [&konst](auto e){
-            static std::map<std::string, int>guards = {{"m", 0}, {"p", 0}, {"n", 0}};
-            event_inspection(e, guards);
-            if(guards[kKeyM] != 0 && guards[kKeyP] != 0){
-                guards[kKeyM] = guards[kKeyP] = guards[kKeyN] = 0;
-                return false;
-            }else if(guards[kKeyP] >= konst[kKeyP]){
-                guards[kKeyP]=guards[kKeyN]= 0;
-                return true;
-            }else if(guards[kKeyM] >= konst[kKeyM]){
-                guards[kKeyM]=guards[kKeyN] = 0;
-                return true;
-            }else if(guards[kKeyN] >= konst[kKeyN]){
-                guards[kKeyN] = 0;
-                return true;
-            }
-            return false;
             };
-    }
-
-    struct fsm {
-        auto operator()()const noexcept {
             using namespace sml;
-            const auto guard_in2post = make_guard_fn(kNumIn2Post);
-            const auto guard_post2none = make_guard_fn(kNumPost2None);
             return make_transition_table(
                 *"init"_s = "none"_s,
                 "none"_s + event<none>/[]{spdlog::info("none -> none");} = "none"_s,
@@ -163,9 +139,11 @@ namespace md{
                 "none"_s + sml::on_entry<_> / [this]{hasMotion = false,timeStart = 0;}
             );
         }
-        public:
-            bool hasMotion = false;
-            uint64_t timeStart =0;
+        private:
+            bool hasMotion{false};
+            int cntMotionOnly = 0;
+            uint64_t timeStart{0};
+            std::map<const std::string, int>guards{{"m", 0}, {"p", 0}, {"n", 0}};
     };
 }
 

hi3518/smart.cc

@@ -40,7 +40,6 @@ XM_S32 MaQue_JpegEnc_getFrame_callback (XM_VOID *pUserArg, MaQueSmartJpegFrame_s
 }
 
 
-
 void maq_smart_task_entry(ev_module_config_t *pArg)
 {
     if(!pArg->module.ai.enabled) {
@@ -173,7 +172,6 @@ void maq_smart_task_entry(ev_module_config_t *pArg)
     extern atomic<uint64_t> frameCntTotal;
     uint64_t frameCntLast = frameCntTotal.load(memory_order_relaxed);
     uint64_t motionCntLast = gMotionCounter.load(memory_order_relaxed);
-<<<<<<< HEAD
     const float kMotionPerSecondThresh = 2.5;
     const int kMotionDetectWindowSeconds = 4;
     const float kMsPerMotion = 1000.0/kMotionPerSecondThresh;
@@ -181,15 +179,6 @@ void maq_smart_task_entry(ev_module_config_t *pArg)
     int windowSecs = kMotionDetectWindowSeconds;
     ev_region_t last_region = {0};
     sml::sm<md::fsm> fsm{md::fsm{}};
-=======
-    const int kMotionPerSecondThresh = 3;
-    const int kMotionDetectWindowSeconds = 4;
-    const float kMsPerMotion = kMotionDetectWindowSeconds*1000.0/kMotionPerSecondThresh;
-    int state = 0;
-    int windowSecs = kMotionDetectWindowSeconds;
-    ev_region_t last_region = {0};
-    sml::sm<md::fsm> fsm;
->>>>>>> 5b92158f6b6f54d698af4a1f24b2fcd7b03b5f5d
     bool hasHuman = false;
     bool hasMotion = false;
     while(1) {
@@ -207,11 +196,7 @@ void maq_smart_task_entry(ev_module_config_t *pArg)
         }
         if (stMaQueSmartTarget.targetFDNum > 0) {
             hasHuman = true;
-<<<<<<< HEAD
             //last_region = {stMaQueSmartTarget.aFDRect[0].s16X1, stMaQueSmartTarget.aFDRect[0].s16Y1, stMaQueSmartTarget.aFDRect[0].s16X2, stMaQueSmartTarget.aFDRect[0].s16Y2};
-=======
-            last_region = {stMaQueSmartTarget.aFDRect[0].s16X1, stMaQueSmartTarget.aFDRect[0].s16Y1, stMaQueSmartTarget.aFDRect[0].s16X2, stMaQueSmartTarget.aFDRect[0].s16Y2};
->>>>>>> 5b92158f6b6f54d698af4a1f24b2fcd7b03b5f5d
         }
         if(windowSecs > 0) {
             windowSecs--;
@@ -219,19 +204,15 @@ void maq_smart_task_entry(ev_module_config_t *pArg)
         else {
             auto frameCntTotal_ = frameCntTotal.load(memory_order_relaxed);
             auto motionCnt_ = gMotionCounter.load(memory_order_relaxed);
-
             float deltaTimeMs = (frameCntTotal_ - frameCntLast) * 1000/pArg->module.sys.fps;
-<<<<<<< HEAD
-            spdlog::info("deltaTimeMs {}", deltaTimeMs);
-=======
->>>>>>> 5b92158f6b6f54d698af4a1f24b2fcd7b03b5f5d
             frameCntLast = frameCntTotal_;
             int motionCntThresh = deltaTimeMs / kMsPerMotion;
             int deltaMotionCnt = motionCnt_ - motionCntLast;
             motionCntLast = motionCnt_;
+
+            spdlog::info("deltaTimeMs {}", deltaTimeMs);
             if(deltaMotionCnt >= motionCntThresh) {
                 hasMotion = true;
-<<<<<<< HEAD
             }
             if(hasMotion && hasHuman) {
                 fsm.process_event(md::mpboth{});
@@ -243,26 +224,13 @@ void maq_smart_task_entry(ev_module_config_t *pArg)
                 fsm.process_event(md::people{});
             }
             else {
-=======
-            }
-            if(hasMotion && hasHuman){
-                fsm.process_event(md::mpboth{});
-            }else if(hasMotion){
-                fsm.process_event(md::motion{});
-            }else if(hasHuman){
-                fsm.process_event(md::people{});
-            }else{
->>>>>>> 5b92158f6b6f54d698af4a1f24b2fcd7b03b5f5d
                 fsm.process_event(md::none{});
             }
             // reset
             windowSecs = kMotionDetectWindowSeconds;
             hasHuman = false;
             hasMotion = false;
-<<<<<<< HEAD
             last_region = {0,0,0,0};
-=======
->>>>>>> 5b92158f6b6f54d698af4a1f24b2fcd7b03b5f5d
         }
     }
 

server/CMakeLists.txt

@@ -18,6 +18,8 @@ link_directories(${COMMON_LIB_DIR})
 
 add_executable(vgw videogateway.cc)
 target_link_libraries(vgw PUBLIC ${VGW_LIBS} ${COMM_LIBS})
+add_executable(test_curl test_curl.cc)
+target_link_libraries(vgw curl})
 
 # add_executable(test_mqtt test_mqtt_rd.cc)
 # target_link_libraries(test_mqtt PUBLIC ${COMM_LIBS})