DIYで作る電子ペーパー額縁に表示する内容を外から変える(AWS編)

2023年4月22日

DIYで作る電子ペーパー額縁の作り方(ハード編)、DIYで作る電子ペーパー額縁の作り方(ソフト編)、DIYで作る電子ペーパー額縁に表示する内容を外から変える(cgi編) の続き

とりあえずこれでラズパイの外から変更できるようにはなりましたが、ラズパイと同一ネットワーク内からでないと変更できません。

というのを解決するためにAWS側に表示内容を置くことにしました。図にするとこんな感じです。

前回はローカルにあるjsonファイルをポーリングしていましたが、そこをS3においてあるjsonファイルをポーリングする形にします。

その上で、S3においてあるjsonファイルをlambdaで更新できるようにしておきます。

ということでコードですが、今回もまた今となっては過去のコードを思い出しながら書いてるので動かなかったらゴメンナサイ。実装イメージと思ってもらえたら幸い。

Rapberry Pi の Pythonによる電子ペーパーの描画

基本的には読み込むjsonファイルをS3側にしたのみです。あと念のため、画像ファイルに変なパスが入ってきた時に動作しないようチェック処理を噛ましています。

S3側では適切なアクセス制限のもとアクセスできるようにポリシーを設定しておきます。私はIPアドレス制限をかけておきました。固定IPではないですが外側IPは滅多に変わらないので。

まー、別に見られても画像ファイル名が入ってるだけなので全然困らないですけどね。

from PIL import Image, ImageDraw, ImageFont

from waveshare_epd import epd7in5b_V2

epd = epd7in5b_V2.EPD()

last_images = []
def display(loop):
    global last_images

    images = json.loads(requests.get("[S3のJSONファイルへのURL]"))

    if images["BnWImage"] != last_image["BnWImage"] or images["RedImage"] != last_image["RedImage"]:
        Himage_bnw = Image.new('1', (epd.width, epd.height), 255)
        Himage_red = Image.new('1', (epd.width, epd.height), 255)
        if images["BnWImage"] != "":
            # 変なパスを指定されておかしくなるのは困るので、一応ディレクトリトラバーサルの
            # チェックをして特定ディレクトリ下のファイルのみを対象とします。
            if os.path.commonprefix((os.path.realpath(images["BnWImage"] ), '/hoge/hoge/')) != "/hoge/hoge/":
                raise ValueError("traversal check error.")
            img = Image.open(images["BnWImage"])
            Himage_bnw.paste(img, (0,0))
        if images["RedImage"] != "":
            if os.path.commonprefix((os.path.realpath(images["RedImage"] ), '/hoge/hoge/')) != "/hoge/hoge/":
                raise ValueError("traversal check error.")
            img = Image.open(images["RedImage"])
            Himage_red.paste(img, (0,0))
        epd.init()
        epd.display(epd.getbuffer(Himage_bnw), epd.getbuffer(Himage_red))
        epd.sleep()
        last_images = images

    # 10秒に1回チェック
    loop.call_later(10, check_calendar, loop)


# メインループ
loop = asyncio.get_event_loop()
loop.call_soon(display_image, loop)

lambda 側

JavaScript で作成しました。今考えたらPythonで作った方がよかったですね。全体的に言語が統一できて。

受け取ったjsonをs3に出力しています。

//@ts-check
"use strict";

const { S3Client, PutObjectCommand } = require("@aws-sdk/client-s3");

module.exports.updateDisplayImageJson = async (event) => {
    // isAuthorized 関数はここには書いてませんが適切な感じで。
    if (isAuthorized(event) === false) {
        console.log("unauthorized access.");
        return {
            statusCode: 403,
        };
    }

    let body;

    // API Gateway で REST にするか http にするかで変わってくるのですが、どちらでも動くようにこう書いてます。
    if (event.isBase64Encoded) {
        body = JSON.parse(Buffer.from(event.body, "base64").toString());
    } else {
        body = JSON.parse(event.body);
    }

    if ("status" in body && "BnWImage" in body && "RedImage" in body) {
        const s3Client = new S3Client({});
        await s3Client.send(
            new PutObjectCommand({
                Bucket: "バケットの名前",
                Key: "jsonファイルのあるパス/display_image.json",
                ContentType: "application/json",
                Body: JSON.stringify(body),
            })
        );
    }

    return {
        statusCode: 200,
        body: "OK",
    };
};

画像の定期的な変更

今回から画像を一定時間ごとに変更する処理を追加しました。

status_image_pattern.json というファイルに複数の画像ファイルを配列で記述したjsonを準備しておき、そこからランダムに選択した画像ファイルの情報をlambdaに渡します。

同じPythonファイル処理内で処理されているのに、一度 AWS を経由するという遠回りでの画像変更w

last_change = time.time()
def randomchange(loop):
    global last_change, images, status_image_pattern

    if (time.time() - last_change) > 900:
        # 前回の変更から15分経過してる時は気分転換に絵を変える
        max_image_count = len(status_image_pattern[images["status"]]) - 1
        while True:
            newimage = status_image_pattern[images["status"]][random.randint(0, max_image_count)]
            if (images["BnWImage"] != newimage["BnWImage"] or images["RedImage"] != newimage["RedImage"]):
                # 画像ファイルが前回と異なれば抜ける。jsonで比較すると一致しない場合があるので要素で比較
                break
        new_image_json = json.dumps(status_image_pattern[images["status"]][random.randint(0, max_image_count)])
        # lambda を呼び出して更新
        requests.post(credentials["url"], headers=credentials["headers"], data=new_image_json)
        last_change = time.time()

    loop.call_later(5, randomchange, loop)

ここに書いてないクレデンシャルの配列を参照してたりするのでそこは適切に要変更です。

IFTTT の設定

画像の変更がlambdaで可能になったのでIFTTTからの変更ができるようになります。

If は SwitchBot なり Webhooks なりで適切にイベントを設定してもらったあとで、Thenはこれ。

設定はこんな感じです。

Header の設定もできるのでセキュリティ的な何かをしている時も指定することができます。

IfをSwitchBot plug にして仕事PCをそのプラグに繋いでおき、電流値の変更をトリガーにして上記の「Then」を実行したら電子ペーパーの画像が「仕事中」に勝手に変わるなんてこともできます。

本当はリアルタイムで変更をやりたかったけど…

今、S3をポーリングして画像の変更を検知しているのですが、これをなんとかできなかなーと思ってます。

リアルタイムで通知するとしたら、WebSocket とかでしょうか。API Gateway を挟んだらできるようなのですが未着手です。

S3、数秒感覚程度のポーリングだったらたいしてお金かからないですし、慌てて画像を変更する必要もありませんしね…。

ということで、電子ペーパー額縁ネタはいったん終了です。

Raspberry Pi

Posted by toshyon