見出し画像

ルンバの掃除タイミングを最適化したい(2025)

これまでの経緯

元々はこんな話でした。

諸々うまく行ったかなと思ったら事故が起きました。

で、凍結になっていたのですが、中途半端な状況を脱したく、再開することに。ただ、改めて上記の記事を読むと、なんで詰まっていたのかよくわからないのですよね。

実現したいこと

5分に一回自宅のネットワークに接続している機器をスキャンして、居住者のモバイル端末が6回連続で見つからなかった場合にお掃除する仕様に変更する。一度実行したら3時間はスキャンを停止する。という2つの仕様を追加するだけのはず。まあ6回はちょっと心許ないので、10回にしてもよいかも。1時間外出してたらお掃除してもらう。

あるいは、承認メッセージをラインで送る形。今はラズパイがIFTTTのwebhookを叩いているので、この向き先をIFTTTではなくてLINEのMessaging APIにするという。LINEでなくてもいいけど。

もう2度と事故は起こしたくないので、ちょっと煩わしいけど当面は承認制で運用してみようかな。IFTTTの契約も切れるし。地味に毎月15ドルはつらい・・・。

ううう・・・こういう無用な出費を人はなぜしてしまうのだろう・・・

webhookを諦めて別の方法(RSSあたりの仕組みを使うとか)で無料でIFTTTを利用する範囲でできればいいのだけど、まずは承認制にするならそれすら不要だしな。

公式が提供している機能

ということでキャンセル。していたら気になる操作を発見。iRobot公式が、家を離れたらルンバを動かすAppletを公開している。いつの間に・・・最初からあったのかな。というかどうやって外出判断しているのだろう。

と思ってみてみると、どうやら一定のエリアから位置情報をみて判断するらしい。

え、どうやって位置情報をとるの?と思ったら、IFTTTのアプリを入れさせたいらしい。なるほど。それなら確かにできますね。

ただ、今回のニーズは複数人が居住している住居で二人とも外出した場合を判断せねばならぬので、この機能だと不十分なのだよな。ということで、利用は見送りに。

LINEbotのアカウントを作る

ということで、不在時に掃除の実行をおすすめしてくれるLINEbotのアカウントを作る。以前に下記でアカウント自体は作ったことがあるので、それほど問題にならないはず。

ということで、記憶を頼りにまずプロバイダーを作成してみると下記の記載が。なんと、公式アカウントを作らなとAPIは利用できないのだそうな。

お気持ちはよく分かります。

24年9月からの変更らしい。なるほど〜

これはつまりあれか。検索するとアカウントが出てきてしまうみたいなことだな。まあいいけども。ということでアカウントを作成。

設定画面からMessaging APIが利用可能に。

ユーザーIDとチャネルアクセストークンを確認して実行。無事にメッセージが配信される。よきよき。

ということで、これを10回スキャンした後に送ってもらう実装に変えてみましょう。10回スキャンが空ぶったら、LINEで通知を送る仕組み。これで1ヶ月くらい運用してみて、問題なさそうなら、IFTTTを使うか、他の方法でリモートで自動的にルンバを作動させる方法を考えてみましょう。

Messaging APIの利用回数というか、月の配信数に制限があるので、通知は1日1回までというスリープモードを設定してもらわねばですね。

ということで下記のようにコードを修正してもらいました。

from scapy.all import ARP, Ether, srp
import requests
import time
import json
from datetime import datetime

# 監視対象のMACアドレス一覧
target_mac_addresses = [
    '自分の端末のMACアドレス',
    '同居人の端末のMACアドレス'
]

# ネットワークスキャン範囲
network_range = '自宅のネットワークのアドレス'

# デバイスが見つからなかった連続回数
consecutive_misses = 0

# LINE API設定
line_api_url = "https://api.line.me/v2/bot/message/push"
line_headers = {
    'Content-Type': 'application/json',
    'Authorization': 'チャネルアクセストークン'
}
line_payload = {
    "to": "ユーザーID",
    "messages": [
        {
            "type": "text",
            "text": "ルンバがお掃除の指示を待っています"
        }
    ]
}

# 今日通知を送信したかを記録
today_last_notification = None


def scan_network():
    """ネットワークスキャンを行い、接続されているデバイスを取得します。"""
    try:
        arp = ARP(pdst=network_range)
        ether = Ether(dst="ff:ff:ff:ff:ff:ff")
        packet = ether / arp

        result = srp(packet, timeout=3, verbose=0)[0]

        devices = []
        for sent, received in result:
            devices.append({'ip': received.psrc, 'mac': received.hwsrc.lower()})
        return devices
    except Exception as e:
        print(f"エラー: ネットワークスキャン中に問題が発生しました: {e}")
        return []


def trigger_line_notification():
    """LINE APIを使用して通知を送信します。"""
    global today_last_notification
    try:
        response = requests.post(line_api_url, headers=line_headers, data=json.dumps(line_payload))
        response.raise_for_status()
        print('LINE通知が送信されました:', response.text)
        today_last_notification = datetime.now().date()
    except requests.exceptions.RequestException as err:
        print('LINE通知送信中にエラーが発生しました:', err)


def check_network():
    """ネットワークをスキャンし、対象デバイスの存在を確認します。"""
    global consecutive_misses, today_last_notification

    print('ネットワークをスキャン中...')
    network_entries = scan_network()

    if not network_entries:
        print('ネットワークスキャン結果なし。')
        consecutive_misses += 1
    else:
        for entry in network_entries:
            print(f"IP: {entry['ip']}, MAC: {entry['mac']}")

        # 監視対象デバイスが存在するか確認
        any_device_found = any(entry['mac'] in target_mac_addresses for entry in network_entries)

        if not any_device_found:
            consecutive_misses += 1
            print(f"デバイスが見つかりません。連続未検出回数: {consecutive_misses}")

            # LINE通知の送信条件を満たし、まだその日に通知を送っていない場合
            if consecutive_misses >= 10 and (today_last_notification is None or today_last_notification != datetime.now().date()):
                print("10回連続でデバイスが見つかりませんでした。LINE通知を送信します。")
                trigger_line_notification()
                consecutive_misses = 0
        else:
            print("監視対象デバイスが見つかりました。")
            consecutive_misses = 0


if __name__ == "__main__":
    try:
        while True:
            check_network()
            time.sleep(5 * 60)  # 5分間隔で実行
    except KeyboardInterrupt:
        print("プログラムが終了しました。")

結果、翌朝目が覚めると、朝の4時にお掃除するかの通知が来ていた。

このメッセージにお掃除実行のURLを設定できるといいですね。
Webhookはお金かかるからあれだけど。

やっぱり夜はスマホもスリープしてるからWi-Fiのアクセス控えてしまうのだなあ(ということで、夜(22:00-8:00の間)は実行しないようにし加筆しました)。

これでひとまずは当面運用してみて、良さそうなら通知内にお掃除実行リンクを仕込む(IFTTTの無料枠内でやりたいのでたぶんRSSの仕組みとかを使う)みたいなことにもチャレンジしたいと思います。おつかれさまでした!

いいなと思ったら応援しよう!

まかない
ご覧いただきありがとうございます。とても嬉しいです。