RtspStreamPredict/communication.py

import time

import lib.ZX.async_communication as cm
import lib.ZX.async_communication_etc_pb2 as rce
import lib.ZX.tool as zx_tool
import os, logging
import detect_people_pb2 as detect_people_info
import google.protobuf.text_format as tf

default_rabbitmq_etc_file = os.path.dirname(os.path.abspath(__file__)) + "/etc/rabbitmq.json"


class rabbitmq_client:
    def rabbitmq_client_exex(self, message):
        info = detect_people_info.DetectPeopleResults()
        tf.Parse(message.statu, info)
        # print("-------------------\n", info)

    def __init__(self, file=default_rabbitmq_etc_file):
        # 加载配置
        rabbitmq_etc = zx_tool.getProtobufJsonConfig(file, rce.RabbitmqEtc())
        print(rabbitmq_etc)
        self.g_rabbitmq = cm.RabbitAsyncCommunicator(rabbitmq_etc.ip, rabbitmq_etc.port,
                                                rabbitmq_etc.user, rabbitmq_etc.password)
        self.statu_ex_keys = []
        for bind in rabbitmq_etc.binds:
            key = [bind.ex, bind.route_key]
            self.statu_ex_keys.append(key)

        self.g_rabbitmq.Init(None, self.statu_ex_keys)
        self.g_rabbitmq.start()

        self.g_rabbitmq.bind_statu_callback(self.statu_ex_keys[0][0], self.statu_ex_keys[0][1], self.rabbitmq_client_exex)

    def sendPredict(self, message):
        return self.g_rabbitmq.publish(self.statu_ex_keys[0][0], self.statu_ex_keys[0][1], message, timeout=1)


if __name__ == '__main__':
    client = rabbitmq_client()
    while True:
        client.sendPredict("test_msg")
        time.sleep(1)