RaspberryPiで監視をお助け
このシリーズでは、RaspberryPiを使ったお知らせランプを製作します。
第3回の今回は、「お知らせランプ」の全体制御を行うメイン部分です。
このシリーズは、今回で完成となります。
Abu(アブー)登場
Detector経由でOpsGenieのAlert情報を取得して、監視対象の状態を特定し、LED表示を設定する仕事は、Alert Buster(略してAbu)が担います。
新たなAlertが発生した時に一度だけチャイムを鳴らすのもAbuの仕事です。
同時に複数存在するかもしれないAlertの状態に対し「お知らせランプ」が表示できるのは一つの状態のみ。
そこで、状態を優先度付けして、最も優先度の高い状態を表示します。
Abuが扱う状態と表示の関係
Abuが扱う状態は、以下の6種類です。
レベルは「エラー」と「ワーニング」の2種類。
「エラー」は急いで対処するもの、「ワーニング」は急ぎではないけど、確認が必要なものを表します。
それぞれ、認知済みかどうかの2状態があります。
「不明」は、起動直後やネットワークの不具合などで、状態がわからない時に使用します。
各状態に割り当てるべき色や、明滅周期も決めておきます。
なお、"Magenta"などの色の名称は、前回の記事のLed.COLORSで定義しているものを使います。
| 優先度 | Alert種別 | 認知状態 | 状態値 | 色 | 明滅周期 | メモ | 
|---|---|---|---|---|---|---|
| 1 | 不明 | - | 1 | Magenta | 0(明滅なし) | |
| 2 | エラー | 未認知 | 2 | Red | 0.5秒 | |
| 3 | ワーニング | 未認知 | 4 | Yellow | 2秒 | |
| 4 | エラー | 認知済み | 8 | Green | 2秒 | |
| 5 | ワーニング | 認知済み | 16 | Green | 6秒 | |
| 6 | Newエラー | - | 32 | - | - | チャイム用 | 
| 7 | Newワーニング | - | 64 | - | - | チャイム用 | 
| 8 | 正常 | - | 0 | Blue | 8秒 | 
Abuはこれらの状態を単一の整数値「状態値」として扱います。
「状態値」は、ビット演算で扱いやすい様に、2進数の各ビット値を割り当てます。
「正常」は、全部のビットが0の状態なので、bit割り当てはありません。
内部では、上表のチャイム制御用のフラグも合わせた7bitのbit単位の論理和を一つの状態値として使用します。
例えば、まだ認知されていない新たなエラー(チャイムを鳴らす前)がある場合の状態値は、0x22です。
状態値はbit毎に意味をもたせた単一の整数値で表す方式としましたが、可読性を重視してPythonのSet型を使っても良いでしょう。
チャイム
鳴らし方
新たなエラーやワーニングが見つかった時は、気付きやすいように一度だけチャイムを鳴らします。
チャイムは、サウンドファイルとして用意したものを、aplayコマンドに渡して再生します。
$ aplay --quiet /usr/share/sounds/alsa/Rear_Center.wav
あらかじめ、本体のイヤホンジャックもしくは、USBポートにスピーカを接続して、音が鳴るように設定しておきます。
参考:https://www.google.co.jp/search?q=raspberrypi+aplay
音源
チャイムの音源は、aplayがサポートする形式ならなんでもOKです。
Abuではfreesound.orgより以下を使いました。
ユーザ登録するとダウンロード可能になりますので、ライセンスを確認して使用します。
- 
Warning 
 気付くことができれば良いので、控えめな音を選びました。
 112860__paulnorthyorks__eighties-synth.wav
 http://www.freesound.org/people/paulnorthyorks/sounds/112860/
- 
Error 
 緊迫感を感じる音を選びました。
 以下の音声を加工して、3回繰り返す音源を作成し、使用しています。
 204425__jaraxe__alarmx3.wav
 http://www.freesound.org/people/JarAxe/sounds/204425/
 加工には、Audacityを使いました。
Abuの動作
Abuの仕事は以下の通りです。
- Detectorを初期化
- LED Deamonを起動
- 以下を繰り返す
- Detectorを呼び出して、Alertのリストを取得する
- Alertリストをまとめて、状態値を作成
- 状態値に応じて、LED表示を設定
- 新たなエラーやワーニングがあれば、チャイムを鳴らす
- 1分間Sleep
 
