streamlit を使って WOL を投げるWebアプリを作ってみる

目的

LANにつながったPCに対して、PINGを使った死活監視とWakeOnLan送信を行うためのアプリをStreamlitを使って作成します。

環境構築

Ubuntu 22.04.1 LTS を使って実行環境を構築します。

必要なPythonモジュールをインストール。

requirements.txt

streamlit
pyyaml
ping3

$ pip3 install -r requirements.txt

環境によってはpingモジュールを使うと PermissionErrorが発生します。

$ python3
Python 3.10.6 (main, Nov  2 2022, 18:53:38) [GCC 11.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from ping3 import ping
>>> ping('google.com')
Traceback (most recent call last):
  File "/home/raisuta/.local/lib/python3.10/site-packages/ping3/__init__.py", line 281, in ping
    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
  File "/usr/lib/python3.10/socket.py", line 232, in __init__
    _socket.socket.__init__(self, family, type, proto, fileno)
PermissionError: [Errno 1] Operation not permitted

下記を参考にnet.ipv4.ping_group_rangeパラメータを変更します。

# リブートするまで有効
$ sudo sysctl net.ipv4.ping_group_range='0   2147483647'

# 永続化
$ echo "# allow all users to create icmp sockets\n net.ipv4.ping_group_range=0 2147483647" | sudo tee -a /etc/sysctl.d/ping_group.conf

WakOnLanはPythonで実装するより、簡単に使えるコマンドがあったのでこれをaptからインストールします。

$ sudo apt install wakeonlan
...
$ wakeonlan
Usage
    wakeonlan [-h] [-v] [-i IP_address] [-p port] [-f file] [[hardware_address] ...]

Options
    -h
        this information
    -v
        displays the script version
    -i ip_address
        set the destination IP address
        default: 255.255.255.255 (the limited broadcast address)
    -p port
        set the destination port
        default: 9 (the discard port)
    -f file
        uses file as a source of hardware addresses

See also
    wakeonlan(1)

アプリの実装

以下にアプリのコード(app.py)と設定用YAMLファイル(hosts.yaml)を記します。

app.py

import streamlit as st
import subprocess
import os
import yaml
import time
from ping3 import ping
from concurrent.futures import ThreadPoolExecutor

hosts = None
def hosts_load():
    global hosts
    hosts = st.session_state.get("hosts")
    if not hosts:
        os.chdir(os.path.dirname(__file__))
        with open('hosts.yaml', 'r') as f:
            hosts = yaml.safe_load(f)
        st.session_state["hosts"] = hosts

def hosts_update_status_text():
    for host in hosts:
        host['status_text'].write(f'{host.get("status", "⚪Unknown")} - {host["ip"]}')

def section_WOL_buttons():
    st.markdown("## WOL buttons")
    for host in hosts:
        host['button'] = st.button(f'WOL - {host["name"]}')
        host['status_text'] = st.text('')
    st.write("")

    hosts_update_status_text()

def checking_button():
    for host in hosts:
        if host['button']:
            subprocess.run(['wakeonlan', host["mac"]])

def checking_ping_status():
    targets = hosts.copy()
    polling_count = 100
    while targets and (polling_count > 0):
        with ThreadPoolExecutor() as executor:
            results = list(executor.map(lambda t : ping(t['ip'], timeout=1), targets))
            for result, host in zip(results, targets):
                if result:
                    host['status'] = '🟢Reachable'
                else:
                    host['status'] = '🔴Unreachable'
            targets = [t for r, t in zip(results, targets) if not r ]

        hosts_update_status_text()
        polling_count -= 1
        time.sleep(1)

# streamlit run <script.py>
if __name__ == '__main__':
    hosts_load()
    section_WOL_buttons()
    checking_button()
    checking_ping_status()

hosts.yaml

- name: DESKTOP-A
  ip:   192.168.0.10
  mac:  aa:aa:aa:aa:aa:aa

- name: DESKTOP-B
  ip:   192.168.0.11
  mac:  bb:bb:bb:bb:bb:bb

streamlit runで実行するとURLが返されるのでここにブラウザでアクセスするとWebアプリが立ち上がります。

$ ls
app.py  hosts.yaml  requirements.txt
$ python3 -m streamlit run ./app.py

  You can now view your Streamlit app in your browser.

  Local URL: http://localhost:8501
  Network URL: http://172.17.68.182:8501

アプリの解説

ページのロードまたはUIボタンが押されるたびに下記順序で関数を実行します。

# streamlit run <script.py>
if __name__ == '__main__':
    hosts_load()
    section_WOL_buttons()
    checking_button()
    checking_ping_status()

hosts_load()

グローバル変数のhostsへ情報を読み込みます。
アプリが初回ロードされた時はst.session_stateにhostsが登録されていないので、yamlモジュールを使ってファイルからロードしてセッション状態に保存されます。
UIのボタンが押されてページが再ロードされるとst.session_state.getにてセッション状態からロードされます。

hosts = None
def hosts_load():
    global hosts
    hosts = st.session_state.get("hosts")
    if not hosts:
        os.chdir(os.path.dirname(__file__))
        with open('hosts.yaml', 'r') as f:
            hosts = yaml.safe_load(f)
        st.session_state["hosts"] = hosts

hosts.yaml

- name: <ホスト名>
  ip:   <IPアドレス>
  mac:  <MACアドレス>

section_WOL_buttons()

hostsの要素ごとにst.button, st.textを追加して、各UIの参照を保存します。
st.textの文字列はhosts_update_status_text()によってhost['status']の状態を反映させますが、初回はstatusキーが無いので、"⚪Unknown"と表示されます。

def hosts_update_status_text():
    for host in hosts:
        host['status_text'].write(f'{host.get("status", "⚪Unknown")} - {host["ip"]}')

def section_WOL_buttons():
    st.markdown("## WOL buttons")
    for host in hosts:
        host['button'] = st.button(f'WOL - {host["name"]}')
        host['status_text'] = st.text('')
    st.write("")

    hosts_update_status_text()

checking_button()

StreamlitはボタンUIが押されるとスクリプトを最初から実行し直すので、全てのhost['button']を見てボタンが押されているかを調べます。
もしボタンが押されていたらsubprocess.runを使ってシェルからwakeonlanコマンドを実行してWakeOnLanを送信します。

def checking_button():
    for host in hosts:
        if host['button']:
            subprocess.run(['wakeonlan', host["mac"]])

checking_ping_status()

全てのhostsへpingを送信して、死活状態を確認します。
このときping3モジュールを使ってpingの応答を調べますが、順番に実行するとhostsを１周するまでに時間がかかるので、ThreadPoolExecutor()を使って並列にping()を実行して、その結果をhost['status']へ反映します。

def checking_ping_status():
    targets = hosts.copy()
    polling_count = 100
    while targets and (polling_count > 0):
        with ThreadPoolExecutor() as executor:
            results = list(executor.map(lambda t : ping(t['ip'], timeout=1), targets))
            for result, host in zip(results, targets):
                if result:
                    host['status'] = '🟢Reachable'
                else:
                    host['status'] = '🔴Unreachable'
            targets = [t for r, t in zip(results, targets) if not r ]

        hosts_update_status_text()
        polling_count -= 1
        time.sleep(1)