Created
December 31, 2020 19:45
-
-
Save advanceboy/b2bb75b1b1b95d7cd2af15c199c4a5b5 to your computer and use it in GitHub Desktop.
GUI 上で指定した Delay と Loss率 を使って、 tc qdisc による遅延・パケロスの設定の追加を行う、 Python コード
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
from typing import List | |
import tkinter as tk | |
import tkinter.messagebox as messagebox | |
import tkinter.ttk as ttk | |
import subprocess | |
ETH0 = 'eth0' | |
ETH1 = 'eth1' | |
TC_PATH = '/usr/sbin/tc' | |
COMBO_DELAY_LIST = ('0ms', '1ms', '2ms', '4ms', '7ms', '10ms', '20ms', '40ms', '70ms', '100ms', '200ms', '400ms', '700ms', '1000ms') | |
COMBO_LOSS_LIST = ('0%', '1%', '2%', '3%', '4%', '5%', '10%', '20%', '50%', '100%') | |
def run_exec(commands: List[str], *, encoding: str='utf-8'): | |
result = subprocess.run( | |
commands, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True | |
) | |
return (result.stdout.decode(encoding), result.stderr.decode(encoding), result.returncode) | |
def has_netem(interface: str) -> bool: | |
stdout, stderr, code = run_exec(f'sudo { TC_PATH } qdisc show dev { interface }'.split()) | |
if code != 0: | |
raise Exception(f'Unexpcedted returncode: {code}\n' + stderr) | |
return 'netem' in stdout | |
def set_netem(interface: str, delay: str, loss: str) -> None: | |
cmd = 'change' if has_netem(interface) else 'add' | |
_, stderr, code = run_exec(f'sudo { TC_PATH } qdisc { cmd } dev { interface } root netem delay { delay } loss { loss }'.split()) | |
if code != 0: | |
raise Exception(f'Unexpcedted returncode: {code}\n' + stderr) | |
def reset_netem(interface: str) -> None: | |
if has_netem(interface): | |
_, stderr, code = run_exec(f'sudo { TC_PATH } qdisc del dev { interface } root netem'.split()) | |
if code != 0: | |
raise Exception(f'Unexpcedted returncode: {code}\n' + stderr) | |
def main(): | |
root = tk.Tk() | |
root.tk.call('tk', 'scaling', 1.5) | |
# Delay | |
tk.Label(root, text='Delay:').grid(column=0, row=0) | |
v_delay = tk.StringVar() | |
cmb_delay = ttk.Combobox(root, state='readonly', values=COMBO_DELAY_LIST, textvariable=v_delay) | |
cmb_delay.current(0) | |
cmb_delay.grid(column=1, row=0) | |
# Loss | |
tk.Label(root, text='Loss:').grid(column=0, row=1) | |
v_loss = tk.StringVar() | |
cmb_loss = ttk.Combobox(root, state='readonly', values=COMBO_LOSS_LIST, textvariable=v_loss) | |
cmb_loss.current(0) | |
cmb_loss.grid(column=1, row=1) | |
btn_apply = tk.Button(root, text='Apply') | |
btn_reset = tk.Button(root, text='Reset') | |
def on_clicked_apply(): | |
btn_apply['state'] = tk.DISABLED | |
try: | |
delay = v_delay.get() | |
loss = v_loss.get() | |
set_netem(interface=ETH0, delay=delay, loss=loss) | |
set_netem(interface=ETH1, delay=delay, loss=loss) | |
finally: | |
btn_apply['state'] = tk.NORMAL | |
def on_clicked_reset(): | |
btn_reset['state'] = tk.DISABLED | |
try: | |
reset_netem(interface=ETH0) | |
reset_netem(interface=ETH1) | |
finally: | |
btn_reset['state'] = tk.NORMAL | |
btn_apply['command'] = on_clicked_apply | |
btn_apply.grid(row=2, columnspan=2) | |
btn_reset['command'] = on_clicked_reset | |
btn_reset.grid(row=3, columnspan=2) | |
root.mainloop() | |
reset_netem(interface=ETH0) | |
reset_netem(interface=ETH1) | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment