commit 6be12dc233645c662ed208ea5d6a1f2c0e6a6a4c Author: Shinuwa Date: Mon Sep 7 16:21:49 2026 +0200 Initial commit diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..f0b22be --- /dev/null +++ b/.env.example @@ -0,0 +1,4 @@ +NTFY_URL= +NTFY_TOPIC= +NTFY_TOKEN= + diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..159d808 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +.env +.codex-home/ +.venv/ +__pycache__/ +*.py[cod] +.pytest_cache/ +dist/ +build/ +*.spec diff --git a/README.md b/README.md new file mode 100644 index 0000000..ff479f8 --- /dev/null +++ b/README.md @@ -0,0 +1,68 @@ +# Stall Watcher + +Stall Watcher is a small passive Windows desktop app that monitors a selected screen region and sends a notification when that region stays visually still for a configured duration. + +It only uses local screenshots and image comparison. It does not click, type, inject code, read another process memory, or interact with the watched application. + +## Install + +Requires Python 3.12+. + +```bash +python -m venv .venv +.venv\Scripts\activate +pip install -r requirements.txt +``` + +## Configure ntfy + +Copy `.env.example` to `.env` and fill in your own values: + +```env +NTFY_URL=https://your.ntfy.server +NTFY_TOPIC=stall-watcher +NTFY_TOKEN=your-token +``` + +The `.env` file is only for ntfy secrets. App preferences are stored separately with Qt settings. + +## Run + +From the repository root: + +```bash +python -m src.main +``` + +## Use + +1. Select the monitor to watch. +2. Click **Select region**. +3. Drag a rectangle over the area to monitor. Press Escape to cancel. +4. Adjust capture interval, alert delay, and movement threshold. +5. Enable Windows and/or ntfy notifications. +6. Click **Start monitoring**. + +The live motion score helps calibrate the threshold. For example, if motion is around `8.4` while moving and `0.7` while stopped, a threshold around `2.0` is a reasonable starting point. + +## Build a Windows exe + +Install PyInstaller in your virtualenv: + +```bash +pip install pyinstaller +pyinstaller --noconsole --name StallWatcher --paths . src/main.py +``` + +The executable will be created under `dist/`. + +## Development + +Run tests: + +```bash +pytest +``` + +The detector tests cover motion scoring, still-time accumulation, reset on movement, one alert per still period, and rearming after movement resumes. + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..fc12eba --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +PySide6>=6.7 +mss>=9.0 +numpy>=1.26 +python-dotenv>=1.0 +requests>=2.31 +pytest>=8.0 + diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ + diff --git a/src/capture.py b/src/capture.py new file mode 100644 index 0000000..4bc0899 --- /dev/null +++ b/src/capture.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from typing import Iterable + +import mss +import numpy as np + +from .models import MonitorInfo, Region + + +class CaptureError(RuntimeError): + pass + + +def list_monitors() -> list[MonitorInfo]: + try: + with mss.mss() as sct: + return [ + MonitorInfo( + id=index, + left=int(monitor["left"]), + top=int(monitor["top"]), + width=int(monitor["width"]), + height=int(monitor["height"]), + ) + for index, monitor in enumerate(sct.monitors[1:], start=1) + ] + except Exception as exc: + raise CaptureError(f"Unable to list monitors: {exc}") from exc + + +def get_monitor(monitors: Iterable[MonitorInfo], monitor_id: int) -> MonitorInfo | None: + return next((monitor for monitor in monitors if monitor.id == monitor_id), None) + + +def validate_region(region: Region, monitor: MonitorInfo) -> None: + if not region.is_valid: + raise CaptureError("Selected region is empty.") + if region.monitor_id != monitor.id: + raise CaptureError("Selected region belongs to a disconnected or different monitor.") + if region.x < monitor.left or region.y < monitor.top: + raise CaptureError("Selected region starts outside the monitor.") + if region.x + region.width > monitor.left + monitor.width: + raise CaptureError("Selected region is wider than the monitor bounds.") + if region.y + region.height > monitor.top + monitor.height: + raise CaptureError("Selected region is taller than the monitor bounds.") + + +def capture_region(region: Region) -> np.ndarray: + bbox = { + "left": region.x, + "top": region.y, + "width": region.width, + "height": region.height, + } + try: + with mss.mss() as sct: + shot = sct.grab(bbox) + return np.asarray(shot, dtype=np.uint8)[:, :, :3] + except Exception as exc: + raise CaptureError(f"Unable to capture selected region: {exc}") from exc + diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..4046ecc --- /dev/null +++ b/src/config.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass + +from dotenv import load_dotenv +from PySide6.QtCore import QSettings + +from .models import DetectionConfig, Region + + +ORG_NAME = "StallWatcher" +APP_NAME = "Stall Watcher" + + +@dataclass(frozen=True) +class NtfyConfig: + url: str + topic: str + token: str + + @property + def is_complete(self) -> bool: + return bool(self.url and self.topic and self.token) + + +def load_ntfy_config() -> NtfyConfig: + load_dotenv() + return NtfyConfig( + url=os.getenv("NTFY_URL", "").rstrip("/"), + topic=os.getenv("NTFY_TOPIC", "").strip("/"), + token=os.getenv("NTFY_TOKEN", ""), + ) + + +class AppSettings: + def __init__(self) -> None: + self._settings = QSettings(ORG_NAME, APP_NAME) + + def monitor_id(self) -> int: + return int(self._settings.value("monitor_id", 1)) + + def set_monitor_id(self, monitor_id: int) -> None: + self._settings.setValue("monitor_id", int(monitor_id)) + + def region(self) -> Region | None: + width = int(self._settings.value("region/width", 0)) + height = int(self._settings.value("region/height", 0)) + if width <= 0 or height <= 0: + return None + return Region( + x=int(self._settings.value("region/x", 0)), + y=int(self._settings.value("region/y", 0)), + width=width, + height=height, + monitor_id=int(self._settings.value("region/monitor_id", self.monitor_id())), + ) + + def set_region(self, region: Region) -> None: + self._settings.setValue("region/x", region.x) + self._settings.setValue("region/y", region.y) + self._settings.setValue("region/width", region.width) + self._settings.setValue("region/height", region.height) + self._settings.setValue("region/monitor_id", region.monitor_id) + + def detection_config(self) -> DetectionConfig: + return DetectionConfig( + movement_threshold=float(self._settings.value("movement_threshold", 2.0)), + alert_after=float(self._settings.value("alert_after", 15.0)), + capture_interval=float(self._settings.value("capture_interval", 1.0)), + ) + + def set_detection_config(self, config: DetectionConfig) -> None: + self._settings.setValue("movement_threshold", config.movement_threshold) + self._settings.setValue("alert_after", config.alert_after) + self._settings.setValue("capture_interval", config.capture_interval) + + def enable_ntfy(self) -> bool: + return self._as_bool(self._settings.value("enable_ntfy", True)) + + def set_enable_ntfy(self, enabled: bool) -> None: + self._settings.setValue("enable_ntfy", enabled) + + def enable_windows_notifications(self) -> bool: + return self._as_bool(self._settings.value("enable_windows_notifications", True)) + + def set_enable_windows_notifications(self, enabled: bool) -> None: + self._settings.setValue("enable_windows_notifications", enabled) + + @staticmethod + def _as_bool(value: object) -> bool: + if isinstance(value, bool): + return value + return str(value).lower() in {"1", "true", "yes"} + diff --git a/src/detector.py b/src/detector.py new file mode 100644 index 0000000..f1dff59 --- /dev/null +++ b/src/detector.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import numpy as np + +from .models import DetectionConfig, DetectionState, DetectionStatus + + +class MotionDetector: + def __init__(self, config: DetectionConfig, downscale_step: int = 2) -> None: + self.config = config + self.downscale_step = max(1, downscale_step) + self.previous_frame: np.ndarray | None = None + self.state = DetectionState.MOVING + self.still_time = 0.0 + + def reset(self) -> None: + self.previous_frame = None + self.state = DetectionState.MOVING + self.still_time = 0.0 + + def update(self, frame: np.ndarray, elapsed: float | None = None) -> DetectionStatus: + current = self._prepare_frame(frame) + if self.previous_frame is None: + self.previous_frame = current + return DetectionStatus(0.0, self.still_time, self.state, False) + + score = self.motion_score_from_prepared(self.previous_frame, current) + self.previous_frame = current + interval = self.config.capture_interval if elapsed is None else elapsed + return self._advance_state(score, interval) + + def _advance_state(self, motion_score: float, elapsed: float) -> DetectionStatus: + moving = motion_score >= self.config.movement_threshold + should_alert = False + + if self.state == DetectionState.MOVING: + if moving: + self.still_time = 0.0 + else: + self.still_time = elapsed + self.state = DetectionState.STILL_WAITING + + elif self.state == DetectionState.STILL_WAITING: + if moving: + self.still_time = 0.0 + self.state = DetectionState.MOVING + else: + self.still_time += elapsed + if self.still_time >= self.config.alert_after: + should_alert = True + self.state = DetectionState.ALERTED + + elif self.state == DetectionState.ALERTED: + if moving: + self.still_time = 0.0 + self.state = DetectionState.MOVING + + return DetectionStatus(motion_score, self.still_time, self.state, should_alert) + + def calculate_motion_score(self, previous: np.ndarray, current: np.ndarray) -> float: + return self.motion_score_from_prepared( + self._prepare_frame(previous), + self._prepare_frame(current), + ) + + def _prepare_frame(self, frame: np.ndarray) -> np.ndarray: + if frame.ndim == 3: + rgb = frame[..., :3].astype(np.float32) + gray = (0.299 * rgb[..., 0]) + (0.587 * rgb[..., 1]) + (0.114 * rgb[..., 2]) + elif frame.ndim == 2: + gray = frame.astype(np.float32) + else: + raise ValueError("Frame must be a 2D grayscale or 3D RGB/BGR array.") + return gray[:: self.downscale_step, :: self.downscale_step] + + @staticmethod + def motion_score_from_prepared(previous: np.ndarray, current: np.ndarray) -> float: + if previous.shape != current.shape: + raise ValueError("Frames must have the same shape.") + diff = np.abs(current.astype(np.float32) - previous.astype(np.float32)) + return float(np.clip((diff.mean() / 255.0) * 100.0, 0.0, 100.0)) + diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..46136bc --- /dev/null +++ b/src/main.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from pathlib import Path +import sys + +from dotenv import load_dotenv +from PySide6.QtWidgets import QApplication + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + from src.config import APP_NAME, ORG_NAME + from src.ui.main_window import MainWindow +else: + from .config import APP_NAME, ORG_NAME + from .ui.main_window import MainWindow + + +def main() -> int: + load_dotenv() + app = QApplication(sys.argv) + app.setOrganizationName(ORG_NAME) + app.setApplicationName(APP_NAME) + window = MainWindow(app) + window.show() + return app.exec() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/models.py b/src/models.py new file mode 100644 index 0000000..6ff5667 --- /dev/null +++ b/src/models.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class DetectionState(str, Enum): + MOVING = "MOVING" + STILL_WAITING = "STILL_WAITING" + ALERTED = "ALERTED" + + +class UiStatus(str, Enum): + STOPPED = "STOPPED" + MOVING = "MOVING" + WAITING = "WAITING" + ALERTED = "ALERTED" + + +@dataclass(frozen=True) +class MonitorInfo: + id: int + left: int + top: int + width: int + height: int + + @property + def label(self) -> str: + return f"Monitor {self.id} - {self.width}x{self.height}" + + +@dataclass(frozen=True) +class Region: + x: int + y: int + width: int + height: int + monitor_id: int + + @property + def is_valid(self) -> bool: + return self.width > 0 and self.height > 0 + + +@dataclass(frozen=True) +class DetectionConfig: + movement_threshold: float = 2.0 + alert_after: float = 15.0 + capture_interval: float = 1.0 + + +@dataclass(frozen=True) +class DetectionStatus: + motion_score: float + still_time: float + state: DetectionState + should_alert: bool = False + diff --git a/src/notifications/__init__.py b/src/notifications/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/notifications/__init__.py @@ -0,0 +1 @@ + diff --git a/src/notifications/ntfy.py b/src/notifications/ntfy.py new file mode 100644 index 0000000..97c79ab --- /dev/null +++ b/src/notifications/ntfy.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import requests + +from ..config import NtfyConfig + + +class NtfyNotifier: + def __init__(self, config: NtfyConfig, timeout: float = 5.0) -> None: + self.config = config + self.timeout = timeout + + def send(self, message: str, title: str = "Stall Watcher") -> tuple[bool, str]: + if not self.config.is_complete: + return False, "ntfy is not configured." + + url = f"{self.config.url}/{self.config.topic}" + headers = { + "Authorization": f"Bearer {self.config.token}", + "Title": title, + "Priority": "high", + } + try: + response = requests.post(url, data=message.encode("utf-8"), headers=headers, timeout=self.timeout) + response.raise_for_status() + return True, "ntfy notification sent." + except requests.RequestException as exc: + return False, f"ntfy notification failed: {exc}" + diff --git a/src/notifications/windows.py b/src/notifications/windows.py new file mode 100644 index 0000000..c6efa39 --- /dev/null +++ b/src/notifications/windows.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from PySide6.QtGui import QIcon +from PySide6.QtWidgets import QApplication, QSystemTrayIcon + + +class WindowsNotifier: + def __init__(self, app: QApplication) -> None: + self._tray = QSystemTrayIcon(QIcon(), app) + self._tray.setToolTip("Stall Watcher") + self._tray.show() + + def send(self, message: str, title: str = "Stall Watcher") -> tuple[bool, str]: + try: + if not QSystemTrayIcon.isSystemTrayAvailable(): + return False, "System tray is not available." + self._tray.showMessage(title, message, QSystemTrayIcon.MessageIcon.Warning, 8000) + return True, "Windows notification sent." + except Exception as exc: + return False, f"Windows notification failed: {exc}" + diff --git a/src/ui/__init__.py b/src/ui/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/ui/__init__.py @@ -0,0 +1 @@ + diff --git a/src/ui/main_window.py b/src/ui/main_window.py new file mode 100644 index 0000000..2bdef61 --- /dev/null +++ b/src/ui/main_window.py @@ -0,0 +1,342 @@ +from __future__ import annotations + +import time + +from PySide6.QtCore import QObject, QThread, Signal, Slot +from PySide6.QtWidgets import ( + QApplication, + QCheckBox, + QComboBox, + QDoubleSpinBox, + QFormLayout, + QGridLayout, + QGroupBox, + QLabel, + QMainWindow, + QMessageBox, + QPushButton, + QSpinBox, + QVBoxLayout, + QWidget, +) + +from ..capture import CaptureError, capture_region, get_monitor, list_monitors, validate_region +from ..config import AppSettings, load_ntfy_config +from ..detector import MotionDetector +from ..models import DetectionConfig, DetectionState, DetectionStatus, MonitorInfo, Region, UiStatus +from ..notifications.ntfy import NtfyNotifier +from ..notifications.windows import WindowsNotifier +from .region_selector import RegionSelector + + +class MonitorWorker(QObject): + status_changed = Signal(object) + error = Signal(str) + alert = Signal(str) + finished = Signal() + + def __init__(self, region: Region, config: DetectionConfig) -> None: + super().__init__() + self.region = region + self.config = config + self._running = False + self._detector = MotionDetector(config) + + @Slot() + def run(self) -> None: + self._running = True + last_capture = time.monotonic() + try: + while self._running: + started = time.monotonic() + frame = capture_region(self.region) + elapsed = max(0.0, started - last_capture) + last_capture = started + status = self._detector.update(frame, elapsed or self.config.capture_interval) + self.status_changed.emit(status) + if status.should_alert: + message = f"No movement detected for {self.config.alert_after:g} seconds." + self.alert.emit(message) + sleep_for = max(0.05, self.config.capture_interval - (time.monotonic() - started)) + end_time = time.monotonic() + sleep_for + while self._running and time.monotonic() < end_time: + QThread.msleep(50) + except CaptureError as exc: + self.error.emit(str(exc)) + except Exception as exc: + self.error.emit(f"Monitoring failed: {exc}") + finally: + self.finished.emit() + + @Slot() + def stop(self) -> None: + self._running = False + + +class MainWindow(QMainWindow): + def __init__(self, app: QApplication) -> None: + super().__init__() + self.setWindowTitle("Stall Watcher") + self.resize(440, 560) + self._settings = AppSettings() + self._monitors: list[MonitorInfo] = [] + self._region: Region | None = self._settings.region() + self._selector: RegionSelector | None = None + self._thread: QThread | None = None + self._worker: MonitorWorker | None = None + self._windows_notifier = WindowsNotifier(app) + self._build_ui() + self._load_monitors() + self._load_settings() + self._refresh_region_label() + self._set_status(UiStatus.STOPPED) + + def closeEvent(self, event) -> None: # type: ignore[no-untyped-def] + self._save_settings() + self.stop_monitoring() + super().closeEvent(event) + + def _build_ui(self) -> None: + root = QWidget() + layout = QVBoxLayout(root) + + title = QLabel("Stall Watcher") + title.setStyleSheet("font-size: 22px; font-weight: 600;") + layout.addWidget(title) + + monitoring_group = QGroupBox("Monitoring") + monitoring_layout = QFormLayout(monitoring_group) + self.monitor_combo = QComboBox() + self.monitor_combo.currentIndexChanged.connect(self._monitor_changed) + self.select_region_button = QPushButton("Select region") + self.select_region_button.clicked.connect(self._select_region) + self.region_label = QLabel("No region selected") + monitoring_layout.addRow("Monitor:", self.monitor_combo) + monitoring_layout.addRow("Region:", self.select_region_button) + monitoring_layout.addRow("Selected region:", self.region_label) + layout.addWidget(monitoring_group) + + detection_group = QGroupBox("Detection") + detection_layout = QFormLayout(detection_group) + self.capture_interval = QDoubleSpinBox() + self.capture_interval.setRange(0.1, 60.0) + self.capture_interval.setSingleStep(0.1) + self.capture_interval.setSuffix(" seconds") + self.alert_after = QSpinBox() + self.alert_after.setRange(1, 3600) + self.alert_after.setSuffix(" seconds") + self.threshold = QDoubleSpinBox() + self.threshold.setRange(0.0, 100.0) + self.threshold.setSingleStep(0.1) + self.threshold.setDecimals(2) + self.motion_score_label = QLabel("0.00") + self.still_for_label = QLabel("0.0 s") + self.state_label = QLabel("STOPPED") + self.status_dot = QLabel() + self.status_dot.setFixedSize(16, 16) + status_row = QWidget() + status_layout = QGridLayout(status_row) + status_layout.setContentsMargins(0, 0, 0, 0) + status_layout.addWidget(self.status_dot, 0, 0) + status_layout.addWidget(self.state_label, 0, 1) + detection_layout.addRow("Capture interval:", self.capture_interval) + detection_layout.addRow("Alert after:", self.alert_after) + detection_layout.addRow("Movement threshold:", self.threshold) + detection_layout.addRow("Motion score:", self.motion_score_label) + detection_layout.addRow("Still for:", self.still_for_label) + detection_layout.addRow("State:", status_row) + layout.addWidget(detection_group) + + notifications_group = QGroupBox("Notifications") + notifications_layout = QFormLayout(notifications_group) + self.windows_checkbox = QCheckBox("Windows notification") + self.ntfy_checkbox = QCheckBox("ntfy") + self.test_button = QPushButton("Test notification") + self.test_button.clicked.connect(self._test_notification) + self.notification_result = QLabel("") + notifications_layout.addRow(self.windows_checkbox) + notifications_layout.addRow(self.ntfy_checkbox) + notifications_layout.addRow(self.test_button) + notifications_layout.addRow(self.notification_result) + layout.addWidget(notifications_group) + + button_row = QWidget() + button_layout = QGridLayout(button_row) + button_layout.setContentsMargins(0, 0, 0, 0) + self.start_button = QPushButton("Start monitoring") + self.stop_button = QPushButton("Stop") + self.stop_button.setEnabled(False) + self.start_button.clicked.connect(self.start_monitoring) + self.stop_button.clicked.connect(self.stop_monitoring) + button_layout.addWidget(self.start_button, 0, 0) + button_layout.addWidget(self.stop_button, 0, 1) + layout.addWidget(button_row) + + self.setCentralWidget(root) + + def _load_monitors(self) -> None: + self.monitor_combo.clear() + try: + self._monitors = list_monitors() + except CaptureError as exc: + self._monitors = [] + QMessageBox.warning(self, "Monitor error", str(exc)) + + for monitor in self._monitors: + self.monitor_combo.addItem(monitor.label, monitor.id) + + if not self._monitors: + self.monitor_combo.addItem("No monitor detected", None) + self.start_button.setEnabled(False) + self.select_region_button.setEnabled(False) + return + + saved_id = self._settings.monitor_id() + index = self.monitor_combo.findData(saved_id) + self.monitor_combo.setCurrentIndex(index if index >= 0 else 0) + + def _load_settings(self) -> None: + config = self._settings.detection_config() + self.capture_interval.setValue(config.capture_interval) + self.alert_after.setValue(int(config.alert_after)) + self.threshold.setValue(config.movement_threshold) + self.ntfy_checkbox.setChecked(self._settings.enable_ntfy()) + self.windows_checkbox.setChecked(self._settings.enable_windows_notifications()) + + def _current_monitor(self) -> MonitorInfo | None: + monitor_id = self.monitor_combo.currentData() + if monitor_id is None: + return None + return get_monitor(self._monitors, int(monitor_id)) + + def _monitor_changed(self) -> None: + monitor = self._current_monitor() + if monitor is not None: + self._settings.set_monitor_id(monitor.id) + + def _select_region(self) -> None: + monitor = self._current_monitor() + if monitor is None: + QMessageBox.warning(self, "Region", "No monitor is available.") + return + self._selector = RegionSelector(monitor) + self._selector.region_selected.connect(self._region_selected) + self._selector.selection_cancelled.connect(lambda: self.notification_result.setText("Region selection cancelled.")) + self._selector.show() + self._selector.activateWindow() + self._selector.raise_() + + def _region_selected(self, region: Region) -> None: + self._region = region + self._settings.set_region(region) + self._refresh_region_label() + + def _refresh_region_label(self) -> None: + if self._region is None: + self.region_label.setText("No region selected") + return + self.region_label.setText( + f"X: {self._region.x}\nY: {self._region.y}\n" + f"Width: {self._region.width}\nHeight: {self._region.height}" + ) + + def _detection_config(self) -> DetectionConfig: + return DetectionConfig( + movement_threshold=float(self.threshold.value()), + alert_after=float(self.alert_after.value()), + capture_interval=float(self.capture_interval.value()), + ) + + def _save_settings(self) -> None: + self._settings.set_detection_config(self._detection_config()) + self._settings.set_enable_ntfy(self.ntfy_checkbox.isChecked()) + self._settings.set_enable_windows_notifications(self.windows_checkbox.isChecked()) + + def start_monitoring(self) -> None: + self._save_settings() + monitor = self._current_monitor() + if monitor is None: + QMessageBox.warning(self, "Monitoring", "No monitor is available.") + return + if self._region is None: + QMessageBox.warning(self, "Monitoring", "Select a region before starting.") + return + try: + validate_region(self._region, monitor) + except CaptureError as exc: + QMessageBox.warning(self, "Monitoring", str(exc)) + return + + self.start_button.setEnabled(False) + self.stop_button.setEnabled(True) + self._set_status(UiStatus.MOVING) + self._thread = QThread(self) + self._worker = MonitorWorker(self._region, self._detection_config()) + self._worker.moveToThread(self._thread) + self._thread.started.connect(self._worker.run) + self._worker.status_changed.connect(self._handle_status) + self._worker.alert.connect(self._send_alert) + self._worker.error.connect(self._handle_worker_error) + self._worker.finished.connect(self._thread.quit) + self._worker.finished.connect(self._worker.deleteLater) + self._thread.finished.connect(self._thread.deleteLater) + self._thread.finished.connect(self._monitoring_finished) + self._thread.start() + + def stop_monitoring(self) -> None: + if self._worker is not None: + self._worker.stop() + if self._thread is not None and self._thread.isRunning(): + self._thread.quit() + self._thread.wait(2000) + + def _monitoring_finished(self) -> None: + self._worker = None + self._thread = None + self.start_button.setEnabled(True) + self.stop_button.setEnabled(False) + self._set_status(UiStatus.STOPPED) + + def _handle_status(self, status: DetectionStatus) -> None: + self.motion_score_label.setText(f"{status.motion_score:.2f}") + self.still_for_label.setText(f"{status.still_time:.1f} s") + if status.state == DetectionState.MOVING: + self._set_status(UiStatus.MOVING) + elif status.state == DetectionState.STILL_WAITING: + self._set_status(UiStatus.WAITING) + else: + self._set_status(UiStatus.ALERTED) + + def _set_status(self, status: UiStatus) -> None: + colors = { + UiStatus.STOPPED: "#666666", + UiStatus.MOVING: "#1f9d55", + UiStatus.WAITING: "#d99000", + UiStatus.ALERTED: "#d92d20", + } + self.state_label.setText(status.value) + self.status_dot.setStyleSheet(f"background: {colors[status]}; border-radius: 8px;") + + def _send_alert(self, message: str) -> None: + results = self._send_notifications(message) + self.notification_result.setText(" | ".join(results)) + + def _test_notification(self) -> None: + self._save_settings() + results = self._send_notifications("Test notification from Stall Watcher") + self.notification_result.setText(" | ".join(results) if results else "No notification channel enabled.") + + def _send_notifications(self, message: str) -> list[str]: + results: list[str] = [] + if self.windows_checkbox.isChecked(): + ok, detail = self._windows_notifier.send(message) + results.append(("OK: " if ok else "Error: ") + detail) + if self.ntfy_checkbox.isChecked(): + ok, detail = NtfyNotifier(load_ntfy_config()).send(message) + results.append(("OK: " if ok else "Error: ") + detail) + return results + + def _handle_worker_error(self, message: str) -> None: + self.notification_result.setText(message) + QMessageBox.warning(self, "Monitoring error", message) + self.stop_monitoring() diff --git a/src/ui/region_selector.py b/src/ui/region_selector.py new file mode 100644 index 0000000..49f2722 --- /dev/null +++ b/src/ui/region_selector.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +from PySide6.QtCore import Qt, QRect, QPoint, Signal +from PySide6.QtGui import QColor, QKeyEvent, QMouseEvent, QPainter, QPen +from PySide6.QtWidgets import QWidget + +from ..models import MonitorInfo, Region + + +class RegionSelector(QWidget): + region_selected = Signal(object) + selection_cancelled = Signal() + + def __init__(self, monitor: MonitorInfo) -> None: + super().__init__() + self.monitor = monitor + self._origin: QPoint | None = None + self._current: QPoint | None = None + self.setWindowFlags(Qt.WindowType.FramelessWindowHint | Qt.WindowType.WindowStaysOnTopHint) + self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground) + self.setGeometry(monitor.left, monitor.top, monitor.width, monitor.height) + self.setCursor(Qt.CursorShape.CrossCursor) + self.setFocusPolicy(Qt.FocusPolicy.StrongFocus) + + def mousePressEvent(self, event: QMouseEvent) -> None: + if event.button() == Qt.MouseButton.LeftButton: + self._origin = event.position().toPoint() + self._current = self._origin + self.update() + + def mouseMoveEvent(self, event: QMouseEvent) -> None: + if self._origin is not None: + self._current = event.position().toPoint() + self.update() + + def mouseReleaseEvent(self, event: QMouseEvent) -> None: + if event.button() != Qt.MouseButton.LeftButton or self._origin is None: + return + self._current = event.position().toPoint() + rect = self._selection_rect() + if rect.width() > 2 and rect.height() > 2: + self.region_selected.emit( + Region( + x=self.monitor.left + rect.x(), + y=self.monitor.top + rect.y(), + width=rect.width(), + height=rect.height(), + monitor_id=self.monitor.id, + ) + ) + else: + self.selection_cancelled.emit() + self.close() + + def keyPressEvent(self, event: QKeyEvent) -> None: + if event.key() == Qt.Key.Key_Escape: + self.selection_cancelled.emit() + self.close() + + def paintEvent(self, event) -> None: # type: ignore[no-untyped-def] + painter = QPainter(self) + painter.fillRect(self.rect(), QColor(0, 0, 0, 90)) + if self._origin is None or self._current is None: + return + rect = self._selection_rect() + painter.fillRect(rect, QColor(60, 140, 255, 55)) + pen = QPen(QColor(60, 140, 255), 2) + painter.setPen(pen) + painter.drawRect(rect) + + def _selection_rect(self) -> QRect: + if self._origin is None or self._current is None: + return QRect() + return QRect(self._origin, self._current).normalized() diff --git a/tests/test_detector.py b/tests/test_detector.py new file mode 100644 index 0000000..b293674 --- /dev/null +++ b/tests/test_detector.py @@ -0,0 +1,67 @@ +import numpy as np + +from src.detector import MotionDetector +from src.models import DetectionConfig, DetectionState + + +def detector() -> MotionDetector: + return MotionDetector(DetectionConfig(movement_threshold=2.0, alert_after=3.0, capture_interval=1.0), downscale_step=1) + + +def test_identical_images_score_near_zero() -> None: + image = np.zeros((10, 10, 3), dtype=np.uint8) + assert detector().calculate_motion_score(image, image) == 0.0 + + +def test_different_images_score_is_higher() -> None: + first = np.zeros((10, 10, 3), dtype=np.uint8) + second = np.full((10, 10, 3), 255, dtype=np.uint8) + assert detector().calculate_motion_score(first, second) > 90.0 + + +def test_still_time_accumulates_below_threshold() -> None: + det = detector() + image = np.zeros((10, 10, 3), dtype=np.uint8) + det.update(image, 1.0) + status = det.update(image, 1.0) + assert status.state == DetectionState.STILL_WAITING + assert status.still_time == 1.0 + status = det.update(image, 1.0) + assert status.still_time == 2.0 + + +def test_still_time_resets_when_movement_returns() -> None: + det = detector() + still = np.zeros((10, 10, 3), dtype=np.uint8) + moved = np.full((10, 10, 3), 255, dtype=np.uint8) + det.update(still, 1.0) + det.update(still, 1.0) + status = det.update(moved, 1.0) + assert status.state == DetectionState.MOVING + assert status.still_time == 0.0 + + +def test_only_one_alert_during_one_still_period() -> None: + det = detector() + image = np.zeros((10, 10, 3), dtype=np.uint8) + det.update(image, 1.0) + assert not det.update(image, 1.0).should_alert + assert not det.update(image, 1.0).should_alert + assert det.update(image, 1.0).should_alert + assert not det.update(image, 1.0).should_alert + assert not det.update(image, 1.0).should_alert + assert det.state == DetectionState.ALERTED + + +def test_alert_rearms_after_movement_resumes() -> None: + det = detector() + still = np.zeros((10, 10, 3), dtype=np.uint8) + moved = np.full((10, 10, 3), 255, dtype=np.uint8) + det.update(still, 1.0) + det.update(still, 1.0) + det.update(still, 1.0) + assert det.update(still, 1.0).should_alert + assert det.update(moved, 1.0).state == DetectionState.MOVING + det.update(moved, 1.0) + det.update(moved, 1.0) + assert det.update(moved, 1.0).should_alert