探究Memo 009
高校生が開発したRaspberry Piを用いた現状のシリンダー錠を電子錠化するシステム。間違いなどがあったらコメントで教えてください。
現状の開発状況のまとめ。
まずは、Raspberry Pi内で動かすPythonコード。
Supabaseのテーブルの状況とかはそのうち載せます。
#モジュールのインポート
import datetime
import os
import pigpio
import DoorSystem
from supabase import create_client
from functools import partial
import setting
#システムのスタートとエンドの時刻の設定
Systemstart = datetime.time(6,30,0)
Systemend = datetime.time(19,30,0)
#現在時刻の取得
dt_now = datetime.datetime.now()
nowtime = dt_now.time()
#他のファイルのクラスの呼び出し
cr = DoorSystem.MyCardReader()
#現在時刻を取得し 、システムを切り替える
while True:
nowtime = dt_now.time()
print(str(nowtime))
#時刻を比較して実行内容を変更する
if nowtime >= Systemstart and nowtime <= Systemend:
#ドアの解錠施錠用のシステムを作動させる
cr.read_id()
else:
#動体検知 +ドアの解錠施錠用のシステムを作動させる
import camera_discord
proc = subprocess.Popen(NightSys)
proc.communicate
Raspberry Pi内部で作動させる、ドアの解錠施錠、深夜帯の不審者検知、解錠に失敗した人物の記録などのシステムを制御している。設定項目を他のファイルから取得している。
#モジュールのインポート
import pigpio
import sys
import binascii
import datetime
import json
import time
import nfc
import subprocess
from supabase import create_client
from functools import partial
import setting
import asyncio
#setting .pyから変数を呼び出している
PROJECT_URL = setting.PROJECT_URL
API_KEY = setting.API_KEY
supabase = create_client(PROJECT_URL, API_KEY)
#認証失敗時のカメラシステムの呼び出し
CameraDis = ["python","Camera.py"]
#サーボモーターの設定
SERVO_PIN = 18
pi = pigpio.pi()
#システムの起動時間の設定
Systemstart = setting.Systemstart
Systemend = setting.Systemend
#サーボモーターを制御する関数の作成
def set_angle(angle):
assert 0 <= angle <= 180.
pulse_width = (angle / 180) * (2500 - 500) + 500
pi.set_servo_pulsewidth(SERVO_PIN, pulse_width)
def afrer(n, started):
return time.time() - started > n
#認証システムを構成しているクラス
class MyCardReader(object):
def on_connect(self, tag):
#一般ユーザーのカードが使える時間を設定する
start = setting.start
end = setting.end
#ローカルファイルにログを記録する
sys.stdout = open("DoorSystem.log", "a")
#カードのid情報を取得する
idm = binascii.hexlify(tag._nfcid)
print("Get IDm")
#現在時刻を取得する
dt_now = datetime.datetime.now()
now = dt_now.time()
print(now)
#一般ユーザーのカードが利用できるか利用できないかを時間を確認して設定する
if now >= start and now <end:
allusers = supabase.table("UserList").select("*").execute()
Memo_log = "No limit"
else:
allusers = supabase.table("MasterList").select("*").execute()
Memo_log = "Master only"
id_list=[f"{user['nfckey']}" for user in allusers.data]
PassData_list=[[f"{user['name']}", f"{user['nfckey']}"] for user in allusers.data]
#ドアの状態を確認する
Door_Log_datalist = supabase.table("viewforgas").select("*").execute()
Door_list=[f"{user['door']}" for user in Door_Log_datalist.data]
Door_list= [s for s in Door_list if s != '操作なし']
Door_condition = (Door_list[0])
#カードを認証できた時の操作
if str(idm) in id_list:
#ドアが解錠されている時の操作
if Door_condition == "解錠":
set_angle(95)
time.sleep(1)
message_door="施錠"
#ドアが施錠されている時の操作
else:
set_angle (0)
message_door="解錠"
Name = [name for [name, nfckey] in PassData_list if str(idm)==nfckey]
print ("同志" + Name[0] + "は認証されたユーザーです。ドアを" + message_door + "します。\n")
message_log = "認証済"
UserName = Name[0]
#カードを認証できなかった時の操作
else:
print ("認証されていないユーザーです。" + str(idm) + "\n")
UserName = "不正なユーザー"
message_log = "認証されません"
message_door = "操作なし"
for a in range(1):
#認証失敗した人物の写真を撮影する (Camera.pyで制御)
if now >=Systemstart and now <= Systemend:
import Camera
proc = subprocess.Popen(CameraDis)
proc.communicate
proc.communicate
sys.stdout = sys.__stdout__
#Supabaseにログを送信する
logdata = {"username": UserName,"certification": message_log, "idm":str(idm), "memo":Memo_log, "door":message_door }
supabase.table("Door_log").insert(logdata).execute()
return True:
#NFCのカードを読みとるための関数
def read_id(self):
started = time.time()
clf = nfc.ContactlessFrontend('usb:000000')
wait_s = 5
#成功した時の処理
try:
clf.connect(rdwr={'on-connect': self.on_connect}, terminate=partial(afrer, wait_s, started))
#失敗した時の処理
finally:
clf.close()
ドアの解錠施錠を行うための関数をまとめたものになっている。このファイルを単独で作動させることはできない。Supabaseを使用するためネット環境がないところでは作動させることができない。ネット環境なしで作動させるためにはローカルにカードのid情報などのJSONリストを作成する必要がある。その場合カメラシステムは一切使用することができない。
from discord import Webhook
import discord
import aiohttp
import asyncio
import cv2
import time
import datetime
import cv2 as cv
import setting
webhook_url = setting.webhook_url
#カメラで撮影用
camera = cv2.VideoCapture(0)
count_number = 0
while count_number == 0:
ret,frame =camera.read()
motion_detected = False
if not ret:
break
dt_now = datetime.datetime.now() #データを取得した時刻
#ファイル名と、画像中に埋め込む日付時刻
dt_format_string = dt_now.strftime('%Y-%m-%d %H:%M:%S')
# 動き検出していれば画像を保存する
if 1+1 == 2:
title="DoorAleart.jpeg"
cv2.imwrite(title, frame)
print(title)
dt_now = datetime.datetime.now()
#メッセージの詳細
username = setting.username
title = (dt_now.strftime('%Y/%m/%d %H:%M:%S') + "部室で不審人物が確認されました")
description = setting.description1
color_hex = setting.color_hex
image_path = "DoorAleart.jpeg"
pathid = "DoorAleart.jpeg"
#discordに画像を送信する
file = discord.File(image_path, filename="image.jpeg")
embed = discord.Embed(
title=title, description=description, color=int(color_hex, 16)
).set_image(url="attachment://"+pathid)
async def foo():
async with aiohttp.ClientSession() as session:
webhookdata = Webhook.from_url(webhook_url, session=session)
await webhookdata.send(username=username, embed=embed, file=file)
asyncio.run(foo())
count_number +=1
key = cv2.waitKey(1)
if key == 27:
break
count_number -=1
camera.release()
cv2.destroyAllWindows()
カードの認証に失敗した人物の写真をDiscordに送信する。動体検知システムが作動している間は作動しない。
camera_discord.py
#モジュールのインポート
from discord import Webhook
import discord
import aiohttp
import asyncio
import cv2
import time
import datetime
import cv2 as cv
import setting
import DoorSystem
import asyncio
import setting
#DiscordのWebhookのURL
webhook_url = setting.webhook_url
#作動時間
start = setting.Systemstart
end = setting.Systemend
#カメラで撮影用
camera = cv2.VideoCapture(0)
camera.set(cv.CAP_PROP_FRAME_WIDTH, 640)
camera.set(cv.CAP_PROP_FRAME_HEIGHT, 480)
DELTA_MAX = 255
DOT_TH = 20
MOTHON_FACTOR_TH = 0.20
avg = None
#現在時刻
dt_now = datetime.datetime.now()
now = dt_now.time()
#動体検知を行う
count_count=0
while now < start or now >end:
dt_now = datetime.datetime.now()
now = dt_now.time()
ret,frame =camera.read()
motion_detected = False
dt_now = datetime.datetime.now() #データを取得した時刻
#画像をモノクロにする
gray = cv.cvtColor(frame, cv.COLOR_BGR2GRAY)
#比較用のフレームを取得する
if avg is None:
avg = gray.copy().astype("float")
continue
#現在のフレームと移動平均との差を計算
cv.accumulateWeighted(gray, avg, 0.6)
frameDelta = cv.absdiff(gray, cv.convertScaleAbs(avg))
#デルタ画像を閾値処理を行う
thresh = cv.threshold(frameDelta, DOT_TH, DELTA_MAX, cv.THRESH_BINARY)[1]
#モーションファクターを計算する。全体としてどれくらいの割合が変化したか。
motion_factor = thresh.sum() * 1.0 / thresh.size / DELTA_MAX
motion_factor_str = '{:.08f}'.format(motion_factor)
#モーションファクターがしきい値を超えていれば動きを検知したことにする
if motion_factor > MOTHON_FACTOR_TH:
motion_detected = True
# 動き検出していれば画像を保存する
if motion_detected == True:
title="image.jpeg"
cv2.imwrite(title, frame)
print(title)
dt_now = datetime.datetime.now()
#メッセージの詳細
username = setting.username
title = (dt_now.strftime('%Y年%m月%d日 %H:%M:%S'))
description = setting.description2
color_hex = setting.color_hex
image_path = "image.jpeg"
pathid = "image.jpeg"
#discordに画像を送信する
file = discord.File(image_path, filename="image.jpeg")
embed = discord.Embed(
title=title, description=description, color=int(color_hex, 16)
).set_image(url="attachment://"+pathid)
async def foo():
async with aiohttp.ClientSession() as session:
webhookdata = Webhook.from_url(webhook_url, session=session)
await webhookdata.send(username=username, embed=embed, file=file)
asyncio.run(foo())
count_count +=1
#ドアの解錠・施錠のシステムを起動する
if count_count%5==0:
cr = DoorSystem.MyCardReader()
cr.read_id()
key = cv2.waitKey(1)
if key == 27:
break
camera.release()
cv2.destroyAllWindows()
・動体検知を行うためのプログラム。日中は作動させない。光のない環境ではうまく作動させることができないので人感センサー付きの照明などを設置するとよい。
import datetime
from supabase import create_client
#Supabaseのリンク
PROJECT_URL = PROJECT_URL
API_KEY = API KEY
#Supabaseから情報を引っ張ってくる
supabase = create_client(PROJECT_URL, API_KEY)
settingdata = supabase.table("setting").select("*").execute()
data, _ = settingdata
settinglist = { d['Item_name']:d for d in data[1]}
#DiscordのWebhook URL
webhook_url = settinglist['webhook_url']['1']
#一般用の鍵が使える時間
start = datetime.time(int(settinglist['start']['1']),int(settinglist['start']['2']),int(settinglist['start']['3']))
end = datetime.time(int(settinglist['end']['1']),int(settinglist['end']['2']),int(settinglist['end']['3']))
#動体検知を行いたい時間
Systemstart = datetime.time(int(settinglist['Systemstart']['1']),int(settinglist['Systemstart']['2']),int(settinglist['Systemstart']['3']))
Systemend = datetime.time(int(settinglist['Systemend']['1']),int(settinglist['Systemend']['2']),int(settinglist['Systemend']['3']))
#認証失敗時のメッセージ
username = settinglist['username']['1']
description1 = settinglist['description1']['1']
#動体検知に引っかかったときのメッセージ
description2 = settinglist['description2']['1']
color_hex = settinglist['color_hex']['1']
#参考にするデータリストを定義している
listname1 = settinglist['listname1']['1']
listname2 = settinglist['listname2']['1']
listname3 = settinglist['listname3']['1']
listname4 = settinglist['listname4']['1']システムの設定用のファイル。Supabase (データベース)のテーブルに書いてあるデータを取得してそれぞれの変数として使用している。システムの起動時に実行するファイルなので設定項目を変更する際にはシステムの再実行が必要となる。
最初の設定時にSupabaseのPROJECT URLとAPI KEYを保存する必要がある。
Supabase SQL
alter database postgres
set timezone to 'Asia/Tokyo';
create view ViewForGAS as
select *
from
"Door_log"
where
created_at > now() - interval '7 days'
limit
100;上のコードがSupabaseのログに保存されている時間をUTC+9 にして保存するもの。
下のコードは、Supabaseに保存されたログデータを全取得し、直近7日以内の最新の100件を表示している。GASでログの検索をするためのデータの取得や、Raspberry Piのドアの解錠施錠の状況を取得するために使用する。
GAS
//Supabaseからのデータ取得を行う関数
function getSupabaseData(tableName) {
const supabaseUrl =SupabaseURL;
const supabaseApiKey = "Supabase API KEY";
const serviceRKey = "Supabase Service KEY";
//SupabaseにGETリクエストを送信している。
const options = {
method: 'GET',
headers: {
'apikey': supabaseApiKey,
'Content-Type': 'application/json',
'Authorization': 'Bearer ' + serviceRKey,
},
};
//諸々の設定(データ形式等)
const response = UrlFetchApp.fetch(supabaseUrl, options);
const data = JSON.parse(response).reverse();
const spreadsheet = SpreadsheetApp.getActiveSpreadsheet();
let number = 1;
let name_of_sheet = 0;
//データの記録が完了するまでループ処理を行う
for(let row of data){
//スプレッドシートの名前を定義する(年月の6桁の数字)
let Sheet_name = row["created_at"].slice(0,4) + row["created_at"].slice(5,7);
let new_sheet = SpreadsheetApp.getActiveSpreadsheet().getSheetByName(Sheet_name);
//読み込んだシート名が存在するかif文で確認
if(!new_sheet){
// シート名指定でシートを取得
let sheet1 = spreadsheet.getSheetByName('000000');
// 『元データ』シートの内容をコピー
new_sheet= sheet1.copyTo(spreadsheet).activate();
// シートの名前を、取得した名前に変更
new_sheet.setName(Sheet_name);
let cell = new_sheet.getRange("A2:F");
cell.clearContent();
//新しく追加したシートを先頭に移動
spreadsheet.moveActiveSheet(1);
}else{
}
let ss = SpreadsheetApp.getActiveSpreadsheet();
ss = ss.getSheetByName(Sheet_name);
let lastRow = ss.getRange(1, 2).getNextDataCell(SpreadsheetApp.Direction.DOWN).getRow();
//A列のみ最終行まで取得してデータが記録された最終行のセルの内容を変数に代入する
let a = ss.getRange("A"+lastRow).getValue();
let log_id = row["log_id"];
console.log(log_id);
//データを入力する行を決定する
number = lastRow + 1
//データを書き込む処理を行う
if(log_id>a){
//ログのidを記録
new_sheet.getRange("A"+ number).setValue(row["log_id"]);
//年月日を記録
new_sheet.getRange("B"+ number).setValue(row["created_at"].slice(0,10));
//時間を記録
new_sheet.getRange("C"+ number).setValue(row["created_at"].slice(11,18));
//利用者を記録
new_sheet.getRange("D"+ number).setValue(row["username"]);
//カードの認証結果を記録
new_sheet.getRange("E"+ number).setValue(row["certification"]);
//カードのidを記録する
new_sheet.getRange("F"+ number).setValue(row["idm"]);
//鍵に対して行った操作を行う
new_sheet.getRange("G"+ number).setValue(row["door"]);
}
number = number +1
name_of_sheet = row["created_at"].slice(0,4) + row["created_at"].slice(5,7);
}
}Googleスプレッドシートを用いたデータの検索システムの元となるデータを取得する。検索等は関数を用いて作成している。関数の詳細はここでは解説しないのでネットで調べてください。空のシートを置いておくので自由に利用してください。
この記事が気に入ったらサポートをしてみませんか？