ラズパイAIカメラ(IMX500)で玄関の人物検知サーバーを自作

sudo apt update
sudo apt install imx500-all

ls /usr/share/imx500-models/imx500_network_ssd_mobilenetv2_fpnlite_320x320_pp.rpk

pip install flask opencv-python numpy requests

mkdir -p /home/<username>/projects/smarthome/ai_camera/images
mkdir -p /home/<username>/projects/smarthome/ai_camera/static

smarthome/
└── ai_camera/
    ├── human_detection_record_and_server.py  # メインプログラム
    ├── images/                               # 検知画像の保存先
    └── static/
        └── latest_detected.jpg              # 最新検知画像（常に上書き）

# -*- coding: utf-8 -*-
import sys
import os
os.chdir(os.path.dirname(os.path.abspath(__file__)))

import cv2
import time
import datetime
import numpy as np
import threading
import requests
import base64
import shutil
from collections import deque
from flask import Flask, Response, jsonify, stream_with_context
from picamera2 import Picamera2
from picamera2.devices.imx500 import IMX500
from libcamera import Transform

app = Flask(__name__)

# --- グローバル変数 ---
frame = None
lock = threading.Lock()
detected_status = 'NotDetected'
is_detected_human = False
stream_bytes_queue = deque()
active_clients = 0
client_lock = threading.Lock()
latest_bandwidth = 0.0

# --- モデルと API 設定 ---
model_file = "/usr/share/imx500-models/imx500_network_ssd_mobilenetv2_fpnlite_320x320_pp.rpk"
record_api_url = 'http://192.168.xxx.xxx:5001/api/record_porch_ai_camera_data'

# --- 画像保存先 ---
images_dir = '/home/<username>/projects/smarthome/ai_camera/images'
latest_detected_path = '/home/<username>/projects/smarthome/ai_camera/static/latest_detected.jpg'

# --- IMX500 初期化 ---
imx500 = IMX500(model_file)

# --- Picamera2 初期化 ---
picam2 = Picamera2()
video_config = picam2.create_video_configuration(
    main={"format": "RGB888", "size": (int(4056/4), int(3040/4))},
    transform=Transform(hflip=True, vflip=True)
)
picam2.configure(video_config)
picam2.start()

# 起動時に前日以前の古い画像を削除
yesterday = datetime.datetime.now().date() - datetime.timedelta(days=1)
threshold_time = time.mktime(
    datetime.datetime.combine(yesterday, datetime.datetime.max.time()).timetuple()
)
for filename in os.listdir(images_dir):
    filepath = os.path.join(images_dir, filename)
    if os.path.isfile(filepath) and os.path.getmtime(filepath) < threshold_time:
        os.remove(filepath)

# --- フレーム取得スレッド（約30FPS）---
def update_frame():
    global frame
    while True:
        new_frame = picam2.capture_array()
        with lock:
            frame = new_frame
        time.sleep(0.03)

# --- AI 推論 & API 送信スレッド（0.5秒ごと）---
def detect_and_post():
    global detected_status, is_detected_human
    last_detect_time = 0

    while True:
        with lock:
            local_frame = frame.copy() if frame is not None else None

        if local_frame is None:
            time.sleep(0.1)
            continue

        now = time.time()
        if now - last_detect_time >= 0.5:
            last_detect_time = now
            metadata = picam2.capture_metadata()
            outputs = imx500.get_outputs(metadata)

            detected_status = 'NotDetected'
            is_detected_human = False
            image_filename = None
            jpg_base64 = None

            if outputs:
                boxes, scores, classes = outputs[0][0], outputs[1][0], outputs[2][0]
                for score, cls in zip(np.atleast_1d(scores), np.atleast_1d(classes)):
                    # クラス0 = 人物、スコアが0.57以上で検知とみなす
                    if cls == 0 and score > 0.57:
                        detected_status = 'Detected'
                        is_detected_human = True

                        # 検知画像を保存
                        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
                        image_filename = f"detected_{timestamp}_[{score}].jpg"
                        save_path = f"{images_dir}/{image_filename}"
                        with lock:
                            cv2.imwrite(save_path, local_frame)
                        # 最新検知画像として static に上書きコピー
                        shutil.copy(save_path, latest_detected_path)

                        # Base64 変換（API 送信用）
                        _, buffer = cv2.imencode('.jpg', local_frame)
                        jpg_base64 = base64.b64encode(buffer).decode('utf-8')
                        break
            else:
                detected_status = 'FailedToGetResults'

            # 検知結果を外部 API に POST
            try:
                response = requests.post(record_api_url, json={
                    'detected_status': detected_status,
                    'is_detected_human': is_detected_human,
                    'image_filename': image_filename,
                    'image_base64': jpg_base64
                })
            except requests.exceptions.RequestException as e:
                print(f'リクエストエラー: {e}')
        else:
            time.sleep(0.05)

# --- ライブストリーム生成 ---
def gen_frames():
    global active_clients
    with client_lock:
        active_clients += 1
    try:
        while True:
            with lock:
                if frame is None:
                    time.sleep(0.01)
                    continue
                _, buffer = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), 50])
                jpg_frame = buffer.tobytes()

            stream_bytes_queue.append(len(jpg_frame))
            try:
                yield (b'--frame\r\n'
                       b'Content-Type: image/jpeg\r\n\r\n' + jpg_frame + b'\r\n')
            except (BrokenPipeError, ConnectionResetError):
                break
            time.sleep(0.03)
    finally:
        with client_lock:
            active_clients = max(0, active_clients - 1)

