|
| 1 | +import os |
| 2 | +import re |
| 3 | + |
| 4 | + |
| 5 | +def discover_hubs(uhubctl_binary: str = "uhubctl"): |
| 6 | + cmd = f"{uhubctl_binary}" |
| 7 | + stream = os.popen(cmd) |
| 8 | + |
| 9 | + hubs = [] |
| 10 | + |
| 11 | + pattern_hub = re.compile("Current status for hub ([\.\d-]+)") |
| 12 | + pattern_port = re.compile(" Port (\d+): \d{4} (power|off)") |
| 13 | + |
| 14 | + for line in stream.readlines(): |
| 15 | + reg_hub = pattern_hub.match(line) |
| 16 | + reg_port = pattern_port.match(line) |
| 17 | + |
| 18 | + if reg_hub: |
| 19 | + hub = Hub(reg_hub.group(1), uhubctl_binary) |
| 20 | + hubs.append(hub) |
| 21 | + elif reg_port: |
| 22 | + # assume that the port is the last detected one |
| 23 | + hub = hubs[-1] |
| 24 | + |
| 25 | + if not hub: |
| 26 | + continue |
| 27 | + |
| 28 | + port = Port(hub, reg_port.group(1)) |
| 29 | + hub.ports.append(port) |
| 30 | + |
| 31 | + return hubs |
| 32 | + |
| 33 | + |
| 34 | +class Hub: |
| 35 | + def __init__(self, path: str, uhubctl_binary: str = "uhubctl") -> None: |
| 36 | + self.path = path |
| 37 | + self.ports = [] |
| 38 | + self.uhubctl = uhubctl_binary |
| 39 | + |
| 40 | + def add_port(self, port_number: int) -> 'Port': |
| 41 | + port = Port(self, port_number) |
| 42 | + self.ports.append(port) |
| 43 | + |
| 44 | + return port |
| 45 | + |
| 46 | + def add_ports(self, port_start: int, port_end: int): |
| 47 | + for port_number in range(port_start, port_end): |
| 48 | + self.add_port(port_number) |
| 49 | + |
| 50 | + def __str__(self) -> str: |
| 51 | + return f"USB Hub {self.path}" |
| 52 | + |
| 53 | + |
| 54 | +class Port: |
| 55 | + def __init__(self, hub: Hub, port_number: int): |
| 56 | + self.hub = hub |
| 57 | + self.port_number = port_number |
| 58 | + |
| 59 | + @property |
| 60 | + def status(self) -> bool: |
| 61 | + status = None |
| 62 | + pattern = re.compile(f" Port {self.port_number}: \d{{4}} (power|off)") |
| 63 | + |
| 64 | + cmd = f"{self.hub.uhubctl} -l {self.hub.path} -p {self.port_number}" |
| 65 | + stream = os.popen(cmd) |
| 66 | + |
| 67 | + for line in stream.readlines(): |
| 68 | + reg = pattern.match(line) |
| 69 | + |
| 70 | + if reg: |
| 71 | + status = (reg.group(1) == "power") |
| 72 | + |
| 73 | + if status is None: |
| 74 | + raise Exception() |
| 75 | + |
| 76 | + return status |
| 77 | + |
| 78 | + @status.setter |
| 79 | + def status(self, status: bool) -> None: |
| 80 | + cmd = f"{self.hub.uhubctl} -l {self.hub.path} -p {self.port_number} -a " |
| 81 | + if status: |
| 82 | + cmd += "on" |
| 83 | + else: |
| 84 | + cmd += "off" |
| 85 | + |
| 86 | + stream = os.popen(cmd) |
| 87 | + stream.readlines() |
| 88 | + |
| 89 | + def __str__(self) -> str: |
| 90 | + return f"USB Port {self.hub.path}.{self.port_number}" |
0 commit comments