Lチカで始めるテスト自動化(18)秋月のIoT学習HATキットの圧電ブザーでテスト終了時にpass/failに応じてメロディを流す #RaspberryPi

1. はじめに

何か作業をしている横で自動テストを実行し、自動テストが終了しているかをときどきポーリングしながら確認するのは、作業に集中しづらいものと思います。そこでIoT学習HATキットに搭載されている圧電ブザーを利用して終了時にpass/failに応じてメロディを鳴らし、終了したことが音で分かるようにします。

秋月のIoT学習HATキットに搭載されている圧電ブザーを利用してテスト終了時にpass/failに応じてメロディを流すようにしてみた。これはpassしたとき。
※音量注意⚠️ pic.twitter.com/X7DQk5Vmiz
— ka’s (@pbjpkas) August 13, 2021

※上記ツイートを別のウィンドウやタブで開くとメヌエット、トッカータとフーガニ短調、エリーゼのためにを試聴できます。

2. メロディ再生プログラムの実装

IoT学習HATキットのサンプルプログラムに倣ってGPIOで矩形波信号を生成するSoftware Tone Libraryを利用します。Pythonで実装します。

2.1 音程と周波数の定義

Piano key frequenciesに掲載されている音程と周波数の対応表を参考に適宜定義します。休符(REST)は0[Hz]とします。Software Tone Libraryの仕様により上限は5,000[Hz]です。

例

#Note, Frequency
R   =   0 # rest
C3  = 131
CS3 = 139 # C♯3
（以下略）

2.2 Software Tone Libraryの初期化

IoT学習HATキットの圧電ブザーはPIN番号16、GPIO23に接続されています。

beepsound.py

import RPi.GPIO as GPIO
import wiringpi

# Akizuki IoT Learning HAT GPIO Number
BUZZER_PIN = 23 #PIN16, GPIO23

class BEEPSOUND:
    def __init__(self, tempo=120.0):
        GPIO.setwarnings(False)
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(BUZZER_PIN, GPIO.OUT)
        wiringpi.wiringPiSetupGpio()
        wiringpi.softToneCreate(BUZZER_PIN)
        self.tempo = tempo

2.3 テンポ設定

2.2節でテンポを120で初期設定していますが、任意の値に設定できるようにします。

呼び出し側

set_tempo(120)

呼ばれる側

from time import sleep

    def set_tempo(self, tempo):
        self.tempo = float(tempo)

2.4 発音

1音ごとに「音程」と「音符の長さ」を指定します。鍵盤楽器であれば鍵盤を「押す」「離す」があり、押しっぱなしにならないよう、押す：95%、離す：5%の時間配分としました。

呼び出し側

generate(C3,  4) # C3(ド)、4分音符

呼ばれる側

    def generate(self, note, length):
        if float(length) != 0:
            duration = float( (4/float(length)) * (60.0/self.tempo) )
            wiringpi.softToneWrite(BUZZER_PIN, int(note))
            sleep(duration*0.95)
            wiringpi.softToneWrite(BUZZER_PIN, 0)
            sleep(duration*0.05)

2.5 楽譜の記述

発音の呼び出し側を「1音＝1行」で記述します。

例

generate(C3,  4) # C3(ド)、4分音符
generate(D3,  4) # D3(レ)、4分音符
generate(E3,  4) # E3(ミ)、4分音符
generate(R,   4) # 4分休符

「音程」と「音符の長さ」の2次元配列を作ってそのアドレス(ポインタ)を引数で渡すような作りにするほどでもないなぁと思い、「1音＝1行」でべた書きとしました。

2.6 楽譜

以下の資料を参考にさせていただきました。pass、failの合計2曲あれば十分なのですが試作を兼ねて3曲あります。

2.7 メロディ再生プログラムの全文

以下にメロディ再生プログラムの全文を示します。

beepsound.py

# -*- coding: utf-8 -*-
# beepsound.py
# python 3.x
#
# This software includes the work that is distributed
# in the Apache License 2.0
# ka's@pbjpkas 2021
#

from time import sleep
import RPi.GPIO as GPIO
import wiringpi

# Akizuki IoT Learning HAT GPIO Number
BUZZER_PIN = 23 #PIN16, GPIO23

#Note, Frequency
R   =   0 #rest
C3  = 131
CS3 = 139
D3  = 147
DS3 = 156
E3  = 165
F3  = 175
FS3 = 185
G3  = 196
GS3 = 208
A3  = 220
AS3 = 233
B3  = 247
C4  = 262
CS4 = 277
D4  = 294
DS4 = 311
E4  = 330
F4  = 349
FS4 = 370
G4  = 392
GS4 = 415
A4  = 440
AS4 = 466
B4  = 494
C5  = 523