# --- 通信量モニタリングスレッド（1秒ごと）---
def monitor_bandwidth():
    global latest_bandwidth
    while True:
        time.sleep(1.0)
        total_bytes = 0
        while stream_bytes_queue:
            total_bytes += stream_bytes_queue.popleft()
        latest_bandwidth = (total_bytes * 8) / 1_000_000

# --- Flask エンドポイント ---
@app.route('/video_feed')
def video_feed():
    return Response(stream_with_context(gen_frames()),
                    mimetype='multipart/x-mixed-replace; boundary=frame')

@app.route('/stats')
def stats():
    return jsonify({'clients': active_clients, 'bandwidth_mbps': round(latest_bandwidth, 2)})

@app.route('/latest_detected_info')
def latest_detected_info():
    if not os.path.exists(latest_detected_path):
        return jsonify({'timestamp': None})
    mtime = os.path.getmtime(latest_detected_path)
    return jsonify({'timestamp': datetime.datetime.fromtimestamp(mtime).isoformat()})

@app.route('/')
def index():
    return '''<html>
    <head>
      <style>
        html,body{margin:0;padding:0;height:100%;background:#000;overflow:hidden;display:flex;flex-direction:row;}
        #left{position:relative;height:100%;display:flex;justify-content:center;align-items:center;overflow:hidden;}
        #stream{width:100%;height:100%;object-fit:cover;display:block;}
        #stats{position:absolute;top:10px;right:10px;color:lime;background:rgba(0,0,0,0.6);padding:6px 12px;border-radius:6px;font-family:monospace;font-size:14px;}
        #right{height:100%;background:#222;display:flex;justify-content:flex-end;align-items:center;}
        #detected-container{position:relative;height:100%;display:flex;align-items:center;justify-content:center;}
        #latest-detected{height:100%;width:auto;object-fit:contain;display:block;}
        #timestamp{position:absolute;top:10px;right:10px;background:rgba(0,0,0,0.5);color:white;padding:4px 8px;font-size:14px;border-radius:4px;font-family:monospace;}
      </style>
      <script>
        function formatTimeAgo(t){
          const d=Math.floor((new Date()-new Date(t))/1000);
          if(d<60)return d+' 秒前';
          if(d<3600)return Math.floor(d/60)+' 分前';
          if(d<86400)return Math.floor(d/3600)+' 時間前';
          return Math.floor(d/86400)+' 日前';
        }
        setInterval(()=>{
          fetch('/latest_detected_info').then(r=>r.json()).then(d=>{
            document.getElementById('timestamp').innerText = d.timestamp ? formatTimeAgo(d.timestamp) : '検出時刻不明';
          });
          fetch('/stats').then(r=>r.json()).then(d=>{
            document.getElementById('stats').innerText = d.bandwidth_mbps.toFixed(2)+' Mbps';
          });
        },1000);
      </script>
    </head>
    <body>
      <div id="left">
        <img id="stream" src="/video_feed"/>
        <div id="stats">取得中...</div>
      </div>
      <div id="right">
        <div id="detected-container">
          <img id="latest-detected" src="/static/latest_detected.jpg"/>
          <div id="timestamp">検出時刻取得中...</div>
        </div>
      </div>
    </body>
    </html>'''

# --- メイン ---
if __name__ == "__main__":
    threading.Thread(target=update_frame,      daemon=True).start()
    threading.Thread(target=detect_and_post,   daemon=True).start()
    threading.Thread(target=monitor_bandwidth, daemon=True).start()
    app.run(host='0.0.0.0', port=5000, threaded=True)

[Unit]
Description=Human Detection Record and Server

[Service]
ExecStart=sudo /usr/bin/python3 /home/<username>/projects/smarthome/ai_camera/human_detection_record_and_server.py
Restart=always
RestartSec=60
Type=simple
User=<username>

[Install]
WantedBy=multi-user.target

# サービスを有効化
sudo systemctl enable human-detection-record-and-server.service
# サービスを起動
sudo systemctl start human-detection-record-and-server.service
# 状態確認
sudo systemctl status human-detection-record-and-server.service

metadata = picam2.capture_metadata()
outputs = imx500.get_outputs(metadata)

if outputs:
    boxes, scores, classes = outputs[0][0], outputs[1][0], outputs[2][0]
    for score, cls in zip(scores, classes):
        if cls == 0 and score > 0.57:   # クラス0 = 人物
            print("人を検出しました。スコア:", score)

yield (b'--frame\r\n'
       b'Content-Type: image/jpeg\r\n\r\n' + jpg_frame + b'\r\n')

video_config = picam2.create_video_configuration(
    main={"format": "RGB888", "size": (int(4056/4), int(3040/4))},  # 1014 x 760
    transform=Transform(hflip=True, vflip=True)  # カメラの取り付け向きに合わせて反転
)

http://<ラズパイのIPアドレス>:5000/

sudo journalctl -u human-detection-record-and-server.service -f