Raspberry Pi PicoのMicroPythonをJupyter Notebookで書けるようにした

WindowsのパソコンとUSBで接続したPicoのMicroPtyhonをJupyter Notebookで書いて実行したかった。

MicroPythonのJupyterカーネルをインストールした。カーネルの候補は2つあった。

今回はjupyter_micropython_upydeviceをインストールした。どうしてこちらを選んだのか今となっては覚えていない。

pipパッケージをインストールする

> pip install jupyter-micropython-upydevice

MS C++ ビルドツールが必要だとエラーになった

Visual StudioインストーラーでC++ビルドツールをインストールした。

io.hが見つからないとエラーになった

Windows SDKをインストールする必要がある。使用しているのはWindows10なので、Visual StudioインストーラーでWindows 10 SDKをインストールした。

インストールしたコンポーネント

成功した。これでPython環境にカーネルモジュールのpipパッケージをインストールできた。次はJupyterにカーネルとして登録する必要がある。

Jupyterにカーネルを登録する

> python -m mpy_kernel_upydevice.install

こんなメッセージが表示される。

Installing IPython kernel spec of micropython
...into C:\Users\azechi\AppData\Roaming\jupyter\kernels\micropython-upydevice

notebookでPicoに接続する

USBシリアルのCOMポートを指定してPicoのMicroPythonと接続する

%serialconnect COM8

DiscordのBotをGoogle Cloud Compute Engineに移した

github.com

自分のDiscord Botがある、何の機能も実装していないただ常時オンラインでいるだけのボット。 Herokuの無料プランで稼動させていたが、Herokuは無料で使えなくなってしまった。

ボットを常時オンラインにしておくにはWebSocketの接続をずっと維持しておく必要がある。 接続を維持しときたいだけだからそういうクラウドサービスがあれば使いたい。 AWSAPI GatewayはWebSocketの接続を低料金の接続時間 + メッセージ課金でずっと維持してくれる。よさそう。 しかし、API GatewayはWebSocketのサーバーにしかなれないので、DiscordのWebSocketサーバーへは接続できない。惜しい。

VPSやEC2みたいにOSレベルからのサーバー利用か、Fargeteみたいなコンテナホスティングを利用してインスタンスを常時起動させておくみたいなことになる。 無料サーバーリソースを配ってるサービスはたくさんある。だいたい常時起動は制限されてるけど。 例えば、shuttleっていうサービスはDiscord Botのホストもできるよってホームページに書いてある。ここは常時起動させてくれるっぽい、たぶん。

どうせだからGoogle Cloudにした。 Google Cloud Compute Engineには無料枠がある。コンテナイメージでのデプロイもできる。

ボットのプログラムはGitHubにある。 GitHub ActionsでコンテナイメージをビルドしてHerokuにデプロイするとこまでやってくれるようにしてあった

Compute Engineへは公開レジストリからなら好きなところに置いたイメージをデプロイできる。 GitHubに無料で使えるコンテナレジストリghcr.ioがある。 GitHub Actionsでghcr.ioにプッシュ(アップロード)するところまで自動化した。

今回は、Compute Engineへデプロイするのは自動化しなった。

この一週間のこと。 先週までにDiscord のOAuth 2.0の仕組みを使ってDiscordの外のアプリケーションのサインアップ、ログインをサーバーを用意せずに無料で使えるリソースで作ってみる、っていうのをやってた。 先週、動くものができた。できたけど、動くものを作ってみたかっただけで自分では使い道がない。作る途中で出会った小さなtipsがいくつかあって、それを誰かに話したいと思っていた。 このブログや、zennやqiitaなどに書いてみようとも思っていた。んで、いざ動くものができてみたら、もう興味を失ってしまって、何もできなくなってしまった。いつもの突然のブロッキング現象だ。 メモは書きながら作業していたけど、ここでは、今思いつくだけのことを確認をとらずに書き出してみる。

  • discordはopenid connectに対応していないので、oauthのアクセストークンで@me apiにアクセスできることをもって認証とする
    • なので、アプリケーションにはサーバーサイドの処理が必要
      • サーバーサイドでの Authorization Code Flow でアクセストークンを受け取る
  • discordの OAuth2.0 エンドポイントは undocumented だけど PKCE に対応している
    • PKCE を利用すれば、client secret は不要
  • IDaaSとしてのFirebase Authentication
    • SMS確認機能を使わないなら無料
    • 無料プランでもユーザー数無制限
    • REST API は identity tool kit という名前が付いていて
    • SDKも identity tool kit という名前だったのが最近 Firebase Authに変更されていた
    • API の認証はサービスアカウントの秘密鍵の署名でできる
      • これは認証APIでアクセストークンを取得しなくてもいい、ということ
  • Firebase Functionsは従量制課金のプランでないと使えない
  • ...

