上个版本: 只是用到ctypes进行传输, 这次将python服务端更改为C++服务端,方便后续维护.

本文实现功能: python传输图片给C++, C++接受图片后对图片进行处理,并将结果返回给python客户端, pass image from python to C++

.h文件

注意文中的model

// .h
#pragma once
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <signal.h>
#include <opencv2/opencv.hpp>

using namespace std;
using namespace cv;

class ModelManager;

class ServerManager
{
private:
    int m_port;
    char *m_addr;
    cv::VideoCapture m_cap;
    int m_server;
    int m_accept; // client conn
public:
    bool initialization(const int &port, const cv::VideoCapture &cap, char *addr = nullptr);
    bool initialization(const int &port, char *addr = nullptr);
    bool build_connect();
    bool acceptClient();
    void error_print(const char *ptr);
    bool free_connect();

    bool send_data_frame(ModelManager& model);
    bool receive_data_frame(cv::Mat &frame, ModelManager& model);
};

.cpp文件

#include "ServerManager.h"
#include "ModelManager.h"
#define BUFFER_SIZE 65538

void ServerManager::error_print(const char * ptr) {
        perror(ptr);
        exit(EXIT_FAILURE);
}

bool ServerManager::initialization(const int& port, const cv::VideoCapture& cap, char* addr){
    m_port = htons(port);
    m_addr = addr;
    m_cap = cap;
    return true;
}

bool ServerManager::initialization(const int& port, char* addr){
    m_port = htons(port);
    m_addr = addr;
    return true;
}

