スマートホームの番人！Pingで家庭内60台のデバイス死活監視とLINE通知を自動化する方法

スマートホームに機器が増えれば増えるほど、「いつの間にかデバイスが落ちていた」という問題が起きやすくなります。
本記事では、Pingを使って家庭内の60台超のデバイスを1秒ごとに死活監視し、異常を検知したらLINEで自動通知するシステムの実装方法を解説します。
デバイスの死活監視に加え、Raspberry Piのsystemdサービスが停止していないかを監視する「サービス死活監視」の仕組みも合わせて紹介します。

なぜこの機能を作ったのか？
実現したいこと
この記事でわかること
必要な準備と用意するもの
完成イメージ
システムの仕組み
1. デバイス死活監視の仕組み
2. サービス死活監視の仕組み
実装のポイント
事前準備
実装方法
コードの解説
動作確認
まとめ

なぜこの機能を作ったのか？

我が家のスマートホームシステムはRaspberry Piと多数のIoT機器で構成されており、それぞれが24時間動き続けています。
最初はデバイスが落ちても気づかないことが多く、「Google Homeが反応しない」「Nature Remoが動いていない」などの問題が後から発覚することが続きました。
早期に異常を検知して対処できるよう、全デバイスの死活監視を自動化することにしました。また、Raspberry Pi上で動くsystemdサービスが落ちた場合も同様にLINEで通知が来るようにしています。

デバイスが落ちてもすぐに気づける仕組みを作りたい！

実現したいこと

家庭内のすべてのIPデバイスをPingで1秒ごとに死活監視する
デバイスへのPingが複数回連続で失敗した場合にLINEで通知する
通知対象かどうかをデバイスごとに設定できるようにする（重要度の低いデバイスは通知しない）
同一デバイスへの通知は24時間に1回に制限する
Raspberry Pi上で動くsystemdサービスが停止していないかも監視する
死活状態をInfluxDBに記録してGrafanaで可視化できるようにする

この記事でわかること

PythonのsubprocessでPingを実行してデバイスの死活を確認する方法
監視対象デバイスをJSONファイルで管理して柔軟に追加・削除する方法
連続失敗回数・通知クールダウンを組み合わせた過剰通知を防ぐ仕組み
systemdサービスの死活状態をPythonで監視してLINE通知する方法
Pythonプログラムをsystemdサービスとして常時起動する方法

必要な準備と用意するもの

ハードウェア

Raspberry Pi（例：Raspberry Pi 5）
監視対象のIPデバイス（IoT機器、スマホ、PCなど）

ソフトウェア/サービス

Python
- subprocess（Ping実行・systemctlコマンド実行）
LINE Notify（障害通知の送信先）
InfluxDB（死活状態の記録・可視化用）

LINE通知の実装（line.py）については別記事「LINEへの通知を自動化する方法」を参照してください。本記事ではline.pyという自作ライブラリが用意されている前提で解説します。

完成イメージ

完成すると、下記のように動作します。

デバイス死活監視：1秒ごとに全デバイスにPingを送り、6回連続で応答がない場合にLINEで通知
サービス死活監視：30秒ごとに指定のsystemdサービスの状態を確認し、3回連続で停止していた場合にLINEで通知
どちらも一度通知したら次の通知まで一定時間（デバイス：24時間、サービス：6時間）待機し、過剰通知を防ぐ

LINEに届く通知のイメージは下記の通りです。

# デバイスへのPingが失敗した場合
Nature Remo E (192.168.xxx.xxx) へのPingがタイムアウトしました。

# systemdサービスが停止した場合
[ALERT] smh-rain-forecast-alert.service が停止しています

システムの仕組み

デバイス死活監視の仕組み

devices.json：監視対象デバイスの一覧（名前・IPアドレス・通知有無・タイムアウト秒数）を定義
device_health_alert.py：起動時にdevices.jsonを読み込み、1秒ごとに全デバイスへPingを実行
Pingが失敗した場合、timeout_counts（連続失敗回数）をカウントアップ
6回連続失敗 + 通知対象の場合、LINEに通知。同時に死活状態をInfluxDBに記録
通知後はnotice_datetimes（最終通知日時）を更新し、24時間は再通知しない

