技术｜在 RV1126 之上使用 rknn_toolkit_lite 和 eKuiper pipeline 进行图片识别推理
本文使用 Base64，OPENCV，Numpy 和 Image 组件进行数据读取转换。 图片格式 一般在边缘计算端进行推理，不会使用一张一张图片，而是采用视频流读帧的方式。 二进制文件格式 file_bytes = "binary file" // 初始化图片二进制 encoded = base64.decodebytes(file_bytes.encode("ascii")) // 转换成base64 img = Image.open(io.BytesIO(encoded)).resize((width, height)) // 打开图片 input_data = np.array(img) // 转换成 numpy image = cv2.cvtColor(input_data, cv2.COLOR_BGR2RGB) // OPENCV 读取图片 图片(二进制)-->base64-->Image-->numpy array-->opencv Base64 编码格式 如果是 Base64 格式输入，处理方法一样，但是，在输入流的地方，要注意 source 采用 json 编码。 // 代码通用 // 输入格式： {"image": "/9j/4AAQSk....X//2Q=="} 图片(json base64）-->Image-->numpy array-->opencv eKuiper source 读取源（输入输入） eKuiper 定义 source 唯一区别就是 format: binary/json，假设都是采用 mqtt broker 作为图片输入。 Binary 二进制格式 输入源定义： { "Name" : "binaryimage", "Options" : { "confKey" : "edgex_mqtt_broker", "datasource" : "image", "format" : "binary", "type" : "mqtt" }, "Statement" : null, "StreamFields" : null, "StreamType" : 0 } 输入图片示例代码, GOLANG，这里的 payload 直接是 ReadFile，二进制： package main import ( "fmt" "io/ioutil" "time" mqtt "github.com/eclipse/paho.mqtt.golang" ) func main() { const TOPIC = "binaryimage" images := []string{ "parrot.jpg", "owl.jpg", "out.jpg", // 其他你需要的图像 } opts := mqtt.NewClientOptions().AddBroker("tcp://192.168.123.184:1883") client := mqtt.NewClient(opts) if token := client.Connect(); token.Wait() && token.Error() != nil { panic(token.Error()) } for _, image := range images { fmt.Println("Publishing " + image) payload, err := ioutil.ReadFile(image) if err != nil { fmt.Println(err) continue } if token := client.Publish(TOPIC, 0, false, payload); token.Wait() && token.Error() != nil { fmt.Println(token.Error()) } else { fmt.Println("Published " + image) } time.Sleep(1 * time.Second) } client.Disconnect(0) } Json 格式 输入源定义： { "Name" : "jsonimage", "Options" : { "confKey" : "edgex_mqtt_broker", "datasource" : "image", "format" : "json", "type" : "mqtt" }, "Statement" : null, "StreamFields" : null, "StreamType" : 0 } 输入图片为 base64 编码后的结果，json 格式如下，发送到 mqtt broker 即可： image.json { "image" : "/9j/4AAQSkZJRgABAQEA8ADwAAD/7Ukit.....ucen55+KEpqqGC7pTdx3X//2Q==" } eKuiper rule 规则定义 labelImageRknn 函数是由 pyairknn 插件（portable）来实现的逻辑处理。 Json 输入格式处理 这里的 FROM jsonimage 来自上面定义的 source，其他根据自身需要修改即可。labelImageRknn(image) 为 MQTT broker 输出内容作为 rule 的输入，这个有些难以理解，多试几遍就明白了。 { "actions" : [ { "mqtt" : { "bufferLength" : 1024, "enableCache" : false, "format" : "json", "insecureSkipVerify" : false, "omitIfEmpty" : false, "protocolVersion" : "3.1.1", "qos" : 0, "resourceId" : "edgex_mqtt_broker", "runAsync" : false, "sendSingle" : true, "server" : "tcp://edgex-mqtt-broker:1883", "topic" : "label" } } ], "id" : "rule-mqtt-image", "options" : { "bufferLength" : 1024, "checkpointInterval" : 300000, "concurrency" : 1, "isEventTime" : false, "lateTolerance" : 1000, "qos" : 0, "restartStrategy" : { "attempts" : 0, "delay" : 1000, "jitter" : 0.1, "maxDelay" : 30000, "multiplier" : 2 }, "sendError" : true, "sendMetaToSink" : false }, "sql" : "SELECT labelImageRknn(image) as label FROM jsonimage" } 二进制输入格式处理 与 Json 格式处理差别只有 labelImageRknn(image) --> labelImageRknn(self)，pyairknn 逻辑并没有变化。 { "sql" : "SELECT labelImageRknn(self) as label FROM jsonimage" } pyairknn 核心处理模块 ekuiper 插件 portable 参照官方 pyai 插件，编写特定格式文件一堆，改几个字符就可以。 实现原理 主要逻辑分析如下几个方面： 初始化 rknn load model； 检查是否符合 RV1126 格式； model 载入时间，如果太长，就需要用整机来预编译model； init runtime，不出错即可。 尤其要注意，这些过程要在 5 秒内完成，因为 eKuiper 超时时间是 5 秒，目前版本还不能对开发者可见修改。这里写一个函数 load_rknn_mode() 来实现清晰的代码。 class LabelImageFunc(Function): def __init__(self): load_rknn_model() pass def validate(self, args: List[Any]): if len(args) != 1: return "invalid arg length" return "" def exec(self, args: List[Any], ctx: Context): logging.debug("executing label") return label(args[0]) def is_aggregate(self): return False 核心推理逻辑： label 函数 注意这里的 return top5 格式就是返回给 rule 里面的结果，不支持动态解析，要做好准备； 中间业务逻辑，就是前面说的载入图片到 numpy； 然后给 rknn 模块进行 inference； 最后对推理结果进行格式化处理； def label(file_bytes): logging.info('image object detecting...') encoded = base64.decodebytes(file_bytes.encode("ascii")) input_data = np.array(Image.open(io.BytesIO(encoded))) image = cv2.cvtColor(input_data, cv2.COLOR_BGR2RGB) image = cv2.resize(image, (224, 224)) t0 = time.time() outputs = rknn_lite.inference(inputs=[image]) print('rknn inference time: {:.3f}ms'.format((time.time() - t0) * 1000)) show_outputs(outputs) output = outputs[0][0] output_sorted = sorted(output, reverse=True) labels = load_labels(LABELS) top5 = [] for i in range(5): value = output_sorted[i] index = np.where(output == value) for j in range(len(index)): if (i + j) >= 5: break if value > 0: for k in range(len(index[j])): top5.append({"index": labels[index[j][k]], "value": float(value)}) else: top5.append({"index": "-1", "value": 0.0}) return top5 最终效果 eKuiper 日志打印 如果有错也会在这里出现，便于查看。可以打开 DEBUG 日志，更加详细。 time="2023-02-17 13:16:23" level=info msg="Start source mqttimage instance 0 successfully" file="node/source_node.go:138" rule=rule-mqtt-image time="2023-02-17 13:16:23" level=info msg="new subscription for topic image, reqId is rule-mqtt-image_mqttimage_0" file="mqtt/mqtt_wrapper.go:220" rule=rule-mqtt-image time="2023-02-17 13:16:23" level=info msg="Successfully subscribed to topic image." file="source/mqtt_source.go:111" rule=rule-mqtt-image I:root:image object detecting... rknn inference time: 7.212ms mobilenet_v1 -----TOP 5----- [ 9 83]: 0.2010498046875 [ 9 83]: 0.2010498046875 2023-02-17 13:16:26,316 - /kuiper/plugins/portable/pyairknn/label.py[line:68] - I: image object detecting... Mqtt 输出 测试阶段，使用 mqtt broker 可以实时看到结果，方便调试。 { "label" : [ { "index" : "84:草原鸡、草原松鸡、草原鸡", "value" : 0.2236328125 }, { "index" : "9:母鸡", "value" : 0.2010498046875 }, { "index" : "83:皱褶松鸡、鹧鸪、Bonasa umbellus", "value" : 0.2010498046875 }, { "index" : "9:母鸡", "value" : 0.2010498046875 }, { "index" : "83:皱褶松鸡、鹧鸪、Bonasa umbellus", "value" : 0.2010498046875 }, { "index" : "139:大鸨", "value" : 0.10638427734375 }, { "index" : "87:鹧鸪", "value" : 0.0626220703125 } ] } YiCLOUD IoT 平台输出 还可以通过脚本或插件方式，将结果输出到 IoT 平台，这里以 YiCLOUD 平台为例。