""" Windows Card Reader (Raw Input API + LL-hook fallback) ======================================================= Drop-in replacement for the Linux `CardReader` class in `src/main.py`. The Linux implementation reads keystrokes from `/dev/input/event*` via `evdev`. On Windows there is no `/dev/input`; instead HID devices (such as USB card readers that emulate a keyboard) are accessed through the Win32 **Raw Input API** (`WM_INPUT`). Design ------ * A dedicated hidden message-only window + message pump runs on a background thread. It registers for Raw Input of all *keyboard* HID devices with `RIDEV_INPUTSINK`, so it receives `WM_INPUT` even though it never has focus. * Device selection mirrors the Linux priority logic: 1. A device whose name contains "card" / "reader" / "rfid". 2. A USB HID keyboard that is *not* the PS/2 system keyboard. 3. Any keyboard device (excluding obvious touchscreens/mice). A config override `card_reader_device` (substring of the device name, e.g. `VID_08FF`) takes precedence. * While reading, only key events from the *selected* device are accepted, so a physical keyboard used for maintenance cannot pollute the card data. * Card data ends on Enter (`VK_RETURN`), mirroring the Linux behaviour. * If Raw Input registration fails (or config `card_reader_mode = "hook"`), it falls back to a low-level keyboard hook (`WH_KEYBOARD_LL`). Config keys (in `config/app_config.json`): "card_reader_device": "VID_08FF" # substring of device name (optional) "card_reader_mode": "auto" # "auto" | "raw" | "hook" "card_reader_timeout": 5 # seconds (optional) """ import ctypes import ctypes.wintypes as wintypes import os import json import threading import time # ── Win32 constants ────────────────────────────────────────────────────────── WM_INPUT = 0x00FF WM_INPUT_DEVICE_CHANGE = 0x00FE WM_KEYDOWN = 0x0100 WM_SYSKEYDOWN = 0x0104 RIM_TYPEMOUSE = 0 RIM_TYPEKEYBOARD = 1 RIM_TYPEHID = 2 RID_INPUT = 0x10000003 RIDI_DEVICENAME = 0x20000007 RIDEV_INPUTSINK = 0x00000100 RIDEV_DEVNOTIFY = 0x00002000 WH_KEYBOARD_LL = 13 HC_ACTION = 0 # Virtual keys we never treat as card data _VK_SHIFT = 0x10 _VK_CONTROL = 0x11 _VK_MENU = 0x12 _VK_CAPITAL = 0x14 _VK_ESCAPE = 0x1B _VK_RETURN = 0x0D _VK_TAB = 0x09 _VK_LSHIFT = 0xA0 _VK_RSHIFT = 0xA1 _VK_LCONTROL = 0xA2 _VK_RCONTROL = 0xA3 _VK_LMENU = 0xA4 _VK_RMENU = 0xA5 # ── ctypes structures ─────────────────────────────────────────────────────── # ctypes.wintypes does not export LRESULT; it is a signed pointer-sized value. # Use ctypes.c_long (standard ctypes workaround; fine for WNDPROC/LL-hook). _LRESULT = ctypes.c_long _WND_PROC = ctypes.WINFUNCTYPE( _LRESULT, wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM, ) _LL_KEYBOARD_PROC = ctypes.WINFUNCTYPE( _LRESULT, ctypes.c_int, wintypes.WPARAM, wintypes.LPARAM, ) class _WNDCLASSW(ctypes.Structure): _fields_ = [ ('style', wintypes.UINT), ('lpfnWndProc', _WND_PROC), ('cbClsExtra', ctypes.c_int), ('cbWndExtra', ctypes.c_int), ('hInstance', wintypes.HINSTANCE), ('hIcon', wintypes.HICON), ('hCursor', ctypes.c_void_p), # HCURSOR (not exported by wintypes) ('hbrBackground', wintypes.HBRUSH), ('lpszMenuName', wintypes.LPCWSTR), ('lpszClassName', wintypes.LPCWSTR), ] class _MSG(ctypes.Structure): _fields_ = [ ('hwnd', wintypes.HWND), ('message', wintypes.UINT), ('wParam', wintypes.WPARAM), ('lParam', wintypes.LPARAM), ('time', wintypes.DWORD), ('pt', wintypes.POINT), ] class _RAWINPUTDEVICE(ctypes.Structure): _fields_ = [ ('usUsagePage', wintypes.USHORT), ('usUsage', wintypes.USHORT), ('dwFlags', wintypes.DWORD), ('hwndTarget', wintypes.HWND), ] class _RAWINPUTDEVICELIST(ctypes.Structure): _fields_ = [ ('hDevice', wintypes.HANDLE), ('dwType', wintypes.DWORD), ] class _RAWINPUTHEADER(ctypes.Structure): _fields_ = [ ('dwType', wintypes.DWORD), ('dwSize', wintypes.DWORD), ('hDevice', wintypes.HANDLE), ('wParam', wintypes.WPARAM), ] class _RAWKEYBOARD(ctypes.Structure): _fields_ = [ ('MakeCode', wintypes.USHORT), ('Flags', wintypes.USHORT), ('Reserved', wintypes.USHORT), ('VKey', wintypes.USHORT), ('Message', wintypes.UINT), ('ExtraInformation', wintypes.ULONG), ] class _RAWINPUT_UNION(ctypes.Union): _fields_ = [('keyboard', _RAWKEYBOARD)] class _RAWINPUT(ctypes.Structure): _fields_ = [ ('header', _RAWINPUTHEADER), ('u', _RAWINPUT_UNION), ] class _KBDLLHOOKSTRUCT(ctypes.Structure): _fields_ = [ ('vkCode', wintypes.DWORD), ('scanCode', wintypes.DWORD), ('flags', wintypes.DWORD), ('time', wintypes.DWORD), ('dwExtraInfo', wintypes.WPARAM), ] # ── Win32 function bindings with explicit signatures ──────────────────────── _user32 = ctypes.windll.user32 _kernel32 = ctypes.windll.kernel32 _user32.RegisterClassW.argtypes = [ctypes.POINTER(_WNDCLASSW)] _user32.RegisterClassW.restype = wintypes.ATOM _user32.CreateWindowExW.argtypes = [ wintypes.DWORD, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int, wintypes.HWND, wintypes.HMENU, wintypes.HINSTANCE, wintypes.LPVOID, ] _user32.CreateWindowExW.restype = wintypes.HWND _user32.DestroyWindow.argtypes = [wintypes.HWND] _user32.UnregisterClassW.argtypes = [wintypes.LPCWSTR, wintypes.HINSTANCE] _user32.GetMessageW.argtypes = [ ctypes.POINTER(_MSG), wintypes.HWND, wintypes.UINT, wintypes.UINT, ] _user32.GetMessageW.restype = wintypes.BOOL _user32.TranslateMessage.argtypes = [ctypes.POINTER(_MSG)] _user32.DispatchMessageW.argtypes = [ctypes.POINTER(_MSG)] _user32.DefWindowProcW.argtypes = [ wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM, ] _user32.DefWindowProcW.restype = _LRESULT _user32.PostMessageW.argtypes = [ wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM, ] _user32.PostMessageW.restype = wintypes.BOOL _user32.RegisterRawInputDevices.argtypes = [ ctypes.POINTER(_RAWINPUTDEVICE), wintypes.UINT, wintypes.UINT, ] _user32.RegisterRawInputDevices.restype = wintypes.BOOL _user32.GetRawInputDeviceList.argtypes = [ ctypes.POINTER(_RAWINPUTDEVICELIST), ctypes.POINTER(wintypes.UINT), wintypes.UINT, ] _user32.GetRawInputDeviceList.restype = wintypes.UINT _user32.GetRawInputDeviceInfoW.argtypes = [ wintypes.HANDLE, wintypes.UINT, wintypes.LPVOID, ctypes.POINTER(wintypes.UINT), ] _user32.GetRawInputDeviceInfoW.restype = wintypes.UINT _user32.GetRawInputData.argtypes = [ wintypes.HANDLE, wintypes.UINT, wintypes.LPVOID, ctypes.POINTER(wintypes.UINT), wintypes.UINT, ] _user32.GetRawInputData.restype = wintypes.UINT _user32.MapVirtualKeyW.argtypes = [wintypes.UINT, wintypes.UINT] _user32.MapVirtualKeyW.restype = wintypes.UINT _user32.SetWindowsHookExW.argtypes = [ ctypes.c_int, _LL_KEYBOARD_PROC, wintypes.HINSTANCE, wintypes.DWORD, ] _user32.SetWindowsHookExW.restype = wintypes.HHOOK _user32.UnhookWindowsHookEx.argtypes = [wintypes.HHOOK] _user32.UnhookWindowsHookEx.restype = wintypes.BOOL _user32.CallNextHookEx.argtypes = [ wintypes.HHOOK, ctypes.c_int, wintypes.WPARAM, wintypes.LPARAM, ] _user32.CallNextHookEx.restype = _LRESULT _kernel32.GetModuleHandleW.argtypes = [wintypes.LPCWSTR] _kernel32.GetModuleHandleW.restype = wintypes.HINSTANCE _WND_CLASS_NAME = 'KiwyWinCardReaderWindow' # The single active reader instance (the Raw Input / LL-hook callbacks are # invoked from a background thread; a module-level ref avoids GC of the proc). _ACTIVE_READER = None # ── Small helpers ─────────────────────────────────────────────────────────── def _log(msg): """Log via Kivy Logger when available, else print.""" try: from kivy.logger import Logger Logger.info(f"CardReaderWin: {msg}") except Exception: try: print(f"[CardReaderWin] {msg}") except Exception: pass def _get_config(): """Read app_config.json (next to exe, or project config in dev mode).""" try: data_dir = os.environ.get('KIWY_DATA_DIR', '') candidates = [] if data_dir: candidates.append(os.path.join(data_dir, 'config', 'app_config.json')) # Dev fallback: /config/app_config.json here = os.path.dirname(os.path.abspath(__file__)) # windows/ candidates.append(os.path.join(os.path.dirname(here), 'config', 'app_config.json')) for path in candidates: if path and os.path.exists(path): with open(path, 'r', encoding='utf-8') as f: return json.load(f) or {} except Exception: pass return {} def _vk_to_char(vk): """Convert a virtual-key code to its character, or None if not data.""" # Numeric keypad 0-9 if 0x60 <= vk <= 0x69: return chr(vk - 0x60 + 0x30) # Main-row digits if 0x30 <= vk <= 0x39: return chr(vk) # Letters (uppercase) — card readers typically emit these if 0x41 <= vk <= 0x5A: return chr(vk) if vk == 0x20: # space return ' ' # Anything else (symbols) via MapVirtualKey try: res = _user32.MapVirtualKeyW(vk, 2) # MAPVK_VK_TO_CHAR if res & 0x80000000: return None # dead key c = res & 0xFFFF if 0x20 <= c <= 0x7E: return chr(c) except Exception: pass return None def _get_device_name(hdev): """Return the Win32 device name (e.g. \\\\?\\HID#VID_08FF&PID_0009#...).""" try: size = wintypes.UINT(0) if _user32.GetRawInputDeviceInfoW(hdev, RIDI_DEVICENAME, None, ctypes.byref(size)) == 0xFFFFFFFF: return '' if not size.value: return '' buf = ctypes.create_unicode_buffer(int(size.value) + 2) if _user32.GetRawInputDeviceInfoW(hdev, RIDI_DEVICENAME, buf, ctypes.byref(size)) == 0xFFFFFFFF: return '' return buf.value or '' except Exception: return '' def _enumerate_keyboards(): """Return [{handle, name}] for all Raw-Input keyboard-type devices.""" result = [] try: count = wintypes.UINT(0) _user32.GetRawInputDeviceList(None, ctypes.byref(count), ctypes.sizeof(_RAWINPUTDEVICELIST)) if not count.value: return result buf = (_RAWINPUTDEVICELIST * count.value)() n = _user32.GetRawInputDeviceList(buf, ctypes.byref(count), ctypes.sizeof(_RAWINPUTDEVICELIST)) for i in range(int(n)): dev = buf[i] if dev.dwType != RIM_TYPEKEYBOARD: continue result.append({'handle': dev.hDevice, 'name': _get_device_name(dev.hDevice)}) except Exception as e: _log(f"enumerate_keyboards error: {e}") return result def _choose_card_reader(override=''): """Pick the most likely card-reader device, mirroring the Linux logic.""" keyboards = _enumerate_keyboards() if not keyboards: _log("No keyboard-type Raw Input devices found") return None for dev in keyboards: _log(f" candidate: {dev['name']}") # Config override (substring of device name, e.g. VID_08FF) if override: for dev in keyboards: if override.lower() in dev['name'].lower(): _log(f"Using config override -> {dev['name']}") return dev _log(f"No device matched config override '{override}'; continuing auto-detect") exclusion = ('touch', 'mouse', 'trackpad', 'pen', 'stylus', 'monitor', 'video') # Priority 1: explicit card / reader / rfid for dev in keyboards: name = dev['name'].lower() if 'card' in name or 'reader' in name or 'rfid' in name: _log(f"Priority 1 (explicit reader) -> {dev['name']}") return dev # Priority 2: USB HID keyboard that is NOT the PS/2 system keyboard for dev in keyboards: name = dev['name'].lower() if 'hid' in name and 'vid' in name: if any(p in name for p in ('pnp0303', 'pnp0c0e', 'pnp0320', 'acpi')): continue if any(k in name for k in exclusion): continue _log(f"Priority 2 (USB HID keyboard) -> {dev['name']}") return dev # Priority 3: any keyboard that isn't obviously a touchscreen/mouse for dev in keyboards: name = dev['name'].lower() if any(k in name for k in exclusion): continue _log(f"Priority 3 (any keyboard) -> {dev['name']}") return dev _log(f"Fallback: using first keyboard device -> {keyboards[0]['name']}") return keyboards[0] # ── WndProc / LL-hook callbacks (called on the pump thread) ───────────────── def _wnd_proc(hwnd, msg, wparam, lparam): """Handle WM_INPUT / WM_INPUT_DEVICE_CHANGE for the hidden window.""" try: reader = _ACTIVE_READER if reader is not None: if msg == WM_INPUT: reader._on_raw_input(lparam) elif msg == WM_INPUT_DEVICE_CHANGE: reader._on_device_change() return _user32.DefWindowProcW(hwnd, msg, wparam, lparam) except Exception: try: return _user32.DefWindowProcW(hwnd, msg, wparam, lparam) except Exception: return 0 def _ll_hook_proc(nCode, wParam, lParam): """Low-level keyboard hook used as a fallback capture path.""" try: if nCode == HC_ACTION: reader = _ACTIVE_READER if reader is not None and wParam in (WM_KEYDOWN, WM_SYSKEYDOWN): kb = _KBDLLHOOKSTRUCT.from_address(lParam) reader._on_key_event(kb.vkCode) return _user32.CallNextHookEx(None, nCode, wParam, lParam) except Exception: try: return _user32.CallNextHookEx(None, nCode, wParam, lParam) except Exception: return 1 # ── Background message-pump thread ────────────────────────────────────────── class _RawInputThread(threading.Thread): def __init__(self, owner): super().__init__(daemon=True) self._owner = owner self._hwnd = None self._hook = None self.ready = threading.Event() def run(self): try: hinst = _kernel32.GetModuleHandleW(None) wndclass = _WNDCLASSW() wndclass.lpfnWndProc = _WND_PROC(_wnd_proc) wndclass.hInstance = hinst wndclass.lpszClassName = _WND_CLASS_NAME if not _user32.RegisterClassW(ctypes.byref(wndclass)): raise ctypes.WinError(ctypes.get_last_error(), "RegisterClassW failed") hwnd = _user32.CreateWindowExW( 0, _WND_CLASS_NAME, 'KiwyWinCardReader', 0, 0, 0, 0, 0, None, None, hinst, None, ) if not hwnd: raise ctypes.WinError(ctypes.get_last_error(), "CreateWindowExW failed") self._hwnd = hwnd self._owner._hwnd = hwnd # Register Raw Input for ALL keyboards (sink = receive w/o focus) rids = (_RAWINPUTDEVICE * 1)() rids[0].usUsagePage = 0x01 rids[0].usUsage = 0x06 # Generic Desktop / Keyboard rids[0].dwFlags = RIDEV_INPUTSINK | RIDEV_DEVNOTIFY rids[0].hwndTarget = hwnd raw_ok = bool(_user32.RegisterRawInputDevices( rids, 1, ctypes.sizeof(_RAWINPUTDEVICE))) self._owner._raw_ok = raw_ok if raw_ok: _log("Raw Input registered for keyboard devices") # LL-hook fallback: use it when config says so, or if raw failed mode = getattr(self._owner, '_mode', 'auto') if mode == 'hook' or not raw_ok: proc = _LL_KEYBOARD_PROC(_ll_hook_proc) hook = _user32.SetWindowsHookExW(WH_KEYBOARD_LL, proc, hinst, 0) if hook: self._hook = hook self._hook_proc = proc # keep reference alive _log("Low-level keyboard hook ACTIVE (fallback capture)") self.ready.set() msg = _MSG() while _user32.GetMessageW(ctypes.byref(msg), None, 0, 0) > 0: _user32.TranslateMessage(ctypes.byref(msg)) _user32.DispatchMessageW(ctypes.byref(msg)) if self._hook: try: _user32.UnhookWindowsHookEx(self._hook) except Exception: pass _user32.DestroyWindow(hwnd) _user32.UnregisterClassW(_WND_CLASS_NAME, hinst) except Exception as e: _log(f"Message pump thread error: {e}") self.ready.set() def stop(self): try: if self._hwnd: _user32.PostMessageW(self._hwnd, 0x0012, 0, 0) # WM_QUIT except Exception: pass # ── Public drop-in replacement for the Linux CardReader ───────────────────── class WindowsCardReader: """Windows-native card reader with the same interface as main.py's CardReader. Interface used by SignagePlayer: read_card_async(callback) -> starts listening; callback(card_data) on Enter, or callback(None) on timeout/cancel stop_reading() -> stops listening """ def __init__(self): self._device = None self._device_name = '' self._reading = False self._finished = False self._callback = None self._card_buffer = [] self._last_activity = 0.0 self._timeout = 5.0 self._mode = 'auto' self._hwnd = None self._raw_ok = False self._thread = None self._lock = threading.Lock() self._last_device_change = 0.0 # debounce WM_INPUT_DEVICE_CHANGE # -- config ------------------------------------------------------- def _load_settings(self): cfg = _get_config() self._mode = str(cfg.get('card_reader_mode', 'auto')).lower().strip() or 'auto' try: self._timeout = float(cfg.get('card_reader_timeout', 5)) except Exception: self._timeout = 5.0 if self._timeout <= 0: self._timeout = 5.0 return str(cfg.get('card_reader_device', '') or '').strip() # -- public API --------------------------------------------------- def read_card_async(self, callback): """Start reading; callback(card_data) on Enter, callback(None) on timeout.""" if self._reading: _log("read_card_async called while already reading — ignoring") return override = self._load_settings() global _ACTIVE_READER _ACTIVE_READER = self self._callback = callback self._reading = True self._finished = False self._card_buffer = [] self._last_activity = time.time() # Choose the target device (raw mode only). In 'hook' mode we capture # from all keyboards via the LL-hook instead. if self._mode != 'hook': self._device = None self._device_name = '' chosen = _choose_card_reader(override=override) if chosen: self._device = chosen['handle'] self._device_name = chosen['name'] _log(f"Selected card reader: {chosen['name']}") else: _log("No dedicated device found — will accept any keyboard input") else: self._device = None self._device_name = '' # Ensure the message pump is running if self._thread is None or not self._thread.is_alive(): self._thread = _RawInputThread(self) self._thread.start() self._thread.ready.wait(timeout=3.0) # Watchdog for the 5-second timeout threading.Thread(target=self._timeout_watchdog, daemon=True).start() _log(f"Waiting for card swipe (mode={self._mode}, timeout={self._timeout}s)") def stop_reading(self): """Stop listening (also stops the timeout watchdog).""" self._reading = False def shutdown(self): """Stop the background pump thread (call on app exit).""" self.stop_reading() global _ACTIVE_READER if _ACTIVE_READER is self: _ACTIVE_READER = None if self._thread is not None: self._thread.stop() # -- internals ---------------------------------------------------- def _timeout_watchdog(self): while self._reading and not self._finished: if time.time() - self._last_activity > self._timeout: _log("Read timeout — sending None") self._finish(None) return time.sleep(0.25) def _on_raw_input(self, lparam): """Process a WM_INPUT message (called on the pump thread).""" try: size = wintypes.UINT(0) _user32.GetRawInputData(lparam, RID_INPUT, None, ctypes.byref(size), ctypes.sizeof(_RAWINPUTHEADER)) if not size.value: return buf = ctypes.create_string_buffer(int(size.value)) got = _user32.GetRawInputData(lparam, RID_INPUT, buf, ctypes.byref(size), ctypes.sizeof(_RAWINPUTHEADER)) if got == 0xFFFFFFFF or got == 0: return raw = ctypes.cast(buf, ctypes.POINTER(_RAWINPUT)).contents if raw.header.dwType != RIM_TYPEKEYBOARD: return # Only accept input from the selected card-reader device. if self._device is not None and raw.header.hDevice != self._device: return kb = raw.u.keyboard if kb.Message in (WM_KEYDOWN, WM_SYSKEYDOWN): self._on_key_event(kb.VKey) except Exception as e: _log(f"_on_raw_input error: {e}") def _on_device_change(self): """A HID device was added/removed — re-select only if the actual choice changes, and never more than once per second (the initial registration/enumeration triggers a spurious change event).""" try: if not self._reading or self._mode == 'hook': return now = time.time() if now - self._last_device_change < 1.0: return # debounce self._last_device_change = now override = str(_get_config().get('card_reader_device', '') or '').strip() chosen = _choose_card_reader(override=override) if chosen is None: return # Only re-select if the winning device actually changed. if self._device is not None and chosen['handle'] == self._device: return self._device = chosen['handle'] self._device_name = chosen['name'] _log(f"Device change — re-selected: {chosen['name']}") except Exception: pass def _on_key_event(self, vk): """Handle a single key event (from Raw Input or LL-hook).""" if not self._reading or self._finished: return if vk == _VK_RETURN: data = ''.join(self._card_buffer).strip() self._card_buffer = [] if data: _log(f"Card read complete: '{data}' (len={len(data)})") self._finish(data) else: # Accidental Enter with no data — keep waiting self._last_activity = time.time() return if vk in (_VK_SHIFT, _VK_LSHIFT, _VK_RSHIFT, _VK_CONTROL, _VK_LCONTROL, _VK_RCONTROL, _VK_MENU, _VK_LMENU, _VK_RMENU, _VK_CAPITAL, _VK_TAB, _VK_ESCAPE): return # modifiers / control keys are not card data ch = _vk_to_char(vk) if ch: self._card_buffer.append(ch) self._last_activity = time.time() def _finish(self, data): """Deliver the result exactly once (thread-safe), on the Kivy thread.""" with self._lock: if self._finished: return self._finished = True self._reading = False cb = self._callback self._callback = None if cb is None: return # Prefer scheduling on the Kivy thread when a Kivy app is running # (the callback touches Kivy widgets). If Kivy isn't running # (manual test / non-GUI context), invoke the callback directly. kivy_running = False try: from kivy.app import App kivy_running = App.get_running_app() is not None except Exception: kivy_running = False if kivy_running: try: from kivy.clock import Clock Clock.schedule_once(lambda dt, d=data: cb(d), 0) return except Exception: pass try: cb(data) except Exception: pass if __name__ == '__main__': # Simple manual test (no Kivy): run for a few seconds and print any card. print("Windows Card Reader - manual test (swipe a card within 10s)") def _cb(data): print(f"CALLBACK: {data!r}") reader = WindowsCardReader() reader.read_card_async(_cb) try: time.sleep(10) except KeyboardInterrupt: pass reader.stop_reading() reader.shutdown() print("Test done.")