サービス死活監視の仕組み

services.json：監視対象のsystemdサービス名のパターン一覧を定義
service_health_alert.py：30秒ごとにsystemctl is-activeで各サービスの状態を確認
3回連続で停止状態の場合、LINEに通知
通知後は6時間は再通知しない

実装のポイント

過剰通知を防ぐ2段階のフィルタリング

Pingの失敗が1回で即通知してしまうと、一時的なネットワークの揺らぎで大量の誤通知が発生します。そのため、下記の2段階でフィルタリングしています。

連続失敗カウント（6回以上）
一時的な通信途絶や再起動中のデバイスで誤通知が出ないよう、6回連続でPingが失敗した場合のみ通知する

24時間クールダウン
同じデバイスへの障害通知は24時間に1回に制限する。長時間停止しているデバイスへの通知が繰り返されるのを防ぐ

デバイスごとのタイムアウト設定

デバイスによってPingのレスポンス速度が大きく異なります。Wi-Fiを長時間スリープするデバイスや応答が遅いデバイスには長めのタイムアウトを設定できます。timeoutがnullの場合はデフォルトの2秒を使用します。

通知対象の選別（is_notification）

スマートホームの核となるデバイス（Google Home、Nature Remo E、ルーターなど）は通知対象（is_notification: true）に設定し、FireTV StickやゲームコンソールなどのデバイスはOFF（is_notification: false）にしています。
使用頻度が低いデバイスや、電源OFFが通常運用のデバイスを通知対象から外すことで、不要な通知を大幅に減らせます。

サービス監視はパターンマッチで柔軟に設定

監視するサービス名をワイルドカードパターン（smh-*など）で指定できるため、新しいサービスを追加した際にservices.jsonを変更しなくても自動的に監視対象に含まれます。

事前準備

監視対象デバイスのIPを固定する

死活監視はIPアドレスをベースに行うため、監視対象デバイスのIPアドレスはルーターでMACアドレスに紐づけてDHCP固定割り当てしておくことを強くおすすめします。IPが変わると監視が正しく機能しなくなります。

スマートフォンなどのデバイスはMACアドレスのランダム化（プライベートアドレス）が有効になっていると固定IPが使えないことがあります。iPhoneの場合はWi-Fiの設定から「プライベートWi-Fiアドレス」をオフにしてください。

devices.jsonの作成

監視対象デバイスを下記のフォーマットでconfig/devices.jsonに定義します。

[
    {"name":"Google Home Mini(リビング)",  "ip":"192.168.xxx.xxx", "is_notification":true,  "timeout":null}
   ,{"name":"Nature Remo E",              "ip":"192.168.xxx.xxx", "is_notification":true,  "timeout":5}
   ,{"name":"LaMetric-LM3262",            "ip":"192.168.xxx.xxx", "is_notification":true,  "timeout":null}
   ,{"name":"DECO XE75(ルーター側)",       "ip":"192.168.xxx.xxx", "is_notification":true,  "timeout":null}
   ,{"name":"Fire TV Stick 4K(テレビ)",   "ip":"192.168.xxx.xxx", "is_notification":false, "timeout":null}
   ,{"name":"Nintendo Switch",            "ip":"192.168.xxx.xxx", "is_notification":false, "timeout":null}
   ,{"name":"Raspberry Pi 4 Model B",     "ip":"192.168.xxx.xxx", "is_notification":true,  "timeout":null}
]

name：デバイスの識別名（LINEの通知メッセージにも使用）
ip：デバイスのIPアドレス（固定IPであること）
is_notification：trueのデバイスのみLINE通知を行う
timeout：Pingのタイムアウト秒数。nullの場合はデフォルト2秒を使用

services.jsonの作成

監視対象のsystemdサービスのパターンをconfig/services.jsonに定義します。ワイルドカード（*）が使用可能です。

[
     "smh-*"
    ,"nginx*"
    ,"grafana*"
    ,"influx*"
    ,"telegraf*"
]

smh-*はスマートホーム関連の自作サービス（smh-rain-forecast-alert、smh-device-health-alertなど）をまとめて監視するパターンです。新しいサービスを追加するたびにservices.jsonを更新しなくて済む設計にしています。