class BEEPSOUND:
    def __init__(self, tempo=120.0):
        GPIO.setwarnings(False)
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(BUZZER_PIN, GPIO.OUT)
        wiringpi.wiringPiSetupGpio()
        wiringpi.softToneCreate(BUZZER_PIN)
        self.tempo = tempo

    def set_tempo(self, tempo):
        self.tempo = float(tempo)

    def generate(self, note, length):
        if float(length) != 0:
            duration = float( (4/float(length)) * (60.0/self.tempo) )
            wiringpi.softToneWrite(BUZZER_PIN, int(note))
            sleep(duration*0.95)
            wiringpi.softToneWrite(BUZZER_PIN, 0)
            sleep(duration*0.05)

    # Für Elise - Ludwig van Beethoven
    # https://ja.wikipedia.org/wiki/エリーゼのために
    def FurElise(self):
        self.generate(E4, 16)
        self.generate(DS4,16)

        self.generate(E4, 16)
        self.generate(DS4,16)
        self.generate(E4, 16)
        self.generate(B3, 16)
        self.generate(D4, 16)
        self.generate(C4, 16)

        self.generate(A3,  8)
        self.generate(R,  16)
        self.generate(C3, 24)
        self.generate(E3, 24)
        self.generate(A3, 24)

        self.generate(B3,  8)
        self.generate(R,  16)
        self.generate(E3, 24)
        self.generate(GS3,24)
        self.generate(B3, 24)

        self.generate(C4,  8)
        self.generate(R,  16)

    # Menuet - Christian Petzold
    # https://ja.wikipedia.org/wiki/メヌエット (ペツォールト)
    def Menuet(self):
        self.generate(D4,  4)
        self.generate(G3,  8)
        self.generate(A3,  8)
        self.generate(B3,  8)
        self.generate(C4,  8)
        self.generate(D4,  4)
        self.generate(G3,  4)
        self.generate(G3,  4)

        self.generate(E4,  4)
        self.generate(C4,  8)
        self.generate(D4,  8)
        self.generate(E4,  8)
        self.generate(FS4, 8)
        self.generate(G4,  4)
        self.generate(G3,  4)
        self.generate(G3,  4)

    # Toccata und Fuge in d-Moll BWV565 - Johann Sebastian Bach
    # https://www.gmajormusictheory.org/Freebies/Level2/2ToccataFugueDm/2ToccataFugueDm.pdf
    def ToccataUndFugeInDMoll(self):
        self.generate(A4, 16)
        self.generate(G4, 16)
        #self.generate(A4,  8)
        self.generate(A4,  2)
        self.generate(G4, 16)
        self.generate(F4, 16)
        self.generate(E4, 16)
        self.generate(D4, 16)
        self.generate(CS4, 4)
        self.generate(D4,  2)
        self.generate(R ,  4)


if __name__ == '__main__':
    beepsound = BEEPSOUND()
    beepsound.set_tempo(60)
    beepsound.FurElise()

    beepsound.set_tempo(120)
    beepsound.Menuet()
    beepsound.ToccataUndFugeInDMoll()

3. テストランナーへの組み込み

テストランナーへ以下の改修を行います。

beepsound.pyをimportする
BEEPSOUND()を実行してインスタンスを生成する
楽譜データのデバッグのためにbeepsoundコマンドを追加する
failして終了する箇所に「エリーゼのために」を追記する
passして終了する箇所に「メヌエット」を追記する

「Lチカで始めるテスト自動化(16)秋月のIoT学習HATキットにBME280を接続してテスト実行環境の温度・湿度・気圧を取得する」の付録.A test-runner.pyのソースとの差分を以下に示します。ソース全文は付録Aをご覧ください。

diff

25a26
> from beepsound import BEEPSOUND
230a232
>     beepsound = BEEPSOUND()
365a368,380
>                 elif cmd[0] == "beepsound":
>                     if cmd[1] == "FurElise":
>                         beepsound.set_tempo(60)
>                         beepsound.FurElise()
>                     elif cmd[1] == "Menuet":
>                         beepsound.set_tempo(120)
>                         beepsound.Menuet()
>                     elif cmd[1] == "ToccataUndFugeInDMoll":
>                         beepsound.set_tempo(120)
>                         beepsound.ToccataUndFugeInDMoll()
>                     else:
>                         is_passed = False
> 
409d423
<                     led_fail()
410a425,427
>                     led_fail()
>                     beepsound.set_tempo(60)
>                     beepsound.FurElise()
416d432
<         led_pass()
417a434,436
>         led_pass()
>         beepsound.set_tempo(120)
>         beepsound.Menuet()

