【Python】EVの充電音を監視し、完了したらLINE通知してみた！

こんにちは！
九保すこひです（フリーランスのITコンサルタント、エンジニア）

【X】九保すこひ＠フリーランスエンジニア

さてさて、4月末にEV（トゥクトゥク。ドアなし）を購入し、生活の質が大幅に向上したのですが、1つ「こうだったらいいのにな…🤔」と思うことがあります。

それは・・・・・・

充電が終わったら通知してほしい。

というものです。

通常のEVならメール通知などをしてくれたりもするようですが、私の車にはそんな機能はついていないので、

「そろそろ充電終わったかな？」

って見に行くと

「あ。まだやったわ…😅」

てことが何回も発生しているわけです。

で、何か対策はないかを考えてたら思いつきました！
私の機体は、充電するときにファンが回る、つまり音が出るので、

ファン音を監視して、静かになったら通知する

という作戦です。

そこで！
今回は以下のような方に記事を書いています。

「Pythonやセンサー技術を学びたい」
「日常の不便をプログラムで解決したい」
「スマートホーム化やDIY電子工作に興味がある人」
「EVの過充電を防ぎたい人」
「LINE APIの活用方法を学びたい人」

ぜひ最後まで読んでくださいね！

「久しぶりのPython
プログラムです！」

1 前提として
2 音を監視して、無音になったらLINE通知する部分をつくる
3 ちなみに：マイク付きイヤホンの設置方法は…？
4 テストしてみる
5 企業様へのご提案
6 おわりに

前提として

今回は以下の流れで音声を監視し、無音になったら通知します。

EV充電を開始：ファンの音が出る
マイク付きイヤホン設置：Bluetoothでパソコンと接続
Pythonを実行：音を監視する
充電が完了：音がなくなる
LINE通知：5秒間の無音をトリガーとして通知

そのため、先にLINE通知するための

アクセストークン：LINE Messaging APIアクセスに必要
送信先のユーザーID：LINE IDではないです。※

※LINE ユーザーIDはWebhookでしか取得できません。

もしまだ取得していない方は以下のページを参考にして用意しておいてください。

アクセストークン：LaravelでLINEにチャットボットをつくる（QRコード作成）
送信先のユーザーID：Laravel + LINE マーケティングで「パーソナライズされたリスト取り」機能をつくる

※Webhookでは以下の部分でLINEの「ユーザーID」を取得しています。

$line_id = $event->getSource()->getUserId();

また、LINEのPUSHメッセージの無料枠は月1,000回までです。
気をつけてくださいね😊

音を監視して、無音になったらLINE通知する部分をつくる

では、メインのPythonコードです。

#!/usr/bin/env python3
import time
import math
import json
import numpy as np
import sounddevice as sd
import requests

# ========= 設定 =========

INPUT_DEVICE_ID = None

SAMPLE_RATE = 16000          # どのくらい細かく音を取るか。サンプリングレート(Hz)
BLOCK_SIZE  = 1024           # 1回のコールバックあたりのサンプル数

THRESHOLD   = 0.0001         # 無音とみなすしきい値（←ここの調整難しいです）
SILENT_SECONDS = 5           # 何秒間無音が続いたら充電完了とみなすか

LINE_CHANNEL_ACCESS_TOKEN = "（ここにアクセストークン）"
LINE_USER_ID = "（ここにLINEのユーザーID）"

# ========= LINE通知関数 =========

def send_line_push(message: str) -> bool:
    url = "https://api.line.me/v2/bot/message/push"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}",
    }
    body = {
        "to": LINE_USER_ID,
        "messages": [
            {"type": "text", "text": message}
        ]
    }
    try:
        resp = requests.post(url, headers=headers, data=json.dumps(body), timeout=10)
        print(f"[LINE] status={resp.status_code}")
        return 200 <= resp.status_code < 300
    except Exception as e:
        print(f"[LINE ERROR] {e}")
        return False

# ========= 無音検出ロジック =========

class SilenceDetector:
    def __init__(self, threshold: float, silent_seconds: int):
        self.threshold = threshold
        self.silent_seconds = silent_seconds
        self.silent_time = 0.0
        self.finished = False  # 繰り返し実行の終了フラグ

    def process_block(self, indata: np.ndarray, frames: int):
        if self.finished:
            return

        try:
            mono = np.mean(indata, axis=1)
            rms = math.sqrt(float(np.mean(mono ** 2)))
            volume = rms

            bar = "|" * int(volume * 50)
            print(f"vol={volume:.4f} {bar}")

            block_duration = frames / SAMPLE_RATE
            if volume < self.threshold:
                self.silent_time += block_duration
            else:
                self.silent_time = 0.0

            if self.silent_time >= self.silent_seconds:
                print(f"[INFO] {self.silent_time:.1f}秒以上無音 -> LINE通知")
                ok = send_line_push("EVの充電が完了しました。確認してください。")
                if ok:
                    print("[INFO] 通知成功。監視を終了します。")
                else:
                    print("[INFO] 通知失敗。監視を終了します。")
                self.finished = True
        except Exception as e:
            print(f"[CALLBACK ERROR] {e}")

# ========= コールバック =========

detector = SilenceDetector(THRESHOLD, SILENT_SECONDS)

def audio_callback(indata, frames, time_info, status):
    try:
        if status:
            print(f"[STATUS] {status}")
        detector.process_block(indata, frames)
    except Exception as e:
        print(f"[AUDIO CALLBACK ERROR] {e}")
        detector.finished = True

# ========= メイン =========

def main():
    print("監視開始!")
    print(f"device={INPUT_DEVICE_ID}, threshold={THRESHOLD}, silent_seconds={SILENT_SECONDS}")

    with sd.InputStream(
        device=INPUT_DEVICE_ID,
        channels=1,
        samplerate=SAMPLE_RATE,
        blocksize=BLOCK_SIZE,
        callback=audio_callback,
    ):
        try:
            while not detector.finished:
                time.sleep(0.5)
        except KeyboardInterrupt:
            print("手動終了")
        finally:
            print("監視終了")

if __name__ == "__main__":
    main()