tray.pyw

"""DX Command tray launcher (CW Studio pattern).

Puts a DX Command icon in the system tray:

  double-click        -> open the dashboard (seamless app window)
  right-click menu    -> Open Dashboard, Start or Stop the server, Exit

The server keeps its own heartbeat watchdog, so even if the dashboard window
is closed without touching the tray, the background service exits by itself
about 90 seconds later. The tray's status line shows whether the server is
currently running.

The dashboard opens in the browser's app mode (no address bar) when a
Chromium browser is available, falling back to the default browser.
"""
import os
import subprocess
import sys
import tempfile
import threading
import time
import traceback
import urllib.request
import webbrowser

import win32api
import win32con
import win32event
import win32gui
import winerror

LOG = os.path.join(tempfile.gettempdir(), 'dxcommand-tray.log')

FROZEN = getattr(sys, 'frozen', False)
BASE = (os.path.dirname(sys.executable) if FROZEN
        else os.path.dirname(os.path.abspath(__file__)))
URL = 'http://localhost:8073'
ICON = os.path.join(BASE, 'dxcommand.ico')

WM_TRAY = win32con.WM_USER + 20
ID_STATUS, ID_DASHBOARD, ID_STARTSTOP, ID_EXIT = 1001, 1002, 1003, 1004


def server_running():
    try:
        urllib.request.urlopen(URL + '/heartbeat', timeout=0.7)
        return True
    except Exception:
        return False


def start_server():
    if server_running():
        return True
    if FROZEN:
        cmd = [os.path.join(BASE, 'DXCommandService.exe'), '--service']
    else:
        pythonw = os.path.join(os.path.dirname(sys.executable), 'pythonw.exe')
        cmd = [pythonw if os.path.exists(pythonw) else sys.executable,
               os.path.join(BASE, 'run_app.py'), '--service']
    env = dict(os.environ, DXDASH_WATCHDOG='1')
    subprocess.Popen(cmd, creationflags=subprocess.CREATE_NO_WINDOW,
                     cwd=BASE, env=env)
    for _ in range(50):          # first start downloads cty.dat etc.
        if server_running():
            return True
        time.sleep(0.2)
    return False


def app_token():
    """The running server's per-launch token, published to its data folder.

    Mutating requests need it (see app/security.py). Data lives beside the exe
    for a portable copy, or under %LOCALAPPDATA% when installed to Program
    Files, so try both.
    """
    candidates = [os.path.join(BASE, 'data', '.dxtoken')]
    local = os.environ.get('LOCALAPPDATA')
    if local:
        candidates.append(os.path.join(local, 'DXCommand', 'data', '.dxtoken'))
    for path in candidates:
        try:
            with open(path, encoding='utf-8') as fh:
                token = fh.read().strip()
            if token:
                return token
        except OSError:
            continue
    return ''


def stop_server():
    try:
        req = urllib.request.Request(URL + '/shutdown', data=b'{}', method='POST')
        req.add_header('X-DX-Token', app_token())
        urllib.request.urlopen(req, timeout=1.5)
    except Exception:
        pass


def _default_browser_exe():
    try:
        import winreg
        with winreg.OpenKey(winreg.HKEY_CURRENT_USER,
                r'Software\Microsoft\Windows\Shell\Associations'
                r'\UrlAssociations\https\UserChoice') as k:
            progid = winreg.QueryValueEx(k, 'ProgId')[0]
        if 'Firefox' in progid or 'Mozilla' in progid:
            return None
        with winreg.OpenKey(winreg.HKEY_CLASSES_ROOT,
                            progid + r'\shell\open\command') as k:
            cmd = winreg.QueryValueEx(k, '')[0]
        exe = cmd.split('"')[1] if cmd.startswith('"') else cmd.split(' ')[0]
        return exe if os.path.exists(exe) else None
    except OSError:
        return None


def find_app_browser():
    exe = _default_browser_exe()
    if exe:
        return exe
    for p in (r'%ProgramFiles(x86)%\Microsoft\Edge\Application\msedge.exe',
              r'%ProgramFiles%\Microsoft\Edge\Application\msedge.exe',
              r'%ProgramFiles%\Google\Chrome\Application\chrome.exe',
              r'%LocalAppData%\Google\Chrome\Application\chrome.exe'):
        p = os.path.expandvars(p)
        if os.path.exists(p):
            return p
    return None


def open_dashboard():
    threading.Thread(target=_open_dashboard, daemon=True).start()


def _open_dashboard():
    start_server()
    browser = find_app_browser()
    if browser:
        subprocess.Popen([browser, '--app=' + URL, '--window-size=1500,940'])
    else:
        webbrowser.open(URL)


