脱・初心者! Python100行で自作ガジェット?! 無線防犯カメラから学ぶ技術
筆者は驚いている、防犯カメラのガジェットを100行のPythonコードで実現できてしまうなんて。
今回、想定しているESP32-S3の開発ボードは以下。筆者もAmazonで購入した。Pythonと表記していますが、ESP32（マイコン）向けのMicroPythonのコードになります。MicroPythonとPythonは異なる部分がありますが、PC向けのPythonにおいても活用できる技術だと思っています。
ESP32を入門するならFeenoveのボードが品質が高いと思う🤩
PCでカメラを動かすだけなら「10行くらいのコード」でできるが、今回は、手のひらにのるマイコンボードを使い、無線Wi-Fiで映像を飛ばし、実際に部屋の隅に設置して、スマホで閲覧できる「物理的なガジェット」が、Pythonコード100行で実現できるという話。
必要なものは以下のボードのみ！2,000円台で買える。
実践で学ぶ
せっかくなので、この実装をケーススタディにして、「Python力」を一段上のレベルへ引き上げましょう。以下に掲載するのが、実際のソースコードです。行数は100行。 この短いコードの中には、HTTPサーバー・通信、HTML、MJPEGによるストリーミング、マルチスレッド処理といった、実務でも有益となる重要概念がぎっしり詰まっています。これらを1行ずつ紐解き、仕組みを完全に理解できたとき、「脱・初心者」を名乗っていいはずです。 しかも、画面の中だけで終わらない「実際に動く自作ガジェット」という最高の成果物付いてきます！（文章書くのに、Geminiに手伝ってもらったら、モリモリになってしまった。もうちょっと自然にしてほしいのだが…）
import network
import socket
import time
import _thread
from camera import Camera, FrameSize, PixelFormat
SSID = 'YOUR_SSID'
PASSWORD = 'YOUR_PASSWORD'
CAM_PIXEL_FORMAT = PixelFormat.JPEG
CURRENT_FRAME_SIZE = FrameSize.VGA
XCLK_FREQ = 10000000
def connect_wifi(ssid, password):
station = network.WLAN(network.STA_IF)
station.active(True)
station.connect(ssid, password)
while not station.isconnected():
time.sleep(1)
return station.ifconfig()[0]
def init_camera(frame_size, pixel_format, xclk_freq):
cam = Camera(frame_size=frame_size, pixel_format=pixel_format, xclk_freq=xclk_freq, init=False)
cam.init()
return cam
def generate_html():
return """<!DOCTYPE html><html><head><meta charset="UTF-8"><title>ESP32 Camera</title></head>
<body><div class="video-container"><h1>ESP32 Camera Live Stream</h1><img src="/stream" alt="Camera Stream">
</div></body></html>
"""
def handle_stream_request(client, cam):
try:
headers = (
b'HTTP/1.1 200 OK\r\n'
b'Content-Type: multipart/x-mixed-replace; boundary=frame\r\n'
b'Cache-Control: no-cache\r\n\r\n')
client.send(headers)
while True:
frame = cam.capture()
if not frame:
time.sleep(0.01)
continue
client.send(b'--frame\r\nContent-Type: image/jpeg\r\nContent-Length: ' + str(len(frame)).encode() + b'\r\n\r\n')
client.send(frame)
client.send(b'\r\n')
except OSError:
pass
def handle_html_request(client, html_content):
response = 'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n' + html_content
client.send(response.encode())
def handle_client(client, cam, html_content):
try:
client.settimeout(5.0)
request = b""
while True:
data = client.recv(1024)
if not data:
break
request += data
if request[-4:] == b"\r\n\r\n":
break
if len(request) > 2048:
print("Request too large, closing connection.")
return
req_text = request.decode('utf-8', 'ignore')
if 'GET /stream' in req_text:
client.settimeout(None)
handle_stream_request(client, cam)
elif 'GET /' in req_text:
handle_html_request(client, html_content)
except OSError:
pass
finally:
try:
client.close()
except:
pass
def main():
ip_address = connect_wifi(SSID, PASSWORD)
print(f'Connected! IP: {ip_address}')
cam = init_camera(CURRENT_FRAME_SIZE, CAM_PIXEL_FORMAT, XCLK_FREQ)
html_content = generate_html()
addr = socket.getaddrinfo('0.0.0.0', 80)[0][-1]
s = socket.socket()
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(addr)
s.listen(2)
while True:
try:
client, addr = s.accept()
_thread.start_new_thread(handle_client, (client, cam, html_content))
except Exception:
time.sleep(1)
main()実務でも使える重要な技術
カメラハードウェアの初期化と設定(MicroPython限定)
MicroPython特有の実装なのであまり重要じゃない。
プログラムの開始時に、cameraモジュールを用いてカメラセンサーを直接制御している。
Wi-FiのAPへ接続(MicroPython限定)
MicroPython特有の実装なのであまり重要じゃない。
networkモジュールを使用し、ESP32をステーション（STA）モードとして動作させる。
指定したSSIDとパスワードを用いて、既存のルーター（アクセスポイント）へ接続を確立する。
接続が完了するまでループで待機し、取得したIPアドレスを返すことで、サーバーがどこで稼働しているかを明確にする。
HTTPサーバー
低レベルなAPIのSocketを使ってWebサーバーを作っている。中級者以上でもやったことがない人がいるかも。ここでHTTPの概要を理解する。
socketモジュールを用いて、標準的なポート80番で待ち受ける簡易的な自前Webサーバーを構築している。
SO_REUSEADDRオプションを設定し、プログラム再起動時に「ポートが使用中」となるエラーを回避している。
メインループ（while True）で常にクライアントからの接続要求（リクエスト）を待ち続ける構成をとっている。
HTMLファイルの配信
💎特に重要な技術。Webサーバー、HTML、マイコン、PCがどのような関係になっているかを理解することが重要。
カメラ映像を表示させるための枠組みとなるHTMLコードを、プログラム内で文字列として生成し保持している。
クライアントがルートURL（GET /）にアクセスした際、標準的なHTTPステータス（200 OK）とtext/htmlのヘッダーを付与してHTMLデータを返却する。
HTML内には<img src="/stream">というタグが含まれており、ブラウザに次の映像リクエストを促す役割を持たせている。
マルチスレッドによる並行処理
💎重要な概念。低レベルなAPIを使っているので、中級者以上でもやったことがない人がいるかも。
ブラウザはHTMLを受信した直後に、画像ソースである/streamへ同時にアクセスを試みる。
ESP32が単一スレッドのままでは、映像を配信し続ける処理（無限ループ）に入ると、他の通信（HTMLの配信など）を一切受け付けなくなってしまう。
リクエストを受け付けるたびに_thread.start_new_threadで独立した処理（スレッド）を立ち上げることで、複数クライアントへの対応や、HTML配信と映像ストリーミングの同時実行を可能にしている。
MJPEGによるストリーミングと持続的接続
💎古い技術だが、まだ使われることが多いので、知っておくと役に立つこともある。HTTPプロトコルという概念の理解は超重要。
映像リクエスト（GET /stream）に対しては、HTTPヘッダーのContent-Type: multipart/x-mixed-replaceを使用し、データが次々と更新されることをブラウザに伝達する。
接続のタイムアウトを無効化（client.settimeout(None)）し、サーバー側から通信を切断しないように設定している。
while Trueの無限ループ内で、カメラから取得した1枚のJPEG画像を区切り文字（今回は「frame」とした）とともに送り続けることで、パラパラ漫画の要領で動画を再生させる（Motion JPEG方式）。
入門するのに最適なLEDや抵抗がセットもある。カメラで撮影中は赤いLEDを点灯させると映えるかも。
コードの解説
関数ごとに解説を書いた。
グローバル変数の意味
# 接続先Wi-FiアクセスポイントのSSIDを設定する。
SSID = 'YOUR_SSID'
# 接続先Wi-Fiのパスワードを設定する。
PASSWORD = 'YOUR_PASSWORD'
# カメラの出力フォーマットを設定する（Webストリーミングに適したJPEGを指定）。
CAM_PIXEL_FORMAT = PixelFormat.JPEG
# カメラの解像度（フレームサイズ）を設定する。
# 【解像度のサンプル設定】
# FrameSize.QQVGA : 160 x 120
# FrameSize.QVGA : 320 x 240
# FrameSize.VGA : 640 x 480 （現在の設定）
# FrameSize.SVGA : 800 x 600
# FrameSize.UXGA : 1600 x 1200
CURRENT_FRAME_SIZE = FrameSize.VGA
# カメラモジュールを駆動する外部クロック周波数を設定する（安定動作のため10MHzを指定）。
XCLK_FREQ = 10000000connect_wifi()
指定されたSSIDとパスワードを使用してWi-Fiネットワークに接続する。
init_camera()
指定された解像度（フレームサイズ）、ピクセルフォーマット（JPEG）、およびクロック周波数でカメラハードウェアを初期化する。
generate_html()
クライアント（ブラウザ）に送信するためのWebページのHTMLデータを文字列として返す。
handle_stream_request()
無限ループの中で、カメラから画像を取得し、クライアントに対して、JPEGデータ（Motion JPEGストリーム）としてクライアントへ送り続ける。
def handle_stream_request(client, cam):
# ストリーム配信（Motion JPEG）の送信処理を開始する。
try:
# 映像ストリーミング用（同一URLでデータを差し替え続ける設定）のHTTPレスポンスヘッダーを作成する。
# 境界線（区切り文字）の名前を「frame」と定義し、ブラウザのキャッシュを無効化（no-cache）する。
headers = (
b'HTTP/1.1 200 OK\r\n'
b'Content-Type: multipart/x-mixed-replace; boundary=frame\r\n'
b'Cache-Control: no-cache\r\n\r\n')
# 定義したHTTPヘッダーをクライアント（ブラウザ）へ送信する。
client.send(headers)
# 映像フレームを無限に配信し続けるためのループ処理を行う。
while True:
# カメラから最新の1フレーム（JPEGデータ）をキャプチャ（取得）する。
frame = cam.capture()
# 画像データが正常に取得できなかった場合の処理を行う。
if not frame:
# 0.01秒待機し、CPUの負荷を下げつつ次のフレーム取得を試みる。
time.sleep(0.01)
continue
# 各画像フレームの区切り情報（ヘッダー）を送信する。
# 区切り文字「--frame」、画像形式「image/jpeg」、および画像のデータサイズを伝える。
client.send(b'--frame\r\nContent-Type: image/jpeg\r\nContent-Length: ' + str(len(frame)).encode() + b'\r\n\r\n')
# キャプチャしたJPEG画像データ本体をクライアントへ送信する。
client.send(frame)
# データの終了を示す改行コードを送信する。
client.send(b'\r\n')
# ブラウザが閉じられたことによる切断など、通信エラー（OSError）が発生した場合は処理を抜ける。
except OSError:
passhandle_html_request()
クライアントに対して、WebページのHTMLデータ（ブラウザの画面）を送信する。
難しかったら、「MicroPythonプログラミング・ガイドブック」などから入るといいかも。今回の話は、複合的な技術を使っていて難易度がそれなりに高い。でも、わかったときの達成感とスキルは価値があると思う。
handle_client()
クライアントからのHTTPリクエストを受信し、内容を解析する。
リクエストURLが GET /stream の場合は、handle_stream_requestを呼び出して映像ストリームを配信する。
リクエストURLが GET / （ルート）の場合は、handle_html_requestを呼び出してHTMLページを配信する。
def handle_client(client, cam, html_content):
# クライアントとの通信処理を開始する。
try:
# データの受信待ちタイムアウトを5秒に設定する。
client.settimeout(5.0)
# 受信データを格納するための空のバイト列を準備する。
request = b""
# リクエストデータを最後まで読み取るためのループ処理を行う。
while True:
# クライアントから最大1024バイトのデータを受信する。
data = client.recv(1024)
# データを受信できなかった場合（切断時など）はループを抜ける。
if not data:
break
# 受信したデータを変数に追加する。
request += data
# HTTPヘッダーの終端（\r\n\r\n）を検出したらループを抜ける。
if request[-4:] == b"\r\n\r\n":
break
# メモリ不足を防ぐため、データが2048バイトを超えた場合は通信を強制終了する。
if len(request) > 2048:
print("Request too large, closing connection.")
return
# 受信したバイト列データをUTF-8の文字列に変換する（変換エラーは無視する）。
req_text = request.decode('utf-8', 'ignore')
# リクエスト内容が映像ストリーム（/stream）を要求しているか判定する。
if 'GET /stream' in req_text:
# 映像を継続的に送信するため、タイムアウトを無効化する。
client.settimeout(None)
# ストリーム配信用の関数を呼び出す。
handle_stream_request(client, cam)
# リクエスト内容がトップページ（/）を要求しているか判定する。
elif 'GET /' in req_text:
# HTMLページ配信用の関数を呼び出す。
handle_html_request(client, html_content)
# 通信中に発生したOSレベルのエラー（切断など）は無視する。
except OSError:
pass
# エラーの有無にかかわらず、最後に必ず実行する。
finally:
try:
# クライアントとの接続（ソケット）を閉じる。
client.close()
# 切断処理中に発生したエラーも無視する。
except:
passmain()
プログラム全体のメイン処理を実行する。
クライアントからの接続要求を受け付けるたびに新しいスレッド（_thread）を生成し、handle_clientによる並行処理を行う。
def main():
# Wi-Fiネットワークに接続し、ESP32のIPアドレスを取得する。
ip_address = connect_wifi(SSID, PASSWORD)
# 接続が完了したことをIPアドレスとともにコンソールに出力する。
print(f'Connected! IP: {ip_address}')
# 指定した解像度やフォーマットでカメラハードウェアを初期化する。
cam = init_camera(CURRENT_FRAME_SIZE, CAM_PIXEL_FORMAT, XCLK_FREQ)
# クライアントに配信するためのWebページのHTMLデータをあらかじめ生成する。
html_content = generate_html()
# 自身のすべてのアドレス（0.0.0.0）で、HTTP標準のポート80番を使用するためのアドレス情報を取得する。
addr = socket.getaddrinfo('0.0.0.0', 80)[0][-1]
# ネットワーク通信を行うためのサーバー用ソケットを作成する。
s = socket.socket()
# プログラム再起動時などに「Address already in use」エラーを防ぐため、ソケットを即座に再利用可能にする設定を行う。
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 取得したアドレス情報（IPとポート）をソケットに紐付ける（バインドする）。
s.bind(addr)
# クライアントからの接続要求の待機（リッスン）を開始する（同時に最大2つまでの接続要求を保留できる）。
s.listen(2)
# サーバーとして常時稼働させるための無限ループ処理を行う。
while True:
try:
# クライアントからの接続要求を受け入れ、通信用のクライアントソケットと相手のアドレスを取得する。
client, addr = s.accept()
# クライアントごとの処理を別スレッドで開始し、複数の接続を並行して処理できるようにする。
_thread.start_new_thread(handle_client, (client, cam, html_content))
# 接続待機中やスレッド作成時にエラーが発生した場合の処理を行う。
except Exception:
# エラーによる無限ループの暴走を防ぐため、1秒間待機する。
time.sleep(1)準備については以下の記事で述べた。
サーボモータなどを接続すれば、カメラの向きをサーボモーターで自由に制御できる。これが自作ガジェットの強み🤩🎈
さらなる高みへ
とりあえず動くものレベルを求めている場合、読まなくていいです。
MicroPythonにおいて _thread モジュールは非推奨のよう。asyncio系を活用した非同期処理に移行するのが良さそう。ただ、Pythonの非同期処理はバージョンによって書き方のが違うことがあり、少し扱いづらい側面がある。
特にMirocoPythonだとバージョンや対応状況が複雑（MicroPythonは基本的なasyncioなら使えると思っていてよさそう）。マルチスレッドなのに cam.capture() へのアクセスにロック（排他制御）がかかっていないため、複数クライントが来たらバグる可能性がある。複数のブラウザで開かない前提の話。
カメラのバッファのゼロコピー
など
