RTSPカメラを使った物体検知AIの構築ガイド

Itoi Yoshihiko

横浜国立大学理工学部建築都市環境系学科卒
一級鉄筋技能士

この記事は、RTSP（Real-Time Streaming Protocol）対応のカメラを使用してリアルタイム物体検知AIを構築する方法について詳しく説明しています。

RTSP対応カメラ

リアルタイム物体検知AIを構築するには、RTSP対応のネットワークカメラが必要です。

適切なRTSPカメラを選ぶポイントを考えましょう。

POCの段階では4000円程度の安価なモノでもOK

4000円程度のものでも、プロジェクトの概念実証（POC）や小規模な監視システムには十分です。

価格と性能のバランスを取りながら、自分のニーズに合ったカメラを選びましょう。

amzn.to

ご迷惑をおかけしています！

~~https://amzn.to/3Um7hgA~~

無線LAN接続

RTSPカメラを無線LANでパソコンに接続するためには、RTSP_URLを取得する必要があります。

RTSP_URLを取得すればパソコンと接続できる

以下はPythonコードの例です。

まず、RTSPカメラのURLを取得し、それを使ってネットワークカメラに接続します。

import cv2

# RTSP URLを指定します。これはカメラによって異なります。
# 例: rtsp://username:password@camera_ip_address:port/stream_path
rtsp_url = "ここにURLを入力"

if __name__ == '__main__':
    # RTSP URLを使ってネットワークカメラに接続
    capture = cv2.VideoCapture(rtsp_url)

    # 「q」が押されるまで繰り返し
    while True:
        # カメラから画像取得q
        ret, frame = capture.read()
        # 取得に成功した場合、表示する
        if ret:
            # 画像表示
            cv2.imshow('camera', frame)
            # キー操作を確認
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

    # 終了処理
    capture.release()
    cv2.destroyAllWindows()

import cv2

# RTSP URLを指定します。これはカメラによって異なります。

# 例: rtsp://username:password@camera_ip_address:port/stream_path

rtsp_url = "ここにURLを入力"

if __name__ == '__main__':

# RTSP URLを使ってネットワークカメラに接続

capture = cv2.VideoCapture(rtsp_url)

# 「q」が押されるまで繰り返し

while True:

# カメラから画像取得q

ret, frame = capture.read()

# 取得に成功した場合、表示する

if ret:

# 画像表示

cv2.imshow('camera', frame)

# キー操作を確認

if cv2.waitKey(1) & 0xFF == ord('q'):

break

# 終了処理

capture.release()

cv2.destroyAllWindows()

画角を調整する

カメラの画角を調整することで、自分の教師データを使ってモデルを学習させる際に、モデルの精度が良くなるように工夫します。

import cv2

# RTSP URLを指定します。これはカメラによって異なります。
# 例: rtsp://username:password@camera_ip_address:port/stream_path
rtsp_url = "ここにURLを入力"

if __name__ == '__main__':
    # RTSP URLを使ってネットワークカメラに接続
    capture = cv2.VideoCapture(rtsp_url)

    # 画質の設定
    # 例: 幅と高さを1280x720に設定
    capture.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
    capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

    # 「q」が押されるまで繰り返し
    while True:
        # カメラから画像取得q
        ret, frame = capture.read()
        # 取得に成功した場合、表示する
        if ret:
            # 画像のクロップ
            # ここでは画像の左上隅から200x200ピクセルの領域をクロップします
            # startX, startY, width, heightを変更して必要な領域を指定してください
            cropped_frame = frame[370:800, 600:1150]

            # クロップした画像表示
            cv2.imshow('camera', cropped_frame)
            # 画像表示
            # cv2.imshow('camera', frame)
            # キー操作を確認
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

    # 終了処理
    capture.release()
    cv2.destroyAllWindows()

import cv2

# RTSP URLを指定します。これはカメラによって異なります。

# 例: rtsp://username:password@camera_ip_address:port/stream_path

rtsp_url = "ここにURLを入力"

if __name__ == '__main__':

# RTSP URLを使ってネットワークカメラに接続

capture = cv2.VideoCapture(rtsp_url)

