# /// script # requires-python = ">=3.11" # dependencies = ["pyserial>=3.5", "pynput>=1.7", "psutil>=5.9", "HardwareMonitor>=1.2; sys_platform == 'win32'", # "pystray>=0.19; sys_platform == 'win32'", "pillow>=10; sys_platform == 'win32'"] # /// """Macropad PC helper for Windows 11 and macOS. Reads K1..K3 from the Arduino Nano, runs the action from config.toml and sends "S:cpu_load,gpu_load,ram_load,cpu_temp,gpu_temp,ram_temp,ram_used,ram_total" back for the OLED every second. uv run macropad_helper.py run (finds the Nano by itself) uv run macropad_helper.py --stats print loads and temperatures once and exit uv run macropad_helper.py --key K1 run the K1 action once, no Nano needed uv run macropad_helper.py --ports list serial ports uv run macropad_helper.py --flash write the bundled PC firmware to the Nano (asks nothing) uv run macropad_helper.py --flash games write the games firmware (no PC needed to play) uv run macropad_helper.py --flash FILE.hex write any .hex (also: drop a .hex onto macropad.exe) If a Nano without the macropad firmware (or with an older one) is plugged in, the helper asks once and flashes macropad_nano.hex itself: no Arduino IDE or avrdude needed. Windows: macropad.exe (built by GitHub Actions, see build_exe.yml) runs without a console, with a tray icon; its config.toml and log live in %APPDATA%/Macropad (win_app.py). build_exe.cmd builds the same exe locally on Windows. Sensors: Windows - LibreHardwareMonitorLib built in (admin rights; CPU temperature also needs the PawnIO driver), or the LibreHardwareMonitor program with its web server on (port 8085) when run without admin; macOS (Apple Silicon) - macmon (brew install macmon). CPU load and RAM fall back to psutil. """ import argparse import json import os import platform import queue import shlex import shutil import subprocess import sys import threading import time import tomllib import urllib.request import webbrowser from pathlib import Path # next to the script, or next to macropad.exe when built with PyInstaller (build_exe.cmd) FROZEN = getattr(sys, 'frozen', False) HERE = Path(sys.executable).resolve().parent if FROZEN else Path(__file__).resolve().parent BUNDLE = Path(getattr(sys, '_MEIPASS', HERE)) HELLO = 'MACROPAD' FIRMWARE_VERSION = 7 # the bundled macropad_nano.hex answers "MACROPAD 7" FIRMWARE_HEX = 'macropad_nano.hex' GAMES_HEX = 'macropad_games.hex' # firmware/macropad_games: 7 games, works without a PC GAMES = -1 # probe() version of a board that answers "MACROPAD-GAMES n" FLASH_REQUEST = {'hex': None} # set from the tray menu; the serial loop flashes it GAMES_PORTS = set() # ports with the games firmware: never probed again while plugged in, # because opening the port resets the Nano and would restart the game BAUD = 115200 KNOWN_USB_VIDS = {0x1A86, 0x0403, 0x2341, 0x2A03, 0x10C4} # CH340, FTDI, Arduino, Arduino.org, CP210x IS_WINDOWS = platform.system() == 'Windows' IS_MAC = platform.system() == 'Darwin' OS_KEY = 'windows' if IS_WINDOWS else 'macos' if IS_MAC else 'linux' TRAY = FROZEN and IS_WINDOWS # macropad.exe: no console, tray icon, settings in %APPDATA% STATS_INTERVAL = 1.0 # seconds between OLED updates unless config.toml sets temp_interval STATE = {'status': 'шукаю плату', 'changed': threading.Event()} if TRAY: import win_app DEFAULT_CONFIG = win_app.APP_DIR / 'config.toml' LOG_PATH = win_app.APP_DIR / 'macropad.log' else: DEFAULT_CONFIG = HERE / 'config.toml' LOG_PATH = None def set_status(text): STATE['status'] = text STATE['changed'].set() def log(msg): print(time.strftime('%H:%M:%S'), msg, flush=True) class Config: def __init__(self, path): self.path = Path(path) self.mtime = None self.data = {} self.reload() def reload(self): mtime = self.path.stat().st_mtime if mtime == self.mtime: return False with open(self.path, 'rb') as f: data = tomllib.load(f) self.data, self.mtime = data, mtime return True def maybe_reload(self): try: if self.reload(): log(f'config reloaded: {self.path.name}') except (OSError, tomllib.TOMLDecodeError) as e: log(f'config not reloaded, keeping the old one: {e}') def get(self, key, default=None): return self.data.get(key, default) def action(self, key): """The [K1] table, with [K1.windows] / [K1.macos] replacing it on that system.""" table = self.data.get(key) if not isinstance(table, dict): return None override = table.get(OS_KEY) if isinstance(override, dict): return override return {k: v for k, v in table.items() if k not in ('windows', 'macos', 'linux')} def key_by_name(name): from pynput.keyboard import Key, KeyCode n = name.strip().lower() aliases = { 'ctrl': Key.ctrl, 'control': Key.ctrl, 'alt': Key.alt, 'option': Key.alt, 'opt': Key.alt, 'shift': Key.shift, 'cmd': Key.cmd, 'command': Key.cmd, 'win': Key.cmd, 'super': Key.cmd, 'enter': Key.enter, 'return': Key.enter, 'esc': Key.esc, 'escape': Key.esc, 'tab': Key.tab, 'space': Key.space, 'backspace': Key.backspace, 'delete': Key.delete, 'del': Key.delete, 'up': Key.up, 'down': Key.down, 'left': Key.left, 'right': Key.right, 'home': Key.home, 'end': Key.end, 'pageup': Key.page_up, 'pagedown': Key.page_down, 'play': Key.media_play_pause, 'playpause': Key.media_play_pause, 'next': Key.media_next, 'prev': Key.media_previous, 'previous': Key.media_previous, 'volup': Key.media_volume_up, 'voldown': Key.media_volume_down, 'mute': Key.media_volume_mute, } if n in aliases: return aliases[n] if n.startswith('f') and n[1:].isdigit() and hasattr(Key, n): return getattr(Key, n) if len(n) == 1: return KeyCode.from_char(n) raise ValueError(f'unknown key name "{name}"') def press_combo(combo): from pynput.keyboard import Controller keys = [key_by_name(part) for part in combo.split('+')] kb = Controller() for k in keys: kb.press(k) time.sleep(0.03) for k in reversed(keys): kb.release(k) def type_text(text): from pynput.keyboard import Controller Controller().type(text) def open_target(target): target = os.path.expandvars(os.path.expanduser(target)) if '://' in target: webbrowser.open(target) elif IS_WINDOWS: os.startfile(target) elif IS_MAC: subprocess.Popen(['open', target]) else: subprocess.Popen(['xdg-open', target]) def run_command(command): if isinstance(command, str): command = shlex.split(command, posix=not IS_WINDOWS) command = [os.path.expandvars(os.path.expanduser(c)) for c in command] flags = subprocess.CREATE_NEW_PROCESS_GROUP if IS_WINDOWS else 0 subprocess.Popen(command, cwd=Path.home(), creationflags=flags, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) ACTIONS = {'combo': press_combo, 'type': type_text, 'open': open_target, 'run': run_command} def do_action(config, key): table = config.action(key) if not table: log(f'{key}: nothing set in {config.path.name}') return for name, func in ACTIONS.items(): if name in table: log(f'{key}: {name} {table[name]!r}') try: func(table[name]) except Exception as e: log(f'{key}: failed: {e}') return log(f'{key}: no combo / type / open / run in the table') STATS = ('cpu_load', 'gpu_load', 'ram_load', 'cpu_temp', 'gpu_temp', 'ram_temp', 'ram_used_gb', 'ram_total_gb') # order of the "S:" line GIB = 1024 ** 3 CPU_TEMP_NAMES = ('CPU Package', 'Core (Tctl/Tdie)', 'Core (Tdie)', 'Core (Tctl)', 'Core Average', 'CPU Cores') GPU_TEMP_NAMES = ('GPU Core', 'GPU Hot Spot') CPU_LOAD_NAMES = ('CPU Total',) GPU_LOAD_NAMES = ('GPU Core', 'D3D 3D') RAM_LOAD_NAMES = ('Memory',) def pick(sensors, preferred): for name in preferred: if name in sensors: return sensors[name] return max(sensors.values()) if sensors else None def device_of(sensor_id): """'/amdcpu/0/temperature/2' -> 'cpu', '/gpu-nvidia/0/load/0' -> 'gpu', '/ram/load/0' -> 'ram'.""" parts = sensor_id.split('/') head = parts[1].lower() if len(parts) > 1 else '' if head.startswith('gpu'): return 'gpu' if 'cpu' in head: return 'cpu' if head in ('ram', 'memory') or 'dimm' in head or 'dimm' in sensor_id.lower(): return 'ram' return None def lhm_tree_readings(root): """LibreHardwareMonitor data.json tree -> [(sensor id, type, name, value)].""" out = [] def walk(node): sid = node.get('SensorId', '') value = node.get('RawValue') if sid and value is None: text = str(node.get('Value', '')).replace(',', '.').split()[0:1] value = float(text[0]) if text else None if sid and value is not None: out.append((sid, node.get('Type'), node.get('Text', ''), float(value))) for child in node.get('Children', []): walk(child) walk(root) return out def summarize(readings): """[(sensor id, type, name, value)] from LibreHardwareMonitor -> loads in % and temperatures in C.""" found = {} for sid, kind, name, value in readings: device = device_of(sid) if device and kind in ('Temperature', 'Load'): found.setdefault((device, kind), {})[name] = value temps = {k: {n: v for n, v in s.items() if v > 0} for k, s in found.items() if k[1] == 'Temperature'} return { 'cpu_load': pick(found.get(('cpu', 'Load'), {}), CPU_LOAD_NAMES), 'gpu_load': pick(found.get(('gpu', 'Load'), {}), GPU_LOAD_NAMES), 'ram_load': pick(found.get(('ram', 'Load'), {}), RAM_LOAD_NAMES), 'cpu_temp': pick(temps.get(('cpu', 'Temperature'), {}), CPU_TEMP_NAMES), 'gpu_temp': pick(temps.get(('gpu', 'Temperature'), {}), GPU_TEMP_NAMES), 'ram_temp': pick(temps.get(('ram', 'Temperature'), {}), ()), } def lhm_stats(url): """LibreHardwareMonitor web server (data.json) -> loads in % and temperatures in C.""" with urllib.request.urlopen(url, timeout=1.5) as r: return summarize(lhm_tree_readings(json.load(r))) def macmon_stats(sample): """One `macmon pipe` JSON line -> loads in % and temperatures in C. Macs have no RAM sensor.""" temp = sample.get('temp') or {} mem = sample.get('memory') or {} gpu = sample.get('gpu_usage') or [None, None] cpu_load = sample.get('cpu_usage_pct') ram_load = 100.0 * mem['ram_usage'] / mem['ram_total'] if mem.get('ram_total') else None return { 'cpu_load': None if cpu_load is None else 100.0 * cpu_load, 'gpu_load': None if gpu[1] is None else 100.0 * gpu[1], 'ram_load': ram_load, 'cpu_temp': temp.get('cpu_temp_avg'), 'gpu_temp': temp.get('gpu_temp_avg'), 'ram_temp': None, 'ram_used_gb': mem['ram_usage'] / GIB if mem.get('ram_usage') is not None else None, 'ram_total_gb': mem['ram_total'] / GIB if mem.get('ram_total') else None, } class MacmonReader: """Keeps `macmon pipe` running and remembers its last sample.""" def __init__(self, interval_ms): self.sample = {} self.updated = 0.0 self.exe = shutil.which('macmon') or next((p for p in ('/opt/homebrew/bin/macmon', '/usr/local/bin/macmon') if os.path.exists(p)), None) if self.exe: threading.Thread(target=self._run, args=(interval_ms,), daemon=True).start() def _run(self, interval_ms): while True: try: proc = subprocess.Popen([self.exe, 'pipe', '-i', str(interval_ms)], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True) for line in proc.stdout: try: self.sample = json.loads(line) except json.JSONDecodeError: continue self.updated = time.time() except OSError as e: log(f'macmon failed: {e}') time.sleep(5) def stats(self): if not self.exe: raise RuntimeError('macmon not found, install it: brew install macmon') if time.time() - self.updated > 10: return {} return macmon_stats(self.sample) class Stats: def __init__(self, config): import psutil self.config = config self.psutil = psutil psutil.cpu_percent(interval=None) # the first call only starts the measurement self.macmon = MacmonReader(int(interval(config) * 1000)) if IS_MAC else None self.lib = None self.last_error = None if IS_WINDOWS and config.get('lhm_url') is None: self.lib = self._open_lib() def _open_lib(self): """Built-in LibreHardwareMonitorLib; needs admin rights, else the web server is used.""" try: import win_app if not win_app.is_admin(): log('no admin rights: sensors from the LibreHardwareMonitor web server') return None lib = win_app.LhmLib() log('sensors: LibreHardwareMonitorLib built in') return lib except Exception as e: log(f'built-in sensors unavailable ({e}); using the LibreHardwareMonitor web server') return None def reopen(self): if self.lib is not None: self.lib.close() self.lib = self._open_lib() def read(self): stats = {} try: if IS_WINDOWS and self.lib is not None: stats = summarize(self.lib.readings()) elif IS_WINDOWS: stats = lhm_stats(self.config.get('lhm_url', 'http://localhost:8085/data.json')) elif IS_MAC: stats = self.macmon.stats() self.last_error = None except Exception as e: if str(e) != self.last_error: hint = ' (is LibreHardwareMonitor running as admin with Options > Remote Web Server on?)' if IS_WINDOWS and self.lib is None else '' log(f'sensors unavailable, CPU load and RAM from the system only: {e}{hint}') self.last_error = str(e) if stats.get('cpu_load') is None: stats['cpu_load'] = self.psutil.cpu_percent(interval=None) if stats.get('ram_load') is None or stats.get('ram_total_gb') is None: vm = self.psutil.virtual_memory() stats.setdefault('ram_load', None) stats['ram_load'] = vm.percent if stats['ram_load'] is None else stats['ram_load'] stats['ram_used_gb'] = (vm.total - vm.available) / GIB stats['ram_total_gb'] = vm.total / GIB stats['ram_temp'] = None # not shown on the macropad (Roman, 12.11); the slot stays in the line return {k: stats.get(k) for k in STATS} def interval(config): return float(config.get('temp_interval', STATS_INTERVAL)) class Sampler: """Reads the sensors in its own thread, so a slow sensor read never delays a key press.""" def __init__(self, stats): self.stats = stats self.latest = None threading.Thread(target=self._run, daemon=True).start() def _run(self): while True: started = time.time() try: self.latest = self.stats.read() except Exception as e: log(f'sensor read failed: {e}') time.sleep(max(0.2, interval(self.stats.config) - (time.time() - started))) def stats_line(stats): """S:cpu_load,gpu_load,ram_load,cpu_temp,gpu_temp,ram_temp,ram_used x10,ram_total x10 (GB in tenths).""" def fmt(v, scale=1): return '-' if v is None else str(int(round(v * scale))) return 'S:' + ','.join(fmt(stats.get(k), 10 if k.endswith('_gb') else 1) for k in STATS) def candidate_ports(): from serial.tools import list_ports ports = [p for p in list_ports.comports() if 'Bluetooth' not in p.device and 'debug' not in p.device] ports.sort(key=lambda p: (p.vid not in KNOWN_USB_VIDS, p.device)) if IS_MAC: ports = [p for p in ports if p.device.startswith('/dev/cu.')] return ports def parse_hex(text): """Intel HEX -> flash image (bytes from address 0). Checks every record checksum.""" data, base = {}, 0 for n, line in enumerate(text.splitlines(), 1): line = line.strip() if not line: continue if not line.startswith(':'): raise ValueError(f'hex line {n}: no ":"') raw = bytes.fromhex(line[1:]) if sum(raw) & 0xFF: raise ValueError(f'hex line {n}: bad checksum') count, addr, kind, payload = raw[0], int.from_bytes(raw[1:3], 'big'), raw[3], raw[4:4 + raw[0]] if kind == 0: for i, b in enumerate(payload): data[base + addr + i] = b elif kind == 1: break elif kind == 2: base = int.from_bytes(payload, 'big') << 4 elif kind == 4: base = int.from_bytes(payload, 'big') << 16 if not data: raise ValueError('hex file has no data') image = bytearray(b'\xff' * (max(data) + 1)) for a, b in data.items(): image[a] = b return bytes(image) def find_firmware(name=FIRMWARE_HEX): """A firmware .hex next to the script or the exe, or bundled inside the exe.""" places = [HERE / name, BUNDLE / name] for f in places: if f.is_file(): return f return None class Stk500: """Arduino Nano bootloader (STK500 v1, optiboot or the old ATmegaBOOT).""" OK, INSYNC = b'\x10', b'\x14' PAGE = 128 # ATmega328P flash page, bytes SIGNATURE = b'\x1e\x95\x0f' # ATmega328P def __init__(self, port, baud): import serial self.ser = serial.Serial(port, baud, timeout=0.5) def close(self): self.ser.close() def reset(self): # DTR low -> high pulses the Nano's RESET through its 100 nF capacitor self.ser.dtr = False self.ser.rts = False time.sleep(0.25) self.ser.dtr = True self.ser.rts = True time.sleep(0.05) self.ser.reset_input_buffer() def cmd(self, payload, reply_len=0): self.ser.write(payload + b'\x20') head = self.ser.read(1) if head != self.INSYNC: raise IOError(f'bootloader not in sync (got {head!r})') body = self.ser.read(reply_len) tail = self.ser.read(1) if len(body) != reply_len or tail != self.OK: raise IOError('bootloader reply incomplete') return body def sync(self, tries=12): for _ in range(tries): self.ser.reset_input_buffer() try: self.cmd(b'\x30') return True except IOError: time.sleep(0.05) return False def load_address(self, byte_addr): word = byte_addr // 2 self.cmd(b'\x55' + bytes([word & 0xFF, word >> 8])) def flash(self, image, progress=None): sig = self.cmd(b'\x75', 3) if sig != self.SIGNATURE: raise IOError(f'not an ATmega328P (signature {sig.hex()})') self.cmd(b'\x50') # enter programming mode pages = range(0, len(image), self.PAGE) for i, addr in enumerate(pages): chunk = image[addr:addr + self.PAGE].ljust(self.PAGE, b'\xff') self.load_address(addr) self.cmd(b'\x64' + len(chunk).to_bytes(2, 'big') + b'F' + chunk) if progress: progress('write', i + 1, len(pages)) for i, addr in enumerate(pages): chunk = image[addr:addr + self.PAGE].ljust(self.PAGE, b'\xff') self.load_address(addr) back = self.cmd(b'\x74' + len(chunk).to_bytes(2, 'big') + b'F', len(chunk)) if back != chunk: raise IOError(f'verify failed at 0x{addr:04x}') if progress: progress('verify', i + 1, len(pages)) self.cmd(b'\x51') # leave programming mode: the sketch starts def flash_board(port, hex_path): """Write hex_path to the Nano on port; tries the new (115200) and the old (57600) bootloader.""" image = parse_hex(Path(hex_path).read_text()) log(f'flashing {Path(hex_path).name} ({len(image)} bytes) to {port}') def progress(stage, i, n): if i == n or i % 32 == 0: log(f' {stage}: {i}/{n} pages') for baud in (115200, 57600): board = Stk500(port, baud) try: board.reset() if not board.sync(): log(f' no bootloader answer at {baud} baud') continue board.flash(image, progress) log(f' done at {baud} baud, written and verified') return True finally: board.close() log('flashing failed: no bootloader answered (close Arduino IDE serial monitor, replug the board, try again)') return False def probe(name): """Open the port, ask "?". Returns (open serial or None, firmware version or None).""" import serial try: ser = serial.Serial(name, BAUD, timeout=0.2) except (serial.SerialException, OSError): return None, None time.sleep(2.0) # opening the port resets the Nano; wait for its bootloader ser.reset_input_buffer() ser.write(b'?\n') deadline = time.time() + 1.5 while time.time() < deadline: line = ser.readline().decode(errors='replace').strip() if line.startswith(HELLO + '-GAMES'): return ser, GAMES if line.startswith(HELLO): parts = line.split() version = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0 return ser, version ser.close() return None, None def ask(question): """Yes/no: a dialog in macropad.exe, the console otherwise; False when nobody can answer.""" if TRAY: return win_app.message(question, question=True) if not sys.stdin or not sys.stdin.isatty(): return False try: return input(question + ' [т/N] ').strip().lower() in ('т', 'так', 'y', 'yes', 'д') except EOFError: return False OFFERED = set() def offer_flash(name, reason): """Ask once per port and session; flash the bundled firmware on yes.""" hex_path = find_firmware() if name in OFFERED or hex_path is None: return False OFFERED.add(name) if not ask(f'{reason} на {name}. Записати прошивку макропада для ПК? Те, що зараз на платі, зітреться.'): log(f'not flashing {name}; run with --flash to do it later') return False return flash_board(name, hex_path) def open_nano(port_name=None): ports = [port_name] if port_name else [p.device for p in candidate_ports()] from serial.tools import list_ports known = {p.device for p in list_ports.comports() if p.vid in KNOWN_USB_VIDS} GAMES_PORTS.intersection_update(p.device for p in list_ports.comports()) # unplugged: forget it for name in ports: if name in GAMES_PORTS: continue ser, version = probe(name) if ser is not None and version == GAMES: ser.close() log(f'{name}: games firmware on the board') set_status('на платі ігри') if offer_flash(name, 'На платі ігрова прошивка макропада'): return open_nano(name) GAMES_PORTS.add(name) continue if ser is not None and version >= FIRMWARE_VERSION: log(f'macropad found on {name}: firmware {version}') return ser if ser is not None: ser.close() if offer_flash(name, f'Стара прошивка макропада ({version}, є {FIRMWARE_VERSION})'): return open_nano(name) continue if name in known and name not in OFFERED: # a board still booting (just flashed, just plugged in) can miss the first "?": ask again ser, version = probe(name) if ser is not None: ser.close() return open_nano(name) if offer_flash(name, 'Плата Arduino без прошивки макропада'): return open_nano(name) return None def board_port(port=None): """The one Arduino-like board, or None (with a log line) when there are none or several.""" if port: return port from serial.tools import list_ports boards = [p.device for p in list_ports.comports() if p.vid in KNOWN_USB_VIDS] if len(boards) != 1: log(f'plug in exactly one Nano (found: {", ".join(boards) or "none"})') return None return boards[0] def request_flash(hex_name): """Tray menu: confirm, then let the serial loop write this firmware (it owns the port).""" text = ('Записати на плату ігри? Прошивка для ПК зітреться; повернути її можна тут же в меню значка.' if hex_name == GAMES_HEX else 'Записати на плату прошивку макропада для ПК?') if ask(text): FLASH_REQUEST['hex'] = hex_name def do_flash_request(port): hex_name = FLASH_REQUEST['hex'] FLASH_REQUEST['hex'] = None hex_path = find_firmware(hex_name) port = board_port(port) if hex_path is None or port is None: message = 'Плату не знайдено: підключіть одну Arduino Nano і спробуйте ще раз.' if hex_path else f'{hex_name} не знайдено.' else: set_status('записую прошивку') GAMES_PORTS.discard(port) ok = flash_board(port, hex_path) OFFERED.add(port) # the user chose this firmware: do not offer to replace it again message = ('Готово: ' + ('ігри на платі. Їх можна живити від будь-якої USB-зарядки.' if hex_name == GAMES_HEX else 'прошивка для ПК на платі.')) if ok else 'Не вдалося записати: перепідключіть плату й спробуйте ще раз.' if TRAY: win_app.message(message) else: log(message) def offer_pawnio(stats, sampler): """macropad.exe: CPU temperature missing and no PawnIO driver -> offer to install it once.""" time.sleep(5) latest = sampler.latest or {} if stats.lib is None or latest.get('cpu_temp') is not None or win_app.pawnio_installed() or not win_app.once('pawnio'): return if not win_app.message('Щоб показувати температуру процесора, Windows потрібен драйвер PawnIO ' '(безкоштовний, його використовує LibreHardwareMonitor). Встановити зараз?', question=True): log('PawnIO declined; CPU temperature stays empty') return set_status('встановлюю драйвер PawnIO') if win_app.install_pawnio(log): stats.reopen() win_app.message('Драйвер встановлено. Температура процесора з\'явиться за кілька секунд.') else: win_app.message('Не вдалося встановити PawnIO. Його можна поставити вручну з pawnio.eu, потім перезапустити макропад.') def run(config, port_name): import serial stats = Stats(config) sampler = Sampler(stats) actions = queue.Queue() if TRAY: threading.Thread(target=offer_pawnio, args=(stats, sampler), daemon=True).start() def worker(): while True: do_action(config, actions.get()) threading.Thread(target=worker, daemon=True).start() waiting_logged = False while True: if FLASH_REQUEST['hex']: do_flash_request(port_name or config.get('port')) ser = open_nano(port_name or config.get('port')) if ser is None: if GAMES_PORTS: set_status('на платі ігри; прошивка для ПК — у цьому меню') elif not waiting_logged: set_status('плату не знайдено, чекаю') if not waiting_logged: log('macropad not found, waiting (plug it in; close Arduino IDE serial monitor)') waiting_logged = True time.sleep(3) continue waiting_logged = False next_temp = 0.0 set_status(f'підключено, {ser.port}') try: while not FLASH_REQUEST['hex']: line = ser.readline().decode(errors='replace').strip() if line in ('K1', 'K2', 'K3'): config.maybe_reload() actions.put(line) elif line: log(f'nano: {line}') if time.time() >= next_temp and sampler.latest is not None: ser.write((stats_line(sampler.latest) + '\n').encode()) next_temp = time.time() + interval(config) except (serial.SerialException, OSError) as e: log(f'connection lost: {e}') set_status('зв\'язок з платою втрачено') try: ser.close() except Exception: pass time.sleep(1) else: ser.close() # a flash request: the next round writes the firmware def setup_tray_files(): """macropad.exe: settings and log in %APPDATA%\\Macropad; a fresh config.toml from the bundled one.""" win_app.APP_DIR.mkdir(parents=True, exist_ok=True) if LOG_PATH.exists() and LOG_PATH.stat().st_size > 1_000_000: LOG_PATH.unlink() sys.stdout = sys.stderr = open(LOG_PATH, 'a', encoding='utf-8', buffering=1) if not DEFAULT_CONFIG.exists(): DEFAULT_CONFIG.write_bytes((BUNDLE / 'config.toml').read_bytes()) def main_tray(config, port): if not win_app.single_instance(): win_app.message('Макропад уже працює: його значок біля годинника (може ховатися під стрілкою ^).') return 0 log(f'system: {OS_KEY}, config: {config.path}, exe: {sys.executable}') if not win_app.is_admin(): log('started without admin rights: CPU temperature will be missing') if win_app.once('autostart') and not win_app.autostart_enabled(): if win_app.message('Запускати макропад разом з Windows?\n\nЦе можна змінити пізніше: значок біля годинника, ' '«Запускати разом з Windows».', question=True): win_app.set_autostart(True) win_app.run_tray(STATE, lambda: run(config, port), config.path, LOG_PATH, request_flash) log('stopped from the tray menu') return 0 def main(): if TRAY: setup_tray_files() ap = argparse.ArgumentParser(description='Macropad helper: keys -> actions, temperatures -> OLED') ap.add_argument('--config', default=str(DEFAULT_CONFIG)) ap.add_argument('--port', help='serial port, e.g. COM5 or /dev/cu.usbserial-1410 (default: search)') ap.add_argument('--stats', '--temps', action='store_true', help='print loads and temperatures once and exit') ap.add_argument('--key', help='run one action (K1, K2 or K3) and exit') ap.add_argument('--ports', action='store_true', help='list serial ports and exit') ap.add_argument('--flash', nargs='?', const='pc', metavar='pc|games|FILE.hex', help='write a firmware to the Nano and exit: pc (default), games or a .hex file') ap.add_argument('hex', nargs='?', help=argparse.SUPPRESS) # a .hex dropped onto macropad.exe args = ap.parse_args() if args.hex and args.hex.lower().endswith('.hex') and not args.flash: args.flash = args.hex if args.ports: for p in candidate_ports(): print(p.device, '-', p.description, f'(VID {p.vid:04X})' if p.vid else '') return 0 if args.flash: named = {'pc': FIRMWARE_HEX, 'games': GAMES_HEX} hex_path = find_firmware(named[args.flash]) if args.flash in named else Path(args.flash) if hex_path is None or not hex_path.is_file(): log(f'{args.flash}: firmware file not found') return 1 if TRAY and not ask(f'Записати {hex_path.name} на плату? Те, що зараз на платі, зітреться.'): return 0 port = board_port(args.port) ok = port is not None and flash_board(port, hex_path) if TRAY: win_app.message(f'Готово: {hex_path.name} на платі.' if ok else 'Не вдалося записати: підключіть одну Arduino Nano і спробуйте ще раз.') return 0 if ok else 1 config = Config(args.config) if args.stats: s = Stats(config) time.sleep(interval(config) + 1.5 if IS_MAC else 1.0) values = s.read() print(' '.join(f'{k} {"-" if v is None else round(v, 1)}' for k, v in values.items())) print(stats_line(values)) return 0 if args.key: do_action(config, args.key.upper()) time.sleep(0.5) return 0 if TRAY: return main_tray(config, args.port) log(f'system: {OS_KEY}, config: {config.path}') try: run(config, args.port) except KeyboardInterrupt: log('stopped') return 0 if __name__ == '__main__': sys.exit(main())