ArduinoとPythonでリアルタイム異常検知システムを作る！Z-scoreでセンサーデータを監視してアラートを飛ばす方法

「センサーの値がおかしい気がするけど、どのタイミングで異常と判断すればいいの？」

こんな疑問を持ったことはありませんか？温度センサーや振動センサーを使っていると、普段と少し違う値が来たとき、それが本当に異常なのか、ただのノイズなのかを判断するのって意外と難しいですよね。

今回はそんな悩みを一気に解決できる方法を紹介します！Arduinoでセンサーデータを取得して、Pythonのシリアル通信でリアルタイムに受け取り、Z-scoreという統計的な手法で異常を自動検知→アラートを飛ばすシステムをゼロから作っていきます 🚨

「機械学習って難しそう…」と思った方、安心してください。Z-scoreはライブラリいらずで実装できる統計的手法で、数学が苦手でも仕組みをざっくり理解できます。Arduinoとシリアル通信の基礎がある方なら、すぐに動かせるはずです。

Contents

📌 この記事の対象読者と難易度
🔍 Z-scoreとは？（異常検知の仕組みをざっくり理解する）
🔧 今回作るシステムの全体像
⚡ STEP 1：Arduinoスケッチを書く（センサー値をシリアル送信）
🐍 STEP 2：Pythonでシリアルデータを受信する
📊 STEP 3：Z-scoreで異常検知ロジックを実装する
🔔 STEP 4：アラートをデスクトップ通知で飛ばす
🛠️ 閾値のチューニングと実運用のコツ
💡 応用アイデア：もっと実用的にしたいなら
✅ まとめ

📌 この記事の対象読者と難易度

arduino sensor monitoring / Photo by Bmonster Lab via Pexels

✅ Arduinoでセンサーを動かしたことがある方
✅ PythonでPySerialを触ったことがある、または興味がある方
✅ IoTデータの「異常」を自動で検知したい方
⚠️ 難易度：中級者向け（コピペして動かすことは初心者でも可能）

🔍 Z-scoreとは？（異常検知の仕組みをざっくり理解する）

まず、Z-scoreの考え方をざっくり説明します。

Z-scoreは「今の値が平均からどれだけ離れているか」を標準偏差で割った数値です。イメージとしては、クラスの平均点から自分の点数がどれだけ外れているかを数値化するような感じです。

式で書くとこうなります：

Z = (現在値 - 平均) / 標準偏差

ポイントをまとめるとこんな感じです：

Z-score が ±2 以内 → 正常範囲（データの約95%が収まる）
Z-score が ±3 を超える → 異常と判断（統計的に非常にまれな値）
閾値（しきい値）は用途によって調整可能

センサーの値がしばらく安定していれば「平均と標準偏差」が蓄積されます。そこに突然おかしな値が来たとき、Z-scoreが跳ね上がって「異常だ！」と検知できるわけです。シンプルですが、IoTの現場でも実際によく使われている手法ですよ。

🔧 今回作るシステムの全体像

構成はこんな感じです：

[温度センサー（LM35など）]
        ↓
[Arduino Uno（アナログ値を読み取ってシリアル送信）]
        ↓  USB シリアル通信
[Python（pyserial でデータ受信 → Z-score 計算 → 異常判定 → アラート表示）]

シンプルな構成ですが、リアルタイムで動き続けながら統計的に異常を検知するという部分が今回の肝です。

使用部品はこちら：

Arduino Uno（またはNano、Mega等でも可）
LM35温度センサー（または任意のアナログセンサー）
USB ケーブル（PCとArduinoを接続）
ブレッドボードとジャンパーワイヤー

⚡ STEP 1：Arduinoスケッチを書く（センサー値をシリアル送信）

まずArduino側のコードです。LM35温度センサーのアナログ値を読み取って、1秒ごとにシリアルポートへ送信します。

// センサーデータをシリアルで送信するスケッチ
// LM35温度センサー → A0ピンに接続

const int sensorPin = A0;  // センサーの入力ピン
float temperature = 0.0;

void setup() {
  Serial.begin(9600);  // シリアル通信を9600bpsで開始
}

void loop() {
  int rawValue = analogRead(sensorPin);  // アナログ値（0〜1023）を読み取る
  
  // LM35の変換式：0.49 = (5.0 / 1024) * 100
  temperature = rawValue * 0.4882;  // ℃ に変換
  
  // 温度値をシリアルに送信（改行付き）
  Serial.println(temperature);
  
  delay(1000);  // 1秒待機
}

ここが重要です：

Serial.println() で改行付きで送ることで、Python側で1行ずつ受け取りやすくなります
ボーレートは 9600bps。PythonとArduinoで同じ値に合わせておくことが必須です
LM35を使わない場合は変換式を変えてください（例：DHT11なら別途ライブラリが必要）

Arduinoに書き込んだら、シリアルモニタを開いて数値が流れてくることを確認しておきましょう ✅

🐍 STEP 2：Pythonでシリアルデータを受信する

次にPython側の準備です。まずライブラリをインストールします。

pip install pyserial

シリアル通信でデータを受信する基本コードはこちらです：

import serial
import time

# Windowsは 'COM3' など、MacやLinuxは '/dev/ttyUSB0' など
PORT = 'COM3'
BAUD_RATE = 9600