Jupyter NotebookでGMail APIにアクセスするためのOAuth 2.0 アクセストークンをAuthorization Code Flowで取得する

Get OAuth 2.0 Refresh Tokens · GitHub

Jupyter Notebookで、一回きりのTCP接続の待ち受けをしてOAuth 2.0 のアクセストークンを取得する。

GMail APIに使用できるアクセストークンはデバイスフローでは取得できない。gmail関連のscopeを受け付けない。 Authorization Code Flow (認可コードフロー)のローカルリダイレクトを使ってNotebookで受け取る。

google api の場合、redirect_urilocalhostは事前に登録しなくても使用できる。ポート番号もauthorization リクエストのときにパラーメターで渡して決められるので、ランダムポートでよい。

Jupyter Notebookは既定でasyncioのループが実行されているので、asyncio.start_serverしてasyncio.wait_forで待てばいい。

async def get_authorization_code(scope, client_id, auth_uri, port=0, timeout=60):
    from asyncio import start_server, wait_for, Event, TimeoutError
    from urllib.parse import urlencode, urlparse, parse_qs

    event = Event()
    
    code: str
    async def handler(reader, writer):
        nonlocal code

        data = await reader.readline()
        data = data.decode()
        code = parse_qs(urlparse(data.split()[1]).query)["code"][0]
    
        writer.write(b"HTTP/1.1 200 OK\nContent-Length: 0\n\n")
        await writer.drain()
        writer.close()
        await writer.wait_closed()
        event.set()
    
    redirect_uri: str
    async with await start_server(handler, '0.0.0.0', port) as srv:
        _, port = srv.sockets[0].getsockname()
        redirect_uri = f"http://localhost:{port}"
        params = urlencode({
            "client_id": client_id,
            "scope": scope,
            "response_type": "code",
            "redirect_uri": redirect_uri
        })

        print(f"{auth_uri}?{params}")

        try:
            await wait_for(event.wait(), timeout=timeout)
        except TimeoutError:
            print("timeout")
            code = None

    print("server closed")
    return code, redirect_uri

認可コードを受信したときのレスポンスに204 No Contentではなくて200を返す。204ではブラウザで認証画面の表示が残ったままになってどうにも気持ち悪いから。

Cloudflare WorkersでHMAC付きCookieを使ったステートレスセッションを実装した

github.com

Cloudflare WorkersのランタイムはNodejsではない。Web Standards API (Crypto, atob, btoa, ... )で実装する必要がある。

有効期限を持ち、MACで改ざんされたら分かるようにしたトークンを発行する。このトークンをクッキーに設定して、ブラウザとサーバーでの連続したラウンドトリップ間にセッションを作る。

この実装ではHMACのハッシュアルゴリズムにはSHA256を選択した。

キーの長さは32bitで十分らしい timing attackへの対策、定数時間でのハッシュ値比較、web cryptoのhmac verifyはどうなってる?

Pythonのurllib.requestでリダイレクトをさせない

urllib.request.urlopenメソッドはレスポンスが300番台だった場合、リダイレクトを実行してそのレスポンスを結果とする。 HTTPのステータスコード302のレスポンスを返すAPIを作ってるときに、リダイレクトを実行せずに、最初に返ってきた302レスポンスを取得したかった。

既定のオープナーではレスポンスが200以外のステータスコードのときHTTPErrorProcessorによって各ハンドラーに振り分けられる。 docs.python.org urllib.request.HTTPErrorProcessor

今回は何か処理をしたいのではなくてレスポンスを結果として取得したいだけ。HTTPErrorProcessorでの処理をステータスコード200のときと同じようにすればいい。

build_openerにHTTPErrorProcessorのサブクラスを渡せば既定のHTTPErrorProcessorの動作を置き換えることができる。

from urllib.request import build_opener, HTTPErrorProcessor

class Handler(HTTPErrorProcessor):
    def http_response(self, request, response):
        return response
    
    https_response = http_response

opener = build_opener(Handler())

with opener.open(req) as res:
    print(res.code)
    print(res.headers)

参考にした