4. おわりに

Software Tone Libraryのおかげで外付けのメロディICを使用せずにメロディを鳴らすことができました。
トッカータとフーガニ短調はfailして終了というよりもゲームオーバーな印象が強く、選曲にセンスが問われると思いました。
任意の音を鳴らせるようになったことでエラーメッセージをモールス符号で発音するような応用もできそうと思いました。

付録.A test-runner.pyのソース

test-runner.py

#!/usr/bin/python3

#
# This software includes the work that is distributed
# in the Apache License 2.0
#

from time import sleep
import random
import sys
import codecs
import csv
import datetime
#import serial
#import pyvisa as visa
#import cv2
#from PIL import Image
#import pyocr
#import pyocr.builders
#import platform
import subprocess
from subprocess import PIPE
from AKI_I2C_AQM0802A import AKI_I2C_AQM0802A
import RPi.GPIO as GPIO
import bme280
from beepsound import BEEPSOUND

UNINITIALIZED = 0xdeadbeef
NUM_OF_SERVO = 6

# Akizuki IoT Learning HAT GPIO Number
LED_GR = 22
LED_Y = 27
LED_R = 17
LED_LCD_BACKLIGHT = 26

def serial_write(h, string):
    if h != UNINITIALIZED:
        string = string + '\n'
        string = str.encode(string)
        h.write(string)
        return True
    else:
        print("UART Not Initialized.")
        return False

def close_uart(h):
    if h != UNINITIALIZED:
        h.close()
    else:
        #print("UART Not Initialized.")
        pass

def open_dso():
    rm = visa.ResourceManager()
    resources = rm.list_resources()
    #print(resources)
    for resource in resources:
        #print(resource)
        try:
            dso = rm.open_resource(resource)
        except:
            print(resource, "Not Found.")
        else:
            print(resource, "Detected.")
            return dso

    #Throw an error to caller if none succeed.
    return dso

def open_cam(camera_number, width, height):
    h = cv2.VideoCapture(camera_number)
    h.set(cv2.CAP_PROP_FRAME_WIDTH, width)
    h.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
    return h

def close_cam(cam):
    if cam != UNINITIALIZED:
        cam.release()

def capture_cam(cam, filename):
    if cam != UNINITIALIZED:
        _, img = cam.read()
        cv2.imwrite(filename, img)
        return True
    else:
        print("CAM Not Ready.")
        return False

def crop_img(filename_in, v, h, filename_out):
    img = cv2.imread(filename_in, cv2.IMREAD_COLOR)
    v0 = int(v.split(':')[0])
    v1 = int(v.split(':')[1])
    h0 = int(h.split(':')[0])
    h1 = int(h.split(':')[1])
    img2 = img[v0:v1, h0:h1]
    cv2.imwrite(filename_out, img2)
    return True

def open_ocr():
    ocr = pyocr.get_available_tools()
    if len(ocr) != 0:
        ocr = ocr[0]
    else:
        ocr = UNINITIALIZED
        print("OCR Not Ready.")
    return ocr

def exec_ocr(ocr, filename):
    try:
        txt = ocr.image_to_string(
            Image.open(filename),
            lang = "eng",
            builder = pyocr.builders.TextBuilder()
        )
    except:
        print("OCR Fail.")
    else:
        return txt

def exec_labelimg(filename, label_string):
    if platform.system() == "Windows" :
        python = "python"
        grep = "findstr"
    else:
        python = "python3"
        grep = "grep"

    cmd = python + \
          " label_image.py \
          --graph=c:\\tmp\\output_graph.pb \
          --labels=c:\\tmp\\output_labels.txt \
          --input_layer=Placeholder \
          --output_layer=final_result \
          --image=" + filename \
          + "|" + grep + " " + label_string
    print(cmd)
    log = subprocess.run(cmd, stdout=subprocess.PIPE, shell=True)
    ret = log.stdout.strip().decode("utf-8").split(" ")[1]
    return ret

