目的
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)