実装方法

デバイス死活監視プログラム (device_health_alert.py)

# -*- coding: utf-8 -*-
__group__       = '死活監視'
__description__ = '家庭内デバイスの死活状態を記録&監視'
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve()
while PROJECT_ROOT.name != "smarthome":
    if PROJECT_ROOT.parent == PROJECT_ROOT:
        raise RuntimeError("smarthome project root not found")
    PROJECT_ROOT = PROJECT_ROOT.parent
sys.path.insert(0, str(PROJECT_ROOT))
from libs.bootstrap import Bootstrap
bootstrap = Bootstrap(__file__)
settings = bootstrap.settings
# ------------------------------------------------------------
import subprocess
import time
from datetime import datetime
from influx.v2 import client as influxclient
from influx.v2 import writer as influxwriter
import line
from enum import Enum
import smarthomelog

try:
    influx_client = influxclient
    influx        = influxwriter.Writer()
    influx.insert_execution_log(__file__, 'Start Device Monitoring!')
    success_log   = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)

    # LINE通知の準備
    line_notify = line.Notify(line.LineChannel.SMARTHOME)

    # 監視対象デバイス一覧を読み込む
    devices = bootstrap.load_local_settings('config/devices.json')

    # デバイスごとの連続失敗回数を初期化
    timeout_counts = [0] * len(devices)
    # デバイスごとの最終通知日時を初期化（遠い過去に設定して初回通知を許可）
    notice_datetimes = [
        datetime.strptime('2000/01/01 00:00:00', '%Y/%m/%d %H:%M:%S')
        for _ in range(len(devices))
    ]

    class ResponseType(Enum):
        NO_RESPONSE = 1  # 応答なし
        TIMEOUT     = 2  # タイムアウト

    def timeout_process(response_type: ResponseType, device_name: str, device_ip: str, is_notification: bool):
        suffix_message = 'は応答がありません。' if response_type == ResponseType.NO_RESPONSE \
                         else 'へのPingがタイムアウトしました。'
        # 前回通知からの経過時間を確認
        before_notice_datetime = datetime.now() - notice_datetimes[device_count]
        if before_notice_datetime.seconds >= 60 * 60 * 24:  # 24時間以上経過していれば通知可能
            timeout_counts[device_count] += 1
            influx.insert_execution_log(__file__, f'{device_name} ({device_ip}) {suffix_message} ---> {timeout_counts[device_count]}回連続')
            influx.insert_device_status(__file__, device_name, device_ip, 0)  # 死：0
            # 6回連続失敗 かつ 通知対象の場合のみLINE通知
            if timeout_counts[device_count] >= 6 and is_notification:
                influx.insert_execution_log(__file__, f'{device_name} ({device_ip}) {suffix_message} ---> LINE通知')
                line_notify.send(f'{device_name} ({device_ip}) {suffix_message}')
                notice_datetimes[device_count] = datetime.now()
                timeout_counts[device_count] = 0
        else:
            influx.insert_execution_log(__file__, f'{device_name} ({device_ip}) {suffix_message} ---> 前回通知から{before_notice_datetime.seconds}秒経過のためスキップ')

    # メインループ（1秒ごとに全デバイスにPingを送る）
    while True:
        device_count = 0
        for device in devices:
            device_name     = device['name']
            device_ip       = device['ip']
            is_notification = device['is_notification']
            timeout         = device['timeout']
            timeout_second  = timeout if timeout is not None else 2  # デフォルト2秒

            try:
                result = subprocess.run(
                    ['ping', '-c', '1', device_ip],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    timeout=timeout_second
                )
                if result.returncode == 0:
                    influx.insert_execution_log(__file__, f'{device_name} ({device_ip}) は生きています。')
                    influx.insert_device_status(__file__, device_name, device_ip, 1)  # 生：1
                    timeout_counts[device_count] = 0
                else:
                    timeout_process(ResponseType.NO_RESPONSE, device_name, device_ip, is_notification)

            except subprocess.TimeoutExpired:
                timeout_process(ResponseType.TIMEOUT, device_name, device_ip, is_notification)

            device_count += 1

        success_log.write('処理が正常に終了しました。', __file__)
        time.sleep(1)