class Tray:
    def __init__(self):
        wc = win32gui.WNDCLASS()
        wc.hInstance = win32gui.GetModuleHandle(None)
        wc.lpszClassName = 'DXCommandTray'
        wc.lpfnWndProc = {
            WM_TRAY: self.on_tray,
            win32con.WM_COMMAND: self.on_command,
            win32con.WM_DESTROY: self.on_destroy,
        }
        atom = win32gui.RegisterClass(wc)
        self.hwnd = win32gui.CreateWindow(atom, 'DX Command', 0, 0, 0, 0, 0,
                                          0, 0, wc.hInstance, None)
        try:
            hicon = win32gui.LoadImage(0, ICON, win32con.IMAGE_ICON, 0, 0,
                                       win32con.LR_LOADFROMFILE | win32con.LR_DEFAULTSIZE)
        except Exception:
            hicon = win32gui.LoadIcon(0, win32con.IDI_APPLICATION)
        self.hicon = hicon
        nid = (self.hwnd, 0,
               win32gui.NIF_ICON | win32gui.NIF_MESSAGE | win32gui.NIF_TIP,
               WM_TRAY, hicon, 'DX Command')
        win32gui.Shell_NotifyIcon(win32gui.NIM_ADD, nid)

    def balloon(self, title, msg):
        nid = (self.hwnd, 0,
               win32gui.NIF_ICON | win32gui.NIF_INFO,
               WM_TRAY, self.hicon, 'DX Command', msg, 10, title)
        try:
            win32gui.Shell_NotifyIcon(win32gui.NIM_MODIFY, nid)
        except Exception:
            pass

    def on_tray(self, hwnd, msg, wparam, lparam):
        if lparam == win32con.WM_LBUTTONDBLCLK:
            open_dashboard()
        elif lparam == win32con.WM_RBUTTONUP:
            self.show_menu()
        return 0

    def show_menu(self):
        running = server_running()
        menu = win32gui.CreatePopupMenu()
        win32gui.AppendMenu(menu, win32con.MF_STRING | win32con.MF_GRAYED, ID_STATUS,
                            'Server: running' if running else 'Server: stopped')
        win32gui.AppendMenu(menu, win32con.MF_SEPARATOR, 0, '')
        win32gui.AppendMenu(menu, win32con.MF_STRING, ID_DASHBOARD, 'Open Dashboard')
        win32gui.AppendMenu(menu, win32con.MF_SEPARATOR, 0, '')
        win32gui.AppendMenu(menu, win32con.MF_STRING, ID_STARTSTOP,
                            'Stop server' if running else 'Start server')
        win32gui.AppendMenu(menu, win32con.MF_SEPARATOR, 0, '')
        win32gui.AppendMenu(menu, win32con.MF_STRING, ID_EXIT, 'Exit')
        pos = win32gui.GetCursorPos()
        win32gui.SetForegroundWindow(self.hwnd)
        win32gui.TrackPopupMenu(menu, win32con.TPM_LEFTALIGN, pos[0], pos[1],
                                0, self.hwnd, None)
        win32gui.PostMessage(self.hwnd, win32con.WM_NULL, 0, 0)

    def on_command(self, hwnd, msg, wparam, lparam):
        cmd = win32gui.LOWORD(wparam)
        if cmd == ID_DASHBOARD:
            open_dashboard()
        elif cmd == ID_STARTSTOP:
            if server_running():
                stop_server()
            else:
                threading.Thread(target=start_server, daemon=True).start()
        elif cmd == ID_EXIT:
            stop_server()
            win32gui.DestroyWindow(self.hwnd)
        return 0

    def on_destroy(self, hwnd, msg, wparam, lparam):
        win32gui.Shell_NotifyIcon(win32gui.NIM_DELETE, (self.hwnd, 0))
        win32gui.PostQuitMessage(0)
        return 0


if __name__ == '__main__':
    try:
        # single instance: a second launch just opens the dashboard.
        # The handle must stay referenced for the lifetime of the process —
        # if it is garbage-collected, the mutex disappears with it.
        _MUTEX = win32event.CreateMutex(None, False, 'DXCommandTrayMutex')
        if win32api.GetLastError() == winerror.ERROR_ALREADY_EXISTS:
            _open_dashboard()
            sys.exit(0)
        tray = Tray()
        # make launching feel immediate: open the dashboard and point at the
        # tray icon (Windows hides new tray icons behind the ^ overflow)
        open_dashboard()
        tray.balloon('DX Command is running',
                     'Find the icon in the system tray — it may be behind '
                     'the ^ overflow arrow. Right-click it for the menu.')
        win32gui.PumpMessages()
    except SystemExit:
        raise
    except Exception:
        with open(LOG, 'a', encoding='utf-8') as f:
            f.write(time.strftime('%Y-%m-%d %H:%M:%S\n'))
            f.write(traceback.format_exc() + '\n')
        raise