【ラズベリーパイ】監視カメラの作り方｜PythonでカメラモジュールV2を自在に操作

sudo raspi-config

sudo python -m pip install --upgrade pip

sudo pip3 install opencv-python==4.5.1.48

pip install numpy==1.23.1

sudo apt update

sudo apt install libatlas3-base

import cv2

camera = cv2.VideoCapture(0)

while True:
  ret, frame = camera.read()
  if not ret:
    break

  cv2.imshow("Frame", frame)
  key = cv2.waitKey(1)
  
  # Escキーを入力されたら画面を閉じる
  if key == 27:
    break

camera.release()
cv2.destroyAllWindows()

import cv2

cap = cv2.VideoCapture(0)
ret, frame = cap.read()
cv2.imwrite('test.jpg', frame)
cap.release()

import cv2
import datetime

dt_now = datetime.datetime.now()
file_name = dt_now.strftime('%Y年%m月%d日%H時%M分%S秒')

cap = cv2.VideoCapture(0)
ret, frame = cap.read()
cv2.imwrite(file_name + '.jpg', frame)
cap.release()

import cv2
import numpy as np
 
cap = cv2.VideoCapture(0)
 
fmt = cv2.VideoWriter_fourcc('m', 'p', '4', 'v')
fps = 20.0
size = (640, 360)
writer = cv2.VideoWriter('test.m4v', fmt, fps, size)
 
while True:
    _, frame = cap.read()
    frame = cv2.resize(frame, size)
    writer.write(frame)
     
    cv2.imshow('frame', frame)
    #escキーで終了
    if cv2.waitKey(1) == 27:
        break
 
writer.release()
cap.release()
cv2.destroyAllWindows()

import RPi.GPIO as GPIO
import cv2
import time
import datetime

GPIO_PIN = 18

GPIO.setmode(GPIO.BCM)
GPIO.setup(GPIO_PIN,GPIO.IN)

while True:
    if(GPIO.input(GPIO_PIN) == GPIO.HIGH):
       
        print("1")
       
        dt_now = datetime.datetime.now()
        file_name = dt_now.strftime('%Y年%m月%d日%H時%M分%S秒')
        print(file_name)
       
        cap = cv2.VideoCapture(0)
        ret, frame = cap.read()
        cv2.imwrite(file_name + '.jpg', frame)
        cap.release()
       
        time.sleep( 10 )
       
    else:
        print("0")
        time.sleep( 1 )
        

import RPi.GPIO as GPIO
import cv2
import time
import datetime
import numpy as np

GPIO_PIN = 18

GPIO.setmode(GPIO.BCM)
GPIO.setup(GPIO_PIN,GPIO.IN)

while True:
    if(GPIO.input(GPIO_PIN) == GPIO.HIGH):
        
        print("1")
        
        dt_now = datetime.datetime.now()
        file_name = dt_now.strftime('%Y年%m月%d日%H時%M分%S秒')
        print(file_name)
        
        #カメラ映像を保存する
        cap = cv2.VideoCapture(0)
        fmt = cv2.VideoWriter_fourcc('m', 'p', '4', 'v')
        fps = 20.0
        size = (640, 360)
        writer = cv2.VideoWriter((file_name + '.m4v'), fmt, fps, size)
        
        i = 0
        #60フレーム撮影する
        while i < 60:
            _, frame = cap.read()
            frame = cv2.resize(frame, size)
            writer.write(frame)
     
            i = i + 1
             
        writer.release()
        cap.release()
        
        time.sleep( 10 )
        
    else:
        print("0")
        time.sleep( 1 )
GPIO.cleanup()

import RPi.GPIO as GPIO
import cv2
import time
import datetime
import requests

#LINEメッセージ送信の関数
def send_message(Discovery_time):
    url = "https://notify-api.line.me/api/notify" 
    token = "トークンをここに入力"
    headers = {"Authorization" : "Bearer "+ token}
    files = {'imageFile': open("image.jpg", "rb")}
    message =  (Discovery_time,"侵入者あり")
    payload = {"message" :  message} 
    r = requests.post(url, headers = headers, params=payload, files=files)

#センサーを使う準備
GPIO_PIN = 18

GPIO.setmode(GPIO.BCM)
GPIO.setup(GPIO_PIN,GPIO.IN)

while True:
    if(GPIO.input(GPIO_PIN) == GPIO.HIGH):
        #センサー検出時の処理
        print("1")
        
        #検出時間の取得
        dt_now = datetime.datetime.now()
        Discovery_time = dt_now.strftime('%Y年%m月%d日%H時%M分%S秒')
        print(Discovery_time)
        
        #カメラ画像を保存する
        cap = cv2.VideoCapture(0)
        ret, frame = cap.read()
        cv2.imwrite("image.jpg", frame)
        cap.release()
        
        #LINEメッセージ送信
        send_message(Discovery_time)
        
        #10秒待機
        time.sleep( 10 )
        
    else:
        #センサー未検出時の処理
        print("0")
        time.sleep( 1 )
        
