はじめに

研究室にあった従来の入退室システムが動かなくなったので，新しく作成しました．

難しい知識が一切なかったので，できるだけ簡単にこなすことを意識し，notionを活用することに決めました(^ ^)．
なにぶん初心者なので，大目に見てください．😀

基本機能

交通系ICカード，学生証で入退室を切り替え
カードに名前を紐づけてNotionデータベースで管理
入退室時teamsに通知
入退室の際に音源再生
０時を超えて入室のままの場合，強制退出させる

開発環境

Ubuntu Desktop 24.04 LTS
NFCリーダー(sony RC-S380)
notion API
python
teams workflows(従来のteams webhook)

初期設定と参考にさせていただいた文献

notionのログインや表の作成方法は省略します．
上記の画像はtableデータベースを作成し，cardIDはhideにしています．

その他コードを作成する前に行った初期設定と参考にさせていただいた記事を添付します．

NFC初期設定

注意
nfc設定のコマンドを実行した後，再起動することを忘れずに！

notionAPI

teams Automate(旧webhook)

ファイル構成

基本的にmain.pyから色々な関数を呼び出しています．
全ての.pyファイルを実行した後にmain.pyを実行したら動くはず...

コードファイル	機能
main.py	NFCの読み取りから全体の流れ
notion.py	notionのデータベースの操作や検索などの命令
notification.py	teamsに送る通知の命令
.env	環境変数

コード

main.py

main.py

import nfc
import notion
import notification
from pygame import mixer
import threading
import time
import schedule
from datetime import datetime, timedelta

def mp3(filename): #音声
    mixer.init()
    mixer.music.load(filename)
    mixer.music.play()

def on_connect(tag):
    card_id = tag.identifier.hex() #カードからID情報を取り出す

    name, state, page_ID = notion.search_student(card_id) #名前，在室or退室，参照先のデータベースIDを名前から入手
    
    #新しいカードの設定
    if name is None: 
        name = input("Enter student name: ")
        notion.add_new_student(name, card_id)
        new_state = '在室'
        
    #既存のカード(入退室の切り替え)
    else:
        new_state = notion.change_state(page_ID, state) 
        #stateを切り替えるには，どの学生のページを編集するかの指定が必要なため，pageIDを使用しています．
        now = datetime.now()
        
        #音
        if new_state == '在室':
            mp3('in.mp3')
        else:
            mp3('out.mp3')
    
    #通知に使用する情報(累計人数)
    students_num = notion.count_students()
    notification.notification(name, new_state, students_num)

    return True

#0時になったらリセット
def schedule_job():
    schedule.every().day.at("00:00").do(notion.change_all_to_absent)
    while True:
        schedule.run_pending()
        time.sleep(60)
        

clf = nfc.ContactlessFrontend()  #nfcのおまじない
if clf.open('your_device_ID'):  #NFCのIDを入力
    schedule_thread = threading.Thread(target=schedule_job)
    schedule_thread.start()

    while True:
        try:
            clf.connect(rdwr={'on-connect': on_connect})
        except nfc.clf.TimeoutError:
            continue
        except KeyboardInterrupt:
            break
    clf.close()
else:
    print("Failed to open NFC device.")

notion.py

notion.py

import requests
from dotenv import load_dotenv
load_dotenv()
import os
import notification

# 必要変数の設定----それぞれ埋めてください(私は.envにまとめています)----------------------
class default:
    NOTION_API_KEY = os.environ['NOTION_API_KEY']
    DATABASE_ID = os.environ['DATABASE_ID']
    PAGE_ID = "null"
    GETurl = os.environ['GET_URL']
    editurl = os.environ['ADD_EDIT_URL'] + PAGE_ID
    addurl = os.environ['ADD_EDIT_URL']
    headers = {
        'Notion-Version': 'your_version',
        'Authorization': 'Bearer ' + NOTION_API_KEY,
        'Content-Type': 'application/json',
    }
