host.match_monitor.tray_app

  1"""
  2System tray application for Match Monitor (Windows).
  3
  4Launch via:  python -m host.match_monitor --tray
  5
  6__main__.py automatically re-launches under pythonw.exe so the originating
  7cmd/PowerShell window closes immediately and no console is left open.
  8
  9The "Open Console" tray menu item allocates a fresh Windows console window
 10(via AllocConsole) on demand.  "Close Console" frees it.  The console runs
 11the interactive monitor> prompt and can be opened/closed independently.
 12"""
 13
 14import ctypes
 15import logging
 16import subprocess
 17import sys
 18import threading
 19import time
 20from pathlib import Path
 21
 22logger = logging.getLogger("match_monitor")
 23
 24if hasattr(sys, '_MEIPASS'):
 25    # Running as a PyInstaller frozen exe — data files land in _MEIPASS
 26    _LOGO_PATH = Path(sys._MEIPASS) / 'images' / 'raptacongear.png'
 27else:
 28    _LOGO_PATH = Path(__file__).parent.parent.parent / 'images' / 'raptacongear.png'
 29
 30_STATUS_COLORS = {
 31    'idle':      '#808080',  # gray  — polling
 32    'connected': '#57F287',  # green — connected
 33    'uploading': '#FEE75C',  # yellow — uploading
 34    'error':     '#ED4245',  # red   — error
 35}
 36
 37_ICON_CACHE: dict = {}
 38
 39
 40def _make_icon(status: str, size: int = 64):
 41    """Raptacon gear logo with a small colored status badge (bottom-right)."""
 42    from PIL import Image, ImageDraw
 43
 44    color = _STATUS_COLORS.get(status, _STATUS_COLORS['idle'])
 45
 46    try:
 47        logo = Image.open(_LOGO_PATH).convert('RGBA').resize((size, size), Image.LANCZOS)
 48    except Exception:  # noqa: BLE001 — fall back to blank icon if logo missing/corrupt
 49        logo = Image.new('RGBA', (size, size), (0, 0, 0, 0))
 50        ImageDraw.Draw(logo).ellipse([4, 4, size - 4, size - 4], fill='#303030')
 51
 52    draw = ImageDraw.Draw(logo)
 53    r = size // 5
 54    x0, y0 = size - r * 2 - 2, size - r * 2 - 2
 55    draw.ellipse([x0, y0, size - 2, size - 2], fill=color, outline='white', width=1)
 56
 57    return logo
 58
 59
 60def _icon_for(status: str):
 61    if status not in _ICON_CACHE:
 62        _ICON_CACHE[status] = _make_icon(status)
 63    return _ICON_CACHE[status]
 64
 65
 66class MatchMonitorTray:
 67    """Wraps a ServerContext in a pystray system-tray icon."""
 68
 69    def __init__(self, ctx) -> None:
 70        import pystray
 71        self._pystray = pystray
 72        self._ctx = ctx
 73        self._stop_requested = False
 74
 75        # Console state — starts closed; opened on demand via AllocConsole()
 76        self._console_open = False
 77        self._console_hwnd = 0
 78        self._console_visible = False
 79        self._console_thread: threading.Thread | None = None
 80
 81        self._icon = pystray.Icon(
 82            'MatchMonitor',
 83            _icon_for('idle'),
 84            'Match Monitor\nPolling for robot...',
 85            menu=pystray.Menu(self._menu_items),
 86        )
 87
 88    # ------------------------------------------------------------------
 89    # Dynamic menu (rebuilt on each right-click)
 90    # ------------------------------------------------------------------
 91
 92    def _menu_items(self):
 93        pystray = self._pystray
 94        info = self._ctx.connector.status_info
 95        state = info['state']
 96        connected = self._ctx.connector.is_connected
 97
 98        uptime_str = ''
 99        if info['connected_since'] is not None:
