tfliteをFlaskで動かしてクライアントからのリクエストで起動させる

tfliteは軽量とはいえ常時稼働させるのは、ラズパイ5のリソース的にも熱的には厳しいだろうから、ワンショットで動かすようにする

手段としてはFlaskでサーバー建てて、クライアントから認識リクエストが来たらカメラから画像取り込み、画像認識してその結果を返すという流れになります

<現状のサーバーで動くコード>

今は全てのリクエスト受けるアドレスになっている(MacBookからリクエスト出すから)けど、最終的にはローカルアドレスからだけ受けるようにします

# work on Flask server
# server will be activated only when client request is occured
#
#
from flask import Flask, jsonify
import cv2
import numpy as np
import tflite_runtime.interpreter as tflite
from picamera2 import Picamera2
from waitress import serve

app = Flask(__name__)

# モデル・ラベル初期化
interpreter = tflite.Interpreter(model_path="efficientdet_lite0.tflite")
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()
with open("coco_labels.txt", "r") as f:
    labels = [line.strip() for line in f.readlines()]

# カメラ初期化
picam2 = Picamera2()
picam2.preview_configuration.main.size = (640, 480)
picam2.preview_configuration.main.format = "RGB888"
picam2.preview_configuration.align()
picam2.configure("preview")
picam2.start()

def preprocess_image(image):
    resized = cv2.resize(image, (320, 320))
    resized = resized[:, :, [2, 1, 0]]  # BGR→RGB
    return np.expand_dims(resized, axis=0).astype(np.uint8)

def postprocess_results(boxes, scores, classes, count, image_shape, labels):
    detections = []
    for i in range(count):
        if scores[i] > 0.4:
            ymin, xmin, ymax, xmax = boxes[i]
            left, right, top, bottom = (
                int(xmin * image_shape[1]),
                int(xmax * image_shape[1]),
                int(ymin * image_shape[0]),
                int(ymax * image_shape[0])
            )
            detections.append({
                'box': [left, top, right, bottom],
                'class_id': int(classes[i]),
                'score': float(scores[i]),
                'label': labels[int(classes[i])] if int(classes[i]) < len(labels) else f"id:{int(classes[i])}"
            })
    return detections

def detect_once():
    frame = picam2.capture_array()
    input_data = preprocess_image(frame)
    interpreter.set_tensor(input_details[0]['index'], input_data)
    interpreter.invoke()
    boxes = interpreter.get_tensor(output_details[0]['index'])[0]
    classes = interpreter.get_tensor(output_details[1]['index'])[0]
    scores = interpreter.get_tensor(output_details[2]['index'])[0]
    count = int(interpreter.get_tensor(output_details[3]['index'])[0])
    detections = postprocess_results(boxes, scores, classes, count, frame.shape, labels)
    return detections

@app.route('/detect', methods=['GET'])
def detect_route():
    detections = detect_once()
    return jsonify(detections)

if __name__ == '__main__':
	# when you limit access only from the local machine, use a loopback address instead of 0.0.0.0
    app.run(host='0.0.0.0', port=5000)

立ち上げのメッセージ

$ python tflite_flask.py
INFO: Created TensorFlow Lite XNNPACK delegate for CPU.
[0:43:36.818382413] [87808]  INFO Camera camera_manager.cpp:327 libcamera v0.4.0+53-29156679
[0:43:36.825774869] [87830]  INFO RPI pisp.cpp:720 libpisp version v1.1.0 e7974a156008 27-01-2025 (21:50:51)
[0:43:36.914007444] [87830]  INFO RPI pisp.cpp:1179 Registered camera /base/axi/pcie@1000120000/rp1/i2c@88000/imx708@1a to CFE device /dev/media0 and ISP device /dev/media1 using PiSP variant BCM2712_C0
[0:43:36.916831533] [87808]  INFO Camera camera.cpp:1202 configuring streams: (0) 640x480-RGB888 (1) 1536x864-BGGR_PISP_COMP1
[0:43:36.916945996] [87830]  INFO RPI pisp.cpp:1484 Sensor: /base/axi/pcie@1000120000/rp1/i2c@88000/imx708@1a - Selected sensor format: 1536x864-SBGGR10_1X10 - Selected CFE format: 1536x864-PC1B
 * Serving Flask app 'tflite_flask'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:5000
 * Running on http://192.168.1.19:5000
Press CTRL+C to quit
192.168.1.18 - - [26/Apr/2025 19:16:35] "GET /detect HTTP/1.1" 200 -

本番運用はWSGI server使ってと言われてますが、ずっとそのままの可能性もあり、最後の行はリクエストに対する応答ですね

<クライアントからのリクエストコード>

import requests

if __name__ == '__main__':
    result = requests.get('http://rasp5.local:5000/detect').json()
    print(result)

クライアントはMacBook Proからリクエストすると、

<人がカメラから見える位置>
% python tfreq.py
[{'box': [3, -1, 636, 481], 'class_id': 0, 'label': 'person', 'score': 0.9375}]


<体をカメラからずらすと>
% python tfreq.py
[{'box': [143, 23, 607, 472], 'class_id': 81, 'label': 'id:81', 'score': 0.4765625}]

こんな感じで帰ってきます、レスポンス時間は体感1秒だから、まあこんなものかというレベルでアプリから考えたらこれで十分だし、ほとんどtfliteに負荷かけないから熱問題も起こらないでしょう

topでCPU負荷見てると、認識処理が走ってる時に瞬間的に%CPUが10%に近づくけどその程度で収まってるということ

 

admin

コメントを残す