# --------------------------------------------------------------

# 新規登録
def add_new_student(name, cardID):
    json_data = {
        'parent': { 'database_id': default.DATABASE_ID },
        'properties': {
            'Name': {
                'title': [
                    {
                        'text': {
                            'content': name
                        }
                    }
                ]
            },
            'State': {
                'multi_select': [
                    {
                        'name': '在室'
                    }
                ]
            },
            'ID': {
                'rich_text': [
                    {
                        'text': {
                            'content': cardID
                        }
                    }
                ]
            }
        },
    }
    response = requests.post(default.addurl, headers=default.headers, json=json_data)
    print("追加したよ")

# stateを変更する
def change_state(PAGE_ID, current_state):
    url = default.addurl + PAGE_ID
    new_state = '不在' if '在室' in current_state else '在室'

    json_data = {
        'properties': {
            'State': {
                'multi_select': [
                    {
                        'name': new_state
                    }
                ]
            }
        }
    }
    response = requests.patch(url, headers=default.headers, json=json_data)
    return(new_state)
    print("変更したよ")

# カードIDから名前を検索
def search_student(cardID):
    json_data = {
        'filter': {
            'property': 'ID',
            'rich_text': {
                'equals': cardID
            }
        }
    }
    response = requests.post(default.GETurl, headers=default.headers, json=json_data)
    response_json = response.json()
    results = response_json.get('results', [])

    # 既に登録されていればin/outを切り替え、なければ新たにデータベースを作成
    if results:
        return get_info(response_json)
    else:
        return None, None, None

# タグと名前とページIDのみを抜き出す
def get_info(response_json):
    for result in response_json.get('results', []):
        properties = result.get('properties', {})
        name = properties.get('Name', {}).get('title', [])
        tags = properties.get('State', {}).get('multi_select', [])
        page_id = result.get('id')

        # Nameのタイトル部分を抽出
        name_text = ''.join([text.get('plain_text', '') for text in name])

        # Tagsのタグ名を抽出
        tags_text = [tag.get('name', '') for tag in tags]

    return name_text, tags_text, page_id

#在室人数を確かめる
def count_students():
    json_data = {
        'filter': {
            'property': 'State',
            'multi_select' :{
                'contains': '在室'
            }
        }
    }
    response = requests.post(default.GETurl, headers=default.headers, json=json_data)
    response_json = response.json()
    results = response_json.get('results', [])
    count = len(results)
    return count

#0時に全員退室にする
def change_all_to_absent():
    count = count_students()
    if count!=0:
        json_data = {
            'filter': {
                'property': 'State',
                'multi_select': {
                    'contains': '在室'
                }
            }
        }
        response = requests.post(default.GETurl, headers=default.headers, json=json_data)
        response_json = response.json()
        notification.go_home()
        results = response_json.get('results', [])

        for result in results:
            page_id = result.get('id')
            current_state = result.get('properties', {}).get('State', {}).get('multi_select', [])
            current_state_names = [state.get('name') for state in current_state]
            change_state(page_id, current_state_names)

notification.py

notification.py


import pymsteams
from dotenv import load_dotenv
load_dotenv()
import os
import pymsteams
import requests

# 作成したIncoming WebhookのURLを指定
incoming_webhook_url = os.environ['TEAMS_URL']

def notification(name,state,num):

    if state == '在室':
        # 送信するメッセージ
        send_message = f"{name}が入室しました。<br>現在は"+str(num)+"人です。"
    else:
        send_message = f"{name}が退室しました。<br>現在は"+str(num)+"人です。"

    # メッセージを送信
    message = {
        "text":send_message
    }

    requests.post(incoming_webhook_url, json=message)

def go_home():
    send_message_title = "0時だよ！全員退散！"
    message = {
        "text":send_message_title
    }
    requests.post(incoming_webhook_url, json=message)

今後の展望

卒業するのでありません！！！！
apple payのモバイルICOCAでも動作するようにしたいですね．