[アドカレ2024]【IoT Core】ピアノの演奏に合わせて光るイルミネーションを作ってみた

はじめに

さてさて、クリスマスまで残り１週間となりました！
DENETアドカレ12月18日を担当させていただきます、新入社員のハヤシです。

毎年恒例（2回目）のDENETアドカレ企画に初参戦！と、いうことで
クリスマスっぽい内容を考えた結果、ピアノの演奏に合わせてLEDを光らせてみることにしました。

やってみた

構成図はこんな感じです。

ピアノの押された鍵盤をラズパイが読み取り、鍵盤に対応したLEDを光らせます。
またIoT Coreを用いてスマホからLEDの光り方をテーマによって変更させる機能も作成しました。

[使ったもの]
・Raspberry Pi（3B+）
・LEDストリップ（WS2812B）
・電子ピアノ

ピアノの鍵盤とLEDを連動させる

まずはラズパイとピアノ・LEDストリップを接続して、ピアノに対応したLEDを光らせてみます。
ピアノの押された鍵盤を認識して処理を実行させるには、MIDI信号を受信して解析する必要があります。

手始めに、ド＝１つ目、ド#＝２つ目、レ＝３つ目...のようにピアノの鍵盤と連動してLEDを光らせることができるかやってみます。
MIDI信号処理用には「mido」ライブラリを、LED制御には「rpi_ws281x」ライブラリを使用しました。

import mido
from rpi_ws281x import PixelStrip, Color

LED_COUNT = 60  #LEDの数
LED_PIN = 18
strip = PixelStrip(LED_COUNT, LED_PIN)
strip.begin()

midi_port_name = 'USB MIDI MIDI 1'  #実際のポート名を指定
midi_port = mido.open_input(midi_port_name)

note_to_led = {note: index for index, note in enumerate(range(60, 60 + LED_COUNT))}

def light_up_led(index, color):
    if 0 <= index < LED_COUNT:
        strip.setPixelColor(index, color)
        strip.show()

def turn_off_led(index):
    if 0 <= index < LED_COUNT:
        strip.setPixelColor(index, Color(0, 0, 0))
        strip.show()

print("Listening for MIDI messages...")
try:
    for msg in midi_port:
        if msg.type == 'note_on' and msg.note in note_to_led:
            light_up_led(note_to_led[msg.note], Color(255, 0, 0))
        elif msg.type == 'note_off' and msg.note in note_to_led:
            turn_off_led(note_to_led[msg.note])
except KeyboardInterrupt:
    print("Stopped listening.")
    for i in range(LED_COUNT):
        strip.setPixelColor(i, Color(0, 0, 0))
    strip.show()

これを書き込んで実際にピアノの鍵盤を押してみると...
※音量注意

無事鍵盤ごとに対応するLEDが光りました！

ここからはLEDの光り方が１種類だと味気ないので、テーマに合わせてスマホからLEDの色を簡単に制御できるよう設定していきます。

WEBからの操作

今回は４つのテーマ（きらきら・るんるん・ぬくぬく・しんしん）を作成しました。
スマホの画面はこんな感じ。

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>LED Control</title>
    <style>※省略※</style>
</head>
<body>
    <h1>LED Control</h1>
    <button onclick="changePattern('kira')">きらきら</button>
    <button onclick="changePattern('run')">るんるん</button>
    <button onclick="changePattern('nuku')">ぬくぬく</button>
    <button onclick="changePattern('shin')">しんしん</button>

    <script>
        async function changePattern(pattern) {
            try {
                const response = await fetch('https://xxx', {　#APIエンドポイント
                    method: 'POST',
                    headers: {
                        'Content-Type': 'application/json'
                    },
                    body: JSON.stringify({ pattern: pattern })
                });
                if (response.ok) {
                    alert('Pattern changed to ' + pattern);
                } else {
                    alert('Error changing pattern');
                }
            } catch (error) {
                console.error('Error:', error);
                alert('Connection failed');
            }
        }
    </script>
</body>
</html>

AWS側の設定

さてここからは、AWSの設定を進めていきます。

まずIoT Coreで新しくモノを作成して、証明書を自動生成します。
※この証明書は後から参照できないので、必ずダウンロードしておいてください

次にLambdaを作成します。トリガーには以下のように設定を追加してIoT Coreと紐づけます。

コードは以下です。
IoT Coreに、選択されたテーマをパブリッシュする処理を行っています。

import json
import boto3

iot_client = boto3.client('iot-data', region_name='ap-northeast-1')
valid_patterns = ["きらきら", "るんるん", "ぬくぬく", "しんしん"]

def lambda_handler(event, context):
    pattern = event.get('pattern', None)

    if pattern not in valid_patterns:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'Invalid pattern'})
        }

    try:
        iot_client.publish(
            topic='iot/topic',
            qos=1,
            payload=json.dumps({"action": "set_pattern", "pattern": pattern})
        )
        return {
            'statusCode': 200,
            'body': json.dumps({'message': f"Pattern '{pattern}' sent successfully!"})
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': f"Failed to send pattern: {str(e)}"})
        }

ラズパイの設定

最後にラズパイの設定です。

まずはIoT Core作成時に生成した証明書を適当なディレクトリ配下に設置しておきます。
そして以下コードを書き込みます。
テーマごとに光らせ方を設定し、IoT Coreより受け取ったメッセージに対応するテーマでLEDを光らせる処理を行っています。

import json
import mido
from rpi_ws281x import PixelStrip, Color
from awscrt.mqtt import QoS
from awsiot import mqtt_connection_builder

ENDPOINT = "xxx-ats.iot.ap-northeast-1.amazonaws.com"  #IoT Coreのエンドポイント
CLIENT_ID = "hayashi"  #IoT Coreの名前
PATH_TO_CERT = "/xxx.pem.crt"  #設置した証明書パス
PATH_TO_KEY = "/xxx.pem.key"  #設置した証明書パス
PATH_TO_ROOT = "/xxx.pem"  #設置した証明書パス
TOPIC_RECEIVE = "iot/topic"  #トピック名

LED_COUNT = 60
LED_PIN = 18
LED_FREQ_HZ = 800000
LED_DMA = 10
LED_BRIGHTNESS = 255
LED_INVERT = False
LED_CHANNEL = 0

strip = PixelStrip(LED_COUNT, LED_PIN, LED_FREQ_HZ, LED_DMA, LED_INVERT, LED_BRIGHTNESS, LED_CHANNEL)
strip.begin()
current_pattern = "きらきら"

patterns = {
    "きらきら": {
        "white_keys": {  #白鍵の色
            0: Color(255, 0, 0),  #ド = 赤
            2: Color(0, 0, 255),  #レ = 青
            4: Color(0, 255, 0),  #ミ = 緑
            5: Color(255, 255, 0),  #ファ = 黄
            7: Color(255, 0, 255),  #ソ = 紫
            9: Color(0, 255, 255),  #ラ = 水色
            11: Color(255, 165, 0),  #シ = オレンジ
        },
        "black_keys": Color(255, 255, 255),  #黒鍵 = 白
    },
    "るんるん": {
    ※以下略
    },
}

def get_led_color(note, pattern):
    key = note % 12
    is_black_key = key in {1, 3, 6, 8, 10}
    if is_black_key:
        return pattern["black_keys"]
    else:
        return pattern["white_keys"].get(key, Color(0, 0, 0))

def set_led_color(index, color):
    if 0 <= index < LED_COUNT:
        strip.setPixelColor(index, color)
    strip.show()

def clear_led():
    for i in range(LED_COUNT):
        strip.setPixelColor(i, Color(0, 0, 0))
    strip.show()

mqtt_connection = mqtt_connection_builder.mtls_from_path(
    endpoint=ENDPOINT,
    cert_filepath=PATH_TO_CERT,
    pri_key_filepath=PATH_TO_KEY,
    client_id=CLIENT_ID,
    ca_filepath=PATH_TO_ROOT,
    clean_session=False,
    keep_alive_secs=30,
)

print("Connecting to AWS IoT Core...")
connect_future = mqtt_connection.connect()
connect_future.result()
print("Connected!")

def on_message_received(topic, payload, **kwargs):
    global current_pattern
    print(f"Message received on topic {topic}: {payload}")
    try:
        message = json.loads(payload)
        if message.get("action") == "set_pattern":
            new_pattern = message.get("pattern")
            if new_pattern in patterns:
                current_pattern = new_pattern
                print(f"Pattern changed to: {current_pattern}")
            else:
                print("Invalid pattern received.")
    except Exception as e:
        print(f"Error processing message: {e}")

subscribe_future, _ = mqtt_connection.subscribe(
    topic=TOPIC_RECEIVE,
    qos=QoS.AT_LEAST_ONCE,
    callback=on_message_received
)
subscribe_future.result()

midi_port_name = 'USB MIDI MIDI 1'
midi_port = mido.open_input(midi_port_name)

try:
    for msg in midi_port:
        if msg.type == 'note_on' and msg.velocity > 0:
            note = msg.note
            color = get_led_color(note, patterns[current_pattern])
            set_led_color(note % LED_COUNT, color)
        elif msg.type == 'note_off':
            note = msg.note
            set_led_color(note % LED_COUNT, Color(0, 0, 0))
except KeyboardInterrupt:
    clear_led()
    disconnect_future = mqtt_connection.disconnect()
    disconnect_future.result()
    print("Disconnected!")

実際に処理を実行してみるとメッセージがちゃんと送られていることが確認できました。