Add event watch functions.

Author: HexDecimal

build_sdl.py

@@ -207,7 +207,10 @@ def on_directive_handle(
 extern "Python" {
 // SDL_AudioCallback callback.
 void _sdl_audio_callback(void* userdata, Uint8* stream, int len);
+// SDL to Python log function.
 void _sdl_log_output_function(void *userdata, int category, SDL_LogPriority priority, const char *message);
+// Generic event watcher callback.
+int _sdl_event_watcher(void* userdata, SDL_Event* event);
 }
 """
 

tcod/cdef.h

@@ -16,7 +16,5 @@ float _pycall_path_dest_only(int x1, int y1, int x2, int y2, void* user_data);
 
 void _pycall_sdl_hook(struct SDL_Surface*);
 
-int _pycall_event_watch(void* userdata, union SDL_Event* event);
-
 void _pycall_cli_output(void* userdata, const char* output);
 }

tcod/event.py

@@ -23,7 +23,7 @@
 
 import enum
 import warnings
-from typing import Any, Callable, Dict, Generic, Iterator, Mapping, NamedTuple, Optional, Tuple, TypeVar, Union
+from typing import Any, Callable, Dict, Generic, Iterator, Mapping, NamedTuple, Optional, Tuple, Type, TypeVar, Union
 
 import numpy as np
 from numpy.typing import NDArray
@@ -238,7 +238,7 @@ def __init__(self, type: Optional[str] = None):
         self.sdl_event = None
 
     @classmethod
-    def from_sdl_event(cls, sdl_event: Any) -> Any:
+    def from_sdl_event(cls, sdl_event: Any) -> Event:
         """Return a class instance from a python-cffi 'SDL_Event*' pointer."""
         raise NotImplementedError()
 
@@ -739,7 +739,7 @@ def __str__(self) -> str:
         return "<Undefined>"
 
 
-_SDL_TO_CLASS_TABLE: Dict[int, Any] = {
+_SDL_TO_CLASS_TABLE: Dict[int, Type[Event]] = {
     lib.SDL_QUIT: Quit,
     lib.SDL_KEYDOWN: KeyDown,
     lib.SDL_KEYUP: KeyUp,
@@ -752,6 +752,13 @@ def __str__(self) -> str:
 }
 
 
+def _parse_event(sdl_event: Any) -> Event:
+    """Convert a C SDL_Event* type into a tcod Event sub-class."""
+    if sdl_event.type not in _SDL_TO_CLASS_TABLE:
+        return Undefined.from_sdl_event(sdl_event)
+    return _SDL_TO_CLASS_TABLE[sdl_event.type].from_sdl_event(sdl_event)
+
+
 def get() -> Iterator[Any]:
     """Return an iterator for all pending events.
 
@@ -1065,10 +1072,63 @@ def get_mouse_state() -> MouseState:
 
 
 @ffi.def_extern()  # type: ignore
-def _pycall_event_watch(userdata: Any, sdl_event: Any) -> int:
+def _sdl_event_watcher(userdata: Any, sdl_event: Any) -> int:
+    callback: Callable[[Event], None] = ffi.from_handle(userdata)
+    callback(_parse_event(sdl_event))
     return 0
 
 
+_EventCallback = TypeVar("_EventCallback", bound=Callable[[Event], None])
+_event_watch_handles: Dict[Callable[[Event], None], Any] = {}  # Callbacks and their FFI handles.
+
+
+def add_watch(callback: _EventCallback) -> _EventCallback:
+    """Add a callback for watching events.
+
+    This function can be called with the callback to register, or be used as a decorator.
+
+    Callbacks added as event watchers can later be removed with :any:`tcod.event.remove_watch`.
+
+    Args:
+        callback (Callable[[Event], None]):
+            A function which accepts :any:`Event` parameters.
+
+    Example::
+
+        import tcod.event
+
+        @tcod.event.add_watch
+        def handle_events(event: tcod.event.Event) -> None:
+            if isinstance(event, tcod.event.KeyDown):
+                print(event)
+
+    .. versionadded:: 13.4
+    """
+    if callback in _event_watch_handles:
+        warnings.warn(f"{callback} is already an active event watcher, nothing was added.", RuntimeWarning)
+        return callback
+    handle = _event_watch_handles[callback] = ffi.new_handle(callback)
+    lib.SDL_AddEventWatch(lib._sdl_event_watcher, handle)
+    return callback
+
+
+def remove_watch(callback: Callable[[Event], None]) -> None:
+    """Remove a callback as an event wacher.
+
+    Args:
+        callback (Callable[[Event], None]):
+            A function which has been previously registered with :any:`tcod.event.add_watch`.
+
+    .. versionadded:: 13.4
+    """
+    if callback not in _event_watch_handles:
+        warnings.warn(f"{callback} is not an active event watcher, nothing was removed.", RuntimeWarning)
+        return
+    handle = _event_watch_handles[callback]
+    lib.SDL_DelEventWatch(lib._sdl_event_watcher, handle)
+    del _event_watch_handles[callback]
+
+
 def get_keyboard_state() -> NDArray[np.bool_]:
     """Return a boolean array with the current keyboard state.