except Exception as e:
    log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
    log.write_error(sys.exc_info())

サービス死活監視プログラム (service_health_alert.py)

# -*- coding: utf-8 -*-
__group__       = '死活監視'
__description__ = 'ラズパイのサービスの死活状態を監視'
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve()
while PROJECT_ROOT.name != "smarthome":
    if PROJECT_ROOT.parent == PROJECT_ROOT:
        raise RuntimeError("smarthome project root not found")
    PROJECT_ROOT = PROJECT_ROOT.parent
sys.path.insert(0, str(PROJECT_ROOT))
from libs.bootstrap import Bootstrap
bootstrap = Bootstrap(__file__)
settings = bootstrap.settings
# ------------------------------------------------------------
import subprocess
import time
from datetime import datetime
from influx.v2 import client as influxclient
from influx.v2 import writer as influxwriter
import line
import smarthomelog

try:
    influx_client = influxclient
    influx        = influxwriter.Writer()
    success_log   = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
    influx.insert_execution_log(__file__, 'Start Service Alive Monitoring!')

    line_notify = line.Notify(line.LineChannel.SMARTHOME)

    # 監視対象サービスのパターン一覧を読み込む
    patterns = bootstrap.load_local_settings('config/services.json')

    CHECK_INTERVAL  = 30         # 秒ごとにチェック
    MAX_FAIL_COUNT  = 3          # 連続何回で通知
    NOTIFY_COOLDOWN = 6 * 60 * 60  # 通知クールダウン（6時間）

    state = {}  # サービスごとの状態管理

    def expand_services(pattern):
        """パターンに一致するサービス名を systemctl から取得"""
        try:
            result = subprocess.run(
                ['systemctl', 'list-unit-files', '--type=service', '--no-legend', '--no-pager', pattern],
                capture_output=True, text=True
            )
            return [line.split()[0] for line in result.stdout.strip().split('\n') if line]
        except Exception:
            return []

    def is_active(service):
        """systemctl is-active でサービスが起動中か確認"""
        try:
            result = subprocess.run(['systemctl', 'is-active', service], capture_output=True, text=True)
            return result.stdout.strip() == 'active'
        except Exception:
            return False

    while True:
        now = datetime.now()

        # パターンからサービスを展開
        all_services = set()
        for pattern in patterns:
            all_services.update(expand_services(pattern))

        for service in all_services:
            if service not in state:
                state[service] = {'fail_count': 0, 'last_notify': datetime.min}

            if is_active(service):
                state[service]['fail_count'] = 0
            else:
                state[service]['fail_count'] += 1
                if state[service]['fail_count'] >= MAX_FAIL_COUNT:
                    elapsed = now - state[service]['last_notify']
                    if elapsed.total_seconds() >= NOTIFY_COOLDOWN:
                        msg = f'[ALERT] {service} が停止しています'
                        line_notify.send(msg)
                        state[service]['last_notify'] = now
                        state[service]['fail_count']  = 0

        success_log.write('処理が正常に終了しました。', __file__)
        time.sleep(CHECK_INTERVAL)

except Exception as e:
    log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
    log.write_error(sys.exc_info())

systemdサービスファイルの作成

2つのプログラムをそれぞれsystemdサービスとして登録します。

[Unit]
Description=Device Monitoring
After=nginx.service influxdb.service webapi.service
Wants=nginx.service influxdb.service webapi.service

[Service]
ExecStartPre=/bin/sleep 10
ExecStart=/home/<username>/Projects/venv/bin/python /home/<username>/Projects/smarthome/core/alert/device_health/device_health_alert.py
Restart=always
RestartSec=60
Type=simple
User=<username>

[Install]
WantedBy=multi-user.target

[Unit]
Description=Service Monitoring
After=nginx.service influxdb.service webapi.service
Wants=nginx.service influxdb.service webapi.service

[Service]
ExecStart=/home/<username>/Projects/venv/bin/python /home/<username>/Projects/smarthome/core/alert/service_health/service_health_alert.py
Restart=always
RestartSec=60
Type=simple
User=<username>

