Skip to content

E
evsuits
提交 ab19868f authored 作者: blu's avatar blu
浏览文件
操作

files ring

上级 0539b12b
隐藏空白字符变更
内嵌 并排
正在显示 包含 138 行增加 和 247 行删除
opencv-motion-detect/evmgr.cpp
浏览文件 @ ab19868f
...@@ -122,7 +122,7 @@ private: ...@@ -122,7 +122,7 @@ private:
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(pRouter, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(pRouter, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(pRouter, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(pRouter, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(pRouter, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(pRouter, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
zmq_setsockopt (pRouter, ZMQ_ROUTER_NOTIFY, &opt_notify, sizeof (opt_notify)); zmq_setsockopt (pRouter, ZMQ_ROUTER_NOTIFY, &opt_notify, sizeof (opt_notify));
......
opencv-motion-detect/evmlmotion.cpp
浏览文件 @ ab19868f
...@@ -323,7 +323,7 @@ private: ...@@ -323,7 +323,7 @@ private:
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size()); ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size());
......
opencv-motion-detect/evpuller.cpp
浏览文件 @ ab19868f
...@@ -266,7 +266,7 @@ private: ...@@ -266,7 +266,7 @@ private:
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size()); ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size());
......
opencv-motion-detect/evpusher.cpp
浏览文件 @ ab19868f
...@@ -119,7 +119,7 @@ private: ...@@ -119,7 +119,7 @@ private:
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0); ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
...@@ -141,7 +141,7 @@ private: ...@@ -141,7 +141,7 @@ private:
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size()); ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size());
......
opencv-motion-detect/evslicer.cpp
浏览文件 @ ab19868f
...@@ -24,6 +24,8 @@ update: 2019/09/10 ...@@ -24,6 +24,8 @@ update: 2019/09/10
#include <functional> #include <functional>
#include <queue> #include <queue>
#include <fstream> #include <fstream>
#include <algorithm>
#include <set>
#include <cstdlib> #include <cstdlib>
#include "inc/zmqhelper.hpp" #include "inc/zmqhelper.hpp"
...@@ -46,9 +48,9 @@ private: ...@@ -46,9 +48,9 @@ private:
void *pSubCtx = nullptr, *pDealerCtx = nullptr; // for packets relay void *pSubCtx = nullptr, *pDealerCtx = nullptr; // for packets relay
void *pSub = nullptr, *pDealer = nullptr, *pDaemonCtx = nullptr, *pDaemon = nullptr; void *pSub = nullptr, *pDealer = nullptr, *pDaemonCtx = nullptr, *pDaemon = nullptr;
string urlOut, urlPub, urlRouter, devSn, mgrSn, selfId, pullerGid, ipcSn; string urlOut, urlPub, urlRouter, devSn, mgrSn, selfId, pullerGid, ipcSn;
int iid, hours, seconds, numSlices, segHead = -1, segHeadP = 0; int iid, hours, seconds, numSlices;
long bootTime = 0; long bootTime = 0;
bool enablePush = false, bSegFull = false; bool enablePush = false;
AVFormatContext *pAVFormatRemux = nullptr; AVFormatContext *pAVFormatRemux = nullptr;
AVFormatContext *pAVFormatInput = nullptr; AVFormatContext *pAVFormatInput = nullptr;
AVDictionary *pOptsRemux = nullptr; AVDictionary *pOptsRemux = nullptr;
...@@ -61,15 +63,14 @@ private: ...@@ -61,15 +63,14 @@ private:
condition_variable cvMsg; condition_variable cvMsg;
mutex mutMsg; mutex mutMsg;
bool gotFormat = false; bool gotFormat = false;
vector<long> vTsOld;
mutex mutTsOld;
vector<long> vTsActive;
mutex mutTsActive;
queue<string> eventQueue; queue<string> eventQueue;
condition_variable cvEvent; condition_variable cvEvent;
mutex mutEvent; mutex mutEvent;
thread thEventHandler; thread thEventHandler;
string videoFileServerApi = "http://139.219.142.18:10008/upload/evtvideos/"; string videoFileServerApi = "http://139.219.142.18:10008/upload/evtvideos/";
set<long> sTsList;
mutex mutTsList;
bool validMsg(json &msg) bool validMsg(json &msg)
{ {
...@@ -190,7 +191,7 @@ private: ...@@ -190,7 +191,7 @@ private:
bProcessed = true; bProcessed = true;
}else if(metaValue == "debug:list_files"){ }else if(metaValue == "debug:list_files"){
// TODO: remove debug feature // TODO: remove debug feature
this->vTsActive = LoadVideoFiles(this->urlOut, this->hours, this->numSlices, this->vTsOld); printVideoFiles(this->sTsList);
bProcessed = true; bProcessed = true;
}else if(metaValue == "debug:toggle_log") { }else if(metaValue == "debug:toggle_log") {
// TODO: remove debug feature // TODO: remove debug feature
...@@ -331,7 +332,7 @@ private: ...@@ -331,7 +332,7 @@ private:
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0); ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
...@@ -355,7 +356,7 @@ private: ...@@ -355,7 +356,7 @@ private:
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(pDealer, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size()); ret = zmq_setsockopt(pDealer, ZMQ_IDENTITY, selfId.c_str(), selfId.size());
...@@ -524,11 +525,8 @@ protected: ...@@ -524,11 +525,8 @@ protected:
spdlog::error("evslicer {} could not open output file {}", selfId, name); spdlog::error("evslicer {} could not open output file {}", selfId, name);
} }
} }
int startIdx = 0; // TODO
if(segHead != -1) { av_dict_set(&pOptsRemux, "segment_start_number", to_string(this->sTsList.size()).data(), 0);
startIdx = segHead;
}
av_dict_set(&pOptsRemux, "segment_start_number", to_string(startIdx).data(), 0);
ret = avformat_write_header(pAVFormatRemux, &pOptsRemux); ret = avformat_write_header(pAVFormatRemux, &pOptsRemux);
if (ret < 0) { if (ret < 0) {
spdlog::error("evslicer {} error occurred when opening output file", selfId); spdlog::error("evslicer {} error occurred when opening output file", selfId);
...@@ -634,7 +632,6 @@ protected: ...@@ -634,7 +632,6 @@ protected:
return ret; return ret;
} }
//spdlog::info("LoadVideoFiles path {}, s {}, e {}", fname, posS, posE);
return fname.substr(posS, posE - posS + 1); return fname.substr(posS, posE - posS + 1);
} }
...@@ -661,122 +658,77 @@ protected: ...@@ -661,122 +658,77 @@ protected:
return string(buffer); return string(buffer);
} }
void debugFilesRing(vector<long> &v) void printVideoFiles(set<long> &list)
{ {
spdlog::info("evslicer {} debug files ring. segHead: {}, isFull: {}, max: {}",this->selfId, this->segHead, this->bSegFull, this->numSlices); spdlog::info("evslicer {} debug files ring. size: {} max: {}",this->selfId, list.size(), this->numSlices);
for(int i = 0; i <= numSlices; i++) { // lock_guard<mutex> lg(mutTsList);
spdlog::info("\tevslicer {} vector[{}] = {}, {}", selfId, i, v[i], videoFileTs2Name(v[i])); for(auto i: list) {
if(v[segToIdx(i)] == 0) { spdlog::info("\tevslicer {} file ts: {}, baseName: {}", selfId, i, videoFileTs2Name(i));
break;
}
} }
} }
vector<long> LoadVideoFiles(string path, int hours, int maxSlices, vector<long> &tsNeedUpload) void loadVideoFiles(string path, int hours, int maxSlices, set<long> &_list)
{ {
tsNeedUpload.clear();
vTsActive.clear();
segHeadP = 0;
segHead = -1;
bSegFull = false;
vector<long> v = vector<long>(maxSlices, 0);
tsNeedUpload = vector<long>(maxSlices, 0);
// get current timestamp
list<long> tsRing;
list<long>tsToProcess;
auto now = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count(); auto now = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count();
try { try {
string fname, baseName; string fname, baseName;
for (const auto & entry : fs::directory_iterator(path)) { for (const auto & entry : fs::directory_iterator(path)) {
fname = entry.path().c_str(); fname = entry.path().c_str();
if(entry.file_size() == 0 || !entry.is_regular_file()||entry.path().extension() != ".mp4") { if(entry.file_size() == 0 || !entry.is_regular_file()||entry.path().extension() != ".mp4") {
spdlog::warn("evslicer {} LoadVideoFiles skipped {} (empty/directory/!mp4)", selfId, entry.path().c_str()); spdlog::warn("evslicer {} loadVideoFiles skipped {} (empty/directory/!mp4)", selfId, entry.path().c_str());
continue; continue;
} }
baseName = getBaseName(fname); baseName = getBaseName(fname);
auto ts = videoFileName2Ts(baseName); auto ts = videoFileName2Ts(baseName);
spdlog::info("evslicer {} LoadVideoFiles basename: {}, ts: {}", selfId, baseName, ts); spdlog::info("evslicer {} loadVideoFiles basename: {}, ts: {}", selfId, baseName, ts);
// check old files // check old files
if(ts - now > hours * 60 * 60) { if(ts - now > hours * 60 * 60) {
spdlog::info("evslicer {} file {} old than {} hours: {}, {}", selfId, entry.path().c_str(), hours, ts, now); spdlog::info("evslicer {} file {} old than {} hours: {}, {}", selfId, entry.path().c_str(), hours, ts, now);
tsToProcess.insert(std::upper_bound(tsToProcess.begin(), tsToProcess.end(), ts), ts); fs::path fname(this->urlOut + "/" + baseName + ".mp4");
fs::remove(fname);
} }
else { else {
tsRing.insert(std::upper_bound(tsRing.begin(), tsRing.end(), ts), ts); insertTsList(_list, ts, maxSlices);
} }
} }
} }
catch(exception &e) { catch(exception &e) {
spdlog::error("LoasdVideoFiles exception : {}", e.what()); spdlog::error("evslicer {} {}:{} loadVideoFiles exception : {}",selfId, __FILE__, __LINE__, e.what());
} }
}
// skip old items void insertTsList(set<long> &_list, long elem, int maxSize) {
list<long>olds; // _list.insert(lower_bound(_list.begin(), _list.end(), elem), elem);
int delta = maxSlices - tsRing.size(); if(_list.size() == 0) {
int skip = delta < 0? (-delta):0; _list.insert(_list.begin(),elem);
spdlog::info("evslicer {} LoasdVideoFiles max: {}, current: {}, skip: {}", selfId, maxSlices, tsRing.size(), skip); return;
int idx = 0;
if(skip > 0) {
this->bSegFull = true;
} }
if(tsRing.size() != 0) { auto itr = _list.rbegin();
segHead = 0;
}
list<long>::iterator pos = tsRing.begin();
for(auto &i:tsRing) {
if(idx < skip) {
idx++;
pos++;
continue;
}
v[segHead] = i;
segHead++;
}
// merge
if(skip > 0) {
tsToProcess.insert(std::upper_bound(tsToProcess.begin(), tsToProcess.end(), tsRing.front()), tsRing.begin(), pos);
}
delta = maxSlices - tsToProcess.size(); for(; itr != _list.rend(); itr++) {
skip = delta < 0? (-delta) : 0; if(*itr < elem){
idx = 0; break;
pos = tsToProcess.begin();
for(auto &i:tsToProcess) {
// remove
fs::path fname(this->urlOut + "/" + videoFileTs2Name(i) + ".mp4");
fs::remove(fname);
// TODO: currently we don't cache event videos. lost on reboot.
// TODO: this behavior will be enhenced later.
continue;
// skip cache operations
if(idx < skip) {
idx++;
pos++;
continue;
} }
tsNeedUpload[segHeadP] = i;
segHeadP++;
} }
if(segHead!=0 && segHeadP != 0) { if(itr == _list.rbegin() ) {
spdlog::info("evslicer {} LoadVideoFiles active:{}, ts1:{}, ts2: {}; toprocess: {}, ts1: {}, ts2:{}", selfId, segHead, v.front(), v.back(), segHeadP, tsNeedUpload.front(), tsNeedUpload.back()); _list.insert(_list.end(), elem);
}else{
_list.insert(itr.base(), elem);
} }
debugFilesRing(v); if(_list.size() > maxSize) {
lock_guard<mutex> lg(mutTsList);
return v; _list.erase(_list.begin());
}
} }
// file monitor callback // file monitor callback
static void fileMonHandler(const std::vector<event>& evts, void *pUserData) static void fileMonHandler(const std::vector<event>& evts, void *pUserData)
{ {
static string lastFile; static string lastFile;
string ext = ".mp4"; string ext = ".mp4";
auto self = static_cast<EvSlicer*>(pUserData); auto self = static_cast<EvSlicer*>(pUserData);
...@@ -788,133 +740,90 @@ protected: ...@@ -788,133 +740,90 @@ protected:
spdlog::debug("evslicer {} invalid file: {}, last: {}", self->selfId, fullPath, lastFile); spdlog::debug("evslicer {} invalid file: {}, last: {}", self->selfId, fullPath, lastFile);
continue; continue;
} }
}
// if(lastFile == i.get_path()) { if(lastFile == i.get_path()) {
// spdlog::debug("evslicer {} skip file : {}, last: {}", self->selfId, fullPath, lastFile); spdlog::debug("evslicer {} skip file : {}, last: {}", self->selfId, fullPath, lastFile);
// continue; continue;
// } }
else if(!lastFile.empty()) {
// else if(!lastFile.empty()) { spdlog::debug("evslicer {} filemon file: {}, ts: {}, last: {}", self->selfId, i.get_path().c_str(), i.get_time(), lastFile);
// // insert into ts active try {
// spdlog::debug("evslicer {} filemon file: {}, ts: {}, last: {}", self->selfId, i.get_path().c_str(), i.get_time(), lastFile); auto baseName = self->getBaseName(lastFile);
// if(self->segHead == 0) { auto ts = self->videoFileName2Ts(baseName);
// //wrap it; if(ts == -1) {
// self->bSegFull = true; spdlog::error("evslicer {} fileMonHandler failed to process file: {}", self->selfId, lastFile);
// } }else{
self->insertTsList(self->sTsList, ts, self->numSlices);
// if(self->bSegFull) { }
// } }
catch(exception &e) {
// try { spdlog::error("evslicer {} fileMonHandler exception: {}", self->selfId, e.what());
// auto baseName = self->getBaseName(lastFile); }
// auto ts = self->videoFileName2Ts(baseName); }
// if(ts == -1) { else {
// spdlog::error("evslicer {} fileMonHandler failed to process file: {}", self->selfId, lastFile); }
// continue;
// }
// auto oldTs = self->vTsActive[self->segHead];
// if(oldTs != 0) {
// // TODO
// fs::path fname(self->urlOut + "/" + self->videoFileTs2Name(oldTs) + ".mp4");
// fs::remove(fname);
// }
// self->vTsActive[self->segHead] = ts;
// self->segHead++;
// self->segHead = self->segToIdx(self->segHead);
// }
// catch(exception &e) {
// spdlog::error("evslicer {} fileMonHandler exception: {}", self->selfId, e.what());
// }
// }
// else {
// //nop
// }
// lastFile = i.get_path();
// }
}
// lastFile = i.get_path();
int segToIdx(int seg)
{
if(seg >= numSlices) {
seg -= numSlices;
}
else if(seg <=-1) {
seg = numSlices + seg;
} }
return seg;
}
int incSegHead(int seg)
{
return segToIdx(++seg);
}
int decSegHead(int seg)
{
return segToIdx(--seg);
} }
// find video files // find video files
vector<string> findSlicesByRange(long tss, long tse, int offsetS, int offsetE) vector<string> findSlicesByRange(long tss, long tse, int offsetS, int offsetE)
{ {
vector<string> ret; vector<string> ret;
int found = 0;
int _itss = 0; lock_guard<mutex> lg(mutTsList);
if(bSegFull) { if(this->sTsList.size() == 0) {
_itss = segHead; return ret;
}
else {
_itss = 1;
} }
if(segHead == -1) { long first = *(this->sTsList.begin());
spdlog::error("evslicer {} no local records."); long end = *(--(this->sTsList.end()));
if(tse < first || tss > end) {
spdlog::info("evslicer {} event range ({}, {}) is in local range ({}, {}).", selfId, tss, tse, first, end);
return ret; return ret;
} }
else {
int idxS, idxE;
int delta = bSegFull? numSlices : 0;
for(int i = segHead - 1 + delta; i >= _itss; i--) {
if(vTsActive[segToIdx(i)] == 0) {
continue;
}
if(tse >= vTsActive[segToIdx(i)]) {
if((found &1) != 1) {
idxE = segToIdx(i);
found |= 1;
}
}
if(tss >= vTsActive[segToIdx(i)]) { first = end = 0;
if((found &2) != 2) { vector<long> tmp;
idxS = segToIdx(i); int found = 0;
found |=2; auto itr = this->sTsList.rbegin();
} for(; itr != this->sTsList.rend(); itr++) {
} if(found == 3) {
break;
}
if(found & 1) {
tmp.push_back(*itr);
spdlog::info("\t matched file: {}, s:{}, e:{}", *itr, tse, tss);
}
if(found == 3) { if(tse >= *itr) {
break; if((found &1) != 1) {
tmp.push_back(*itr);
spdlog::info("\t matched file: {}, s:{}, e:{}", *itr, tse, tss);
found |= 1;
} }
} }
if(found ==3) {
if(idxS > idxE) { if(tss >= *itr) {
idxE += numSlices; if((found &2) != 2) {
} tmp.push_back(*itr);
string sf; spdlog::info("\t matched file: {}, s:{}, e:{}", *itr, tse, tss);
for(int i = idxS; i <= idxE; i++) { found |=2;
int idx = segToIdx(i);
long ts = vTsActive[idx];
string fname = videoFileTs2Name(ts, true);
sf += "\n\t" + this->urlOut + "/" + fname + ".mp4, " + to_string(ts) + ", " + to_string(idx);
ret.push_back(fname);
} }
spdlog::info("evslicer {} event {} - {} files to upload: {}", selfId, videoFileTs2Name(tss), videoFileTs2Name(tse), sf); }
}
if(found ==3) {
string sf;
auto itr = tmp.rbegin();
for(; itr != tmp.rend(); itr++) {
string fname = videoFileTs2Name(*itr, true);
sf += "\n\t" + this->urlOut + "/" + fname + ".mp4, " + to_string(*itr);
ret.push_back(fname);
} }
spdlog::info("evslicer {} event {} - {} files to upload: {}", selfId, videoFileTs2Name(tss), videoFileTs2Name(tse), sf);
} }
return ret; return ret;
...@@ -989,24 +898,23 @@ public: ...@@ -989,24 +898,23 @@ public:
}); });
thCloudMsgHandler.detach(); thCloudMsgHandler.detach();
//
this->loadVideoFiles(this->urlOut, this->hours, this->numSlices, this->sTsList);
// thread for slicer maintenace
thSliceMgr = thread([this]() {
// get old and active slices
monitor * m = nullptr;
// TODO: bugfix CreateDirMon(&m, this->urlOut, ".mp4", vector<string>(), EvSlicer::fileMonHandler, (void *)this);
// // thread for slicer maintenace });
// thSliceMgr = thread([this]() { thSliceMgr.detach();
// // get old and active slices
// this->vTsActive = this->LoadVideoFiles(this->urlOut, this->hours, this->numSlices, this->vTsOld);
// spdlog::info("evslicer {} will store slice from index: {}", selfId, this->segHead);
// monitor * m = nullptr;
// CreateDirMon(&m, this->urlOut, ".mp4", vector<string>(), EvSlicer::fileMonHandler, (void *)this);
// });
// thSliceMgr.detach();
// event thread // event thread
thEventHandler = thread([this] { thEventHandler = thread([this] {
while(true) while(true)
{ {
string evt; string evt;
int ret = 0;
unique_lock<mutex> lk(this->mutEvent); unique_lock<mutex> lk(this->mutEvent);
this->cvEvent.wait(lk, [this] {return !(this->eventQueue.empty());}); this->cvEvent.wait(lk, [this] {return !(this->eventQueue.empty());});
...@@ -1028,45 +936,24 @@ public: ...@@ -1028,45 +936,24 @@ public:
long offsetE = 0; long offsetE = 0;
// TODO: async // TODO: async
this->vTsActive = this->LoadVideoFiles(this->urlOut, this->hours, this->numSlices, this->vTsOld);
if(this->segHead == -1) {
spdlog::error("evslicer {} no local video files");
continue;
}
if(tss < this->bootTime) { if(tss < this->bootTime) {
spdlog::warn("evslicer {} should we discard old msg? {} < bootTime {}", selfId, evt, this->bootTime); spdlog::warn("evslicer {} should we discard old msg? {} < bootTime {}", selfId, evt, this->bootTime);
} }
// TODO: scheduled task long first = 0, end = 0;
// check tss, tse if(this->sTsList.size()!=0 ) {
int sIdx = 0, eIdx = 0; first = *(this->sTsList.begin());
if(this->bSegFull) { end = *(--(this->sTsList.end()));
sIdx = this->incSegHead(this->segHead);
eIdx = this->decSegHead(this->segHead);
}else{
sIdx = 0;
eIdx = this->decSegHead(this->segHead);
} }
if(sIdx > eIdx || tss > this->vTsActive[eIdx] || tse < this->vTsActive[sIdx]) { if(first == 0||tse < first || tss > end) {
spdlog::error("evslicer {} event range ({}, {}) not in local range ({}, {}).", selfId, tss, tse, this->vTsActive[sIdx], this->vTsActive[eIdx]); spdlog::error("evslicer {} event range ({}, {}) is in local range ({}, {}).", selfId, tss, tse, first, end);
continue; continue;
} }
if(sIdx == eIdx)
{
spdlog::info("evslicer {} wait for {}s to matching event videos", this->selfId, this->seconds + 5);
this_thread::sleep_for(chrono::seconds(this->seconds + 5));
}else{
spdlog::info("evslicer {} event range ({}, {}) is in local range ({}, {}).", selfId, tss, tse, this->vTsActive[sIdx], this->vTsActive[eIdx]);
}
auto v = findSlicesByRange(tss, tse, offsetS, offsetE); auto v = findSlicesByRange(tss, tse, offsetS, offsetE);
if(v.size() == 0) { if(v.size() == 0) {
spdlog::error("evslicer {} ignore upload videos in range ({}, {})", this->selfId, this->videoFileTs2Name(tss), this->videoFileTs2Name(tse)); spdlog::error("evslicer {} ignore upload videos in range ({}, {}): not found", this->selfId, this->videoFileTs2Name(tss), this->videoFileTs2Name(tse));
} }
else { else {
vector<tuple<string, string> > params= {{"startTime", to_string(tss)},{"endTime", to_string(tse)},{"cameraId", ipcSn}, {"headOffset", to_string(offsetS)},{"tailOffset", to_string(offsetE)}}; vector<tuple<string, string> > params= {{"startTime", to_string(tss)},{"endTime", to_string(tse)},{"cameraId", ipcSn}, {"headOffset", to_string(offsetS)},{"tailOffset", to_string(offsetE)}};
...@@ -1078,13 +965,13 @@ public: ...@@ -1078,13 +965,13 @@ public:
sf+="\tfile\t" + fname + "\n"; sf+="\tfile\t" + fname + "\n";
} }
spdlog::info("evslicer {} file upload range:{} - {} ({} - {}), url: {}", selfId, tss, tse, this->videoFileTs2Name(tss), this->videoFileTs2Name(tse), this->videoFileServerApi); spdlog::info("evslicer {} file upload range:({},{}) = ({}, {}), url: {}", selfId, tss, tse, this->videoFileTs2Name(tss), this->videoFileTs2Name(tse), this->videoFileServerApi);
// TODO: check result and reschedule it // TODO: check result and reschedule it
if(netutils::postFiles(std::move(this->videoFileServerApi), std::move(params), std::move(fileNames)) != 0) { if(netutils::postFiles(std::move(this->videoFileServerApi), std::move(params), std::move(fileNames)) != 0) {
spdlog::error("evslicer {} failed to upload files:\n{}", selfId, sf); spdlog::error("evslicer {} failed to upload files:\n{}", selfId, sf);
} }
else { else {
spdlog::info("evslicer {} successfully uploaded {} - {} ({} - {})files:\n{}", selfId, tss, tse, this->videoFileTs2Name(tss), this->videoFileTs2Name(tse), sf); spdlog::info("evslicer {} successfully uploaded ({}, {}) = ({}, {}) files:\n{}", selfId, tss, tse, this->videoFileTs2Name(tss), this->videoFileTs2Name(tse), sf);
} }
} }
} }
......
opencv-motion-detect/inc/zmqhelper.cpp
浏览文件 @ ab19868f
...@@ -102,7 +102,7 @@ int setupRouter(void **ctx, void **s, string addr){ ...@@ -102,7 +102,7 @@ int setupRouter(void **ctx, void **s, string addr){
//ZMQ_TCP_KEEPALIVE_INTVL //ZMQ_TCP_KEEPALIVE_INTVL
ret = 1; ret = 1;
zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret)); zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 20; ret = 5;
zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret)); zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret)); zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
zmq_setsockopt(*s, ZMQ_ROUTER_NOTIFY, &opt_notify, sizeof (opt_notify)); zmq_setsockopt(*s, ZMQ_ROUTER_NOTIFY, &opt_notify, sizeof (opt_notify));
...@@ -119,6 +119,11 @@ int setupDealer(void **ctx, void **s, string addr, string ident) { ...@@ -119,6 +119,11 @@ int setupDealer(void **ctx, void **s, string addr, string ident) {
int ret = 0; int ret = 0;
*ctx = zmq_ctx_new(); *ctx = zmq_ctx_new();
*s = zmq_socket(*ctx, ZMQ_DEALER); *s = zmq_socket(*ctx, ZMQ_DEALER);
ret = 1;
zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
ret = 5;
zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_setsockopt(*s, ZMQ_IDENTITY, ident.c_str(), ident.size()); ret = zmq_setsockopt(*s, ZMQ_IDENTITY, ident.c_str(), ident.size());
ret += zmq_setsockopt (*s, ZMQ_ROUTING_ID, ident.c_str(), ident.size()); ret += zmq_setsockopt (*s, ZMQ_ROUTING_ID, ident.c_str(), ident.size());
if(ret < 0) { if(ret < 0) {
......
opencv-motion-detect/test_rlist.cpp
浏览文件 @ ab19868f
...@@ -14,7 +14,6 @@ void insertTsList(set<long> &_list, long elem, int maxSize) { ...@@ -14,7 +14,6 @@ void insertTsList(set<long> &_list, long elem, int maxSize) {
} }
auto itr = _list.rbegin(); auto itr = _list.rbegin();
for(; itr != _list.rend(); itr++) { for(; itr != _list.rend(); itr++) {
if(*itr < elem){ if(*itr < elem){
break; break;
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论