Last active
August 12, 2026 17:07
-
-
Save deekb/fec5f160c9074bfd784607ce55d4c55d to your computer and use it in GitHub Desktop.
KVM switch control utility for "MT-VIKI HDMI QUAD MULTIVIEWER SYNCHRONIZER" KVM switch model: "MT-SW041S" when used from the TinyPilot
This file contains hidden or 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 | |
| """ | |
| kvm_switch.py - Control a KVM switch's [*] + <key> hotkeys via TinyPilot's | |
| USB HID gadget (/dev/hidg0). | |
| The switch's hotkeys all follow the same pattern: | |
| 1. Press and HOLD [*] | |
| 2. Press (and release) the command key(s) | |
| 3. Release [*] | |
| This uses the KEYPAD versions of [*], [+], and the digits 0-9, matching | |
| the switch's own reference card. | |
| Run with no arguments (or `tui`) for an interactive keypress-driven UI. | |
| Run `--help`, or any subcommand, for the scriptable CLI. | |
| Requires root (or write access to /dev/hidg0). Install click if needed: | |
| pip install click --break-system-packages | |
| """ | |
| from __future__ import annotations | |
| import curses | |
| import sys | |
| import time | |
| from dataclasses import dataclass | |
| import click | |
| HIDG_DEVICE = "/dev/hidg0" | |
| KEY_DELAY = 0.05 # seconds between individual key events | |
| # --- HID keycodes (USB HID Usage Tables, keyboard/keypad page) --- | |
| LETTER = { | |
| "A": 0x04, "B": 0x05, "C": 0x06, "D": 0x07, "E": 0x08, "F": 0x09, | |
| "G": 0x0A, "H": 0x0B, "I": 0x0C, "J": 0x0D, "K": 0x0E, "L": 0x0F, | |
| "M": 0x10, "N": 0x11, "O": 0x12, "P": 0x13, "Q": 0x14, "R": 0x15, | |
| "S": 0x16, "T": 0x17, "U": 0x18, "V": 0x19, "W": 0x1A, "X": 0x1B, | |
| "Y": 0x1C, "Z": 0x1D, | |
| } | |
| FKEY = {"F1": 0x3A, "F2": 0x3B, "F3": 0x3C, "F4": 0x3D} | |
| # Keypad digits (NOT top-row digits) | |
| KEYPAD_DIGIT = { | |
| "0": 0x62, "1": 0x59, "2": 0x5A, "3": 0x5B, "4": 0x5C, | |
| "5": 0x5D, "6": 0x5E, "7": 0x5F, "8": 0x60, "9": 0x61, | |
| } | |
| KEY_ASTERISK = 0x55 # keypad * | |
| KEY_PLUS = 0x57 # keypad + | |
| # --- named option tables (from the switch's reference card) --- | |
| RESOLUTION_MODES = { | |
| 1: "4K@30 (default)", | |
| 2: "1080P@60", | |
| 3: "720P@60", | |
| 4: "2560x1440P@60Hz", | |
| 5: "1440x900P@60Hz", | |
| 6: "1600x1200P@60Hz", | |
| } | |
| VIDEOMODE_MODES = { | |
| 0: "4 screen segmentation mode", | |
| 1: "Single screen mode", | |
| 2: "Horizontal preview mode (16:9)", | |
| 3: "PIP mode", | |
| 4: "Horizontal POP mode (16:9)", | |
| 5: "Up and down POP mode (16:9)", | |
| 6: "Horizontal preview mode (full screen)", | |
| 7: "Horizontal POP mode (full screen)", | |
| 8: "Up and down POP mode (full screen)", | |
| } | |
| class HidError(Exception): | |
| """Raised when the HID gadget device can't be written to.""" | |
| # --- low-level HID helpers --- | |
| class HidWriter: | |
| """Writes raw 8-byte HID keyboard reports to the gadget device.""" | |
| def __init__(self, device_path: str = HIDG_DEVICE): | |
| self.device_path = device_path | |
| self.held_keys: list[int] = [] | |
| def _write(self, report: bytes) -> None: | |
| try: | |
| with open(self.device_path, "wb") as hidg: | |
| hidg.write(report) | |
| except PermissionError as exc: | |
| raise HidError( | |
| f"Permission denied writing to {self.device_path} (try sudo)." | |
| ) from exc | |
| except FileNotFoundError as exc: | |
| raise HidError( | |
| f"{self.device_path} not found. Is the HID gadget set up?" | |
| ) from exc | |
| except OSError as exc: | |
| raise HidError(f"Error writing to {self.device_path}: {exc}") from exc | |
| def _send_held(self) -> None: | |
| # 8-byte boot keyboard report: | |
| # byte 0 = modifiers | |
| # byte 1 = reserved | |
| # bytes 2-7 = up to six held keys | |
| report = bytes([0, 0, *self.held_keys[:6]]) | |
| self._write(report) | |
| time.sleep(KEY_DELAY) | |
| def press(self, keycode: int) -> None: | |
| if keycode not in self.held_keys: | |
| self.held_keys.append(keycode) | |
| self._send_held() | |
| def release(self, keycode: int) -> None: | |
| if keycode in self.held_keys: | |
| self.held_keys.remove(keycode) | |
| self._send_held() | |
| def release_all(self) -> None: | |
| self.held_keys.clear() | |
| self._send_held() | |
| def tap(self, keycode: int) -> None: | |
| """Press and release one key without disturbing other held keys.""" | |
| self.press(keycode) | |
| self.release(keycode) | |
| def hotkey(self, *keycodes: int) -> None: | |
| """Hold * while tapping each command key, then release *.""" | |
| self.press(KEY_ASTERISK) | |
| for code in keycodes: | |
| self.tap(code) | |
| self.release(KEY_ASTERISK) | |
| # --- actions shared by CLI and TUI --- | |
| @dataclass | |
| class Action: | |
| key: str # TUI keypress | |
| label: str # short description for menus/help | |
| hotkey_desc: str # e.g. "* + 1" | |
| run: "callable" # fn(hid, prompt_fn) -> None | |
| def switch_port(hid: HidWriter, port: int) -> None: | |
| hid.hotkey(KEYPAD_DIGIT[str(port)]) | |
| def switch_fkey(hid: HidWriter, fkey: str) -> None: | |
| hid.hotkey(FKEY[fkey.upper()]) | |
| def sync_mode(hid: HidWriter) -> None: | |
| hid.hotkey(KEYPAD_DIGIT["0"]) | |
| def lrpass_mode(hid: HidWriter) -> None: | |
| hid.hotkey(LETTER["X"]) | |
| def traversal_mode(hid: HidWriter) -> None: | |
| hid.hotkey(LETTER["Y"]) | |
| def roaming_mode(hid: HidWriter) -> None: | |
| hid.hotkey(LETTER["Z"]) | |
| def toggle_buzzer(hid: HidWriter) -> None: | |
| hid.hotkey(LETTER["B"]) | |
| def toggle_scan(hid: HidWriter) -> None: | |
| hid.hotkey(LETTER["S"]) | |
| def set_scantime(hid: HidWriter, seconds: int) -> None: | |
| hid.hotkey(LETTER["I"], *digits_to_codes(str(seconds))) | |
| def set_audio(hid: HidWriter, channel: int) -> None: | |
| hid.hotkey(LETTER["A"], KEYPAD_DIGIT[str(channel)]) | |
| def set_resolution(hid: HidWriter, mode: int) -> None: | |
| hid.hotkey(LETTER["R"], KEYPAD_DIGIT[str(mode)]) | |
| def set_videomode(hid: HidWriter, mode: int) -> None: | |
| hid.hotkey(LETTER["M"], KEYPAD_DIGIT[str(mode)]) | |
| def modify_vidpid(hid: HidWriter) -> None: | |
| hid.hotkey(KEY_PLUS) | |
| # ============================== CLI (Click) ============================== | |
| CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]} | |
| @click.group( | |
| context_settings=CONTEXT_SETTINGS, | |
| invoke_without_command=True, | |
| help=( | |
| "Control a KVM switch's [*] + <key> hotkeys via TinyPilot's HID gadget.\n\n" | |
| "Run with no command for the interactive TUI, or use a subcommand " | |
| "below to fire a single hotkey from scripts/cron." | |
| ), | |
| ) | |
| @click.option( | |
| "--device", | |
| default=HIDG_DEVICE, | |
| show_default=True, | |
| help="HID gadget device path.", | |
| ) | |
| @click.pass_context | |
| def cli(ctx: click.Context, device: str) -> None: | |
| ctx.ensure_object(dict) | |
| ctx.obj["hid"] = HidWriter(device) | |
| if ctx.invoked_subcommand is None: | |
| ctx.invoke(tui) | |
| @cli.command() | |
| @click.argument("port", type=click.IntRange(1, 4)) | |
| @click.pass_context | |
| def port(ctx: click.Context, port: int) -> None: | |
| """Switch to port 1-4. (hotkey: * + [1-4])""" | |
| click.echo(f"Switching to port {port} ...") | |
| run_safely(lambda: switch_port(ctx.obj["hid"], port)) | |
| @cli.command() | |
| @click.argument("fkey", type=click.Choice(["F1", "F2", "F3", "F4"], case_sensitive=False)) | |
| @click.pass_context | |
| def fkey(ctx: click.Context, fkey: str) -> None: | |
| """Switching mode 1. (hotkey: * + F1..F4)""" | |
| click.echo(f"Sending switching mode 1: * + {fkey.upper()} ...") | |
| run_safely(lambda: switch_fkey(ctx.obj["hid"], fkey)) | |
| @cli.command() | |
| @click.pass_context | |
| def sync(ctx: click.Context) -> None: | |
| """Synchronization mode. (hotkey: * + 0)""" | |
| click.echo("Entering synchronization mode (* + 0) ...") | |
| run_safely(lambda: sync_mode(ctx.obj["hid"])) | |
| @cli.command() | |
| @click.pass_context | |
| def lrpass(ctx: click.Context) -> None: | |
| """Left/right pass mode. (hotkey: * + X)""" | |
| click.echo("Entering left/right pass mode (* + X) ...") | |
| run_safely(lambda: lrpass_mode(ctx.obj["hid"])) | |
| @cli.command() | |
| @click.pass_context | |
| def traversal(ctx: click.Context) -> None: | |
| """Traversal mode. (hotkey: * + Y)""" | |
| click.echo("Entering traversal mode (* + Y) ...") | |
| run_safely(lambda: traversal_mode(ctx.obj["hid"])) | |
| @cli.command() | |
| @click.pass_context | |
| def roaming(ctx: click.Context) -> None: | |
| """Roaming mode. (hotkey: * + Z)""" | |
| click.echo("Entering roaming mode (* + Z) ...") | |
| run_safely(lambda: roaming_mode(ctx.obj["hid"])) | |
| @cli.command() | |
| @click.pass_context | |
| def buzzer(ctx: click.Context) -> None: | |
| """Toggle the buzzer. (hotkey: * + B)""" | |
| click.echo("Toggling buzzer (* + B) ...") | |
| run_safely(lambda: toggle_buzzer(ctx.obj["hid"])) | |
| @cli.command() | |
| @click.pass_context | |
| def scan(ctx: click.Context) -> None: | |
| """Toggle automatic scanning. (hotkey: * + S)""" | |
| click.echo("Toggling automatic scanning (* + S) ...") | |
| run_safely(lambda: toggle_scan(ctx.obj["hid"])) | |
| @cli.command("scantime") | |
| @click.argument("seconds", type=click.IntRange(5, 999)) | |
| @click.pass_context | |
| def scantime_cmd(ctx: click.Context, seconds: int) -> None: | |
| """Set scan time in seconds (5-999). (hotkey: * + I + n)""" | |
| click.echo(f"Setting scan time to {seconds}s (* + I + {seconds}) ...") | |
| run_safely(lambda: set_scantime(ctx.obj["hid"], seconds)) | |
| @cli.command() | |
| @click.argument("channel", type=click.IntRange(0, 4)) | |
| @click.pass_context | |
| def audio(ctx: click.Context, channel: int) -> None: | |
| """Set audio output channel 0-4. (hotkey: * + A + n)""" | |
| click.echo(f"Setting audio channel to {channel} (* + A + {channel}) ...") | |
| run_safely(lambda: set_audio(ctx.obj["hid"], channel)) | |
| @cli.command() | |
| @click.argument("mode", type=click.IntRange(1, 6)) | |
| @click.pass_context | |
| def resolution(ctx: click.Context, mode: int) -> None: | |
| """Set video output resolution 1-6. (hotkey: * + R + n) | |
| \b | |
| 1: 4K@30 (default) | |
| 2: 1080P@60 | |
| 3: 720P@60 | |
| 4: 2560x1440P@60Hz | |
| 5: 1440x900P@60Hz | |
| 6: 1600x1200P@60Hz | |
| """ | |
| desc = RESOLUTION_MODES[mode] | |
| click.echo(f"Setting video resolution mode {mode} ({desc}) (* + R + {mode}) ...") | |
| run_safely(lambda: set_resolution(ctx.obj["hid"], mode)) | |
| @cli.command("videomode") | |
| @click.argument("mode", type=click.IntRange(0, 8)) | |
| @click.pass_context | |
| def videomode_cmd(ctx: click.Context, mode: int) -> None: | |
| """Set video output/split mode 0-8. (hotkey: * + M + n) | |
| \b | |
| 0: 4 screen segmentation mode | |
| 1: Single screen mode | |
| 2: Horizontal preview mode (16:9) | |
| 3: PIP mode | |
| 4: Horizontal POP mode (16:9) | |
| 5: Up and down POP mode (16:9) | |
| 6: Horizontal preview mode (full screen) | |
| 7: Horizontal POP mode (full screen) | |
| 8: Up and down POP mode (full screen) | |
| """ | |
| desc = VIDEOMODE_MODES[mode] | |
| click.echo(f"Setting video output mode {mode} ({desc}) (* + M + {mode}) ...") | |
| run_safely(lambda: set_videomode(ctx.obj["hid"], mode)) | |
| @cli.command() | |
| @click.pass_context | |
| def vidpid(ctx: click.Context) -> None: | |
| """Trigger a VID&PID change. (hotkey: * + +)""" | |
| click.echo("Triggering VID&PID modification (* + +) ...") | |
| run_safely(lambda: modify_vidpid(ctx.obj["hid"])) | |
| @cli.command() | |
| @click.pass_context | |
| def tui(ctx: click.Context) -> None: | |
| """Launch the interactive keypress-driven UI.""" | |
| hid: HidWriter = ctx.obj["hid"] | |
| try: | |
| curses.wrapper(lambda stdscr: run_tui(stdscr, hid)) | |
| except HidError as exc: | |
| raise click.ClickException(str(exc)) | |
| def run_safely(action) -> None: | |
| try: | |
| action() | |
| except HidError as exc: | |
| raise click.ClickException(str(exc)) | |
| click.echo("Done.") | |
| # ============================== TUI (curses) ============================== | |
| MENU = [ | |
| ("1", "Switch to port 1", lambda hid: switch_port(hid, 1)), | |
| ("2", "Switch to port 2", lambda hid: switch_port(hid, 2)), | |
| ("3", "Switch to port 3", lambda hid: switch_port(hid, 3)), | |
| ("4", "Switch to port 4", lambda hid: switch_port(hid, 4)), | |
| ("F1-F4", "Switching mode 1 (press F1..F4)", None), # handled specially | |
| ("s", "Toggle automatic scanning", lambda hid: toggle_scan(hid)), | |
| ("y", "Synchronization mode", lambda hid: sync_mode(hid)), | |
| ("x", "Left/right pass mode", lambda hid: lrpass_mode(hid)), | |
| ("t", "Traversal mode", lambda hid: traversal_mode(hid)), | |
| ("o", "Roaming mode", lambda hid: roaming_mode(hid)), | |
| ("b", "Toggle buzzer", lambda hid: toggle_buzzer(hid)), | |
| ("i", "Set scan time (prompts)", "prompt_scantime"), | |
| ("a", "Set audio channel (prompts)", "prompt_audio"), | |
| ("r", "Set video output resolution (shows options)", "prompt_resolution"), | |
| ("m", "Set video output mode (shows options)", "prompt_videomode"), | |
| ("v", "Modify VID&PID", lambda hid: modify_vidpid(hid)), | |
| ("q", "Quit", None), | |
| ] | |
| FKEY_CURSES = { | |
| curses.KEY_F1: "F1", | |
| curses.KEY_F2: "F2", | |
| curses.KEY_F3: "F3", | |
| curses.KEY_F4: "F4", | |
| } | |
| def draw_menu(stdscr, status: str) -> None: | |
| stdscr.erase() | |
| stdscr.addstr(0, 0, "KVM Switch Control", curses.A_BOLD) | |
| stdscr.addstr(1, 0, "Press a key to send the hotkey. 'q' to quit.") | |
| row = 3 | |
| for key, label, _handler in MENU: | |
| stdscr.addstr(row, 2, f"[{key:>5}] {label}") | |
| row += 1 | |
| row += 1 | |
| stdscr.addstr(row, 0, "-" * 60) | |
| stdscr.addstr(row + 1, 0, f"Status: {status}"[: curses.COLS - 1]) | |
| stdscr.refresh() | |
| def prompt_number(stdscr, prompt: str, lo: int, hi: int) -> int | None: | |
| curses.echo() | |
| curses.curs_set(1) | |
| height, _width = stdscr.getmaxyx() | |
| input_row = height - 1 | |
| stdscr.move(input_row, 0) | |
| stdscr.clrtoeol() | |
| stdscr.addstr(input_row, 0, f"{prompt} ({lo}-{hi}): ") | |
| stdscr.refresh() | |
| try: | |
| raw = stdscr.getstr(input_row, len(prompt) + len(str(lo)) + len(str(hi)) + 5, 6) | |
| value = int(raw.decode().strip()) | |
| except (ValueError, curses.error): | |
| value = None | |
| curses.noecho() | |
| curses.curs_set(0) | |
| if value is None or not (lo <= value <= hi): | |
| return None | |
| return value | |
| def prompt_choice(stdscr, title: str, choices: dict[int, str]) -> int | None: | |
| """Show a numbered list of named options and read back the user's pick.""" | |
| stdscr.erase() | |
| stdscr.addstr(0, 0, title, curses.A_BOLD) | |
| row = 2 | |
| for num in sorted(choices): | |
| stdscr.addstr(row, 2, f"[{num}] {choices[num]}") | |
| row += 1 | |
| row += 1 | |
| prompt_label = "Enter choice (blank to cancel): " | |
| stdscr.addstr(row, 0, prompt_label) | |
| curses.echo() | |
| curses.curs_set(1) | |
| stdscr.refresh() | |
| try: | |
| raw = stdscr.getstr(row, len(prompt_label), 6) | |
| value = int(raw.decode().strip()) | |
| except (ValueError, curses.error): | |
| value = None | |
| curses.noecho() | |
| curses.curs_set(0) | |
| if value is None or value not in choices: | |
| return None | |
| return value | |
| def run_tui(stdscr, hid: HidWriter) -> None: | |
| curses.curs_set(0) | |
| stdscr.keypad(True) | |
| status = "Ready." | |
| while True: | |
| draw_menu(stdscr, status) | |
| ch = stdscr.getch() | |
| try: | |
| if ch in FKEY_CURSES: | |
| fkey_name = FKEY_CURSES[ch] | |
| switch_fkey(hid, fkey_name) | |
| status = f"Sent switching mode 1: * + {fkey_name}" | |
| continue | |
| key = chr(ch) if 0 <= ch < 256 else "" | |
| if key == "q": | |
| break | |
| elif key in ("1", "2", "3", "4"): | |
| switch_port(hid, int(key)) | |
| status = f"Switched to port {key}" | |
| elif key == "s": | |
| toggle_scan(hid) | |
| status = "Toggled automatic scanning" | |
| elif key == "y": | |
| sync_mode(hid) | |
| status = "Entered synchronization mode" | |
| elif key == "x": | |
| lrpass_mode(hid) | |
| status = "Entered left/right pass mode" | |
| elif key == "t": | |
| traversal_mode(hid) | |
| status = "Entered traversal mode" | |
| elif key == "o": | |
| roaming_mode(hid) | |
| status = "Entered roaming mode" | |
| elif key == "b": | |
| toggle_buzzer(hid) | |
| status = "Toggled buzzer" | |
| elif key == "v": | |
| modify_vidpid(hid) | |
| status = "Sent VID&PID modification" | |
| elif key == "i": | |
| value = prompt_number(stdscr, "Scan time seconds", 5, 999) | |
| if value is not None: | |
| set_scantime(hid, value) | |
| status = f"Set scan time to {value}s" | |
| else: | |
| status = "Cancelled (invalid input)" | |
| elif key == "a": | |
| value = prompt_number(stdscr, "Audio channel", 0, 4) | |
| if value is not None: | |
| set_audio(hid, value) | |
| status = f"Set audio channel to {value}" | |
| else: | |
| status = "Cancelled (invalid input)" | |
| elif key == "r": | |
| value = prompt_choice(stdscr, "Video output resolution", RESOLUTION_MODES) | |
| if value is not None: | |
| set_resolution(hid, value) | |
| status = f"Set resolution mode to {value} ({RESOLUTION_MODES[value]})" | |
| else: | |
| status = "Cancelled (invalid input)" | |
| elif key == "m": | |
| value = prompt_choice(stdscr, "Video output mode", VIDEOMODE_MODES) | |
| if value is not None: | |
| set_videomode(hid, value) | |
| status = f"Set video output mode to {value} ({VIDEOMODE_MODES[value]})" | |
| else: | |
| status = "Cancelled (invalid input)" | |
| # any other key: ignore, keep last status | |
| except HidError as exc: | |
| status = f"ERROR: {exc}" | |
| if __name__ == "__main__": | |
| try: | |
| cli(obj={}) | |
| except HidError as exc: | |
| click.echo(f"Error: {exc}", err=True) | |
| sys.exit(1) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment