feat(app): add main application loop, distraction detection, and window closing logic
This commit is contained in:
+207
@@ -0,0 +1,207 @@
|
||||
"""
|
||||
Main Application Engine for Catser.
|
||||
|
||||
Ties together the transparent overlay, ActivityWatch window detection,
|
||||
cat kinematics & state machine, user interactions, and window closing logic.
|
||||
"""
|
||||
|
||||
import time
|
||||
import logging
|
||||
import winsound
|
||||
from typing import Optional
|
||||
|
||||
from .config import Config
|
||||
from .assets_manager import AssetsManager
|
||||
from .window_manager import WindowManager, WindowInfo
|
||||
from .activitywatch import ActivityWatchClient, DistractionEvent
|
||||
from .overlay import OverlayWindow
|
||||
from .cat_controller import CatController, CatState
|
||||
|
||||
logger = logging.getLogger("catser.app")
|
||||
|
||||
|
||||
class CatserApp:
|
||||
"""
|
||||
Core application manager running the 60 FPS animation loop and window monitoring.
|
||||
"""
|
||||
|
||||
def __init__(self, config: Optional[Config] = None):
|
||||
self.config = config or Config()
|
||||
self.assets = AssetsManager(cat_width=self.config.cat_width)
|
||||
self.aw_client = ActivityWatchClient(self.config)
|
||||
self.controller: Optional[CatController] = None
|
||||
self.overlay: Optional[OverlayWindow] = None
|
||||
|
||||
self.is_running = False
|
||||
self.last_aw_check_time = 0.0
|
||||
self.last_closed_time: float = 0.0
|
||||
self.last_closed_hwnd: int = 0
|
||||
|
||||
def initialize(self) -> None:
|
||||
"""Initializes assets, window overlay, and controller."""
|
||||
logger.info("Initializing Catser application...")
|
||||
# 1. Assets
|
||||
self.assets.initialize()
|
||||
|
||||
# 2. Controller
|
||||
self.controller = CatController(self.config, self.assets)
|
||||
|
||||
# 3. Transparent Overlay Window
|
||||
self.overlay = OverlayWindow(
|
||||
width=self.config.cat_width,
|
||||
height=self.config.cat_height,
|
||||
initial_x=int(self.controller.x),
|
||||
initial_y=int(self.controller.y),
|
||||
on_mouse_down=self.controller.on_mouse_down,
|
||||
on_mouse_move=self.controller.on_mouse_move,
|
||||
on_mouse_up=self.controller.on_mouse_up,
|
||||
)
|
||||
|
||||
# 4. ActivityWatch Initial Ping
|
||||
aw_online = self.aw_client.check_connection()
|
||||
if aw_online:
|
||||
logger.info(f"ActivityWatch is active! Bucket: {self.aw_client.bucket_id}")
|
||||
else:
|
||||
logger.info("ActivityWatch offline. Using native Win32 window tracker.")
|
||||
|
||||
def run(self, max_seconds: Optional[float] = None) -> None:
|
||||
"""
|
||||
Runs the main animation loop at ~60 FPS.
|
||||
If max_seconds is provided, exits automatically after the duration (useful for automated tests).
|
||||
"""
|
||||
if not self.overlay or not self.controller:
|
||||
self.initialize()
|
||||
|
||||
self.is_running = True
|
||||
logger.info("Catser is now running! Look for the cat on your desktop.")
|
||||
|
||||
fps_target = 60.0
|
||||
frame_time = 1.0 / fps_target
|
||||
start_time = time.time()
|
||||
last_tick = time.time()
|
||||
|
||||
try:
|
||||
while self.is_running:
|
||||
now = time.time()
|
||||
dt = max(0.001, min(0.1, now - last_tick))
|
||||
last_tick = now
|
||||
|
||||
# 1. Process Windows OS events
|
||||
if not self.overlay.process_messages():
|
||||
logger.info("Overlay window closed.")
|
||||
break
|
||||
|
||||
# 2. Periodic ActivityWatch / Window Distraction Check
|
||||
if now - self.last_aw_check_time >= self.config.poll_interval_sec:
|
||||
self.last_aw_check_time = now
|
||||
self._check_distractions()
|
||||
|
||||
# 3. Update Cat Controller Kinematics & Render Active Frame
|
||||
frame = self.controller.update(dt)
|
||||
|
||||
# 4. Update Layered Window Position & Pixels
|
||||
self.overlay.draw_frame(frame, self.controller.x, self.controller.y)
|
||||
|
||||
# 5. Check duration limit
|
||||
if max_seconds and (now - start_time >= max_seconds):
|
||||
logger.info(f"Reached max duration of {max_seconds}s. Stopping.")
|
||||
break
|
||||
|
||||
# 6. Sleep to maintain ~60 FPS
|
||||
elapsed = time.time() - now
|
||||
sleep_sec = max(0.001, frame_time - elapsed)
|
||||
time.sleep(sleep_sec)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Catser stopped by user.")
|
||||
finally:
|
||||
self.shutdown()
|
||||
|
||||
def _check_distractions(self) -> None:
|
||||
"""Checks for active distraction windows and commands cat to close them."""
|
||||
# Avoid targeting if cat is already attacking or being dragged
|
||||
if self.controller.state in (CatState.ALERT, CatState.RUN, CatState.PAW, CatState.DRAG):
|
||||
return
|
||||
|
||||
now = time.time()
|
||||
distraction = self.aw_client.check_for_distraction()
|
||||
if not distraction:
|
||||
return
|
||||
|
||||
hwnd = distraction.window_info.hwnd
|
||||
|
||||
# Check cooldown to prevent repeatedly attacking the same window
|
||||
if hwnd == self.last_closed_hwnd and (now - self.last_closed_time < self.config.cooldown_after_close_sec):
|
||||
return
|
||||
|
||||
logger.info(
|
||||
f"SPOTTED DISTRACTION! Target: '{distraction.title}' "
|
||||
f"(App: {distraction.app}, Rule: {distraction.matched_rule})"
|
||||
)
|
||||
|
||||
# Command the cat to target this window
|
||||
self.controller.target_window_for_close(
|
||||
distraction.window_info,
|
||||
on_contact=self._on_cat_strike_window,
|
||||
)
|
||||
|
||||
def _on_cat_strike_window(self, target_window: WindowInfo) -> None:
|
||||
"""
|
||||
Executed at exact PAW_CONTACT_FRAME (frame 14) when the paw strikes [X].
|
||||
"""
|
||||
logger.info(f"PAW CONTACT! Closing window '{target_window.title}' (HWND {target_window.hwnd})")
|
||||
self.last_closed_hwnd = target_window.hwnd
|
||||
self.last_closed_time = time.time()
|
||||
|
||||
# Play impact sound if enabled
|
||||
if self.config.sound_enabled:
|
||||
try:
|
||||
# Standard Windows exclamation / hand sound
|
||||
winsound.MessageBeep(winsound.MB_ICONEXCLAMATION)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Execute configured action
|
||||
if self.config.action_on_hit == "close":
|
||||
WindowManager.close_window(target_window.hwnd)
|
||||
elif self.config.action_on_hit == "minimize":
|
||||
WindowManager.minimize_window(target_window.hwnd)
|
||||
else:
|
||||
logger.info(f"[TEST MODE] Simulating close on HWND {target_window.hwnd}")
|
||||
|
||||
def trigger_test_attack(self, target_coords: Optional[tuple] = None) -> None:
|
||||
"""
|
||||
Manually triggers an attack sprint & paw strike on demand.
|
||||
Used for verification and user demonstrations.
|
||||
"""
|
||||
if not self.controller:
|
||||
return
|
||||
|
||||
if target_coords is None:
|
||||
# Default to top right corner of screen if no window is specified
|
||||
w_area = WindowManager.get_work_area()
|
||||
target_coords = (w_area[2] - 100, w_area[1] + 100)
|
||||
|
||||
mock_window = WindowInfo(
|
||||
hwnd=0,
|
||||
title="Demo Test Window",
|
||||
process_name="test.exe",
|
||||
rect=(target_coords[0] - 300, target_coords[1] - 20, target_coords[0] + 30, target_coords[1] + 250),
|
||||
close_button=target_coords,
|
||||
is_visible=True,
|
||||
is_minimized=False,
|
||||
)
|
||||
|
||||
logger.info(f"Triggering manual test attack to {target_coords}")
|
||||
self.controller.target_window_for_close(
|
||||
mock_window,
|
||||
on_contact=lambda w: logger.info("Manual test attack PAW SWIPE completed!"),
|
||||
)
|
||||
|
||||
def shutdown(self) -> None:
|
||||
"""Destroys overlay window and releases resources."""
|
||||
self.is_running = False
|
||||
if self.overlay:
|
||||
self.overlay.destroy()
|
||||
self.overlay = None
|
||||
logger.info("Catser shutdown cleanly.")
|
||||
Reference in New Issue
Block a user