mirror of
https://github.com/AG7GN/nexus-utilities.git
synced 2025-05-15 22:20:09 -07:00
367 lines
15 KiB
Python
Executable File
367 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import sys
|
|
import signal
|
|
import os
|
|
import subprocess
|
|
import tkinter as tk
|
|
from tkinter import ttk
|
|
import tkinter.font as tkfont
|
|
import collections
|
|
|
|
__author__ = "Steve Magnuson AG7GN"
|
|
__copyright__ = "Copyright 2020, Steve Magnuson"
|
|
__credits__ = ["Steve Magnuson"]
|
|
__license__ = "GPL"
|
|
__version__ = "1.0.2"
|
|
__maintainer__ = "Steve Magnuson"
|
|
__email__ = "ag7gn@arrl.net"
|
|
__status__ = "Production"
|
|
|
|
|
|
class UsbWindow(object):
|
|
"""
|
|
GUI that allows user to click on a USB device to toggle it's state
|
|
between 'bound' (enabled) or 'unbound' (disabled).
|
|
"""
|
|
treeview_font = (None, 12,)
|
|
treeview_header_font = (None, 12, 'bold')
|
|
label_font = (None, 11, 'bold')
|
|
button_font = (None, 12, 'bold')
|
|
max_label_width = 350
|
|
min_window_width = max_label_width + 35
|
|
max_window_height = 380
|
|
|
|
def __init__(self, master):
|
|
self.master = master
|
|
master.title(f"USB Device Manager - version {__version__}")
|
|
ws = master.winfo_screenwidth()
|
|
hs = master.winfo_screenheight()
|
|
pos_x = (ws // 2) - (self.min_window_width // 2)
|
|
pos_y = (hs // 2) - (self.max_window_height // 2)
|
|
master.geometry(f"+{pos_x}+{pos_y}")
|
|
# master.config(bg="skyblue")
|
|
# Make the label frame & label
|
|
self.header = ["ID", "Tag", "Device", "State"]
|
|
self.label_frame = tk.Frame(master=master, borderwidth=5)
|
|
self.label_frame.pack(side='top', fill='both', padx=5, pady=5, expand=True)
|
|
s = """Click on a USB device to toggle enable/disable state.
|
|
(Enabled = bound, Disabled = unbound)"""
|
|
msg = tk.Label(master=self.label_frame,
|
|
wraplength=self.max_label_width,
|
|
justify="center",
|
|
font=self.label_font, fg='blue',
|
|
anchor="center", text=s)
|
|
msg.pack(side='top', padx=5, pady=5)
|
|
self.list_frame = tk.Frame(master=master, borderwidth=5)
|
|
self.list_frame.pack(side='top', fill='both', padx=5, pady=5,
|
|
expand=True)
|
|
style = ttk.Style()
|
|
# Set table header font. 'None' means use the default font.
|
|
style.configure("Treeview.Heading", font=self.treeview_header_font)
|
|
# Set table contents font.
|
|
style.configure("Treeview", font=self.treeview_font)
|
|
self.tree = ttk.Treeview(master=self.list_frame,
|
|
columns=self.header,
|
|
show="headings", style="Treeview")
|
|
self.tree.bind('<ButtonRelease-1>', self._select_item)
|
|
# NOTE: tags are broken in tkinter 8.6.9!
|
|
self.tree.tag_configure('odd', background='#E8E8E8')
|
|
self.tree.tag_configure('even', background='#DFDFDF')
|
|
vsb = tk.Scrollbar(master=self.list_frame,
|
|
orient="vertical",
|
|
command=self.tree.yview)
|
|
# hsb = tk.Scrollbar(master=self.list_frame,
|
|
# orient="horizontal",
|
|
# command=self.tree.xview)
|
|
# self.tree.configure(yscrollcommand=vsb.set,
|
|
# xscrollcommand=hsb.set)
|
|
self.tree.configure(yscrollcommand=vsb.set)
|
|
self.tree.grid(column=0, row=0,
|
|
sticky='nsew', in_=self.list_frame)
|
|
vsb.grid(column=1, row=0, sticky='ns', in_=self.list_frame)
|
|
# hsb.grid(column=0, row=1, sticky='ew', in_=self.list_frame)
|
|
self.list_frame.grid_columnconfigure(0, weight=1)
|
|
self.list_frame.grid_rowconfigure(0, weight=1)
|
|
|
|
self.button_frame = tk.Frame(master=master, borderwidth=5)
|
|
self.button_frame.pack(side='top', fill='both', padx=5, pady=5,
|
|
expand=True)
|
|
# self.refresh_button = tk.Button(master=self.button_frame,
|
|
# text='Refresh List',
|
|
# command=lambda: self._build_tree())
|
|
# self.refresh_button.pack(padx=5, pady=5, side='left',
|
|
# fill='both', expand=True)
|
|
self.quit_button = tk.Button(master=self.button_frame,
|
|
text='Quit',
|
|
font=self.button_font,
|
|
command=lambda:
|
|
self.master.quit())
|
|
self.quit_button.pack(side='left',
|
|
fill='both', expand=True)
|
|
self.current_list = None
|
|
self._update_tree()
|
|
|
|
def _build_headers(self):
|
|
"""
|
|
Constructs the Treeview table headers and auto-adjusts the
|
|
width of each header using the number of characters in the
|
|
column heading.
|
|
|
|
:return: None
|
|
"""
|
|
for col in self.header:
|
|
self.tree.heading(col, text=col.title(),
|
|
command=lambda c=col: self._sort_by(c, 0))
|
|
# Adjust the column's width to the header string
|
|
_width = tkfont.Font(font=self.treeview_header_font).measure(col.title() + '__')
|
|
self.tree.column(col, width=_width)
|
|
|
|
def _build_tree(self):
|
|
"""
|
|
Calls get_usb_devices and inserts results into the tree. Column
|
|
width auto-adjusts based on the character count in the row
|
|
field with the greatest number of characters.
|
|
|
|
:return: None
|
|
"""
|
|
self.tree.delete(*self.tree.get_children())
|
|
i = 0
|
|
tree_width = 0
|
|
for item in self.current_list:
|
|
if i % 2 == 0:
|
|
self.tree.insert('', 'end', values=item, tags=('even',))
|
|
else:
|
|
self.tree.insert('', 'end', values=item, tags=('odd',))
|
|
i += 1
|
|
# adjust column's width if necessary to fit each value
|
|
# Use the column's header string as a minimum width
|
|
tree_width = 0
|
|
for ix, val in enumerate(item):
|
|
min_col_w = self.tree.column(self.header[ix], width=None)
|
|
col_w = tkfont.Font(font=self.treeview_font).measure(val + '__')
|
|
if min_col_w < col_w:
|
|
self.tree.column(self.header[ix], width=col_w)
|
|
tree_width += col_w
|
|
else:
|
|
tree_width += min_col_w
|
|
# Update root window width to accommodate the new tree width
|
|
self.master.update_idletasks()
|
|
self.master.update()
|
|
x = self.master.winfo_width()
|
|
y = self.master.winfo_height()
|
|
window_width = max([tree_width + 35, self.min_window_width])
|
|
if x != window_width:
|
|
self.master.geometry(f"{window_width}x{y}")
|
|
|
|
def _select_item(self, _):
|
|
selected = self.tree.focus()
|
|
if not selected:
|
|
return
|
|
_device = self.tree.item(selected)['values'][2]
|
|
_state = self.tree.item(selected)['values'][3].casefold()
|
|
if _state == "enabled":
|
|
target_state = "unbind"
|
|
else:
|
|
target_state = "bind"
|
|
set_usb_device_state(_device, target_state)
|
|
self._build_tree()
|
|
|
|
def _update_tree(self):
|
|
"""
|
|
Checks to see if the size of the list of USB devices has changed
|
|
and if it has, refresh the tree header and list of devices.
|
|
|
|
:return: None
|
|
"""
|
|
_latest_list = get_usb_devices()
|
|
if collections.Counter(_latest_list) != collections.Counter(self.current_list):
|
|
self.current_list = _latest_list
|
|
self._build_headers()
|
|
self._build_tree()
|
|
self.list_frame.after(1000, self._update_tree)
|
|
|
|
def _sort_by(self, col, descending):
|
|
"""
|
|
Sorts tree contents when a column header is clicked on.
|
|
|
|
:param col: The column to sort on
|
|
:param descending: True if descending, False if ascending sort
|
|
desired
|
|
:return: None
|
|
"""
|
|
data = [(self.tree.set(child, col), child) for child in self.tree.get_children('')]
|
|
# Sort the data in place
|
|
# if the data to be sorted is numeric change to float
|
|
# data = change_numeric(data)
|
|
data.sort(reverse=descending)
|
|
for ix, item in enumerate(data):
|
|
self.tree.move(item[1], '', ix)
|
|
# Switch the sort direction
|
|
self.tree.heading(col,
|
|
command=lambda c=col:
|
|
self._sort_by(col, int(not descending)))
|
|
|
|
|
|
def get_usb_devices() -> list:
|
|
"""Returns list of USB devices that are eligible for binding
|
|
and unbinding. Devices with 'hub' in the description are excluded
|
|
from the list. The returned list consists of tuples, with each
|
|
tuple containing the USB device ID, Tag (Product description), Device
|
|
number, and state. State is "Enabled" if device is bound, "Disabled"
|
|
if unbound.
|
|
|
|
:return: List of tuples of USB devices. If no devices were found,
|
|
returns empty list.
|
|
"""
|
|
import re
|
|
device_re = re.compile("Bus\\s+(?P<bus>\\d+)\\s+"
|
|
"Device\\s+(?P<device>\\d+).+"
|
|
"ID\\s(?P<id>\\w+:\\w+)\\s(?P<tag>.+)$", re.I)
|
|
df = subprocess.check_output("lsusb").decode('utf-8')
|
|
devices = []
|
|
for i in df.split('\n'):
|
|
if i:
|
|
info = device_re.match(i)
|
|
if info:
|
|
dinfo = info.groupdict()
|
|
if " hub" not in dinfo['tag'].casefold():
|
|
_bus = dinfo.pop('bus')
|
|
_device = dinfo.pop('device')
|
|
cmd = f"grep -l {_bus}/{_device} /sys/bus/usb/devices/*/uevent 2>/dev/null | tail -1"
|
|
product = subprocess.check_output(cmd, shell=True).decode('utf-8')
|
|
if product:
|
|
_p = product.split('/')[5]
|
|
if os.path.islink(f"/sys/bus/usb/drivers/usb/{_p}"):
|
|
status = "Enabled"
|
|
else:
|
|
status = 'Disabled'
|
|
devices.append((dinfo['id'], dinfo['tag'],
|
|
_p, status))
|
|
return devices
|
|
|
|
|
|
def set_usb_device_state(_device: str, _action: str) -> bool:
|
|
"""
|
|
Binds (enables) or unbinds (disables) a USB device
|
|
|
|
:param _device: Device designation
|
|
:param _action: bind|unbind Desired setting for device
|
|
:return True if _action was successful, False otherwise
|
|
"""
|
|
cmd = f"echo {_device} | sudo tee /sys/bus/usb/drivers/usb/{_action} 2>/dev/null"
|
|
try:
|
|
_result = subprocess.check_output(cmd, shell=True).decode('utf-8')
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"ERROR: {e}. Do you have sudo permissions to run "
|
|
f"'sudo tee /sys/bus/usb/drivers/usb/{_action}'?",
|
|
file=sys.stderr)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
def find_usb_device(_device_string: str, _action: str) -> bool:
|
|
"""
|
|
Searches the ID and tag for device_string, and if a match is
|
|
found, calls set_usb_device_state to bind/unbind that USB device.
|
|
|
|
:param _device_string: String to search for in USB device ID or Tag
|
|
(Product string)
|
|
:param _action: bind|unbind Desired setting for device
|
|
:return True if device was found AND _action was successful, False
|
|
otherwise
|
|
"""
|
|
devices = get_usb_devices()
|
|
if devices:
|
|
for d in devices:
|
|
if _device_string in d[0] or \
|
|
_device_string.casefold() in d[1].casefold():
|
|
if (_action == "bind" and d[3] == "Enabled") or \
|
|
(_action == "unbind" and d[3] == "Disabled"):
|
|
print(f"ERROR: {_action} requested but device is already {d[3]}",
|
|
file=sys.stderr)
|
|
return False
|
|
_answer = set_usb_device_state(d[2], _action)
|
|
if _answer:
|
|
return True
|
|
else:
|
|
return False
|
|
return False
|
|
|
|
|
|
def sigint_handler(sig, frame):
|
|
print(f"Signal handler caught {sig} {frame}")
|
|
root.quit()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
parser = argparse.ArgumentParser(prog='usb_control.py',
|
|
description=f"USB Device Control")
|
|
parser.add_argument('-v', '--version', action='version',
|
|
version=f"Version: {__version__}")
|
|
parser.add_argument("-l", "--list", action='store_true',
|
|
help="List available USB devices")
|
|
parser.add_argument("-b", "--bind",
|
|
type=str, metavar="STRING",
|
|
help="bind (enable) a usb device containing "
|
|
"STRING (case-insensitive) in 'lsusb' "
|
|
"output ID or Tag fields")
|
|
parser.add_argument("-u", "--unbind",
|
|
type=str, metavar="STRING",
|
|
help="unbind (disable) a usb device containing "
|
|
"STRING (case-insensitive) in 'lsusb' "
|
|
"output ID or Tag fields")
|
|
arg_info = parser.parse_args()
|
|
if not sys.platform.startswith('linux'):
|
|
print(f"This application only works on Linux", file=sys.stderr)
|
|
sys.exit(1)
|
|
if arg_info.list:
|
|
dev_list = get_usb_devices()
|
|
if dev_list:
|
|
try:
|
|
from tabulate import tabulate
|
|
except ModuleNotFoundError:
|
|
print("Python 'tabulate' module required. Run 'sudo "
|
|
"apt update && sudo apt install python3-tabulate' "
|
|
"to install it.", file=sys.stderr)
|
|
sys.exit(1)
|
|
else:
|
|
print(tabulate(dev_list, headers=["ID", "Tag", "Device", "State"]))
|
|
sys.exit(0)
|
|
else:
|
|
print("No USB devices found", file=sys.stderr)
|
|
sys.exit(0)
|
|
if arg_info.bind:
|
|
answer = find_usb_device(arg_info.bind, "bind")
|
|
if answer:
|
|
sys.exit(0)
|
|
else:
|
|
sys.exit(1)
|
|
if arg_info.unbind:
|
|
answer = find_usb_device(arg_info.unbind, "unbind")
|
|
if answer:
|
|
sys.exit(0)
|
|
else:
|
|
sys.exit(1)
|
|
|
|
# If we made it this far, no arguments were supplied. Attempt
|
|
# to open GUI.
|
|
if os.environ.get('DISPLAY', '') == '':
|
|
print(f"ERROR: No $DISPLAY environment. "
|
|
f"Must supply argument to run without X", file=sys.stderr)
|
|
sys.exit(1)
|
|
# os.environ.__setitem__('DISPLAY', ':0.0')
|
|
|
|
root = tk.Tk()
|
|
root.resizable(width=True, height=True)
|
|
signal.signal(signal.SIGINT, sigint_handler)
|
|
# Stop program if Esc key pressed
|
|
root.bind('<Escape>', lambda _: root.quit())
|
|
# Stop program if window is closed at OS level ('X' in upper right
|
|
# corner or red dot in upper left on Mac)
|
|
root.protocol("WM_DELETE_WINDOW", lambda: root.quit())
|
|
UsbWindow(root)
|
|
root.mainloop()
|
|
sys.exit(0)
|