def set_servo(uart, servo_id, servo_pos):
    if servo_id < 0 or servo_id >= NUM_OF_SERVO:
        print("Invalid Servo ID")
        return False

    if servo_pos < 0 or servo_pos > 180:
        print("Invalid Servo Position")
        return False

    if uart != UNINITIALIZED:
        serial_write(uart, "servoread")

        servo_position = uart.readline().strip().decode('utf-8')
        print(servo_position)

        # discard "OK"
        devnull = uart.readline().strip().decode('utf-8')

        current_pos = int(servo_position.split(' ')[servo_id])
        #print(servo_id, current_pos)

        if current_pos < servo_pos:
            start = current_pos +1
            stop  = servo_pos +1
            step  = 1
        else:
            start = current_pos -1
            stop  = servo_pos -1
            step  = -1

        for i in range(start, stop, step):
            command = "servo " + str(servo_id) + " " + str(i)
            print(command)
            serial_write(uart, command)
            # discard "OK"
            devnull = uart.readline().strip().decode('utf-8')
            sleep(0.2) # sec.

        return True
    else:
        print("UART Not Initialized.")
        return False

def led_init():
    GPIO.setmode(GPIO.BCM)

    GPIO.setwarnings(False)
    GPIO.setup(LED_GR, GPIO.OUT)
    GPIO.setup(LED_Y, GPIO.OUT)
    GPIO.setup(LED_R, GPIO.OUT)
    GPIO.setup(LED_LCD_BACKLIGHT, GPIO.OUT)

    GPIO.output(LED_GR, False)
    GPIO.output(LED_Y, True)
    GPIO.output(LED_R, False)
    GPIO.output(LED_LCD_BACKLIGHT, True)

def led_pass():
    GPIO.output(LED_GR, True)
    GPIO.output(LED_Y, False)
    GPIO.output(LED_R, False)

def led_fail():
    GPIO.output(LED_GR, False)
    GPIO.output(LED_Y, False)
    GPIO.output(LED_R, True)

def lcd_init():
    lcd = AKI_I2C_AQM0802A()
    lcd.Init_LCD()
    lcd.ClearDisplay()
    lcd.NewClearDisplay(0, 0)
    lcd.NewClearDisplay(1, 0)
    return lcd

def lcd_str(lcd, row, col, str):
    lcd.NewClearDisplay(row, 0)
    lcd.WritePos(row, col)
    lcd.WriteStr(str)

def main():
    is_passed = True
    val = str(UNINITIALIZED)
    fval = 0.0
    uart = UNINITIALIZED
    dso = UNINITIALIZED
    cam = UNINITIALIZED
    ocr = UNINITIALIZED
    led_init()
    lcd = lcd_init()
    beepsound = BEEPSOUND()

    if len(sys.argv) == 2:
        script_file_name = sys.argv[1] + ".csv"
        result_file_name = sys.argv[1] + "_result.csv"
    else:
        script_file_name = "script.csv"
        result_file_name = "result.csv"

    script_total_line = subprocess.check_output(['wc', '-l', script_file_name]).decode().split(' ')[0]
    lcd_str(lcd, 0, 0, script_total_line)

    with codecs.open(script_file_name, 'r', 'utf-8') as file:
        script = csv.reader(file, delimiter=',', lineterminator='\r\n', quotechar='"')

        with codecs.open(result_file_name, 'w', 'utf-8') as file:
            result = csv.writer(file, delimiter=',', lineterminator='\r\n', quotechar='"')

            script_line = 0
            for cmd in script:
                timestamp = datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
                script_line += 1
                lcd_str(lcd, 1, 0, str(script_line))
                print("########## " + str(script_line) + " ##########")
                print(cmd)

                if "#" in cmd[0]:
                    pass

                elif cmd[0] == "sleep":
                    sleep(float(cmd[1]))

                elif cmd[0] == "rusleep":
                    fval = random.uniform(float(cmd[1]), float(cmd[2]))
                    cmd.append(str(fval))
                    sleep(fval)

                elif cmd[0] == "open_uart":
                    if len(cmd) == 2:
                        dsrdtr_val = 1
                    else:
                        dsrdtr_val = int(cmd[2])
                    try:
                        uart = serial.Serial(cmd[1], 115200, timeout=1.0, dsrdtr=dsrdtr_val)
                    except:
                        is_passed = False

                elif cmd[0] == "send":
                    ret = serial_write(uart, cmd[1])
                    if ret == False:
                        is_passed = False

                elif cmd[0] == "rcvd":
                    try:
                        val = uart.readline().strip().decode('utf-8')
                        cmd.append(val)
                    except:
                        is_passed = False

                elif cmd[0] == "open_dso":
                    try:
                        dso = open_dso()
                    except:
                        is_passed = False

                elif cmd[0] == "dso":
                    try:
                        if "?" in cmd[1]:
                            val = dso.query(cmd[1]).rstrip().replace(",", "-")
                            cmd.append(val)
                        else:
                            dso.write(cmd[1])
                    except:
                        is_passed = False

                elif cmd[0] == "open_cam":
                    if len(cmd) == 2:
                        cam = open_cam(int(cmd[1]), 640, 480)
                    else:
                        cam = open_cam(int(cmd[1]), int(cmd[2]), int(cmd[3]))

                elif cmd[0] == "close_cam":
                    close_cam(cam)
                    cam = UNINITIALIZED

                elif cmd[0] == "capture_cam":
                    ret = capture_cam(cam, cmd[1])
                    if ret == False:
                        is_passed = False

                elif cmd[0] == "crop_img":
                    crop_img(cmd[1], cmd[2], cmd[3], cmd[4])

                elif cmd[0] == "open_ocr":
                    ocr = open_ocr()
                    if ocr == UNINITIALIZED:
                        is_passed = False

                elif cmd[0] == "exec_ocr":
                    try:
                        val = exec_ocr(ocr, cmd[1])
                    except:
                        is_passed = False
                    else:
                        cmd.append(str(val))

                elif cmd[0] == "exec_labelimg":
                    try:
                        val = exec_labelimg(cmd[1], cmd[2])
                    except:
                        is_passed = False
                    else:
                        cmd.append(str(val))

                elif cmd[0] == "set_servo":
                    ret = set_servo(uart, int(cmd[1]), int(cmd[2]))
                    if ret == False:
                        is_passed = False

                elif cmd[0] == "run":
                    ret = subprocess.run(cmd[1], shell=True, stdout=PIPE, stderr=PIPE, universal_newlines=True)
                    val = ret.stdout.strip()
                    print(ret)
                    if ret.returncode != 0:
                        is_passed = False

                elif cmd[0] == "env":
                    try:
                        val = bme280.readData(int(cmd[1],16))
                    except:
                        is_passed = False
                    else:
                        val = val.split(',')
                        for i in val:
                            cmd.append(i)

                elif cmd[0] == "beepsound":
                    if cmd[1] == "FurElise":
                        beepsound.set_tempo(60)
                        beepsound.FurElise()
                    elif cmd[1] == "Menuet":
                        beepsound.set_tempo(120)
                        beepsound.Menuet()
                    elif cmd[1] == "ToccataUndFugeInDMoll":
                        beepsound.set_tempo(120)
                        beepsound.ToccataUndFugeInDMoll()
                    else:
                        is_passed = False

                elif cmd[0] == "eval_str_eq":
                    if str(val) != str(cmd[1]):
                        is_passed = False

                elif cmd[0] == "eval_int_eq":
                    if int(val) != int(cmd[1]):
                        is_passed = False

                elif cmd[0] == "eval_int_gt":
                    if int(val) < int(cmd[1]):
                        is_passed = False

                elif cmd[0] == "eval_int_lt":
                    if int(val) > int(cmd[1]):
                        is_passed = False

                elif cmd[0] == "eval_dbl_eq":
                    if float(val) != float(cmd[1]):
                        is_passed = False

                elif cmd[0] == "eval_dbl_gt":
                    if float(val) < float(cmd[1]):
                        is_passed = False

                elif cmd[0] == "eval_dbl_lt":
                    if float(val) > float(cmd[1]):
                        is_passed = False

                else:
                    cmd.append("#")

                if is_passed == True:
                    cmd.append("OK")
                    cmd.insert(0,timestamp)
                    print(cmd)
                    result.writerow(cmd)
                else:
                    cmd.append("NG")
                    cmd.insert(0,timestamp)
                    print(cmd)
                    result.writerow(cmd)
                    close_uart(uart)
                    close_cam(cam)
                    print("FAIL")
                    led_fail()
                    beepsound.set_tempo(60)
                    beepsound.FurElise()
                    sys.exit(1)

    if is_passed == True:
        close_uart(uart)
        close_cam(cam)
        print("PASS")
        led_pass()
        beepsound.set_tempo(120)
        beepsound.Menuet()
        sys.exit(0)

main()

付録B. Lチカで始めるテスト自動化・記事一覧

電子書籍化したものを技術書典で頒布しています。
1. Lチカで始めるテスト自動化
2. Lチカで始めるテスト自動化 -2- リレー駆動回路の設計 (※書き下ろし)
3. Lチカで始めるテスト自動化 -3- Raspberry Pi Zero WHとIoT学習HATキットで作るテストランナー