概要

人「NFCを読み取るアプリ欲しい」
蝦「はい」
人「見た目はリッチにしたい」
蝦「はい」
人「NFCを読むところまではPythonでできたのでいい感じにしてほしい」
蝦「はい」

動作環境

ASUS Chromebook C434TA +crostini(Debian)
python3.7以降(asyicioを使用します。)
NFCリーダー: RC-S380/S

基本構成

Uvicornでhttpリクエストの受付
FastAPIでリクエストを処理する。
nfcpyでUSB接続したNFCリーダーを操作する。

実際のコード

読み取り処理部分

読み取り部分は至極単純。Pythonにはアロー関数が無いので関数内で関数を宣言している。

nfc.py

import nfc

# NFCでデータを読み込むための関数
def read_nfc_tag(clf):
    data = {}
    # アロー関数的な処理
    def connected(tag):
        # スコープの扱いに注意
        nonlocal data
        data = tag
    # 接続したらこの行で止まる
    clf.connect(rdwr ={'on-connect': connected})
    # 読み込んだらconnected関数が実行されて次の行でreturn
    return data

非同期処理で使えるようにする

上で作った関数をasyncioで同期処理にする。NFCリーダーに接続したままだと次のリクエストを受付できないので、finaly節で必ず接続を閉じる。

nfc.py

import asyncio

# 上の読み込み処理を非同期処理にした関数
async def read_nfc_tag_async():
    clf=  nfc.ContactlessFrontend('usb')    
    d={}
    try:
        # イベントループを取得して
        loop = asyncio.get_running_loop()
        # そこに非同期にしたい処理を入れる感じ
        d = await loop.run_in_executor(None, read_nfc_tag,clf)
    finally:
        # 必ず接続を閉じる
        print("close")
        clf.close()
    return d

FastAPIでAPI化

/nfcをエンドポイントにする。asyncioのwait_forメソッドで、一定時間返答が無ければ例外を投げるようにする。
読み込み待機中にリクエストが来た場合はNFCリーダーに接続できないため例外が発生する。

app.py

from fastapi import FastAPI

import asyncio
from nfc_run import read_nfc_tag_async

app = FastAPI()

@app.get("/nfc")
async def nfc():
    try:
        res= await asyncio.wait_for(read_nfc_tag_async(), timeout=5)
        return {"result":str(res)}
        
    except asyncio.TimeoutError:
        return {"result":"timeout"}

if __name__ == "__main__":
    app.run()

実際に呼び出す場合

uvicornでappを起動させ、httpリクエストを受け付ける

uvicorn app:app

curlでAPIをたたいてみる。今回はポート指定が無いので"localhost:8000/nfc"にGETリクエストを送れば動くはず

curl http://localhost:8000

読込できれば成功。

HTML上から呼び出してみる

FastAPIからhtmlを配信させて、そこからAPIでリクエストを投げてみる。

app.py

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import RedirectResponse

import asyncio

from nfc_run import read_nfc_tag_async

app = FastAPI()

@app.get("/")
def index_redirect():
    return RedirectResponse("/index.html")


@app.get("/nfc")
async def nfc():
    try:
        res= await asyncio.wait_for(read_nfc_tag_async(), timeout=5)
        return {"result":str(res)}
        
    except asyncio.TimeoutError:
        return {"result":"待ち受け時間を超過しました。"}
        
app.mount("/", StaticFiles(directory="static"), name="static")

if __name__ == "__main__":
    app.run()

HTMLのコードはこんな感じ。axiosで/nfcにリクエストを投げるだけ

index.html

<!DOCTYPE html>
<html lang="ja">
    <head>
        <meta charset="UTF-8" />
        <meta http-equiv="X-UA-Compatible" content="IE=edge" />
        <meta name="viewport" content="width=device-width, initial-scale=1.0" />
        <script src="https://cdn.jsdelivr.net/npm/axios/dist/axios.min.js"></script>
        <title>NFCで読み込むやつ</title>
    </head>
    <body>
        <form id="nfc">
            <p>下のボタンを押すと読み込みを開始します。待機時間は5秒です。</p>
            <input type="submit" value="NFC読み込み開始" />
        </form>
        <p id="result">ボタンを押すと読み込みを待機します。</p>
    </body>
    <script>
        const form = document.querySelector('#nfc');
        const result = document.querySelector('#result');

        form.addEventListener('submit', (e) => {
            result.textContent = '読み込み待ちです。';
            // フォームの送信イベントを止めて
            e.preventDefault();
            // axiosでのHTTPリクエストに切り替える
            axios
                .get('/nfc')
                .then((res) => {
                    console.log(res);
                    const msg = res?.data?.result;
                    result.textContent = msg;
                })
                .catch((e) => {
                    console.log(e);
                });
        });
    </script>
</html>

色々応用できそう

NFC以外にも標準入力もAPIで呼び出せるので、使い方次第では色々できそう。

input.py

# input関数を非同期処理化した関数
async def aio_input():
    loop = asyncio.get_running_loop()
    d = await loop.run_in_executor(None,input)
    loop.close()
    return d

# 上記の独自関数をから受け取った文字列を返すAPI
@app.get("/input")
async def input_original():
    try:
        res_str= await asyncio.wait_for(aio_input(), timeout=10)
        return res_str
    except asyncio.TimeoutError:
        return {"result":"待ち受け時間を超過しました。"}

あとは以前書いたジェネレータ関数でリクエストを継続的に投げる処理を使えば、見た目上の待機時間を制御できるようになる。今回は5秒でタイムアウトなので、例えば3回リクエストを投げるなら15秒くらい待っているように見える。(実際は接続を閉じたりレスポンスを返すまでの時間があるのでずれる。)

あとがき

だいぶ前に作ったサンプルを元に記事を書いたので、動かなかったらコメントください。
あとコードに問題があってもコメントください。

リポジトリ

Github

注意

意図しない場所からリクエストが来ないように気を付けましょう。