システム自体は出来上がっておりますが、この記事はまだ完成していません。
完成形のソースをGitHubにて公開＆各所の解説を追記する予定です。
qiitaさん、間違って公開してしまっても非公開にできるようにしてください。m(_ _)m
はじめに
自宅ラズパイとUSBカメラを使ってペット監視システムを自作したので投稿します。
開発背景（読み飛ばし推奨）
社会人になってから癒される機会って減りましたよね。
10代のころは好きな部活に励んだり、友達と遊んだり、恋人とデートしたり(？)、
してましたよね。楽しいことが多かったですよね。
学生というボーナスタイムを終え、
- 職場と自宅の往復
- コロナで人と接する機会が減少
こんな状況にしばらく身を置くうちに私の心はいつからか癒しを求めるように。。。。
そしてついに半年前くらいからハムスターを飼い始めてしまいました。
とにかく、かわいい！！
小学生くらいの時に飼っていたことはあったんですが
当時は「うんち汚い」「ケージが臭い」とか思って
あんまりかわいいと思いませんでした。
でもこの年になって、飼ってみると不思議とめちゃくちゃかわいい！
やはり私も大人になったのでしょうね。。
というわけで、かわいいハムちゃんたちを
いつでもどこでも観察できるようにします。
（仕事の昼休みにスマホから、旅行先から、etc）
環境
ハードウェア
| 項目 | 型式 | 備考 |
|---|---|---|
| --- | RaspberryPi 4B | --- |
| USBカメラ | --- | 暗い時には自動で暗視モードになるのでおすすめ 夜行性だからねっ★ |
ソフトウェア
| 項目 | ライブラリ名 | バージョン | 備考 |
|---|---|---|---|
| webアプリケーション フレームワーク |
Django | ?? | はじめはFlask使ってたが、 謎のチャレンジ精神でDjangoに乗り換えたw |
| デプロイサービス | Ngrok | ?? | （代替案模索中…） |
開発手順
ラズパイの準備
まずはラズパイに64bitOSを入れて、起動
pythonの準備
パッケージリストを最新に更新してDjangoをインストール
ngrokの準備
Ngrokをインストール
トンネル登録
(TBD)
サーバ設定
(TBD)
日本のサーバを使うと通信速度が段違いに速くなるので設定しておく
Djangoアプリケーションを作成
いよいよDjangoでwebアプリケーションを開発していきます。
ファイル構成はこんな感じ（一部余計なファイルも表示されてしまっているかも、、）
└── pt_streamer
├── __init__.py
├── __pycache__
│ ├── __init__.cpython-39.pyc
│ ├── admin.cpython-39.pyc
│ ├── apps.cpython-39.pyc
│ ├── models.cpython-39.pyc
│ ├── urls.cpython-39.pyc
│ └── views.cpython-39.pyc
├── admin.py
├── apps.py
├── migrations
│ ├── 0001_initial.py
│ ├── 0002_alter_dummymodel_pub_date.py
│ ├── __init__.py
│ └── __pycache__
│ ├── 0001_initial.cpython-39.pyc
│ ├── 0002_alter_dummymodel_pub_date.cpython-39.pyc
│ └── __init__.cpython-39.pyc
├── models.py
├── myModules
│ ├── __pycache__
│ │ ├── base_camera.cpython-39.pyc
│ │ └── camera.cpython-39.pyc
│ ├── base_camera.py
│ └── camera.py
├── static
│ └── pt_streamer
│ ├── Blue_Sky_and_Clouds_Timelapse_0892__Videvo_preview.mp4
│ ├── Hamstar.png
│ ├── IMG_0184.JPG
│ ├── IMG_0211.JPG
│ ├── IMG_0212.JPG
│ ├── IMG_0279.JPG
│ ├── IMG_0317.JPG
│ ├── MVI_0321.MP4
│ ├── base.css
│ ├── base.js
│ ├── css
│ │ ├── bootstrap.min.css
│ │ └── bootstrap.min.css.map
│ ├── experiment.css
│ ├── index.css
│ ├── js
│ │ ├── bootstrap.bundle.min.js
│ │ └── bootstrap.bundle.min.js.map
│ └── stream.css
├── templates
│ └── pt_streamer
│ ├── base.html
│ ├── base.html-old
│ ├── base.html_old2
│ ├── experiment.html
│ ├── header.html
│ ├── index.html
│ ├── index_puni.html
│ ├── index_robo.html
│ └── stream.html
├── tests.py
├── urls.py
└── views.py
Python
バックエンドはpythonで実装しました。
ルーティング
views.py
from django.http import HttpResponse, HttpResponseRedirect, StreamingHttpResponse
from django.shortcuts import get_object_or_404, render
from django.urls import reverse
from django.views import generic
from django.views.decorators.csrf import csrf_exempt
from .models import DummyModel
from .myModules.camera import Camera
import cv2
import copy
# Create your views here.
def index(request):
latest_question_list = DummyModel.objects.order_by("-pub_date")[:5]
context = {"latest_question_list": latest_question_list}
return render(request, "pt_streamer/index.html", context)
# class IndexView(generic.ListView):
# template_name = "pt_streamer/index.html"
# context_object_name = "d"
# class DetailView(generic.DetailView):
# model = DummyModel
# template_name = "pt_streamer/index.html"
@csrf_exempt #この関数はCSRF検証を無効化する
def stream(request):
return render(
request,
'pt_streamer/stream.html',
{},
)
def stream_puni(request):
return render(
request,
'pt_streamer/index_puni.html',
{}
)
def stream_robo(request):
return render(
request,
'pt_streamer/index_robo.html',
{},
)
def video_feed_puni(request):
#return HttpResponse(gen(Camera(1, 0, False)),
return StreamingHttpResponse(gen(Camera(1, 0, False)),
content_type="multipart/x-mixed-replace; boundary=frame")
def video_feed_robo(request):
#return HttpResponse(gen(Camera(2, 2, True)),
return StreamingHttpResponse(gen(Camera(2, 2, True)),
content_type="multipart/x-mixed-replace; boundary=frame")
def gen(camera):
#img_no_image = cv2.imread('pt_streamer/Hamstar.png')
#img_no_image = cv2.resize(src=img_no_image, dsize=[1280, 720])
yield b'--frame\r\n'
#previous = time.time()
while True:
frame = camera.get_frame()
if frame is None:
print("frame is none")
#frame = copy.deepcopy(img_no_image)
yield b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n--frame\r\n'
def experiment(request):
return render(request, "pt_streamer/experiment.html", {})
カメラコントロールクラス
base_camera.py
import copy
import time
import datetime
import threading
try:
from greenlet import getcurrent as get_ident
except ImportError:
try:
from thread import get_ident
except ImportError:
from _thread import get_ident
class CameraEvent(object):
"""An Event-like class that signals all active clients when a new frame is
available.
"""
def __init__(self):
self.events = {}
def wait(self):
"""Invoked from each client's thread to wait for the next frame."""
ident = get_ident()
if ident not in self.events:
# this is a new client
# add an entry for it in the self.events dict
# each entry has two elements, a threading.Event() and a timestamp
self.events[ident] = [threading.Event(), time.time()]
return self.events[ident][0].wait()
def set(self):
"""Invoked by the camera thread when a new frame is available."""
now = time.time()
#remove = []
remove = None
for ident, event in self.events.items():
if not event[0].isSet():
# if this client's event is not set, then set it
# also update the last set timestamp to now
event[0].set()
event[1] = now
#print(ident)
else:
print('({0}) Process({1}) is already set.'.format(datetime.datetime.now(), ident))
# if the client's event is already set, it means the client
# did not process a previous frame
# if the event stays set for more than 5 seconds, then assume
# the client is gone and remove it
if now - event[1] > 5:
print('({0}) Detected process({1}) should be removed. Time:{2}'.format(datetime.datetime.now(), ident, now - event[1]))
#remove.append(ident)
remove = ident
#for ident in remove:
#del self.events[ident]
if remove:
del self.events[remove]
def clear(self):
"""Invoked from each client's thread after a frame was processed."""
self.events[get_ident()][0].clear()
class BaseCamera(object):
thread = [None, None]#None # background thread that reads frames from camera
frame = [None, None]#None # current frame is stored here by background thread
last_access = [0, 0]#0 # time of last client access to the camera
event = [CameraEvent(), CameraEvent()]#CameraEvent()
def __init__(self, camIndex, capIndex, vertFlip):
self.cam_index = camIndex
self.cap_index = capIndex
self.isVertFlip = vertFlip
print('({3}) Init BaseCamera cam_idx:{0}, cap_idx:{1}, flip:{2}'.format(self.cam_index, self.cap_index, self.isVertFlip, datetime.datetime.now()))
"""Start the background camera thread if it isn't running yet."""
if BaseCamera.thread[self.cam_index - 1] is None:
BaseCamera.last_access[self.cam_index - 1] = time.time()
# start background frame thread
BaseCamera.thread[self.cam_index - 1] = threading.Thread(target=self._thread, args=(self.cam_index, self.cap_index, self.isVertFlip,))
BaseCamera.thread[self.cam_index - 1].start()
# wait until frames are available
#while self.get_frame() is None:
#print('[CAM{0}]self.get_frame() is None'.format(self.cam_index))
#time.sleep(0)
BaseCamera.event[self.cam_index - 1].wait()
def get_frame(self):
"""Return the current camera frame."""
BaseCamera.last_access[self.cam_index - 1] = time.time()
# wait for a signal from the camera thread
BaseCamera.event[self.cam_index - 1].wait()
BaseCamera.event[self.cam_index - 1].clear()
return BaseCamera.frame[self.cam_index - 1]
@staticmethod
def frames():
""""Generator that returns frames from the camera."""
raise RuntimeError('Must be implemented by subclasses.')
@classmethod
def _thread(cls, cam_index, cap_index, flip):
"""Camera background thread."""
print('({3}) Starting camera thread. cam_index:{0}, cap_index:{1}, flip:{2}'.format(cam_index, cap_index, flip, datetime.datetime.now()))
frames_iterator = cls.frames(cap_index, flip)
for frame in frames_iterator:
BaseCamera.frame[cam_index - 1] = frame
BaseCamera.event[cam_index - 1].set() # send signal to clients
time.sleep(0)
# if there hasn't been any clients asking for frames in
# the last 10 seconds then stop the thread
if time.time() - BaseCamera.last_access[cam_index - 1] > 10:
frames_iterator.close()
print('({0}) Stopping camera thread due to inactivity.'.format(datetime.datetime.now()))
break
BaseCamera.thread[cam_index - 1] = None
camera.py
#import os
#os.environ['OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS'] = 0
import cv2
from .base_camera import BaseCamera
import time
import datetime
class Camera(BaseCamera):
def __init__(self, camIndex, capIndex, vertFlip):
super().__init__(camIndex, capIndex, vertFlip)
#self.index = deviceIndex #いらんかも
#self.isVertFlip = vertFlip #いらんかも
@staticmethod
def frames(cap_index, flip):
#camera = cv2.VideoCapture(2)
camera = cv2.VideoCapture(cap_index, cv2.CAP_V4L2)
if not camera.isOpened():
raise RuntimeError('Could not start camera.')
camera.set(cv2.CAP_PROP_BUFFERSIZE, 1)
camera.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc('M', 'J', 'P', 'G'))
camera.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
#camera.set(cv2.CAP_PROP_FPS, 30)
print("({5}) Width:{0}, Height:{1}, BufSize:{2}, Format:{3}, Fps:{4}".format(camera.get(cv2.CAP_PROP_FRAME_WIDTH), camera.get(cv2.CAP_PROP_FRAME_HEIGHT), camera.get(cv2.CAP_PROP_BUFFERSIZE), int(camera.get(cv2.CAP_PROP_FOURCC)).to_bytes(4, 'little').decode('utf-8'), camera.get(cv2.CAP_PROP_FPS), datetime.datetime.now()))
#camera.read()
#time.sleep(0)
while True:
#time_sta = time.time()
#time.sleep(0.01)
# read current frame
_, img = camera.read()
#time_end = time.time()
#print('[CAP{0}]camera.read():{1}'.format(cap_index, time_end - time_sta))
if not img is None and flip == True:
img = cv2.rotate(img, cv2.ROTATE_180)
# encode as a jpeg image and return it
yield cv2.imencode('.jpg', img)[1].tobytes()
#time.sleep(1 / camera.get(cv2.CAP_PROP_FPS))
html
フロントエンドはhtml（＋後述のcssスタイルシートで装飾）で実装しました。
ハムスターケージは2つあるのでそれぞれのカメラ映像が1ページに収まるように
htmlをカスタマイズ
(TBD)
css
cssスタイルシートでhtmlを装飾します。
(TBD)
単体デバッグ
Djangoアプリを起動し、ローカルで映像配信されていることを確認
デプロイ
DjangoのポートをNgrokで世界へ開放！（自己責任でお願いします。）
システム動作確認
宅内ネットワーク外から接続し、ハムちゃんたちが見れたら成功
※IPアドレスがバレてしまうのでGIFの下の方はマスクしています（Ngrokへの懸念点①）
わぁ〜い
ちなみにうちにはハムスターゲージが２棟ありまして、
１棟目⇨ジャンガリアンハムスター ぷに♂
２棟目⇨ロボロフスキーハムスター ３匹とも♀（名前はまだ無い）
という体制をとっております。
本来ならページ上部には１棟目が写し出されるはずなのですが
奥さんが里帰り出産した際に実家に連れ去られてしまったのでピンクのイカちゃんが場を和ませてくれています。。
感想
やりたいことは最低限実現できたが、時々動作が重くなるなど
まだまだ改善点はある。
考えられる要因は以下
- Ngrokのスループットが足りてない（AWSとか使えば速くなる？）
- 実力不足で最適なPythonプログラムが書けていない
- 外部からアクセスしている端末で何かしら問題がある（処理が間に合っていないなど。そんなことあるか？？）
ローカルではサクサク動くことを考えるとNgrokが原因かな～
ネットで検索してみても結構遅いと言われているし、、、
まあ、ひとまず今回の目標は達成したので
高速化は次回のテーマということで！
FlaskからDjangoに乗り換えた際にやった作業なんかも記事にしたいなぁ
最後まで読んでいただきありがとうございました。
アドバイスや質問など、コメントお待ちしております。