""" 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.canvas_width, height=self.config.canvas_height, initial_x=int(self.controller.x - self.config.pad_x), initial_y=int(self.controller.y - self.config.pad_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, on_display_change=self.controller.refresh_display_bounds, ) # 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.config.pad_x, self.controller.y - self.config.pad_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.aw_client: self.aw_client.stop() if self.overlay: self.overlay.destroy() self.overlay = None logger.info("Catser shutdown cleanly.")