起動方法
$ ./abu.py ********-****-****-****-************
********-****-****-****-************の部分には、前回の記事で紹介した、OpsGenieのAPI Keyを指定します。
OS起動時に自動起動するように、cronに登録する場合は、例えば以下のようにcronに追記すると良いでしょう。
$ crontab -e
@reboot cd path-to-module-dir;./abu.py ********-****-****-****-************ >/dev/nul 2>&1
ソースコード
# !/usr/bin/env python
# coding:utf-8
import time
import os
from opsgenie import OpsGenie
from led_deamon import LedDeamon
class Abu(object):
    DOWN = 1
    ERR_UNACKED = 2
    WARN_UNACKED = 4
    ERR_ACKED = 8
    WARN_ACKED = 16
    ERR_NEW = 32
    WARN_NEW = 64
    NORMAL = 0
    ERROR_SOUND = "./204425__jaraxe__alarm-2.wav"
    WARNING_SOUND = "./112860__paulnorthyorks__eighties-synth.wav"
    STATE_MAP = {
        DOWN: {"color": "Magenta", "interval": "0"},
        ERR_UNACKED: {"color": "Red", "interval": "0.5"},
        WARN_UNACKED: {"color": "Yellow", "interval": "4"},
        ERR_ACKED: {"color": "Green", "interval": "4"},
        WARN_ACKED: {"color": "Green", "interval": "8"},
        NORMAL: {"color": "Blue", "interval": "8"}}
    def _state2mode(self, state):
        if state == Abu.NORMAL:
            return Abu.STATE_MAP[Abu.NORMAL]
        for s in (Abu.DOWN, Abu.ERR_UNACKED, Abu.WARN_UNACKED,
                  Abu.ERR_ACKED, Abu.WARN_ACKED):
            if state & s:
                return Abu.STATE_MAP[s]
        else:
            return Abu.STATE_MAP[Abu.NORMAL]
            return None
    def __init__(self, url, api_key):
        self.opsgenie = OpsGenie(url, api_key)
    def _summarize(self, list):
        state = Abu.NORMAL
        if list is None:
            state = Abu.DOWN
        else:
            for l in list:
                if l["error"]:
                    if l["acknowledged"]:
                        state |= Abu.ERR_ACKED
                    else:
                        state |= Abu.ERR_UNACKED
                        if l["new"]:
                            state |= Abu.ERR_NEW
                else:
                    if l["acknowledged"]:
                        state |= Abu.WARN_ACKED
                    else:
                        state |= Abu.WARN_UNACKED
                        if l["new"]:
                            state |= Abu.WARN_NEW
        return state
    def start(self):
        ld = LedDeamon()
        ld.set_mode(self._state2mode(Abu.DOWN))
        ld.start()
        while True:
            alert_list = self.opsgenie.detector()
            # print alert_list
            state = self._summarize(alert_list)
            # print hex(state)
            mode = self._state2mode(state)
            if mode:
                ld.set_mode(mode)
            if state & Abu.ERR_NEW:
                os.system("aplay " + Abu.ERROR_SOUND + "&")
            else:
                if state & Abu.WARN_NEW:
                    os.system("aplay " + Abu.WARNING_SOUND + "&")
            time.sleep(60)
if __name__ == '__main__':
    import sys
    if len(sys.argv) != 2:
        print("Usage: %s 'api-key for your OpsGenie account'." % sys.argv[0])
        print("Example: %s ********-****-****-****-************." % sys.argv[0])
        print("You can get your OpsGenie account at https://www.opsgenie.com.")
        exit()
    apiKey = sys.argv[1]
    a = Abu(apiKey)
    a.start()
振り返り
Pythonの勉強方々RaspberryPiを使ってみました。
安価でPowerfulそしてGPIOを手軽に使えるライブラリもあり、こうした用途にはとてもマッチしていることがわかりました。
PWMによるLEDのカラー制御、非同期処理、状態値の取り扱い、OpsGenie連携など、いくつか悩むポイントはありましたが、幸いにして、それぞれ無理のない解決策を見出すことが出来、完成に至ることができました。
今回製作した「お知らせランプ」は、職場の雰囲気にもあまり違和感なく溶け込んで、実運用でも役立っています。
興味のある方はお試しになってみてください。
このシリーズは、この記事で完結となりますが、次のアイディアが固まったら、また投稿しようと思います。
