ラズベリーパイで顔認識（改-2）ではストリーミングにMjpg-Streamer＋Bullseye ＋CSIカメラモジュールの組み合わせでした。
OSにTrixieを使う場合はどうでしょう。
（改-2）でBullseyeを使っていたのはneuralassemblyのMjpg-StreamerだとTrixieでエラーが出るからでしたが、jacksonliamのMjpg-StreamerならTrixieでもエラーになりません。
このやり方の場合、カメラはCSIカメラモジュールではなくUSBカメラを使うことになります。
こんな感じです。
Trixie Lite（32-bit）ではIPアドレスを固定する方法も従来とは異なりNetwork Managerを使うようになっています。
やってみましょう。
ストリーミング側
普通にRaspberry Pi Imager の最新版でRaspberr Pi OS Lite (32-bit) Trixie をMicro SD に焼きます。
ここでユーザー名とパスワード、SSIDとパスワード、SSHを設定しておきます。
ZeroにMicroSDをセットして起動。
設定しておいたユーザー名とパスワードでログインします。
ip a コマンドでIPアドレスを確認してSSHで接続すれば、以降の作業が楽になります。
リポジトリをアップデートしておきます。
|
1 |
sudo apt update |
Lite版の各種設定画面を起動します。
|
1 |
sudo raspi-config |
Interface Optionsで以下を設定
SPI
I2C
Localisation Optionsで以下を設定
Locale
Ja_JP.UTF-8
最後にFinishします。
再起動して日本語環境にしておきましょう。
|
1 |
sudo reboot |
次にIPアドレスを固定します。
以前はdhcpcd.confを編集していましたが、新しいOSではNetwork Managerを使います。dhcpcd.confを編集しても反映されません。
Network Managerを起動
|
1 |
sudo nmtui |
OSを焼く時に最新のImagerではWireless LAN設定になっているので以下のような画面になっています。
「接続の編集」でリターキーを押します。
Wi-Fiアダプターの編集を選択してリターンキー。
静的IPアドレスを設定するのでタブキーでここをフォーカスします。
リターンキーを押して、自動を手作業に変更します。
次に「表示する」にフォーカスしてリターンキーを押します。
編集画面に入ったら、アドレスの「追加」をフォーカスしてリターンキーを押します。
設定画面に入ったら、アドレス・ゲートウェイ・DNSサーバーに入力して、「IPv4アドレス設定が必要になります」にチェックを入れてOKで抜けます。
「戻る」でリターキーをおします。
最初の画面に戻るので「終了」で設定を終わります。
再起動してアドレスが変わっているのを確認します。
|
1 2 3 |
sudo reboot ip a |
ついでにSAMBAも設定しておきましょう。
|
1 |
sudo apt install samba |
共有用のディレクトリを作っておきます。
|
1 |
sudo mkdir ~/Public |
編集
|
1 |
sudo nano /etc/samba/smb.conf |
最後尾に以下を追加
[Public]
path = /home/＜ユーザー名＞/Public
read only = no
guest ok = yes
public = yes
writable = Yes
browsable = yes
force user = ＜ユーザー名＞
directory mode = 0774
create mode = 0774
パスワード設定
|
1 |
sudo pdbedit -a ＜ユーザー名＞ |
サービスを起動
|
1 2 |
sudo service smbd restart sudo service nmbd restart |
確認
|
1 |
smbstatus |
以下が表示されれば完了
smbstatus only works as root!
Mjpg-Streamerをインストール
パッケージやライブラリをインストール
|
1 |
sudo apt install -y build-essential git cmake pkg-config libjpeg-dev imagemagick libv4l-dev liblog4cpp5-dev |
ユーザールートに戻って jacksonliamのMjpg-streamerをクローンしてビルド。
|
1 2 3 4 5 6 7 8 9 |
cd ~/ git clone https://github.com/jacksonliam/mjpg-streamer.git cd mjpg-streamer/mjpg-streamer-experimental make sudo make install |
Mjpg-Streamer 起動
|
1 2 3 |
cd ~/mjpg-streamer/mjpg-streamer-experimental ./mjpg_streamer -i "./input_uvc.so -d /dev/video0 -r 640x480 -f 30" -o "./output_http.so -w ./www" |
Errorが出ていますが問題ないようです。
ストリームされている画像を外部ブラウザーで見てみます。
http://＜IPアドレス＞:8080
自動起動用のスクリプト
|
1 2 3 |
cd ~/ sudo nano stream.sh |
シェルスクリプトコード
|
1 2 3 4 5 |
#!/bin/bash cd /home/＜ユーザー名＞/mjpg-streamer/mjpg-streamer-experimental ./mjpg_streamer -i "./input_uvc.so -d /dev/video0 -rot 0 -r 640x480 -f 10 -y -n" -o "./output_http.so -w ./www -p 8080" |
カメラの傾きはrot (回転) オプションで設定（0 ～ 360）。
/etc/rc.localに以下を追加して再起動
systemdで自動起動
ユーザーホームディレクトリに起動用シェルスクリプト作成
|
1 |
sudo nano /home/＜ユーザー名＞/STREAM.sh |
以下を記述
|
1 2 3 4 |
#!/bin/sh cd /home/＜ユーザー名＞/mjpg-streamer/mjpg-streamer-experimental mjpg_streamer -i "./input_uvc.so -d /dev/video0 -r 640x480 -f 30" -o "./output_http.so -w ./www" |
モード変更
|
1 |
sudo chmod 755 /home/＜ユーザー名＞/STREAM.sh |
サービス名を例えば myservice とします。
/etc/systemd/systemにmyservice.serviceというサービスファイルを作成。
|
1 |
sudo nano /etc/systemd/system/myservice.service |
以下を記述
|
1 2 3 4 5 6 7 8 9 10 |
[Unit] Description=My Service [Service] WorkingDirectory=/home/＜ユーザー名＞ Type=simple ExecStart=/home/＜ユーザー名＞/STREAM.sh [Install] WantedBy=multi-user.target |
サービス開始
|
1 2 |
sudo systemctl enable myservice sudo systemctl start myservice |
再起動
|
1 |
sudo reboot |
ステータスを確認する場合
|
1 |
sudo systemctl status myservice.service |
エラーが無ければOK。
顔認識側
ストリームされている画像をキャプチャして、顔認識（識別）をやってみます。
基本的な動作は（改-1）でやった単体用のコードを少し変更するだけです。仮想環境を作ってface_recognitionをインストールしておきます。
単体動作の場合のコードはfaceディレクトリにあるfacial_recognition.pyです。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 |
#! /usr/bin/python """ 2025/03/08 Bullseyeへ変換 終了はesc 2025/03/11 画像サイズをconfigで設定 2025/03/12 toleranceパラメータを設定 2025/03/14 ・multiprocessingによる高速化 倍近いスピードアップした マルチプロセスを使う関係上 if __name__ == "__main__":に プログラムを書く必要があり、改造がしにくい感がある ・検出する顔の大きさを設定できるようにする これにより小さい顔は検出しない ひとりで大きく映る場合はちょうど良い 2025/03/25 日本人の精度を上げるため田口モデルを適用 データ置き場を整理 処理軽減のパラメータをconfig化 2025/03/30 モデルを選択できるようにした 2025/03/31 コメント整理 2025/04/17 プログラム名を顔識別 face_recognitionとした 2025/07/25 カメラタイプ対応 *1 2025/09/08 取説とプログラム名が違っているので、facial_recognitionとした facial_recognition.py 01 """ import config # tolerance_parameterの取得 精度調整用 tolerance_para = config.tolerance_parameter() print("tolerance_parameter=",tolerance_para) print("ウォームアップ中ちょっと待って") print() import dlib import cv2 import time from imutils.video import FPS import pickle import multiprocessing from picamera2 import Picamera2 from scipy.spatial.distance import euclidean # Euclidean distanceを使用 # *1 from libcamera import controls # 設定の取得 # カメラタイプの取り込み *1 camera_type = config.camera_type() # カメラから取り込む画像の大きさの設定 camera_width_x, camera_width_y = config.camera_width() # 処理用の画像の大きさの設定 disp_width_x, disp_width_y = config.disp_width() # 現在使っていない # 検出する顔の最小大きさ face_size_para = config.face_size() # 必要な連続検出回数 necessary_count = config.necessary_count() # 現在使っていない # 処理性能軽減のための画像サイズ縮小 speed_size = config.speed_size() # Dlibの顔検出器 face_detector = dlib.get_frontal_face_detector() # model_choice = config.model_choice() # 汎用モデル face_rec_model_path = "./face_dat/dlib_face_recognition_resnet_model_v1.dat" # 田口モデル if model_choice == 1: face_rec_model_path = "./face_dat/taguchi_face_recognition_resnet_model_v1.dat" face_rec_model = dlib.face_recognition_model_v1(face_rec_model_path) # 顔検出器 特徴量を取得 shape_predictor = dlib.shape_predictor("./face_dat/shape_predictor_68_face_landmarks.dat") # 初期化 currentname = "unknown" # encodingsP = "./face_dat/encodings_taguchi.pickle" encodingsP = "./face_dat/encodings.pickle" print("[INFO] loading encodings + face detector...") # pickleの読み込み data = pickle.loads(open(encodingsP, "rb").read()) def process_faces(rgb_frame, gray_frame, tolerance, face_size): """顔検出と顔認識を並列処理""" # 顔検出 faces = face_detector(gray_frame, 1) encodings = [] names = [] for face in faces: # 顔の領域の幅と高さを取得 width = face.right() - face.left() height = face.bottom() - face.top() # 最小顔サイズより小さい顔を無視 if width < face_size or height < face_size: continue # 顔を検出 shape = shape_predictor(gray_frame, face) encoding = face_rec_model.compute_face_descriptor(rgb_frame, shape) encodings.append(encoding) # 顔識別 distances = [] for known_encoding in data["encodings"]: # Euclidean距離を計算 distance = euclidean(encoding, known_encoding) distances.append(distance) # 最も近い顔を選択 min_distance = min(distances) if min_distance < tolerance: name = data["names"][distances.index(min_distance)] else: name = "Unknown" names.append(name) return faces, names if __name__ == "__main__": # カメラ設定 picam2 = Picamera2() cam_config = picam2.create_preview_configuration(main={"size": (camera_width_x, camera_width_y), "format": "RGB888"}) picam2.configure(cam_config) if camera_type == 3: # オートフォーカスを有効にする *1 picam2.set_controls({"AfMode": controls.AfModeEnum.Continuous}) picam2.start() time.sleep(2.0) # Camera warm-up # FPS計測開始 fps = FPS().start() # 並列処理用プールを作成 num_processes = multiprocessing.cpu_count() pool = multiprocessing.Pool(processes=num_processes) # 処理負荷を下げるため縮小 # デフォルトでは、640*480の画像だが、処理時にリサイズして処理 # 小さい値にするとスピードアップになるが、やりすぎると精度が落ちる proces_width_x = int(640 * speed_size) proces_width_y = int(480 * speed_size) # face_sizeとの関係があるので調整 face_size_para = face_size_para * speed_size while True: frame = picam2.capture_array() # small_frame = cv2.resize(frame, (320, 240)) # 処理負荷を下げるため縮小 small_frame = cv2.resize(frame, (proces_width_x, proces_width_y)) # 処理負荷を下げるため縮小 gray = cv2.cvtColor(small_frame, cv2.COLOR_BGR2GRAY) rgb = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB) # 並列処理で顔を検出・認識 result_async = pool.apply_async(process_faces, (rgb, gray, tolerance_para, face_size_para)) faces, names = result_async.get() for (face, name) in zip(faces, names): (top, right, bottom, left) = (face.top(), face.right(), face.bottom(), face.left()) # 座標を元のサイズにスケールアップ # top, right, bottom, left = [int(val * (camera_width_x / 320)) for val in [top, right, bottom, left]] top, right, bottom, left = [int(val * (camera_width_x / proces_width_x)) for val in [top, right, bottom, left]] y = top - 15 if top - 15 > 15 else top + 15 if name != "Unknown":# 名前が判別できた時は 緑の枠 cv2.rectangle(frame, (left, top), (right, bottom), (0, 255, 0), 4) cv2.putText(frame, name, (left, y), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2) else: # Unknown の場合は 赤の枠 cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 225), 6) cv2.putText(frame, name, (left, y), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) if name != currentname: currentname = name print(currentname) """ """ """ """ """ """ """ """ """ """ """ """ """ ここで、名前を確定した場合の処理を入れることが可能 """ if name == currentname and name != "Unknown":# 2回 連続して同じ名前ならokとプリントする print("ok") """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ """ cv2.imshow("Facial Recognition is Running", frame) # q or ESCキーが押されたら終了 key = cv2.waitKey(1) & 0xFF if key == ord("q") or key == 27: break # 'p' キーが押されたら画像保存 if key == ord("p"): cv2.imwrite("face_" + name +".jpg", frame) fps.update() # FPS計測終了 fps.stop() print("[INFO] elapsed time: {:.2f}".format(fps.elapsed())) print("[INFO] approx. FPS: {:.2f}".format(fps.fps())) cv2.destroyAllWindows() pool.close() pool.join() |
ここではCSIカメラモジュールからの画像を得るのに以下のコードを使っています。
picam2 = Picamera2()
さらに、ここからフレームを切り出すのは以下のコードです。
frame = picam2.capture_array()
変更をかけるのはこの２か所ということになります。
こんな感じです。
オリジナルのコメントは外していますが、改変部分のみStreamingというコメントを入れています。
IPアドレスの部分は各自の環境で書き換えてください。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
#! /usr/bin/python import config tolerance_para = config.tolerance_parameter() print("tolerance_parameter=",tolerance_para) print("ウォームアップ中ちょっと待って") print() import dlib import cv2 import time from imutils.video import FPS import pickle import multiprocessing from scipy.spatial.distance import euclidean camera_width_x, camera_width_y = config.camera_width() disp_width_x, disp_width_y = config.disp_width() face_size_para = config.face_size() necessary_count = config.necessary_count() speed_size = config.speed_size() face_detector = dlib.get_frontal_face_detector() model_choice = config.model_choice() face_rec_model_path = "./face_dat/dlib_face_recognition_resnet_model_v1.dat" if model_choice == 1: face_rec_model_path = "./face_dat/taguchi_face_recognition_resnet_model_v1.dat" face_rec_model = dlib.face_recognition_model_v1(face_rec_model_path) shape_predictor = dlib.shape_predictor("./face_dat/shape_predictor_68_face_landmarks.dat") currentname = "unknown" encodingsP = "./face_dat/encodings.pickle" print("[INFO] loading encodings + face detector...") data = pickle.loads(open(encodingsP, "rb").read()) def process_faces(rgb_frame, gray_frame, tolerance, face_size): faces = face_detector(gray_frame, 1) encodings = [] names = [] for face in faces: width = face.right() - face.left() height = face.bottom() - face.top() if width < face_size or height < face_size: continue shape = shape_predictor(gray_frame, face) encoding = face_rec_model.compute_face_descriptor(rgb_frame, shape) encodings.append(encoding) distances = [] for known_encoding in data["encodings"]: distance = euclidean(encoding, known_encoding) distances.append(distance) min_distance = min(distances) if min_distance < tolerance: name = data["names"][distances.index(min_distance)] else: name = "Unknown" names.append(name) return faces, names if __name__ == "__main__": # Streamingされた画像をキャプチャ URL = "http://＜IPアドレス＞:8080/?action=stream" s_cap = cv2.VideoCapture(URL) time.sleep(2.0) fps = FPS().start() num_processes = multiprocessing.cpu_count() pool = multiprocessing.Pool(processes=num_processes) proces_width_x = int(640 * speed_size) proces_width_y = int(480 * speed_size) face_size_para = face_size_para * speed_size while True: # Streamingされた画像をフレームに切り出す ret,frame = s_cap.read() # 時計回りに９０度回転 #frame = cv2.rotate(frame,cv2.ROTATE_90_CLOCKWISE) # 180度回転 #frame = cv2.rotate(frame,cv2.ROTATE_180) # 反時計回りに９０度回転 #frame = cv2.rotate(frame,cv2.ROTATE_90_COUNTERCLOCKWISE) small_frame = cv2.resize(frame, (proces_width_x, proces_width_y)) gray = cv2.cvtColor(small_frame, cv2.COLOR_BGR2GRAY) rgb = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB) result_async = pool.apply_async(process_faces, (rgb, gray, tolerance_para, face_size_para)) faces, names = result_async.get() for (face, name) in zip(faces, names): (top, right, bottom, left) = (face.top(), face.right(), face.bottom(), face.left()) top, right, bottom, left = [int(val * (camera_width_x / proces_width_x)) for val in [top, right, bottom, left]] y = top - 15 if top - 15 > 15 else top + 15 if name != "Unknown": cv2.rectangle(frame, (left, top), (right, bottom), (0, 255, 0), 4) cv2.putText(frame, name, (left, y), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2) else: cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 225), 6) cv2.putText(frame, name, (left, y), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) if name != currentname: currentname = name print(currentname) if name == currentname and name != "Unknown": print("ok") cv2.imshow("Facial Recognition is Running", frame) key = cv2.waitKey(1) & 0xFF if key == ord("q") or key == 27: break if key == ord("p"): cv2.imwrite("face_" + name +".jpg", frame) fps.update() fps.stop() print("[INFO] elapsed time: {:.2f}".format(fps.elapsed())) print("[INFO] approx. FPS: {:.2f}".format(fps.fps())) cv2.destroyAllWindows() pool.close() pool.join() |
実行手順は（改-1）と同じです。
source V_face/bin/activate
cd face
python3 facial_recognition.py
Leave a Reply