Win32 Message Monitor

A few years ago, I wrote a tool that replicated serial port and Bluetooth connections across different sites, often in different countries. The tool was based on ZeroMQ and worked quite well. I was proud of myself until some users started reporting rare, intermittent problems. Under certain circumstances, the tool couldn’t resume normal operation after a laptop was awakened from sleep.

Why would anyone put their laptop to sleep in the middle of an ongoing serial port operation is a separate mystery… but the user is always right, right? ZeroMQ is quite resilient and almost always recovered from such abuse, but not always.

So I had to deal with some rare edge cases. I could have tried to fix ZeroMQ itself, but that would have been too much work. Instead, I opted for an easier, lazier solution. I wrote a Python tool that monitored Windows Win32 application messages and captured power events. This way, the tool was immediately notified when the laptop went to sleep or woke up, and it could react accordingly by gracefully resetting the state of the messaging queue. This fixed all the issues, and the tool became a success within our department.

I had almost forgotten about all of that until a new challenge came up last week. I needed to run a script every time my laptop was plugged into or unplugged from an external display. There are a few ways to accomplish this, but then I remembered my old Python script that monitored Win32 messages. Plugging or unplugging a display is just another Windows application message, so here we are.

I asked ChatGPT to extend my old script, and the new class DisplayChangeEvent was born. Challenge solved. Thanks, Chatty.

And one more thing: Yes, I still use CamelCase in Python. Yes, I am that old.

import atexit
import logging
import win32api
import win32con
import win32gui
import win32ts
from evtSessionEvent import SessionEvent
from evtTimeChangeEvent import TimeChangeEvent
from evtPowerEvent import PowerEvent
from evtDisplayChangeEvent import DisplayChangeEvent
from msgDict import MSG_DICT as knownWinMessages
class WorkstationMessageMonitor(object):
    CLASS_NAME = "WorkstationMessageMonitor"
    WINDOW_TITLE = "Workstation Message Monitor"
    def __init__(self):
        self.windowHandle = None
        self._registerListener()
        atexit.register(self.stop)
    def _registerListener(self):
        wndClass = win32gui.WNDCLASS()
        wndClass.hInstance = handleInstance = win32api.GetModuleHandle(None)
        wndClass.lpszClassName = self.CLASS_NAME
        wndClass.lpfnWndProc = self._windowProcedure
        window_class = win32gui.RegisterClass(wndClass)
        style = 0
        self.windowHandle = win32gui.CreateWindow(
                window_class,         # Window class
                self.WINDOW_TITLE,    # Window text
                style,                # Window style
                # size and position
                0, 0, win32con.CW_USEDEFAULT, win32con.CW_USEDEFAULT,
                # Parent, Menu, Instance handle, Additional appl. data
                0, 0, handleInstance, None)
        win32gui.UpdateWindow(self.windowHandle)
        scope = win32ts.NOTIFY_FOR_ALL_SESSIONS
        win32ts.WTSRegisterSessionNotification(self.windowHandle, scope)
    def listen(self):
        logging.info('Listening')
        try:
            win32gui.PumpMessages()
        except KeyboardInterrupt:
            logging.info('Interrupted by user')
        finally:
            self.stop()
    def stop(self):
        logging.info('Exiting')
        exitCode = 0
        win32ts.WTSUnRegisterSessionNotification(self.windowHandle)
        win32gui.DestroyWindow(self.windowHandle)
        win32gui.PostQuitMessage(exitCode)
    @staticmethod
    def _windowProcedure(hWnd: int, uMsg: int, wParam, lParam) -> int:
        """
        WindowProc callback function.
        :param hWnd: A handle to the window.
        :param uMsg: The message.
        :param wParam: Additional message information.
        :param lParam: Additional message information.
        Returns an LRESULT (int).
        """
        if uMsg == PowerEvent.MESSAGE:
            logging.info(f'{PowerEvent.EVENTS.get(wParam, "Unknown power event")}, lParam: {lParam}')
            return 1
        elif uMsg == TimeChangeEvent.MESSAGE:
            logging.info('WM_TIMECHANGE')
            return 0
        elif uMsg == SessionEvent.MESSAGE:
            logging.info(f'{SessionEvent.EVENTS.get(wParam, "Unknown session event")}, lParam: {lParam}')
            return 0
        elif uMsg == DisplayChangeEvent.MESSAGE:
            logging.info(f'WM_DISPLAYCHANGE: New primary display {DisplayChangeEvent.describe(wParam, lParam)}')
            return 0
        else:
            if uMsg in knownWinMessages:
                logging.info(f'{knownWinMessages[uMsg]}, '
                             f'wParam: {wParam} ({hex(wParam)}), '
                             f'lParam: {lParam}')
            else:
                logging.info(f'Unknown message {uMsg} ({hex(uMsg)}), '
                             f'wParam: {wParam} ({hex(wParam)}), '
                             f'lParam: {lParam}')
            return win32gui.DefWindowProc(hWnd, uMsg, wParam, lParam)
if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    m = WorkstationMessageMonitor()
    m.listen()

class PowerEvent(object):
    MESSAGE = 0x218     # WM_POWERBROADCAST, winuser.h
    EVENTS = {
        0x04: 'PBT_APMSUSPEND',
        0x07: 'PBT_APMRESUMESUSPEND',
        0x0a: 'PBT_APMPOWERSTATUSCHANGE',
        0x12: 'PBT_APMRESUMEAUTOMATIC',
        0x8013: 'PBT_POWERSETTINGCHANGE',
    }

class SessionEvent(object):
    MESSAGE = 0x2B1     # WM_WTSSESSION_CHANGE, wtsapi32.h
    EVENTS = {
        0x1: 'WTS_CONSOLE_CONNECT',
        0x2: 'WTS_CONSOLE_DISCONNECT',
        0x3: 'WTS_REMOTE_CONNECT',
        0x4: 'WTS_REMOTE_DISCONNECT',
        0x5: 'WTS_SESSION_LOGON',
        0x6: 'WTS_SESSION_LOGOFF',
        0x7: 'WTS_SESSION_LOCK',
        0x8: 'WTS_SESSION_UNLOCK',
        0x9: 'WTS_SESSION_REMOTE_CONTROL',
    }

class TimeChangeEvent(object):
    MESSAGE = 0x001E     # WM_TIMECHANGE, winuser.h
    EVENTS = None

class DisplayChangeEvent(object):
    # windows message
    MESSAGE = 0x007E     # WM_DISPLAYCHANGE
    @staticmethod
    def bitsPerPixel(wParam: int) -> int:
        return int(wParam)
    @staticmethod
    def width(lParam: int) -> int:
        return int(lParam) & 0xFFFF
    @staticmethod
    def height(lParam: int) -> int:
        return (int(lParam) >> 16) & 0xFFFF
    @staticmethod
    def decode(wParam: int, lParam: int) -> tuple[int, int, int]:
        """
        Returns (width, height, bitsPerPixel).
        """
        width = DisplayChangeEvent.width(lParam)
        height = DisplayChangeEvent.height(lParam)
        bpp = DisplayChangeEvent.bitsPerPixel(wParam)
        return width, height, bpp
    @staticmethod
    def describe(wParam: int, lParam: int) -> str:
        """
        Returns a human-readable description, e.g., '1920x1080 @ 32bpp'.
        """
        w, h, bpp = DisplayChangeEvent.decode(wParam, lParam)
        return f"{w}x{h} @ {bpp}bpp"