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

# 目的

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/8721/d039746b-df15-fb8f-a53b-1ce0f24bccb4.png)

# 環境構築

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

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

```requirements.txt
streamlit
pyyaml
ping3
```

```sh
$ 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`パラメータを変更します。

https://github.com/kyan001/ping3/blob/master/TROUBLESHOOTING.MD#permission-denied-on-linux

```sh
# リブートするまで有効
$ 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`からインストールします。

```sh
$ 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アプリが立ち上がります。

```sh
$ 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ボタンが押されるたびに下記順序で関数を実行します。

```python
# 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`にてセッション状態からロードされます。
```python
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"`と表示されます。

```python
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()

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/8721/b4992eb5-26ab-d148-55c8-18c228341677.png)

## checking_button()

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

```python
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']`へ反映します。

```python
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)

```