141 lines
4.5 KiB
Python
141 lines
4.5 KiB
Python
# event_captor.py
|
|
"""
|
|
Hooks souris pour agent_v0.
|
|
|
|
v0+ :
|
|
- écoute des clics souris globaux
|
|
- écoute du scroll (molette)
|
|
- détecte les périodes d'immobilité (hover) > hover_min_idle_ms
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from typing import Callable, Optional
|
|
|
|
from pynput import mouse
|
|
|
|
MouseClickCallback = Callable[[str, int, int], None]
|
|
MouseScrollCallback = Callable[[int, int, int, int], None] # x, y, dx, dy
|
|
MouseHoverCallback = Callable[[int, int, int], None] # x, y, idle_ms
|
|
|
|
|
|
class EventCaptor:
|
|
"""
|
|
Capture les événements souris globaux et les relaie à des callbacks.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
on_mouse_click: MouseClickCallback,
|
|
on_scroll: Optional[MouseScrollCallback] = None,
|
|
on_hover: Optional[MouseHoverCallback] = None,
|
|
hover_min_idle_ms: int = 700,
|
|
) -> None:
|
|
"""
|
|
:param on_mouse_click: fonction appelée sur clic souris,
|
|
signature (button_str, x, y)
|
|
:param on_scroll: fonction appelée sur scroll (molette), ou None
|
|
signature (x, y, dx, dy)
|
|
:param on_hover: fonction appelée quand la souris est immobile
|
|
depuis hover_min_idle_ms, ou None
|
|
signature (x, y, idle_ms)
|
|
"""
|
|
self._on_mouse_click = on_mouse_click
|
|
self._on_scroll = on_scroll
|
|
self._on_hover = on_hover
|
|
self._hover_min_idle_ms = hover_min_idle_ms
|
|
|
|
self._listener: mouse.Listener | None = None
|
|
|
|
# Pour le suivi du déplacement souris (hover)
|
|
self._last_move_time: Optional[float] = None
|
|
self._last_move_pos: Optional[tuple[int, int]] = None
|
|
|
|
self._hover_thread: Optional[threading.Thread] = None
|
|
self._hover_thread_running: bool = False
|
|
|
|
def start(self) -> None:
|
|
"""Démarre l'écoute globale des événements souris."""
|
|
if self._listener is not None:
|
|
return
|
|
|
|
self._listener = mouse.Listener(
|
|
on_click=self._handle_click,
|
|
on_scroll=self._handle_scroll,
|
|
on_move=self._handle_move,
|
|
)
|
|
self._listener.start()
|
|
|
|
if self._on_hover is not None:
|
|
self._hover_thread_running = True
|
|
self._hover_thread = threading.Thread(
|
|
target=self._hover_loop, name="HoverDetector", daemon=True
|
|
)
|
|
self._hover_thread.start()
|
|
|
|
def stop(self) -> None:
|
|
"""Arrête l'écoute globale et le thread hover."""
|
|
if self._listener is not None:
|
|
self._listener.stop()
|
|
self._listener = None
|
|
|
|
self._hover_thread_running = False
|
|
self._hover_thread = None
|
|
|
|
self._last_move_time = None
|
|
self._last_move_pos = None
|
|
|
|
# --- internes ---
|
|
|
|
def _handle_click(self, x: int, y: int, button: mouse.Button, pressed: bool) -> None:
|
|
# On ne loggue que l'événement "pressed"
|
|
if not pressed:
|
|
return
|
|
|
|
button_str = getattr(button, "name", str(button))
|
|
self._on_mouse_click(button_str, x, y)
|
|
|
|
def _handle_scroll(self, x: int, y: int, dx: int, dy: int) -> None:
|
|
if self._on_scroll is not None:
|
|
self._on_scroll(x, y, dx, dy)
|
|
|
|
def _handle_move(self, x: int, y: int) -> None:
|
|
now = time.monotonic()
|
|
self._last_move_time = now
|
|
self._last_move_pos = (x, y)
|
|
|
|
def _hover_loop(self) -> None:
|
|
"""
|
|
Thread qui surveille l'immobilité de la souris.
|
|
Si la souris reste immobile plus de hover_min_idle_ms,
|
|
on déclenche un callback on_hover(x, y, idle_ms).
|
|
"""
|
|
last_hover_trigger_time: float | None = None
|
|
|
|
while self._hover_thread_running:
|
|
time.sleep(0.1)
|
|
|
|
if self._on_hover is None:
|
|
continue
|
|
|
|
if self._last_move_time is None or self._last_move_pos is None:
|
|
continue
|
|
|
|
now = time.monotonic()
|
|
idle_ms = int((now - self._last_move_time) * 1000)
|
|
|
|
if idle_ms < self._hover_min_idle_ms:
|
|
continue
|
|
|
|
# Éviter de déclencher en boucle pour la même "immobilité"
|
|
if last_hover_trigger_time is not None:
|
|
# Si on a déjà déclenché récemment, on attend un nouveau mouvement
|
|
if self._last_move_time <= last_hover_trigger_time:
|
|
continue
|
|
|
|
x, y = self._last_move_pos
|
|
self._on_hover(x, y, idle_ms)
|
|
last_hover_trigger_time = now
|