Skip to content

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

init

上级 eee69bf3
显示空白字符变更
内嵌 并排
正在显示 包含 217 行增加 和 201 行删除
opencv-motion-detect/evpuller.cpp
浏览文件 @ ed3f32a9
......@@ -20,20 +20,28 @@ namespace fs = std::filesystem;
using namespace std;
class PacketProducer: public TinyThread {
class EvPuller: public TinyThread {
private:
void *pPubContext = NULL; // for packets publishing
void *pPublisher = NULL;
void *pPubCtx = NULL; // for packets publishing
void *pPub = NULL;
void *pRepCtx = NULL; // for packets REP
void *pRep = NULL;
AVFormatContext *pAVFormatInput = NULL;
string urlIn;
string urlIn, urlPub;
int *streamList = NULL, numStreams = 0;
public:
PacketProducer(string urlIn):urlIn(urlIn){
setupMq();
EvPuller()
{
int ret = 0;
do {
init();
ret = setupMq();
}while(ret < 0);
}
~PacketProducer(){
~EvPuller()
{
}
protected:
......@@ -41,7 +49,6 @@ protected:
void run()
{
int ret = 0;
setupMq();
if ((ret = avformat_open_input(&pAVFormatInput, urlIn.c_str(), NULL, NULL)) < 0) {
spdlog::error("Could not open input file {}", urlIn);
}
......@@ -109,7 +116,7 @@ protected:
char * data = NULL;
int size = AVPacketSerializer::encode(packet, &data);
zmq_msg_init_data(&msg, (void*)data, size, mqPacketFree, NULL);
zmq_send_const(pPublisher, zmq_msg_data(&msg), size, 0);
zmq_send_const(pPub, zmq_msg_data(&msg), size, 0);
av_packet_unref(&packet);
}
......@@ -117,137 +124,84 @@ protected:
// TODO:
if(ret < 0 && !bStopSig) {
// reconnect
}else {
}
else {
std::cout << "Task End" << std::endl;
}
}
private:
int setupMq()
private:
int init()
{
teardownMq();
pPubContext = zmq_ctx_new();
pPublisher = zmq_socket(pPubContext, ZMQ_PUB);
int rc = zmq_bind(pPublisher, "tcp://0.0.0.0:5556");
if(rc != 0) {
spdlog::error("failed create pub");
bool inited = false;
while(!inited) {
// TODO: read db to get sn
const char* sn = "ILS-2";
// req config
json jr = cloudutils::registry(sn, "evpuller", 0);
bool bcnt = false;
try {
spdlog::info("registry: {:s}", jr.dump());
string ipc = jr["data"]["ipc"];
string user = jr["data"]["username"];
string passwd = jr["data"]["password"];
json data = jr["data"]["services"]["evpuller"];
urlIn = "rtsp://" + user + ":" + passwd + "@"+ ipc + "/h264/ch1/sub/av_stream";
urlPub = string("tcp://") +data["addr"].get<string>() + ":" + to_string(data["port-pub"]);
}
catch(exception &e) {
bcnt = true;
spdlog::error(e.what());
}
if(bcnt) {
this_thread::sleep_for(chrono::milliseconds(1000*10));
continue;
}
return 0;
inited = true;
}
int teardownMq()
{
if(pPublisher != NULL) {
zmq_close(pPublisher);
}
if(pPubContext != NULL) {
zmq_ctx_destroy(pPubContext);
}
return 0;
}
};
class EdgeVideoMgr {
private:
#define SECS_SLICE (60*5/2)
AVFormatContext *pAVFormatInput = NULL, *pAVFormatRemux = NULL;
AVCodec *pCodec = NULL;
AVDictionary *pOptsRemux = NULL, *pOptsInput = NULL, *pOptsOutput = NULL;
int idxVideo = -1, idxAudio = -1, numStreams = 0, numSlices = 6, secsSlice = SECS_SLICE;
int *streamList = NULL;
bool bPush = true, bRecord = false;
string urlIn, urlOut, pathSlice;
unordered_map<string, string> envParams = unordered_map<string, string>();
// mq
void *pRepContext = NULL; // for msg from edge gateway
void *pReqContext = NULL; // for msg to edge gateway
private:
void setupParams()
int setupMq()
{
char *tmp = getenv("URL_IN");
urlIn = (tmp == NULL?string(""): string(tmp));
tmp= getenv("URL_OUT");
urlOut = (tmp == NULL?string(""): string(tmp));
tmp = getenv("SLICE_NUM");
numSlices = (tmp == NULL?6:atoi(tmp));
if(numSlices <=2) {
numSlices = 6;
}
spdlog::info("in: {}", urlIn);
tmp = getenv("SLICE_PATH");
pathSlice = (tmp == NULL?string("slices"):string(tmp));
// OSX XCode doesn't ship with the filesystem header as of version 10.x
#ifdef __LINUX___
if (!fs::exists(pathSlice.c_str())) {
if (!fs::create_directory(pathSlice.c_str())) {
spdlog::error("can't create directory: {}", pathSlice.c_str());
exit(1);
}
fs::permissions(pathSlice.c_str(), fs::perms::all);
}
#endif
tmp = getenv("PUSH");
bPush = (tmp == NULL?false: (string(tmp) == string("false")?false:true));
teardownMq();
pPubCtx = zmq_ctx_new();
pPub = zmq_socket(pPubCtx, ZMQ_PUB);
tmp = getenv("SLICE_SECS");
secsSlice = (tmp == NULL?SECS_SLICE:atoi(tmp));
if(secsSlice < SECS_SLICE) {
secsSlice = SECS_SLICE;
int rc = zmq_bind(pPub, urlPub.c_str());
if(rc != 0) {
spdlog::error("failed create pub: {}, {}", zmq_strerror(rc), urlPub.c_str());
this_thread::sleep_for(chrono::milliseconds(100*10));
return -1;
}
if(urlIn == "" or urlOut == "") {
spdlog::error("no input/output url");
exit(1);
}
return 0;
}
int setupStreams()
int teardownMq()
{
int ret = 0;
PacketProducer packetProducer(urlIn);
packetProducer.join();
// std::this_thread::sleep_for(std::chrono::milliseconds(30000));
// packetProducer.stop();
return ret;
if(pPub != NULL) {
zmq_close(pPub);
}
public:
// ctor
EdgeVideoMgr()
{
setupParams();
setupStreams();
if(pPubCtx != NULL) {
zmq_ctx_destroy(pPubCtx);
}
// dtor
~EdgeVideoMgr()
{
avformat_close_input(&pAVFormatInput);
/* close output */
if (pAVFormatRemux && !(pAVFormatRemux->oformat->flags & AVFMT_NOFILE))
avio_closep(&pAVFormatRemux->pb);
avformat_free_context(pAVFormatRemux);
av_freep(&streamList);
return 0;
}
};
int main(int argc, char **argv)
{
av_log_set_level(AV_LOG_INFO);
spdlog::set_level(spdlog::level::info);
DB::exec(NULL, NULL, NULL ,NULL);
spdlog::info("hello");
auto vp = EdgeVideoMgr();
DB::exec(NULL, NULL, NULL,NULL);
auto evp = EvPuller();
evp.join();
return 0;
}
opencv-motion-detect/inc/common.hpp
浏览文件 @ ed3f32a9
#ifndef __COMMON_H__
#define __COMMON_H__
extern "C" {
extern "C"
{
#include <libavformat/avformat.h>
#include <libavutil/time.h>
}
......@@ -10,91 +11,100 @@ extern "C" {
using json = nlohmann::json;
#undef av_err2str
#define av_err2str(errnum) av_make_error_string((char*)__builtin_alloca(AV_ERROR_MAX_STRING_SIZE), AV_ERROR_MAX_STRING_SIZE, errnum)
#define av_err2str(errnum) av_make_error_string((char *)__builtin_alloca(AV_ERROR_MAX_STRING_SIZE), AV_ERROR_MAX_STRING_SIZE, errnum)
#define PS_MARK_E "DEADBEEF"
#define PS_MARK_S "BEEFDEAD"
void avlogThrow(void * avcl, int lvl, const char *fmt, ...)
void avlogThrow(void *avcl, int lvl, const char *fmt, ...)
{
(void) avcl;
(void) lvl;
(void)avcl;
(void)lvl;
va_list args;
va_start( args, fmt );
va_start(args, fmt);
av_log(NULL, AV_LOG_FATAL, fmt, args);
va_end( args );
va_end(args);
throw fmt;
}
namespace AVPacketSerializer {
int encode(AVPacket &pkt, char **bytes) {
// AVPacketSerializer
namespace AVPacketSerializer
{
int encode(AVPacket &pkt, char **bytes)
{
int cnt = 0;
//data
int wholeSize = strlen(PS_MARK_S) + sizeof(pkt.size) + pkt.size;
//side data
wholeSize +=sizeof(pkt.side_data_elems);
if(pkt.side_data_elems != 0) {
for(int i = 0; i < pkt.side_data_elems; i++) {
wholeSize += sizeof(pkt.side_data_elems);
if (pkt.side_data_elems != 0)
{
for (int i = 0; i < pkt.side_data_elems; i++)
{
wholeSize += pkt.side_data[i].size + sizeof(AVPacketSideData);
}
}
// 4 + 8: wholeSize + DEADBEAF
wholeSize += sizeof(pkt.pts) * 4 + sizeof(pkt.flags) + sizeof(pkt.stream_index) + sizeof(wholeSize) + strlen(PS_MARK_E);
*bytes = (char*)malloc(wholeSize);
*bytes = (char *)malloc(wholeSize);
memcpy((*bytes)+cnt, PS_MARK_S, strlen(PS_MARK_S));
memcpy((*bytes) + cnt, PS_MARK_S, strlen(PS_MARK_S));
cnt += strlen(PS_MARK_S);
// data
memcpy((*bytes)+cnt, &(pkt.size), sizeof(pkt.size));
cnt +=sizeof(pkt.size);
memcpy((*bytes )+cnt, pkt.data, pkt.size);
memcpy((*bytes) + cnt, &(pkt.size), sizeof(pkt.size));
cnt += sizeof(pkt.size);
memcpy((*bytes) + cnt, pkt.data, pkt.size);
cnt += pkt.size;
//side data
memcpy((*bytes )+cnt, &(pkt.side_data_elems), sizeof(pkt.side_data_elems));
memcpy((*bytes) + cnt, &(pkt.side_data_elems), sizeof(pkt.side_data_elems));
cnt += sizeof(pkt.side_data_elems);
if(pkt.side_data_elems != 0) {
for(int i = 0; i < pkt.side_data_elems; i++) {
memcpy((*bytes )+cnt, &(pkt.side_data[i].size), sizeof(pkt.side_data[i].size));
cnt+=sizeof(pkt.side_data[i].size);
memcpy((*bytes )+cnt, pkt.side_data[i].data, pkt.side_data[i].size);
cnt+=pkt.side_data[i].size;
memcpy((*bytes )+cnt, &(pkt.side_data[i].type), sizeof(pkt.side_data[i].type));
cnt+=sizeof(pkt.side_data[i].type);
if (pkt.side_data_elems != 0)
{
for (int i = 0; i < pkt.side_data_elems; i++)
{
memcpy((*bytes) + cnt, &(pkt.side_data[i].size), sizeof(pkt.side_data[i].size));
cnt += sizeof(pkt.side_data[i].size);
memcpy((*bytes) + cnt, pkt.side_data[i].data, pkt.side_data[i].size);
cnt += pkt.side_data[i].size;
memcpy((*bytes) + cnt, &(pkt.side_data[i].type), sizeof(pkt.side_data[i].type));
cnt += sizeof(pkt.side_data[i].type);
}
}
// other properties
memcpy((*bytes )+cnt, &(pkt.pts), sizeof(pkt.pts));
cnt+=sizeof(pkt.pts);
memcpy((*bytes )+cnt, &(pkt.dts), sizeof(pkt.dts));
cnt+=8;
memcpy((*bytes )+cnt, &(pkt.pos), sizeof(pkt.pos));
cnt+=sizeof(pkt.pos);
memcpy((*bytes )+cnt, &(pkt.duration), sizeof(pkt.duration));
cnt+=sizeof(pkt.duration);
memcpy((*bytes) + cnt, &(pkt.pts), sizeof(pkt.pts));
cnt += sizeof(pkt.pts);
memcpy((*bytes) + cnt, &(pkt.dts), sizeof(pkt.dts));
cnt += 8;
memcpy((*bytes) + cnt, &(pkt.pos), sizeof(pkt.pos));
cnt += sizeof(pkt.pos);
memcpy((*bytes) + cnt, &(pkt.duration), sizeof(pkt.duration));
cnt += sizeof(pkt.duration);
// deprecated
//memcpy((*bytes )+cnt, &(pkt.convergence_duration), sizeof(pkt.convergence_duration));
//cnt+=sizeof(pkt.convergence_duration);
memcpy((*bytes )+cnt, &(pkt.flags), sizeof(pkt.flags));
cnt+=sizeof(pkt.flags);
memcpy((*bytes )+cnt, &(pkt.stream_index), sizeof(pkt.stream_index));
cnt+=sizeof(pkt.stream_index);
memcpy((*bytes )+cnt,&wholeSize, sizeof(wholeSize));
cnt+=sizeof(wholeSize);
memcpy((*bytes )+cnt, PS_MARK_E, strlen(PS_MARK_E));
cnt+=strlen(PS_MARK_E);
memcpy((*bytes) + cnt, &(pkt.flags), sizeof(pkt.flags));
cnt += sizeof(pkt.flags);
memcpy((*bytes) + cnt, &(pkt.stream_index), sizeof(pkt.stream_index));
cnt += sizeof(pkt.stream_index);
memcpy((*bytes) + cnt, &wholeSize, sizeof(wholeSize));
cnt += sizeof(wholeSize);
memcpy((*bytes) + cnt, PS_MARK_E, strlen(PS_MARK_E));
cnt += strlen(PS_MARK_E);
assert(cnt == wholeSize);
spdlog::debug("pkt origin size {:d}, serialized size: {:d}, elems: {:d}", pkt.size, wholeSize, pkt.side_data_elems);
return wholeSize;
}
}
int decode(char * bytes, int len, AVPacket *pkt) {
int decode(char *bytes, int len, AVPacket *pkt)
{
// allocate packet mem on heap
//AVPacket *pkt = (AVPacket*)malloc(sizeof(AVPacket));
int ret = 0;
int got = 0;
if(memcmp(PS_MARK_E, bytes + len - strlen(PS_MARK_E), strlen(PS_MARK_E)) != 0 || memcmp(PS_MARK_S, bytes, strlen(PS_MARK_S))) {
if (memcmp(PS_MARK_E, bytes + len - strlen(PS_MARK_E), strlen(PS_MARK_E)) != 0 || memcmp(PS_MARK_S, bytes, strlen(PS_MARK_S)))
{
spdlog::error("invalid packet");
return -1;
}
......@@ -107,10 +117,11 @@ namespace AVPacketSerializer {
got += pkt->size;
memcpy(&pkt->side_data_elems, bytes + got, sizeof(pkt->side_data_elems));
got += sizeof(pkt->side_data_elems);
for(int i = 0; i < pkt->side_data_elems; i++) {
memcpy(&(pkt->side_data[i].size), bytes+got, sizeof(pkt->side_data[i].size));
for (int i = 0; i < pkt->side_data_elems; i++)
{
memcpy(&(pkt->side_data[i].size), bytes + got, sizeof(pkt->side_data[i].size));
got += sizeof(pkt->side_data[i].size);
memcpy(pkt->side_data[i].data,bytes + got ,pkt->side_data[i].size);
memcpy(pkt->side_data[i].data, bytes + got, pkt->side_data[i].size);
got += pkt->side_data[i].size;
memcpy(&(pkt->side_data[i].type), bytes + got, sizeof(pkt->side_data[i].type));
got += sizeof(pkt->side_data[i].type);
......@@ -140,15 +151,19 @@ namespace AVPacketSerializer {
spdlog::debug("wholeSize: {:d}, {:d}", wholeSize, got);
return ret;
}
}
} // namespace AVPacketSerializer
void mqPacketFree(void *data, void*hint) {
void mqPacketFree(void *data, void *hint)
{
free(data);
}
namespace AVFormatCtxSerializer {
int encode(AVFormatContext *ctx, char **bytes) {
// AVFormatCtxSerializer
namespace AVFormatCtxSerializer
{
int encode(AVFormatContext *ctx, char **bytes)
{
int ret = 0;
int wholeSize = 0;
int got = 0;
......@@ -157,7 +172,8 @@ namespace AVFormatCtxSerializer {
// num streams
wholeSize += sizeof(ctx->nb_streams);
wholeSize += sizeof(ctx->nb_streams);
for(int i = 0; i < ctx->nb_streams; i++) {
for (int i = 0; i < ctx->nb_streams; i++)
{
wholeSize += sizeof(AVStream);
wholeSize += sizeof(AVCodecParameters);
}
......@@ -165,13 +181,14 @@ namespace AVFormatCtxSerializer {
wholeSize += strlen(PS_MARK_E);
// alloc memory
*bytes = (char *) malloc(wholeSize);
*bytes = (char *)malloc(wholeSize);
// populate
memcpy((*bytes) + got, PS_MARK_S, strlen(PS_MARK_S));
got += strlen(PS_MARK_S);
memcpy((*bytes) + got, (void*)&(ctx->nb_streams), sizeof(ctx->nb_streams));
memcpy((*bytes) + got, (void *)&(ctx->nb_streams), sizeof(ctx->nb_streams));
got += sizeof(ctx->nb_streams);
for(int i = 0; i < ctx->nb_streams; i++) {
for (int i = 0; i < ctx->nb_streams; i++)
{
//
memcpy((*bytes) + got, ctx->streams[i], sizeof(AVStream));
got += sizeof(AVStream);
......@@ -184,65 +201,110 @@ namespace AVFormatCtxSerializer {
memcpy((*bytes) + got, PS_MARK_E, strlen(PS_MARK_E));
return wholeSize;
}
}
int decode(char *bytes, int len, AVFormatContext *pCtx) {
int decode(char *bytes, int len, AVFormatContext *pCtx)
{
int ret = 0;
int got = 0;
if(memcmp(PS_MARK_S, bytes+got, strlen(PS_MARK_S)) !=0 && memcmp(PS_MARK_E, bytes+len-strlen(PS_MARK_E), strlen(PS_MARK_E)) != 0) {
if (memcmp(PS_MARK_S, bytes + got, strlen(PS_MARK_S)) != 0 && memcmp(PS_MARK_E, bytes + len - strlen(PS_MARK_E), strlen(PS_MARK_E)) != 0)
{
spdlog::error("invalid packet");
return -1;
}
got+=strlen(PS_MARK_S);
memcpy(&ret, bytes+got, sizeof(ret));
got +=sizeof(ret);
got += strlen(PS_MARK_S);
memcpy(&ret, bytes + got, sizeof(ret));
got += sizeof(ret);
pCtx->streams = (AVStream **)malloc(sizeof(AVStream *) * ret);
pCtx->nb_streams = ret;
for(int i = 0; i < ret; i++) {
pCtx->streams[i] = (AVStream*)malloc(sizeof(AVStream));
memcpy(pCtx->streams[i], bytes+got, sizeof(AVStream));
got+=sizeof(AVStream);
pCtx->streams[i]->codecpar = (AVCodecParameters*)malloc(sizeof(AVCodecParameters));
memcpy(pCtx->streams[i]->codecpar, bytes+got, sizeof(AVCodecParameters));
got +=sizeof(AVCodecParameters);
for (int i = 0; i < ret; i++)
{
pCtx->streams[i] = (AVStream *)malloc(sizeof(AVStream));
memcpy(pCtx->streams[i], bytes + got, sizeof(AVStream));
got += sizeof(AVStream);
pCtx->streams[i]->codecpar = (AVCodecParameters *)malloc(sizeof(AVCodecParameters));
memcpy(pCtx->streams[i]->codecpar, bytes + got, sizeof(AVCodecParameters));
got += sizeof(AVCodecParameters);
}
memcpy(&ret, bytes+got, sizeof(ret));
memcpy(&ret, bytes + got, sizeof(ret));
assert(ret == len);
return ret;
}
}
void freeCtx(AVFormatContext *pCtx) {
for(int i = 0; i < pCtx->nb_streams; i++) {
void freeCtx(AVFormatContext *pCtx)
{
for (int i = 0; i < pCtx->nb_streams; i++)
{
free(pCtx->streams[i]->codecpar);
free(pCtx->streams[i]);
}
free(pCtx->streams);
}
} // namespace AVFormatCtxSerializer
// cloudutils
namespace cloudutils
{
/**
* scn:
* evpuller, evmgr, evmotion, evslicer
*
* */
/*
{
"code":0,
"time":0,
"data":{
"ipc":"192.168.0.23",
"username":"admin",
"password":"FWBWTU",
"services":{
"evmgr":{
"sn":"ILS-1",
"addr":"0.0.0.0",
"port-pub":5556,
"port-rep":5557,
"iid":1
},
"evpuller":{
"sn":"ILS-2",
"addr":"0.0.0.0",
"port-pub":5556,
"port-rep":5557,
"iid":2
},
"evslicer":[
{
"sn":"ILS-3",
"addr":"192.168.0.25",
"iid":3
}
],
"evml":[
{
"feature":"motion",
"sn":"ILS-4",
"addr":"192.168.0.26",
"iid":4
}
]
}
}
}
*/
const char *config = "{\"code\":0,\"time\":0,\"data\":{\"ipc\":\"192.168.0.23\",\"username\":\"admin\",\"password\":\"FWBWTU\",\"services\":{\"evmgr\":{\"sn\":\"ILS-1\",\"addr\":\"0.0.0.0\",\"port-pub\":5556,\"port-rep\":5557,\"iid\":1},\"evpuller\":{\"sn\":\"ILS-2\",\"addr\":\"0.0.0.0\",\"port-pub\":5556,\"port-rep\":5557,\"iid\":2},\"evslicer\":[{\"sn\":\"ILS-3\",\"addr\":\"192.168.0.25\",\"iid\":3}],\"evml\":[{\"feature\":\"motion\",\"sn\":\"ILS-4\",\"addr\":\"192.168.0.26\",\"iid\":4}]}}}";
namespace cloudutils {
json registry(const char *gn, const char *scn, int iid) {
json jret;
json registry(const char *sn, const char *scn, int iid)
{
// find local info in db
// request cloud info
// moc
jret["iid"] = "MOCK_IID";
jret["scn"] = string(scn);
jret["ipc"] = "172.31.0.51";
jret["gn"] = "MOCK_GN";
jret["ga"] = "localhost";
if(scn != NULL && strcmp(scn, "evmgr")) {
jret["evmgr"] = "tcp://localhost:5556";
}else{
jret["evmgr"] = "tcp://0.0.0.0:5556";
}
jret["code"] = 0;
json ret = json::parse(config);
return jret;
}
return ret;
}
} // namespace cloudutils
#endif
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论