GPIO.cleanup()

import cv2 as cv

HAAR_FILE = \
"/usr/local/lib/python3.9/dist-packages/cv2/data/"\
"haarcascade_frontalface_default.xml"
cascade = cv.CascadeClassifier(HAAR_FILE)

cap = cv.VideoCapture(0)

while(True):
    ret, frame = cap.read()
    
    face = cascade.detectMultiScale(frame)

    for x, y, w, h in face:
        cv.rectangle(frame,(x,y),(x+w,y+h),(0,0,255),1)

    cv.imshow('Capture',frame)
    
    if cv.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv.destroyAllWindows()

import cv2 as cv
from gpiozero import LED
import time

led = LED(18)

HAAR_FILE = \
"/usr/local/lib/python3.9/dist-packages/cv2/data/"\
"haarcascade_frontalface_default.xml"
cascade = cv.CascadeClassifier(HAAR_FILE)

cap = cv.VideoCapture(0)

while(True):
    ret, frame = cap.read()
    
    face = cascade.detectMultiScale(frame)
    
    if len(face) > 0:
        led.on()

        for x, y, w, h in face:
            cv.rectangle(frame,(x,y),(x+w,y+h),(0,0,255),1)
            
    else:
        led.off()
        
    cv.imshow('Capture',frame)
    
    if cv.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv.destroyAllWindows()

import cv2 as cv
import requests
import datetime
import time

# LINEメッセージ送信の関数
def send_message(discovery_time):
    url = "https://notify-api.line.me/api/notify"
    token = "トークンをここに入力"
    headers = {"Authorization": "Bearer " + token}
    files = {'imageFile': open("image.jpg", "rb")}
    message = f"{discovery_time} 顔を検出しました。"
    payload = {"message": message}
    r = requests.post(url, headers=headers, params=payload, files=files)
    print("送信完了")

# 顔認識用のカスケード分類器のパス
HAAR_FILE = "/usr/local/lib/python3.9/dist-packages/cv2/data/haarcascade_frontalface_default.xml"
cascade = cv.CascadeClassifier(HAAR_FILE)

# メッセージ送信の間隔時間（秒）
wait_time = 30  # 30秒間隔

#連続して顔を検出する必要があるフレーム数
consecutive_frames_required = 2  # 2フレーム連続

cap = cv.VideoCapture(0)

# 最後にメッセージを送信した時間を記録
last_notification_time = time.time() - wait_time  # 初回実行時にすぐ送信されるように調整

# 連続して顔が検出されたフレーム数を記録
consecutive_frames = 0

while True:
    ret, frame = cap.read()

    # 顔を検出
    face = cascade.detectMultiScale(frame)

    # 顔を検出した場合
    if len(face) > 0:
        consecutive_frames += 1
        for x, y, w, h in face:
            cv.rectangle(frame, (x, y), (x + w, y + h), (0, 0, 255), 1)

        # 連続して顔が検出されたフレーム数が設定値を超えた場合
        if consecutive_frames >= consecutive_frames_required:
            current_time = time.time()
            # 30秒間隔でメッセージを送信する
            if current_time - last_notification_time > wait_time:
                cv.imwrite("image.jpg", frame)

                # 検出時間の取得
                dt_now = datetime.datetime.now()
                discovery_time = dt_now.strftime('%Y年%m月%d日%H時%M分%S秒')
                print(discovery_time)

                # LINEメッセージ送信
                send_message(discovery_time)

                last_notification_time = current_time
                consecutive_frames = 0  # カウントリセット

    else:
        consecutive_frames = 0  # 顔が検出されなかったらカウントリセット

    cv.imshow('Capture', frame)

    # 'q'キーで終了
    if cv.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv.destroyAllWindows()

sudo apt install libcap-dev

git clone https://github.com/googlesamples/mediapipe.git

cd mediapipe/examples/object_detection/raspberry_pi

sh setup.sh

#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import sys
import time
import cv2
import mediapipe as mp
from mediapipe.tasks import python
from mediapipe.tasks.python import vision
from utils import visualize

# FPSを計算し推論を制御するためのグローバル変数
COUNTER, FPS = 0, 0
START_TIME = time.time()
is_inference_in_flight = False  # 推論中かどうかを制御するフラグ
latest_detection_result = None  # 最新の検出結果を格納する変数