100            uptime = time.monotonic() - info['connected_since']
101            mins, secs = divmod(int(uptime), 60)
102            hrs, mins = divmod(mins, 60)
103            uptime_str = f'  ({hrs}h {mins:02d}m {secs:02d}s)'
104
105        if not self._console_open:
106            console_item = pystray.MenuItem('Open Console', self._open_console)
107        elif self._console_visible:
108            console_item = pystray.MenuItem('Hide Console', self._hide_console)
109        else:
110            console_item = pystray.MenuItem('Show Console', self._show_console)
111
112        return (
113            pystray.MenuItem(f'● {state}{uptime_str}', None, enabled=False),
114            pystray.Menu.SEPARATOR,
115            pystray.MenuItem('Robot', pystray.Menu(
116                pystray.MenuItem('Force Upload',   self._force_upload,   enabled=connected),
117                pystray.MenuItem('Stop Upload',    self._stop_upload,    enabled=connected),
118                pystray.MenuItem('Clear Manifest', self._clear_manifest, enabled=connected),
119                pystray.MenuItem('List Logs',      self._list_logs,      enabled=connected),
120            )),
121            pystray.MenuItem('Open Log Folder', self._open_log_folder),
122            pystray.MenuItem('Set Code Directory...', self._set_code_dir),
123            pystray.Menu.SEPARATOR,
124            console_item,
125            pystray.Menu.SEPARATOR,
126            pystray.MenuItem('Quit', self._quit),
127        )
128
129    # ------------------------------------------------------------------
130    # Robot actions
131    # ------------------------------------------------------------------
132
133    def _force_upload(self):
134        self._ctx.connector.send_to_robot({'type': 'FORCE_UPLOAD'})
135
136    def _stop_upload(self):
137        self._ctx.connector.send_to_robot({'type': 'STOP_UPLOAD'})
138
139    def _clear_manifest(self):
140        self._ctx.connector.send_to_robot({'type': 'CLEAR_MANIFEST'})
141
142    def _list_logs(self):
143        self._ctx.connector.send_to_robot({'type': 'LIST_LOGS'})
144
145    def _open_log_folder(self):
146        subprocess.Popen(['explorer', str(self._ctx.out_path)])
147
148    def _set_code_dir(self):
149        """Open a folder picker dialog to set the robot code directory.
150
151        Runs the tkinter dialog in a subprocess because pystray callbacks
152        execute on a background thread, and tkinter requires the main thread.
153        """
154        import subprocess
155        import sys
156        from pathlib import Path
157        from .receiver import _save_config_key
158
159        initial = str(self._ctx.code_dir_holder[0]) if self._ctx.code_dir_holder[0] else ''
160        script = (
161            "import tkinter as tk; from tkinter.filedialog import askdirectory; "
162            "r=tk.Tk(); r.withdraw(); r.attributes('-topmost',True); "
163            f"print(askdirectory(title='Select Robot Code Directory', initialdir={initial!r}) or ''); "
164            "r.destroy()"
165        )
166        result = subprocess.run([sys.executable, '-c', script],
167                                capture_output=True, text=True)
168        chosen = result.stdout.strip()
169
170        if not chosen:
171            return
172
173        new_path = Path(chosen).resolve()
174        self._ctx.code_dir_holder[0] = new_path
175
176        notifier = self._ctx.discord_holder[0]
177        if notifier:
178            notifier._repo_root = new_path
179
180        _save_config_key(self._ctx.config_path, 'robot_code_dir', str(new_path))
181        is_git = (new_path / '.git').exists()
182        suffix = '' if is_git else ' (no .git)'
183        logger.info(f"Code directory set to: {new_path}{suffix}")
184
185    # ------------------------------------------------------------------
186    # Console management
187    # ------------------------------------------------------------------
188
189    def _open_console(self):
190        """Allocate a new Windows console window and start the monitor> prompt."""
191        if self._console_open:
192            self._show_console()
193            return
194
195        ctypes.windll.kernel32.AllocConsole()
196        ctypes.windll.kernel32.SetConsoleTitleW("Match Monitor Console")
197
198        # Redirect Python I/O to the new console
199        try:
200            sys.stdout = open('CONOUT$', 'w', buffering=1, encoding='utf-8',
201                              errors='replace')
202            sys.stderr = open('CONOUT$', 'w', buffering=1, encoding='utf-8',
203                              errors='replace')
204            sys.stdin = open('CONIN$', 'r', encoding='utf-8', errors='replace')
205        except OSError as e:
206            logger.warning(f"Could not redirect streams to console: {e}")
207            return
208
209        # Update the logging console handler to write to the new console
210        self._ctx.console_handler.stream = sys.stdout
211        self._ctx.console_handler.setLevel(logging.INFO)
212
213        self._console_hwnd = ctypes.windll.kernel32.GetConsoleWindow()
214        self._console_open = True
215        self._console_visible = True
216
217        # Disable the console close (X) button.
218        # CTRL_CLOSE_EVENT cannot be safely handled without Windows eventually
219        # terminating the process — disabling the button avoids the problem
220        # entirely.  Users close the console with 'exit' instead.
221        _SC_CLOSE = 0xF060
222        _MF_BYCOMMAND = 0
223        _hmenu = ctypes.windll.user32.GetSystemMenu(self._console_hwnd, False)
224        ctypes.windll.user32.DeleteMenu(_hmenu, _SC_CLOSE, _MF_BYCOMMAND)
225
226        # Start the interactive monitor> prompt in the console
227        from .receiver import _console_loop
228        self._console_thread = threading.Thread(
229            target=_console_loop,
230            args=(self._ctx.connector, self._ctx.console_handler,
231                  self._ctx.file_handler, self._ctx.log_dir,
232                  self._ctx.registry, self._ctx.discord_holder,
233                  self._ctx.config_path, self._ctx.code_dir_holder),
234            kwargs={'analyzer': self._ctx.analyzer,
235                    'out_path': self._ctx.out_path,
236                    'match_client': self._ctx.match_client,
237                    'quit_callback': self._quit,
238                    'close_callback': self._on_console_closed},
239            daemon=True,
240        )
241        self._console_thread.start()
242
243        print("Match Monitor Console — type 'help' for commands.")
244
245    def _on_console_closed(self):
246        """Free the Windows console and reset state so it can be re-opened.
247
248        Called after the console loop exits (exit command or stop-server).
249        Must be idempotent.
250        """
251        if not self._console_open:
252            return
253
254        self._console_open = False
255        self._console_visible = False
256        self._console_hwnd = 0
257        self._console_thread = None
258
259        # Redirect logging handler away from the (about-to-be-freed) console
260        import os
261        _devnull = open(os.devnull, 'w')
262        self._ctx.console_handler.stream = _devnull
263
264        # Close stdin first — this unblocks any pending input() in the console
265        # thread so it exits cleanly via EOFError/OSError.
266        try:
267            if sys.stdin is not None:
268                sys.stdin.close()
269            sys.stdin = None
270        except Exception:
271            logger.debug("Error closing stdin during console teardown", exc_info=True)
272
273        # Detach from the console — window closes (we are the only process).
274        ctypes.windll.kernel32.FreeConsole()
275
276        # Close stdout/stderr after detaching (handles are now invalid)
277        for attr in ('stdout', 'stderr'):
278            try:
279                s = getattr(sys, attr, None)
280                if s is not None:
281                    s.close()
282            except Exception:
283                logger.debug("Error closing %s during console teardown", attr, exc_info=True)
284            setattr(sys, attr, None)
285
286        # Force the tray menu to rebuild so "Open Console" replaces "Hide Console"
287        try:
288            self._icon.update_menu()
289        except Exception:
290            logger.debug("Error updating tray menu after console close", exc_info=True)
291
292    def _hide_console(self):
293        if self._console_hwnd:
294            ctypes.windll.user32.ShowWindow(self._console_hwnd, 0)  # SW_HIDE
295        self._console_visible = False
296
297    def _show_console(self):
298        if self._console_hwnd:
299            ctypes.windll.user32.ShowWindow(self._console_hwnd, 5)  # SW_SHOW
300            ctypes.windll.user32.SetForegroundWindow(self._console_hwnd)
301        self._console_visible = True
302
303    # ------------------------------------------------------------------
304    # Quit
305    # ------------------------------------------------------------------
306
307    def _quit(self):
308        """Stop the tray icon — cleanup runs in run() after icon.run() returns."""
309        self._stop_requested = True
310        self._icon.stop()
311
312    # ------------------------------------------------------------------
313    # Background update loop
314    # ------------------------------------------------------------------
315
316    def _update_loop(self) -> None:
317        while not self._stop_requested:
318            time.sleep(2)
319            try:
320                info = self._ctx.connector.status_info
321                state_str = info['state'].lower()
322
323                if 'uploading' in state_str:
324                    icon_state = 'uploading'
325                    tooltip = 'Match Monitor\nUploading logs...'
326                elif 'connected' in state_str:
327                    icon_state = 'connected'
328                    robot = info.get('robot_address') or ''
329                    uptime = ''
330                    if info['connected_since'] is not None:
331                        secs = int(time.monotonic() - info['connected_since'])
332                        m, s = divmod(secs, 60)
333                        h, m = divmod(m, 60)
334                        uptime = f'  {h}h {m:02d}m {s:02d}s'
335                    tooltip = f'Match Monitor\nConnected: {robot}{uptime}'
336                else:
337                    icon_state = 'idle'
338                    tooltip = 'Match Monitor\nPolling for robot...'
339
340                self._icon.icon = _icon_for(icon_state)
341                self._icon.title = tooltip
342            except Exception:
343                logger.debug("Error updating tray icon", exc_info=True)
344
345    # ------------------------------------------------------------------
346    # Entry point
347    # ------------------------------------------------------------------
348
349    def run(self) -> None:
350        """Start the HTTP server in background, run the tray on the main thread."""
351        threading.Thread(
352            target=self._ctx.server.serve_forever,
353            daemon=True,
354        ).start()
355
356        threading.Thread(target=self._update_loop, daemon=True).start()
357
358        logger.info("Match Monitor tray started")
359        try:
360            self._icon.run()  # blocks until _quit() calls icon.stop()
361        except KeyboardInterrupt:
362            logger.info("Keyboard interrupt received, shutting down")
363
364        # --- Cleanup after tray icon stops ---
365        self._stop_requested = True
366        self._ctx.connector.stop()
367        logger.info("Match Monitor tray shutting down")
368        threading.Thread(target=self._ctx.server.shutdown, daemon=True).start()
369        sys.exit(0)
370
371
372def run_server_tray(bind: str, port: int, output_dir: str = None,
373                    debug: bool = False) -> None:
374    """Set up the server and run it as a system tray application."""
375    from .receiver import setup_server
376
377    ctx = setup_server(bind, port, output_dir, debug)
378    logger.info(f"Match Monitor tray on {bind}:{port}, saving to {ctx.out_path}")
379
380    MatchMonitorTray(ctx).run()