# 画質の設定

# 例: 幅と高さを1280x720に設定

capture.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)

capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

# 「q」が押されるまで繰り返し

while True:

# カメラから画像取得q

ret, frame = capture.read()

# 取得に成功した場合、表示する

if ret:

# 画像のクロップ

# ここでは画像の左上隅から200x200ピクセルの領域をクロップします

# startX, startY, width, heightを変更して必要な領域を指定してください

cropped_frame = frame[370:800, 600:1150]

# クロップした画像表示

cv2.imshow('camera', cropped_frame)

# 画像表示

# cv2.imshow('camera', frame)

# キー操作を確認

if cv2.waitKey(1) & 0xFF == ord('q'):

break

# 終了処理

capture.release()

cv2.destroyAllWindows()

保存間隔を設定して、アノテーション用の画像を自動保存

データを保存するために、画像の自動保存設定を行いましょう。

以下のPythonコードは、一定の間隔で画像を自動的に保存する例です。

保存された画像は、後でアノテーションとして使用できます。

import cv2
import time
import os

# RTSP URLを指定します。これはカメラによって異なります。
# 例: rtsp://username:password@camera_ip_address:port/stream_path
rtsp_url = "ここにURLを入力"

# 画像を保存するフォルダのパス（Windowsの例、ユーザー名は適宜変更してください）
save_path = 'C:\gazou'

# 保存間隔を秒単位で設定（5分 = 300秒）
save_interval = 150

# 最後に画像を保存した時刻を追跡
last_saved_time = time.time()

if __name__ == '__main__':
    # RTSP URLを使ってネットワークカメラに接続
    capture = cv2.VideoCapture(rtsp_url)

    # 画質の設定
    # 例: 幅と高さを1280x720に設定
    capture.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
    capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

    while True:
        # カメラから画像取得
        ret, frame = capture.read()

        if ret:
            # 画像のクロップ
            # 例: 画像の特定の領域をクロップ
            cropped_frame = frame[370:800, 600:1150]

            # クロップした画像表示
            cv2.imshow('camera', cropped_frame)

            # 現在時刻を取得
            current_time = time.time()

            # 最後に保存してから5分経過したか確認
            if current_time - last_saved_time >= save_interval:
                # 画像をPNG形式で保存
                filename = os.path.join(save_path, f"saved_image_{int(current_time)}.png")
                cv2.imwrite(filename, cropped_frame)
                print(f"Saved: {filename}")

                # 最後に保存した時刻を更新
                last_saved_time = current_time

            # キー操作を確認
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

    # 終了処理
    capture.release()
    cv2.destroyAllWindows()

import cv2

import time

import os

# RTSP URLを指定します。これはカメラによって異なります。

# 例: rtsp://username:password@camera_ip_address:port/stream_path

rtsp_url = "ここにURLを入力"

# 画像を保存するフォルダのパス（Windowsの例、ユーザー名は適宜変更してください）

save_path = 'C:\gazou'

# 保存間隔を秒単位で設定（5分 = 300秒）

save_interval = 150

# 最後に画像を保存した時刻を追跡

last_saved_time = time.time()

if __name__ == '__main__':

# RTSP URLを使ってネットワークカメラに接続

capture = cv2.VideoCapture(rtsp_url)

# 画質の設定

# 例: 幅と高さを1280x720に設定

capture.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)

capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

while True:

# カメラから画像取得

ret, frame = capture.read()

if ret:

# 画像のクロップ

# 例: 画像の特定の領域をクロップ

cropped_frame = frame[370:800, 600:1150]

# クロップした画像表示

cv2.imshow('camera', cropped_frame)

# 現在時刻を取得

current_time = time.time()

# 最後に保存してから5分経過したか確認

if current_time - last_saved_time >= save_interval:

# 画像をPNG形式で保存

filename = os.path.join(save_path, f"saved_image_{int(current_time)}.png")

cv2.imwrite(filename, cropped_frame)

print(f"Saved: {filename}")

# 最後に保存した時刻を更新

last_saved_time = current_time

# キー操作を確認

if cv2.waitKey(1) & 0xFF == ord('q'):

