Python样例

def SendProtobuf(streamName: bytes, inPluginId: int, protobufVec: list) -> int:

# -*- coding:utf-8 -*-
import json
# import StreamManagerApi.py
from StreamManagerApi import *
import MxpiDataType_pb2 as MxpiDataType
if __name__ == '__main__':
    # init stream manager
    streamManagerApi = StreamManagerApi()
    ret = streamManagerApi.InitManager()
    if ret != 0:
        print("Failed to init Stream manager, ret=%s" % str(ret))
        exit()

    # create streams by pipeline config file
    pipeline = {
        "classification+detection": {
            "appsrc0": {
                "props": {
                    "blocksize": "409600"
                },
                "factory": "appsrc",
                "next": "appsink0"
             },
            "appsink0": {
                "factory": "appsink"
            }
        }
    }
    pipelineStr = json.dumps(pipeline).encode()
    ret = streamManagerApi.CreateMultipleStreams(pipelineStr)
    if ret != 0:
        print("Failed to create Stream, ret=%s" % str(ret))
        exit()

    # Construct the input of the stream
    dataInput = MxDataInput()
    with open("test.jpg", 'rb') as f:
        dataInput.data = f.read()
    # Inputs data to a specified stream based on streamName.
    streamName = b'classification+detection'
    inPluginId = 0
    key = b'mxpi_modelinfer2'
    visionList =  MxpiDataType.MxpiVisionList()
    visionVec = visionList.visionVec.add()
    visionVec.visionData.deviceId = 0
    visionVec.visionData.memType = 0
    visionVec.visionData.dataStr = dataInput.data
    protobuf = MxProtobufIn()
    protobuf.key = key
    protobuf.type = b'MxTools.MxpiVisionList'
    protobuf.protobuf = visionList.SerializeToString()
    # print(protobuf.key)
    # print(protobuf.protobuf)

    protobufVec = InProtobufVector()
    protobufVec.push_back(protobuf)
    # print(streamName)
    errorCode = streamManagerApi.SendProtobuf(streamName, inPluginId, protobufVec)
    if errorCode < 0:
        print("Failed to send data to stream.")
        exit()
    keyVec = StringVector()
    keyVec.push_back(key)
    inferResult = streamManagerApi.GetProtobuf(streamName, inPluginId, keyVec)
    if inferResult.size() == 0:
        print("inferResult is null")
        exit()
    if inferResult[0].errorCode != 0:
        print("GetResultWithUniqueId error. errorCode=%d" % (
        inferResult[0].errorCode))
        exit()
    #print the infer result
    print("GetProtobuf errorCode=%d" % (inferResult[0].errorCode))
    print("key:"+str(inferResult[0].messageName))
    result =  MxpiDataType.MxpiVisionList()
    result.ParseFromString(inferResult[0].messageBuf)
    fo = open("result.jpg", "wb")
    fo.write(result.visionVec[0].visionData.dataStr)
    # print(result.visionVec[0].visionData.dataSize)
    # print(result.visionVec[0].visionData.dataStr)
    # destroy streams
    streamManagerApi.DestroyAllStreams()

def SendProtobuf(streamName: bytes, elementName: bytes, protobufVec: list) -> int:

ELEMENT_NAME = b'appsrc0'
KEY = b'appsrc0'
protobuf_vec = prepare_data()
error_code = stream_manager_api.SendProtobuf(STREAM_NAME, ELEMENT_NAME, protobuf_vec)
if error_code < 0:
    print("Failed to send data to stream.")
    exit()

key_vec_sample = StringVector()
key_vec_sample.push_back(key)
infer_result_sample = stream_manager_api.GetProtobuf(STREAM_NAME, inplugin_id, key_vec_sample)
if infer_result_sample.size() == 0:
    print("infer_result is null")
    exit()

result_sample = MxpiDataType.MxpiVisionList()
result_sample.ParseFromString(infer_result_sample[0].messageBuf)
print("result: {}".format(result_sample.visionVec[0].visionData.dataStr))