あなたへの挑戦状
実は先日、会社でプロトタイプの発表会があり、Mr.-Variable-Pickerというプロトタイプを作りました
簡単に説明すると、カメラに映ったお客様の分析をするハードウェアです。
私の会社は普通の小売業なのでこの製作物は個人的にレベルが高いかななんて思ってました
しかし！発表会では最優秀所に私の名前がありませんでした…
悔しい！！
しかも私に勝利した方はシステム担当者ではなく、勝手にライバル視していた相手でした。
ここまで負けっぱなしで終われません
勝手に延長戦としてプロトタイプの改良案を記事にします。
改良案
・属性判断モデルをage-gender-recognition-retail-0013からDeepFaceに変更
・ラズパイ5上の映像をPCに送信し、PC側でAI処理（ラズパイ5DeepFaceのリアルタイム処理が重過ぎるため）
AWSとか使えばよくねって思った方。その通りです。
ただAWSがわからない！！いつか勉強します…（こうやって重要な技術後回しにするから負けるんですよね…）
アーキテクチャ
1.ラズパイ (Client): カメラで人を撮影し、顔を見つけたらその画像をPCに送信します。
2.ハイスペックPC (Server): 送られてきた画像を受け取り、DeepFaceで「年齢・性別・感情」を分析して結果を返します。
3.ラズパイ (Client): 結果を受け取り、画面に表示＆CSVに記録します。
準備するもの
・Raspberry Pi 5（4でも動きますが5推奨）
・USB Webカメラ（ラズパイに接続）
・PC（WindowsでもMacでもOK。AI処理を担当します）
※ 同じWi-Fiネットワークに繋がっている必要があります
ステップ1 PC側（サーバー）の準備
ずは、重い処理を担当するPC側の準備です。 ここでは「送られてきた画像を分析して返す」Webサーバーを作ります。
1. ライブラリのインストール
PCのコマンドプロンプト（またはターミナル）を開き、以下のコマンドを入力して必要なライブラリを入れます。
pip install fastapi uvicorn python-multipart opencv-python deepface numpy
2. サーバーのプログラム作成
適当なフォルダを作り、その中に server.pyというファイルを作成して、以下のコードをコピペしてください。
# server.py
import uvicorn
from fastapi import FastAPI, File, UploadFile
import numpy as np
import cv2
from deepface import DeepFace
app = FastAPI()
print("AIモデルを読み込んでいます...少々お待ちください")
try:
# ダミー画像で一度実行してモデルをメモリに展開
dummy_img = np.zeros((100, 100, 3), dtype=np.uint8)
DeepFace.analyze(dummy_img, actions=['age', 'gender', 'emotion'], enforce_detection=False)
print("モデルの読み込み完了！準備OKです。")
except:
pass
@app.post("/analyze")
async def analyze_face(file: UploadFile = File(...)):
try:
# 1. 送られてきたデータを画像形式に変換
contents = await file.read()
nparr = np.frombuffer(contents, np.uint8)
img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
if img is None:
return {"error": "画像が壊れています"}
# 2. DeepFaceで分析実行！
# detector_backend='skip' にして、顔検出を省略
analysis_results = DeepFace.analyze(
img,
actions=['age', 'gender', 'emotion'],
enforce_detection=False,
detector_backend='skip'
)
# 3. 結果を整理して返す
if analysis_results:
result = analysis_results[0]
return {
"success": True,
"age": result['age'],
"gender": result['dominant_gender'],
"emotion": result['dominant_emotion']
}
else:
return {"success": False}
except Exception as e:
return {"success": False, "error": str(e)}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
3. PCのIPアドレスを調べる
ラズパイが「どこに画像を送ればいいか」を知るために、PCの住所（IPアドレス）が必要です。
Windowsの場合: コマンドプロンプトで ipconfig と入力。IPv4 アドレス の横にある数字（例: 192.168.1.10）をメモします。
Macの場合: ターミナルで ifconfig | grep "inet " と入力するか、システム設定のネットワークから確認します。
4. サーバー起動
PCで以下のコマンドを実行します。
python server.py
「モデルの読み込み完了！」と表示されたら、PC側は準備万端で待機状態です。
ステップ2：ラズパイ側（クライアント）の準備
1. フォルダ構成を作る
ラズパイのデスクトップなどに customer_analytics というフォルダを作り、その中を以下のようにします。
haarcascadeはこちらで公開されています。
customer_analytics/
├─ models/
│ └─ haarcascade_frontalface_default.xml ←これをダウンロードして配置
├─ logs/ ←自動で生成される
├─ tracker.py
├─ analytics.py
└─ main_rpi.py
2. ライブラリのインストール
ラズパイのターミナルで実行します。
pip install opencv-python requests scipy numpy
3. プログラムの作成
以下の3つのファイルを作成して、さきほどのフォルダに保存してください。
① tracker.py （人の追跡を行うプログラム）
顔が動いても「同じ人だ」と認識し続けるためのロジックです。
# tracker.py
import time
from collections import OrderedDict
from scipy.spatial import distance as dist
import numpy as np
class CentroidTracker:
def __init__(self, max_disappeared=30):
self.next_id = 0
self.objects = OrderedDict()
self.bboxes = OrderedDict()
self.disappeared = OrderedDict()
self.max_disappeared = max_disappeared
def register(self, centroid, bbox):
self.objects[self.next_id] = centroid
self.bboxes[self.next_id] = bbox
self.disappeared[self.next_id] = 0
self.next_id += 1
def deregister(self, obj_id):
del self.objects[obj_id]
del self.bboxes[obj_id]
del self.disappeared[obj_id]
def update(self, rects):
deregistered_ids = []
if len(rects) == 0:
for obj_id in list(self.disappeared.keys()):
self.disappeared[obj_id] += 1
if self.disappeared[obj_id] > self.max_disappeared:
deregistered_ids.append(obj_id)
self.deregister(obj_id)
return self.bboxes, deregistered_ids
input_centroids = np.zeros((len(rects), 2), dtype="int")
for (i, (startX, startY, endX, endY)) in enumerate(rects):
cX = int((startX + endX) / 2.0)
cY = int((startY + endY) / 2.0)
input_centroids[i] = (cX, cY)
if len(self.objects) == 0:
for i in range(len(input_centroids)):
self.register(input_centroids[i], rects[i])
else:
obj_ids = list(self.objects.keys())
obj_centroids = list(self.objects.values())
D = dist.cdist(np.array(obj_centroids), input_centroids)
rows = D.min(axis=1).argsort()
cols = D.argmin(axis=1)[rows]
used_rows, used_cols = set(), set()
for (row, col) in zip(rows, cols):
if row in used_rows or col in used_cols: continue
obj_id = obj_ids[row]
self.objects[obj_id] = input_centroids[col]
self.bboxes[obj_id] = rects[col]
self.disappeared[obj_id] = 0
used_rows.add(row)
used_cols.add(col)
unused_rows = set(range(0, D.shape[0])).difference(used_rows)
unused_cols = set(range(0, D.shape[1])).difference(used_cols)
if D.shape[0] >= D.shape[1]:
for row in unused_rows:
obj_id = obj_ids[row]
self.disappeared[obj_id] += 1
if self.disappeared[obj_id] > self.max_disappeared:
deregistered_ids.append(obj_id)
self.deregister(obj_id)
else:
for col in unused_cols:
self.register(input_centroids[col], rects[col])
return self.bboxes, deregistered_ids
② analytics.py （データの集計・保存を行うプログラム）
分析結果をまとめたり、CSVファイルに保存したりします。
# analytics.py
import csv
import time
from collections import deque, Counter
from pathlib import Path
from typing import Dict, Any
import numpy as np
DWELL_THRESHOLD = 2.0 # 何秒以上いたら「滞在」とみなすか
class RetailMetrics:
def __init__(self):
self.face_records = {}
def update(self, obj_id, expression, age, gender):
if obj_id not in self.face_records:
self.face_records[obj_id] = deque(maxlen=150)
# 表記ゆれを統一
gender = gender.replace("Man", "Male").replace("Woman", "Female")
self.face_records[obj_id].append((time.time(), expression, age, gender))
def get_person_summary(self, obj_id):
records = self.face_records.get(obj_id)
if not records: return {}
timestamps, expressions, ages, genders = zip(*records)
dwell_sec = timestamps[-1] - timestamps[0]
return {
"gender": Counter(genders).most_common(1)[0][0],
"age": int(np.median(ages)),
"expression": Counter(expressions).most_common(1)[0][0],
"result": 'stay' if dwell_sec >= DWELL_THRESHOLD else 'pass',
"dwell_sec": dwell_sec
}
def get_current_stable_attributes(self, obj_id):
records = self.face_records.get(obj_id)
if not records: return {}
# 直近5件のデータから多数決をとる
recent = list(records)[-5:]
_, expressions, ages, genders = zip(*recent)
return {
"age": int(np.median(ages)),
"gender": Counter(genders).most_common(1)[0][0],
"expression": Counter(expressions).most_common(1)[0][0]
}
def finalize_person(self, obj_id):
if obj_id in self.face_records: del self.face_records[obj_id]
class CsvLogger:
def __init__(self, out_dir='logs'):
Path(out_dir).mkdir(parents=True, exist_ok=True)
ts = time.strftime('%Y%m%d_%H%M')
self.path = Path(out_dir) / f'analytics_{ts}.csv'
self.file = open(self.path, 'w', newline='', encoding='utf-8')
self.writer = csv.writer(self.file)
self.writer.writerow(['timestamp', 'gender', 'age', 'expression', 'result', 'dwell_sec'])
def log(self, summary):
self.writer.writerow([
time.strftime('%Y-%m-%d %H:%M:%S'),
summary.get("gender"), summary.get("age"),
summary.get("expression"), summary.get("result"),
f'{summary.get("dwell_sec", 0):.2f}'
])
self.file.flush() # すぐに書き込む
def close(self): self.file.close()
def __enter__(self): return self
def __exit__(self, exc_type, exc, tb): self.close()
③ main_rpi.py （メイン実行プログラム）
これがラズパイ側の主役です。以下のコードの SERVER_URL を、ステップ1で調べたPCのIPアドレスに書き換えてください！
# main_rpi.py
import cv2
import threading
import requests
from tracker import CentroidTracker
from analytics import RetailMetrics, CsvLogger
# ここを自分のPCのIPアドレスに書き換えてください！
SERVER_URL = "http://192.168.X.X:8000/analyze"
# 例: "http://192.168.1.10:8000/analyze"
face_cascade_path = 'models/haarcascade_frontalface_default.xml'
face_cascade = cv2.CascadeClassifier(face_cascade_path)
# 分析リクエストを別スレッド（裏側）で送る関数
def request_analysis(crop_img, obj_id, metrics_obj):
try:
_, img_encoded = cv2.imencode('.jpg', crop_img)
files = {'file': ('image.jpg', img_encoded.tobytes(), 'image/jpeg')}
# タイムアウトを2秒に設定して、遅い場合は無視
response = requests.post(SERVER_URL, files=files, timeout=2.0)
if response.status_code == 200:
data = response.json()
if data.get("success"):
metrics_obj.update(obj_id, data['emotion'], data['age'], data['gender'])
except Exception:
pass # 通信エラー時は何もしない
tracker = CentroidTracker()
metrics = RetailMetrics()
cap = cv2.VideoCapture(0) # カメラ起動
if face_cascade.empty():
print("エラー: modelsフォルダにxmlファイルが見つかりません！")
exit()
with CsvLogger('logs') as logger:
print(f'接続先サーバー: {SERVER_URL}')
print('[q] キーを押すと終了します')
frame_counter = 0
while True:
ret, frame = cap.read()
if not ret: break
# 1. 顔検出
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
faces = face_cascade.detectMultiScale(gray, 1.1, 8, minSize=(80, 80))
rects = [(x, y, x+w, y+h) for (x, y, w, h) in faces]
# 2. トラッキング（ID付与）
objects, deregistered_ids = tracker.update(rects)
# 3. いなくなった人をCSVに記録
for obj_id in deregistered_ids:
summary = metrics.get_person_summary(obj_id)
if summary: logger.log(summary)
metrics.finalize_person(obj_id)
# 4. 画面内の顔を処理
for obj_id, (x1, y1, x2, y2) in objects.items():
# 座標のはみ出し防止
h, w = frame.shape[:2]
x1, y1, x2, y2 = max(0, x1), max(0, y1), min(w, x2), min(h, y2)
crop = frame[y1:y2, x1:x2]
if crop.size == 0: continue
# 10フレームに1回だけサーバーに送る
if obj_id % 10 == frame_counter % 10:
# メイン処理を止めないように「別スレッド」で送信
threading.Thread(target=request_analysis, args=(crop.copy(), obj_id, metrics), daemon=True).start()
# 現在の結果を取得して表示
attrs = metrics.get_current_stable_attributes(obj_id)
label = f"ID:{obj_id} {attrs.get('age','?')} {attrs.get('gender','?')} {attrs.get('expression','?')}"
cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
cv2.putText(frame, label, (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
frame_counter += 1
cv2.imshow('RPi Camera Client', frame)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
cap.release()
cv2.destroyAllWindows()
実行！
1.必ず PC → ラズパイ の順番で起動します。
2.PC側: python server.py を実行中であることを確認。
ラズパイ側: 以下のコマンドで実行！
python main_rpi.py
成功すると…？
ラズパイの画面にカメラ映像が映ります。 顔をカメラに向けると、最初は ? ですが、一瞬遅れてPCから分析結果が届き、頭の上に「ID:0 25 Male happy」のように表示されます！
logs フォルダにCSVファイルが溜まっていくので、後でExcelで開いて分析してみてくださいね。
うまくいかないときは？
・PCのファイアウォール、Windows Defenderなどが「ポート8000」の通信をブロックしている可能性があります。「pythonの通信を許可しますか？」というポップアップが出たら「許可」を押してください。
・ラズパイのコード内の SERVER_URL が正しいかもう一度確認しましょう。
・PCとラズパイは同じWi-Fiに繋がっていますか？
私に勝ったあなたへ
見てますか？あなたです。
そう、あなたです。あなたなのです。あなたに言っています。
世の中にはこんな言葉があります。
「男子、三日会わざれば刮目して見よ」
ここでいう「男子」とは、性別のことではなく「努力する人」のことらしいです。
しかし！私は「鬼のように努力する人」なので一日おき位に刮目したほうがいいですよ
次の戦いを楽しみにしています🔥
という事で皆さんもレッツトライ！