bool ServerManager::build_connect() {
    struct sockaddr_in server_addr;
    bzero(&server_addr,sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = m_addr?inet_addr(m_addr):INADDR_ANY;
    server_addr.sin_port = m_port;
    // create socket
    m_server = socket(AF_INET, SOCK_STREAM, 0);
    if(m_server < 0)
        error_print("socket bind error");
    // can reuse port
    int on = 1;
    if(setsockopt(m_server,SOL_SOCKET,SO_REUSEADDR,&on,sizeof(on)) < 0)
        error_print("setsockopt error");
    // bind addr
    if(bind(m_server, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0)
        error_print("bind error");
    // listen only one client
    if(listen(m_server, 1) < 0)
        error_print("listen failed");

    cout << "ServerManager is listening, plesae wait..." << endl;

    return true;
}

bool ServerManager::acceptClient(){
    struct sockaddr_in accept_addr;
    socklen_t accept_len = sizeof(accept_addr);
    bzero(&accept_addr,sizeof(accept_addr));
    // accept client connection
    if((m_accept = accept(m_server,(struct sockaddr*)&accept_addr,&accept_len)) < 0)
        error_print("accept error");
    std::cout << "Connection established" << std::endl;
    return true;
}

bool ServerManager::send_data_frame(ModelManager& model) {
    char *json_output = nullptr;
    json_output = model.createJson();
    if (json_output == nullptr) {
        return false;
    }

    // printf("send data %s\n", json_output);
    // just send json_output, dont memcpy new char*!!! it wastes me two hours
    // send json
    int result = send(m_accept, json_output, strlen(json_output), 0);
    if (result == -1) {
        cout << "send fail" << endl;
        return false;
    }
    return true;
}

bool ServerManager::receive_data_frame(Mat& frame, ModelManager& model) {
    // recv frame size
    int data_size;
    if (recv(m_accept, &data_size, sizeof(data_size), 0) != sizeof(data_size)) {
        // when client close, then close connection
        close(m_accept);
        cout << "close connection to client" << endl;
        acceptClient(); // restart a new accept, to accept new connection
        return false;
    }
    cout << data_size << endl;
    // recv frame data

    // char buf[data_size];
    // std::vector<uchar> decode;
    // int bytes_received = 0;
    // do
    // {
    //  int nBytes = recv(m_accept, buf, data_size - bytes_received, 0);
    //  for (int i = 0; i < nBytes; i++)  // maybe can use memcpy, maybe faster
    //  {
    //      decode.emplace_back(buf[i]);
    //  }
    //  cout << bytes_received << endl;
    //  bytes_received += nBytes;
    // } while (bytes_received < data_size);
    char *recv_char = new char[data_size];
    std::vector<uchar> decode(data_size, 0);
    int index = 0;
    int bytes_received = 0;
    int count = data_size;
    while (count > 0)// if count >= 0, dead loop
    {
        int iRet = recv(m_accept, recv_char, count, 0);
        if (index >= data_size) index = data_size;
        memcpy(&decode[index], recv_char , iRet);
        index += iRet;
        if (!iRet) { return -1; }
        count -= iRet;
    }
    // decode message
    frame = imdecode(decode, cv::IMREAD_COLOR);
    // push into Model's queueMat
    model.mtxQueueImg.lock();
    model.queueMat.push(frame);
    model.mtxQueueImg.unlock();
    return true;
}

bool ServerManager::free_connect() {
    m_cap.release();
    close(m_accept);
    close(m_server);
    return true;
}

.h文件

#pragma once
#include "CV_Classify.h"
#include "CV_Detect.h"
#include "ServerManager.h"
#include <opencv2/opencv.hpp>
#include <mutex>
#include <queue>
#include <unistd.h> // usleep
#include <thread>
#include "cJSON.h"
#include <string>

using namespace std;
using namespace cv;

class ModelManager{
public:
    Detect objdetect;
    Classify objclassify;
    std::mutex mtxQueueDet;  // mutex for detect queue
    std::mutex mtxQueueImg;  // mutex for image queue
    std::mutex mtxQueueCls;  // mutex for classify queue
    std::queue<cv::Mat> queueMat;
    std::queue<ObjDetectOutput> queueDetOut;// Detect queue
    std::queue<ObjClassifyOutput> queueClsOut;// Classify queue

    bool DetectFlag = true;
    bool ClassifyFlag = true;
    bool empty_flag = false;
    friend class ServerManager;
public:
    void initDetectModel() ;
    void initClassifyModel() ;
    void DetectImg();
    void ClassifyImg();
    void getClsResult(ObjClassifyOutput &output);
    // ObjClassifyOutput getClsResult();
    char* createJson();
};

.cpp文件

部分有删减,createJson可参考使用,利用json来传递值

#include "ModelManager.h"

void ModelManager::initDetectModel()
{
    std::string config_path = "DetectConfig.yaml";
    objdetect.Init(config_path, 1);
}
void ModelManager::initClassifyModel()
{
    std::string config_path = "ClassiflyConfig.yaml";
    objclassify.Init(config_path, 1);
}

void ModelManager::DetectImg()
{
    DetectInput detect_input;
    DetectOutput detect_output;
    cv::Mat frame;
    size_t mm = 0;
    while(1)
    {
        if (queueMat.empty())
        {
            if(!DetectFlag)
            {
                break;
            }
            usleep(2000);
            continue;
        }
        // get image from queueMat
        mtxQueueImg.lock();
        frame = queueMat.front();
        queueMat.pop();
        mtxQueueImg.unlock();
        // run model
        objdetect.Run(detect_input, detect_output);
        // push detect result into queueDetOut
        mtxQueueDet.lock();
        queueDetOut.push(detect_output);
        // cout << "detect run !!" << endl;
        mtxQueueDet.unlock();
    }
    return;
}
void ModelManager::ClassifyImg()
{
    ObjClassifyInput input;
    ObjClassifyOutput output;
    cv::Mat frame;
    Detoutput detect_result;
    while(1)
    {
        if (queueDetOut.empty())
        {
            if(!ClassifyFlag)
            {
                break;
            }
            usleep(2000);
            continue;
        }
        // get detect from queueDetOut
        mtxQueueDet.lock();
        detect_result = queueDetOut.front();
        queueDetOut.pop();
        mtxQueueDet.unlock();
        // run model
        objclassify.Run(input, output);
        // push cls result into queueClsOut
        mtxQueueCls.lock();
        queueClsOut.push(output);
        mtxQueueCls.unlock();
    }
    return;
}

void ModelManager::getClsResult(ObjClassifyOutput& output){
    if (queueClsOut.empty()){
        output.object_list.object_num = -1;  // -1 is now empty;
        return; // must return in thread otherwise cant use &output
    }
    output = queueClsOut.front();
    queueClsOut.pop();
    return;
}

char* ModelManager::createJson()  // dont know why cant use &value, need return value
{
    mtxQueueCls.lock();
    ObjClassifyOutput output;
    getClsResult(output);
    mtxQueueCls.unlock();

    if (output.object_list.object_num == -1){
        return nullptr;
    }
    // prepare send data json
    cJSON* json_object_list = NULL;
    cJSON* json_ObjClassifyOutput = NULL;

    json_ObjClassifyOutput = cJSON_CreateObject();
    json_object_list = cJSON_CreateObject();
    cJSON_AddItemToObject(json_ObjClassifyOutput, "object_list", json_object_list);

    int obj_num = output.object_list.object_num;
    cJSON_AddNumberToObject(json_object_list, "object_num", obj_num);
    for (int i = 0; i < obj_num; ++i){
        cJSON* json_object = cJSON_CreateObject();
        cJSON* json_box = cJSON_CreateObject();
        cJSON_AddNumberToObject(json_box,"x", output.object_list.object[i].bbox.x);
        cJSON_AddNumberToObject(json_box,"y", output.object_list.object[i].bbox.y);
        cJSON_AddNumberToObject(json_box,"w", output.object_list.object[i].bbox.w);
        cJSON_AddNumberToObject(json_box,"h", output.object_list.object[i].bbox.h);
        cJSON_AddItemToObject(json_object,"bbox", json_box);
        cJSON_AddNumberToObject(json_object, "classes", output.object_list.object[i].classes);
        cJSON_AddNumberToObject(json_object, "objectness", output.object_list.object[i].objectness);
        // double prob = output.object_list.object[i].prob;
        // cJSON_AddNumberToObject(json_object, "prob", prob); // pointer cant use?
        string str = "object" + to_string(i);
        cJSON_AddItemToObject(json_object_list, str.c_str(), json_object);
        // printf("prob: %f", output.object_list.object[i].prob);
    }

    char* json_output = cJSON_Print(json_ObjClassifyOutput);
    cJSON_Delete(json_ObjClassifyOutput);
    return json_output;
}



#include <../include/ServerManager.h>
#include <../include/ModelManager.h>
#include <thread>
#define PORT 8080
void recvServer(ServerManager& s, ModelManager& model){
    int idx = 0;

    while (true){
        // auto start = std::chrono::steady_clock::now();
        cv::Mat frame;
        s.receive_data_frame(frame, model);
        // cal time cost
        // auto end = std::chrono::steady_clock::now();
        // std::chrono::duration<double, std::milli> elapsed = end - start;
        // std::cout << "recv execution time: " << elapsed.count() << " ms\n";
        if (frame.empty()) {
            usleep(2000);
            continue;
        }
        // cv::imwrite("image"+to_string(idx++)+".jpg", frame);
        std::cout << "Image " << idx++ <<" received !!" << std::endl;
    }
}

void sendServer(ServerManager& s, ModelManager& model){
    while (true){
        if (s.send_data_frame(model)) {
            cout << "send success!!" << endl;
            cout << endl;
        }else{
            // cout << "send fail!!" << endl;
            usleep(2000);
        }
    }
}

int main()
{
    ServerManager s;
    ModelManager model;
    model.initDetectModel();
    model.initClassifyModel();
    cout << endl;
    s.initialization(PORT);
    s.build_connect();
    s.acceptClient();

    thread recv_server(recvServer, std::ref(s), std::ref(model));
    thread send_server(sendServer, std::ref(s), std::ref(model));
    thread detect(&ModelManager::DetectImg, &model);
    thread classfy(&ModelManager::ClassifyImg, &model);
    detect.join();
    classfy.join();
    recv_server.join();
    send_server.join();
    return 0;
}


import json
import socket
import struct
import time
from multiprocessing import JoinableQueue
from threading import Thread

import os
from natsort import ns, natsorted

host = '192.168.0.2'  # '192.168.0.2' 'localhost'
port = 8080

def img_encode(img_path):
    img = cv2.imread(img_path)
    # img = cv2.resize(img, (500, 500), interpolation=cv2.INTER_CUBIC)
    img_param = [95]  # 图片压缩率0-100
    _, img = cv2.imencode('.jpg', img, img_param)
    img = img.tobytes()
    return img

def img_product(img_queue, path, path_mode='image'):
    if path_mode == 'image':
        image = img_encode(path)
        img_queue.put(image)
    elif path_mode == 'dir':
        dir_list = os.listdir(path)
        files = natsorted(dir_list, alg=ns.PATH)  # 顺序读取文件名
        for filename in files:
            img_path = path + '/' + filename
            image = img_encode(img_path)
            img_queue.put(image)
        img_queue.put('E')
    img_queue.join()

def server_consumer(img_queue):
    while True:
        start = int(round(time.time() * 1000))
        # 1. get img from queue
        img_obj = img_queue.get()
        img_queue.task_done()
        # get end signal
        if img_obj[0] == 'E':
        client.close()
        break
        # 2. send package(img_bytes_size, img_bytes)
        pack_size = struct.pack("i", len(img_obj))
        client.send(pack_size + img_obj)
        end = int(round(time.time() * 1000))

        data = client.recv(65536)
        json_str = data.decode('utf8', 'ignore').strip(b'\x00'.decode())
        results = json.loads(json_str)
        end = int(round(time.time() * 1000))
        end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        print('send and recv cost time: ', (end - start))
        print(results)

        if __name__ == '__main__':
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.connect((host, port))
        img_dir = 'data'
        one_img = './data/image.jpg'
        mode = 'dir'

        img_jq = JoinableQueue()
        producer = Thread(target=img_product, args=(img_jq, img_dir, mode,))
        consumer = Thread(target=server_consumer, args=(img_jq,))
        producer.daemon = True  # set daemon but not set join()

        producer.start()
        consumer.start()

    # producer.join() // 让生产者先关闭,防止close错误
        consumer.join()

• 其实这个项目真正做完感觉还是挺简单, 就是对socket通信不太熟悉, 以及传图片没做过.

• 实际上传图片只需要读取图片后,imencode,然后tobytes,最后发送size和data即可.而接受端只需要拼接数组,然后imdecode即可.

• 另外传输结果的话利用json传输可以让结果可读性可高, 传输也比较方便, 当时copy别人的发送代码, 没有细看,导致使用memcpy让json格式乱码,导致无法解码json.

• 如果你感觉接收端没问题,一定要看看发送端.

• 之后的新任务就是视频传输利用rtsp流,敬请期待

• use memcpy in receive_frame function if you want,对应github地址

• a pure client and server code to create a simple demo

• 如果用ctrl+z中断,可能导致address in use,使用bg/fg

  • 用户可以使用fg/bg操作继续前台或后台的任务，fg命令重新启动前台被中断的任务，bg命令把被中断的任务放在后台执行

• zmq send img

• 优化imdecode速度,本代码未使用

• 视频传输跟图片传输差不多

• Linux c++获取本地毫秒级精确时间

• 如果不想用json,可以试试struct

• 利用json来传输分类结果