# シリアルポートを開く
ser = serial.Serial(PORT, BAUD_RATE, timeout=1)
time.sleep(2)  # Arduinoのリセット待機（重要！）

print("受信開始...")

while True:
    line = ser.readline().decode('utf-8').strip()  # 1行読み取り
    if line:
        try:
            value = float(line)  # 数値に変換
            print(f"受信値: {value:.2f} ℃")
        except ValueError:
            pass  # 変換できない文字列はスキップ

ポイントをまとめるとこんな感じです：

time.sleep(2)：Arduinoは接続直後にリセットがかかるため、2秒待つのがコツです
.decode(‘utf-8’).strip()：バイト列を文字列に変換して前後の空白・改行を除去
接続できない場合はポート番号を確認（Windowsはデバイスマネージャー、MacLinuxは ls /dev/tty*）

📊 STEP 3：Z-scoreで異常検知ロジックを実装する

ここからが本題です！受信したデータをリストに蓄積して、Z-scoreを計算し、閾値を超えたら異常と判定するロジックを加えます。

import serial
import time
import math
import statistics
from collections import deque

# === 設定 ===
PORT = 'COM3'          # ご自身の環境に合わせて変更
BAUD_RATE = 9600
WINDOW_SIZE = 30       # 統計計算に使う直近のデータ件数
Z_THRESHOLD = 2.5      # 異常とみなすZ-scoreの閾値
MIN_DATA_COUNT = 10    # この件数に達するまで判定しない（学習フェーズ）

# 直近データを保持するキュー（最大WINDOW_SIZE件）
data_window = deque(maxlen=WINDOW_SIZE)

def calculate_z_score(value, data):
    """現在値のZ-scoreを計算する関数"""
    if len(data) < 2:
        return 0.0  # データが少なすぎる場合は0を返す
    
    mean = statistics.mean(data)      # 平均
    stdev = statistics.stdev(data)    # 標準偏差
    
    if stdev == 0:
        return 0.0  # 標準偏差が0（全て同じ値）の場合は0を返す
    
    z_score = (value - mean) / stdev  # Z-score計算
    return z_score

def check_anomaly(value, z_score):
    """Z-scoreに基づいて異常を判定してアラートを出す関数"""
    if abs(z_score) > Z_THRESHOLD:
        print(f"🚨 異常検知！  値: {value:.2f} ℃  |  Z-score: {z_score:.2f}")
        print(f"   → 平均から {abs(z_score):.1f} 標準偏差分離れています")
        return True
    return False

# === メインループ ===
ser = serial.Serial(PORT, BAUD_RATE, timeout=1)
time.sleep(2)  # Arduinoリセット待機

print("🔍 異常検知システム起動！データ収集を開始します...")
print(f"   最初の{MIN_DATA_COUNT}件は学習フェーズです")
print("-" * 50)

data_count = 0

while True:
    line = ser.readline().decode('utf-8').strip()
    if not line:
        continue
    
    try:
        value = float(line)
        data_count += 1
        data_window.append(value)  # キューに追加（自動で古いデータを削除）
        
        if data_count < MIN_DATA_COUNT:
            # 学習フェーズ中はデータ蓄積のみ
            print(f"[学習中 {data_count}/{MIN_DATA_COUNT}]  値: {value:.2f} ℃")
        else:
            # 判定フェーズ：Z-scoreを計算して異常チェック
            z = calculate_z_score(value, list(data_window))
            is_anomaly = check_anomaly(value, z)
            
            if not is_anomaly:
                print(f"[正常]  値: {value:.2f} ℃  |  Z-score: {z:.2f}")
    
    except ValueError:
        pass  # 数値以外のデータはスキップ

このコードの重要なポイントをまとめます：

deque(maxlen=WINDOW_SIZE)：直近30件のデータだけを保持する「スライディングウィンドウ」。これにより時間経過による正常範囲の変化にも対応できます
MIN_DATA_COUNT：最初の10件はデータを蓄積するだけの「学習フェーズ」。少ないデータで誤検知しないための工夫です
abs(z_score) > Z_THRESHOLD：Z-scoreの絶対値が閾値を超えたら異常。プラス方向・マイナス方向どちらも検知できます

🔔 STEP 4：アラートをデスクトップ通知で飛ばす

異常を検知したとき、ターミナルに表示するだけでなくデスクトップ通知として飛ばすと実用性がぐっと上がります。

Windowsでは plyer ライブラリを使うのが手軽です：

pip install plyer

from plyer import notification

def send_alert(value, z_score):
    """デスクトップ通知でアラートを送る関数"""
    notification.notify(
        title='🚨 センサー異常検知！',
        message=f'値: {value:.2f} ℃  Z-score: {z_score:.2f}\n平均から大きく外れています！',
        app_name='Arduino異常検知システム',
        timeout=10  # 10秒後に自動で消える
    )

# check_anomaly 関数内に組み込む場合
def check_anomaly(value, z_score):
    if abs(z_score) > Z_THRESHOLD:
        print(f"🚨 異常検知！  値: {value:.2f} ℃  |  Z-score: {z_score:.2f}")
        send_alert(value, z_score)  # 通知を飛ばす！
        return True
    return False

MacOSやLinuxでもplyerは動作しますが、環境によっては別途設定が必要な場合があります。plyerが使いにくい場合は subprocess でOSネイティブの通知コマンドを呼ぶ方法もありますよ。