def run(model: str, max_results: int, score_threshold: float, 
        camera_id: int, width: int, height: int) -> None:
    """カメラから取得した映像に対して継続的に推論を実行します。

    引数:
        model: TFLite物体検出モデルの名前。
        max_results: 検出結果の最大数。
        score_threshold: 検出結果のスコア閾値。
        camera_id: OpenCVに渡すカメラID。
        width: カメラから取得するフレームの幅。
        height: カメラから取得するフレームの高さ。
    """
    global is_inference_in_flight, latest_detection_result

    # カメラからの映像入力を開始
    cap = cv2.VideoCapture(camera_id)
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)

    # 表示用パラメータ
    row_size = 50  # ピクセル
    left_margin = 24  # ピクセル
    text_color = (0, 0, 0)  # 黒
    font_size = 1
    font_thickness = 1
    fps_avg_frame_count = 10

    def save_result(result: vision.ObjectDetectorResult, unused_output_image: mp.Image, timestamp_ms: int):
        """推論が完了した際に呼び出されるコールバック関数。"""
        global is_inference_in_flight, latest_detection_result, COUNTER, START_TIME, FPS

        # 推論が完了したことをマーク
        is_inference_in_flight = False

        # FPSを更新
        if COUNTER % fps_avg_frame_count == 0:
            current_time = time.time()
            FPS = fps_avg_frame_count / (current_time - START_TIME)
            START_TIME = current_time

        latest_detection_result = result
        COUNTER += 1

    # 物体検出モデルを初期化
    base_options = python.BaseOptions(model_asset_path=model)
    options = vision.ObjectDetectorOptions(
        base_options=base_options,
        running_mode=vision.RunningMode.LIVE_STREAM,
        max_results=max_results,
        score_threshold=score_threshold,
        result_callback=save_result
    )
    detector = vision.ObjectDetector.create_from_options(options)

    # カメラから継続的に映像を取得し推論を実行
    while cap.isOpened():
        success, image = cap.read()
        if not success:
            sys.exit(
                'エラー: ウェブカメラから映像を読み取れません。カメラ設定を確認してください。'
            )

        # TFLiteモデルが要求する形式に合わせて映像をBGRからRGBに変換
        rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb_image)

        # 前回の推論が完了している場合のみ新しい推論を開始
        if not is_inference_in_flight:
            detector.detect_async(mp_image, time.time_ns() // 1_000_000)
            is_inference_in_flight = True

        # FPSを表示
        fps_text = 'FPS = {:.1f}'.format(FPS)
        text_location = (left_margin, row_size)
        current_frame = image
        cv2.putText(current_frame, fps_text, text_location, cv2.FONT_HERSHEY_DUPLEX,
                    font_size, text_color, font_thickness, cv2.LINE_AA)

        # 最新の検出結果がある場合は可視化
        if latest_detection_result is not None:
            current_frame = visualize(current_frame, latest_detection_result)

        cv2.imshow('object_detection', current_frame)

        # ESCキーが押されたらプログラムを終了
        if cv2.waitKey(1) == 27:
            break

    detector.close()
    cap.release()
    cv2.destroyAllWindows()


def main():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument(
        '--model',
        help='物体検出モデルのパス。',
        required=False,
        default='efficientdet.tflite')
    parser.add_argument(
        '--maxResults',
        help='検出結果の最大数。',
        required=False,
        default=5)
    parser.add_argument(
        '--scoreThreshold',
        help='検出結果のスコア閾値。',
        required=False,
        type=float,
        default=0.25)
    parser.add_argument(
        '--cameraId', help='カメラのID。', required=False, type=int, default=0)
    parser.add_argument(
        '--frameWidth',
        help='カメラから取得するフレームの幅。',
        required=False,
        type=int,
        default=800)
    parser.add_argument(
        '--frameHeight',
        help='カメラから取得するフレームの高さ。',
        required=False,
        type=int,
        default=600)
    args = parser.parse_args()

    run(args.model, int(args.maxResults),
        args.scoreThreshold, int(args.cameraId), args.frameWidth, args.frameHeight)


if __name__ == '__main__':
    main()

#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import sys
import time
import cv2
import mediapipe as mp
from mediapipe.tasks import python
from mediapipe.tasks.python import vision
from utils import visualize

# 任意の物体名を指定する変数（ここで変更可能）
target_object = "car"  # ここを好きな物体名に変更できる

# FPSを計算し推論を制御するためのグローバル変数
COUNTER, FPS = 0, 0
START_TIME = time.time()
is_inference_in_flight = False  # 推論中かどうかを制御するフラグ
latest_detection_result = None  # 最新の検出結果を格納する変数


def run(model: str, max_results: int, score_threshold: float, 
        camera_id: int, width: int, height: int) -> None:
    """カメラから取得した映像に対して継続的に推論を実行します。

    引数:
        model: TFLite物体検出モデルの名前。
        max_results: 検出結果の最大数。
        score_threshold: 検出結果のスコア閾値。
        camera_id: OpenCVに渡すカメラID。
        width: カメラから取得するフレームの幅。
        height: カメラから取得するフレームの高さ。
    """
    global is_inference_in_flight, latest_detection_result

    # カメラからの映像入力を開始
    cap = cv2.VideoCapture(camera_id)
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)

    # 表示用パラメータ
    row_size = 50  # ピクセル
    left_margin = 24  # ピクセル
    text_color = (0, 0, 0)  # 黒
    font_size = 1
    font_thickness = 1
    fps_avg_frame_count = 10

    def save_result(result: vision.ObjectDetectorResult, unused_output_image: mp.Image, timestamp_ms: int):
        """推論が完了した際に呼び出されるコールバック関数。"""
        global is_inference_in_flight, latest_detection_result, COUNTER, START_TIME, FPS

        # 推論が完了したことをマーク
        is_inference_in_flight = False

        # FPSを更新
        if COUNTER % fps_avg_frame_count == 0:
            current_time = time.time()
            FPS = fps_avg_frame_count / (current_time - START_TIME)
            START_TIME = current_time
            
        # target_objectの数をカウント
        target_count = sum(1 for detection in result.detections
                           if detection.categories[0].category_name == target_object)
        
        print(f"検出された{target_object}の数: {target_count}")           

        latest_detection_result = result
        COUNTER += 1

    # 物体検出モデルを初期化
    base_options = python.BaseOptions(model_asset_path=model)
    options = vision.ObjectDetectorOptions(
        base_options=base_options,
        running_mode=vision.RunningMode.LIVE_STREAM,
        max_results=max_results,
        score_threshold=score_threshold,
        result_callback=save_result
    )
    detector = vision.ObjectDetector.create_from_options(options)

    # カメラから継続的に映像を取得し推論を実行
    while cap.isOpened():
        success, image = cap.read()
        if not success:
            sys.exit(
                'エラー: ウェブカメラから映像を読み取れません。カメラ設定を確認してください。'
            )

        # TFLiteモデルが要求する形式に合わせて映像をBGRからRGBに変換
        rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb_image)

        # 前回の推論が完了している場合のみ新しい推論を開始
        if not is_inference_in_flight:
            detector.detect_async(mp_image, time.time_ns() // 1_000_000)
            is_inference_in_flight = True

        # FPSを表示
        fps_text = 'FPS = {:.1f}'.format(FPS)
        text_location = (left_margin, row_size)
        current_frame = image
        cv2.putText(current_frame, fps_text, text_location, cv2.FONT_HERSHEY_DUPLEX,
                    font_size, text_color, font_thickness, cv2.LINE_AA)

        # 最新の検出結果がある場合は可視化
        if latest_detection_result is not None:
            # 可視化処理を実行
            current_frame = visualize(current_frame, latest_detection_result)

            # target_objectのカウントを計算
            target_count = sum(1 for detection in latest_detection_result.detections
                               if detection.categories[0].category_name == target_object)

            # target_objectのカウントを画面に表示
            count_text = f'{target_object} = {target_count}'
            count_location = (left_margin, row_size * 2)
            cv2.putText(current_frame, count_text, count_location, cv2.FONT_HERSHEY_DUPLEX,
                        font_size, text_color, font_thickness, cv2.LINE_AA)            

        cv2.imshow('object_detection', current_frame)

        # ESCキーが押されたらプログラムを終了
        if cv2.waitKey(1) == 27:
            break

    detector.close()
    cap.release()
    cv2.destroyAllWindows()


def main():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument(
        '--model',
        help='物体検出モデルのパス。',
        required=False,
        default='efficientdet.tflite')
    parser.add_argument(
        '--maxResults',
        help='検出結果の最大数。',
        required=False,
        default=5)
    parser.add_argument(
        '--scoreThreshold',
        help='検出結果のスコア閾値。',
        required=False,
        type=float,
        default=0.25)
    parser.add_argument(
        '--cameraId', help='カメラのID。', required=False, type=int, default=0)
    parser.add_argument(
        '--frameWidth',
        help='カメラから取得するフレームの幅。',
        required=False,
        type=int,
        default=800)
    parser.add_argument(
        '--frameHeight',
        help='カメラから取得するフレームの高さ。',
        required=False,
        type=int,
        default=600)
    args = parser.parse_args()

    run(args.model, int(args.maxResults),
        args.scoreThreshold, int(args.cameraId), args.frameWidth, args.frameHeight)


if __name__ == '__main__':
    main()