[Install]
WantedBy=multi-user.target

# サービスファイルをコピー
sudo cp smh-device-health-alert.service /etc/systemd/system/
sudo cp smh-service-health-alert.service /etc/systemd/system/

# systemdに読み込ませる
sudo systemctl daemon-reload

# 自動起動を有効化 & 開始
sudo systemctl enable smh-device-health-alert.service
sudo systemctl start smh-device-health-alert.service

sudo systemctl enable smh-service-health-alert.service
sudo systemctl start smh-service-health-alert.service

コードの解説

Pingの実行と判定（device_health_alert.py）

Pythonのsubprocess.runでシステムのpingコマンドを実行します。returncode == 0であれば疎通成功、それ以外は失敗です。timeoutパラメータを指定することで、応答が遅いデバイスへの対策もできます。

result = subprocess.run(
    ['ping', '-c', '1', device_ip],   # 1回だけPingを送る
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    timeout=timeout_second            # タイムアウト秒数
)
if result.returncode == 0:
    # 疎通OK → 連続失敗カウントをリセット
    timeout_counts[device_count] = 0
else:
    # 疎通NG → timeout_process で失敗カウントを増やす
    timeout_process(ResponseType.NO_RESPONSE, ...)

通知条件の判定ロジック

「連続失敗6回以上」と「前回通知から24時間以上経過」の両方を満たした場合のみLINE通知を送ります。

# 24時間以上経過しているか確認
before_notice_datetime = datetime.now() - notice_datetimes[device_count]
if before_notice_datetime.seconds >= 60 * 60 * 24:
    timeout_counts[device_count] += 1
    # 6回連続失敗 かつ 通知対象の場合のみLINE通知
    if timeout_counts[device_count] >= 6 and is_notification:
        line_notify.send(f'{device_name} ({device_ip}) {suffix_message}')
        notice_datetimes[device_count] = datetime.now()  # 最終通知日時を更新
        timeout_counts[device_count] = 0                 # 連続カウントをリセット

サービスパターンの展開（service_health_alert.py）

systemctl list-unit-filesコマンドにパターン文字列を渡すことで、ワイルドカードに一致するサービス名の一覧を取得できます。

result = subprocess.run(
    ['systemctl', 'list-unit-files', '--type=service', '--no-legend', '--no-pager', 'smh-*'],
    capture_output=True, text=True
)
# 出力例：
# smh-device-health-alert.service   enabled
# smh-rain-forecast-alert.service   enabled
# smh-service-health-alert.service  enabled

ExecStartPreで起動遅延を設ける（device health service）

サービスファイルのExecStartPre=/bin/sleep 10は、OS起動直後にネットワークやInfluxDBの準備が整う前にプログラムが起動して即エラーになるのを防ぐための遅延です。

After=influxdb.serviceの依存関係だけでは「起動処理が始まった」ことしか保証されず、「完全に使える状態になった」ことは保証されないため、念のため10秒の遅延を入れています。

動作確認

まずはコマンドラインから手動で実行して動作を確認します。

# デバイス死活監視の手動実行
python /home/<username>/Projects/smarthome/core/alert/device_health/device_health_alert.py

# サービス死活監視の手動実行
python /home/<username>/Projects/smarthome/core/alert/service_health/service_health_alert.py

下記を確認してください。

デバイス死活監視：ターミナルに「〜は生きています。」が流れ続けること
テストとして監視対象のデバイスの電源を落とし、しばらく後にLINEに通知が来ること（6回失敗で通知）
サービス死活監視：監視対象のサービスを手動で停止して3回チェック後にLINEに通知が来ること

手動実行で問題なければ、systemdサービスの状態を確認します。

# サービスの状態確認
sudo systemctl status smh-device-health-alert.service
sudo systemctl status smh-service-health-alert.service

# ログのリアルタイム確認
journalctl -u smh-device-health-alert.service -f

LINE通知が来ない場合は下記を確認してください。
・devices.jsonのIPアドレスが正しいか、かつ対象デバイスのIPが固定されているか
・is_notification: trueになっているか
・LINE Notify のトークンが有効か
・InfluxDBが起動しているか（サービスがAfter=influxdb.serviceに依存しているため）