break

# 終了処理

capture.release()

cv2.destroyAllWindows()

物体検知モデルを適用する

最後に、物体検知モデルを導入し、リアルタイムで物体を検知しましょう。

YoloV5モデルなどの学習済みモデルだけではなく、独自の自社製品を検知したい場合は、自分でモデルを学習させる必要があります。

"""
launch_web_camera.pyとtest_yolov5.pyを組み合わせることで、
リアルタイムに検知できるシステムができます。

"""
import glob
import shutil

import torch
import cv2

# メールの送信
from launch_RTSPweb_camera import rtsp_url
from mail import send_mail

def detect_img(model, image):
    # 推論の実行
    model.conf = 0.6  # confidence threshold (0-1)
    results = model(image)

    # 結果を返す
    return results

def add_result(image, results, x, y):
    # 結果から検出された物体の名前を取得する
    results_name = [row["name"] for ind, row in results.pandas().xyxy[0].iterrows()]
    # リストにある重複をカウントする
    results_count = {i: results_name.count(i) for i in results_name}
    str = ""
    offset = 0
    for key in results_count.keys():
        str = f"{key}:{results_count[key]}"
        # 文字を追加する
        image = cv2.putText(image, str, (x, y+offset), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
        offset+=30
    return image

if __name__ == '__main__':
    weights_dir = "weights/best.pt"
    model = torch.hub.load('yolov5', 'custom', path=weights_dir, source='local')
    # model = torch.hub.load('ultralytics/yolov5', 'yolov5s', pretrained=True)

    capture = cv2.VideoCapture(rtsp_url)

    while (True):
        ret, frame = capture.read()
        if ret:

            # 位置検出
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            results = detect_img(model, frame)
            # 結果保存
            results.save(save_dir="result")
            # 結果読み込み
            img = cv2.cvtColor(cv2.imread("result/image0.jpg"), cv2.COLOR_BGR2RGB)
            shutil.rmtree("result")
            # 文字の追加
            img = add_result(img, results, 10, 30)
            # 結果表示
            results.print()
            cv2.imshow('camera', img)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

    capture.release()
    cv2.destroyAllWindows()

"""

launch_web_camera.pyとtest_yolov5.pyを組み合わせることで、

リアルタイムに検知できるシステムができます。

"""

import glob

import shutil

import torch

import cv2

# メールの送信

from launch_RTSPweb_camera import rtsp_url

from mail import send_mail

def detect_img(model, image):

# 推論の実行

model.conf = 0.6 # confidence threshold (0-1)

results = model(image)

# 結果を返す

return results

def add_result(image, results, x, y):

# 結果から検出された物体の名前を取得する

results_name = [row["name"] for ind, row in results.pandas().xyxy[0].iterrows()]

# リストにある重複をカウントする

results_count = {i: results_name.count(i) for i in results_name}

str = ""

offset = 0

for key in results_count.keys():

str = f"{key}:{results_count[key]}"

# 文字を追加する

image = cv2.putText(image, str, (x, y+offset), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)

offset+=30

return image

if __name__ == '__main__':

weights_dir = "weights/best.pt"

model = torch.hub.load('yolov5', 'custom', path=weights_dir, source='local')

# model = torch.hub.load('ultralytics/yolov5', 'yolov5s', pretrained=True)

capture = cv2.VideoCapture(rtsp_url)

while (True):

ret, frame = capture.read()

if ret:

# 位置検出

frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

results = detect_img(model, frame)

# 結果保存

results.save(save_dir="result")

# 結果読み込み

img = cv2.cvtColor(cv2.imread("result/image0.jpg"), cv2.COLOR_BGR2RGB)

shutil.rmtree("result")

# 文字の追加

img = add_result(img, results, 10, 30)

# 結果表示

results.print()

cv2.imshow('camera', img)

if cv2.waitKey(1) & 0xFF == ord('q'):

break

capture.release()

cv2.destroyAllWindows()

まとめ

この記事では、RTSPカメラを活用したリアルタイム物体検知AIの構築手順を紹介しました。

個々のニーズに合わせて、このシステムをカスタマイズしてみてください。