rattatwinko/mucapy
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

alert for persons in camera view

Browse Source
This commit is contained in:
rattatwinko
2025-11-01 02:30:20 +01:00
parent d30a55fb0b
commit c227beeaca
4 changed files with 431 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
.gitea/workflows/run-mucapy.yml
View File

@@ -1,45 +0,0 @@
name: Build MuCaPy Executable
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
build-windows-exe:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v3
- name: Install dependencies for Wine
run: |
sudo dpkg --add-architecture i386
sudo apt update
sudo apt install -y wine64 wine32 unzip wget cabextract
- name: Install Windows Python under Wine
run: |
wget https://www.python.org/ftp/python/3.13.9/python-3.13.9-amd64.exe -O python_installer.exe
wine python_installer.exe /quiet InstallAllUsers=1 PrependPath=1
- name: Upgrade pip and install PyInstaller (Windows)
run: |
wine python -m pip install --upgrade pip
wine python -m pip install pyinstaller
- name: Build Windows executable
run: |
wine pyinstaller --onefile --windowed mucapy/main.py \
--add-data "mucapy/styling;styling" \
--add-data "mucapy/models;models" \
--add-data "mucapy/todopackage;todopackage"
- name: Upload Windows executable
uses: actions/upload-artifact@v3
with:
name: mucapy-windows-exe
path: dist/

46
mucapy/compile.py Normal file
View File

@@ -0,0 +1,46 @@
import os
from PIL import Image
import PyInstaller.__main__
import PyQt5
# Paths
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
MAIN_SCRIPT = os.path.join(ROOT_DIR, "main.py")
STYLING_DIR = os.path.join(ROOT_DIR, "styling")
# Icon paths
PNG_ICON = os.path.join(STYLING_DIR, "logo.png")
ICO_ICON = os.path.join(STYLING_DIR, "logo.ico")
# Convert PNG to ICO
img = Image.open(PNG_ICON)
img.save(ICO_ICON, format="ICO", sizes=[(256,256), (128,128), (64,64), (32,32), (16,16)])
print(f"Converted {PNG_ICON} to {ICO_ICON}")
# Detect PyQt5 platforms folder automatically
pyqt_dir = os.path.dirname(PyQt5.__file__)
platforms_path = None
# Walk recursively to find the 'platforms' folder
for root, dirs, files in os.walk(pyqt_dir):
if 'platforms' in dirs:
platforms_path = os.path.join(root, 'platforms')
break
if platforms_path is None or not os.path.exists(platforms_path):
raise FileNotFoundError(f"Could not locate PyQt5 'platforms' folder under {pyqt_dir}")
print(f"Using PyQt5 platforms folder: {platforms_path}")
# Build EXE with PyInstaller
PyInstaller.__main__.run([
MAIN_SCRIPT,
'--noconfirm',
'--onefile',
'--windowed',
f'--icon={ICO_ICON}',
# Only include the platforms folder (minimal requirement for PyQt5)
'--add-data', f'{platforms_path};PyQt5/Qt/plugins/platforms',
])
print("Build complete! Check the 'dist' folder for the executable.")

401
mucapy/main.py
View File

@@ -10,6 +10,7 @@ import sys
import time
import urllib.parse
import ctypes
import shutil
import cv2
import numpy as np
import psutil # Add psutil import
@@ -37,10 +38,18 @@ except ImportError as e:
"3) Install VC++ runtime: https://aka.ms/vs/17/release/vc_redist.x64.exe\n"
"4) Restart the app after reinstalling.\n\n"
f"Original error: {e}")
sys.exit(1)
import todopackage.todo as todo # This shit will fail eventually | Or not IDK
# Audio alert dependencies
import wave
try:
import simpleaudio as sa
except Exception:
sa = None
# Force-disable simpleaudio to avoid potential native backend crashes; use OS-native players only
sa = None
def bytes_to_human(n: int) -> str:
"""Convert a byte value to a human-readable string using base 1024.
@@ -117,7 +126,7 @@ class Config:
with open(self.config_file, 'w') as f:
json.dump(self.settings, f, indent=4)
except FileNotFoundError:
exit(1)
pass
except Exception as e:
print(f"Error saving config: {e}")
@@ -374,6 +383,172 @@ class CameraScanThread(QThread):
except Exception as e:
print(f"CameraScanThread error: {e}")
self.scan_finished.emit([], {})
class AlertWorker(QThread):
"""Worker thread to play an alert sound safely without blocking UI.
Uses winsound on Windows, external system players on Unix (afplay/paplay/aplay/ffplay),
and falls back to simpleaudio if available. Supports cooperative stop.
"""
finished = pyqtSignal(bool, str) # success, message
def __init__(self, wav_path: str, parent=None):
super().__init__(parent)
self.wav_path = wav_path
self._stop = False
self._subproc = None
self._play_obj = None
def stop(self):
"""Request the worker to stop early."""
try:
self._stop = True
if self._play_obj is not None:
try:
self._play_obj.stop()
except Exception:
pass
if self._subproc is not None:
try:
self._subproc.terminate()
except Exception:
pass
except Exception:
pass
def _find_unix_player(self):
"""Return (cmd_list, name) for an available player on Unix or (None, None)."""
try:
if sys.platform.startswith('darwin'):
if shutil.which('afplay'):
return (['afplay'], 'afplay')
# Linux and others
if shutil.which('paplay'):
return (['paplay'], 'paplay')
if shutil.which('aplay'):
return (['aplay', '-q'], 'aplay')
if shutil.which('ffplay'):
return (['ffplay', '-nodisp', '-autoexit', '-loglevel', 'error'], 'ffplay')
except Exception:
pass
return (None, None)
def run(self):
try:
if not os.path.exists(self.wav_path):
self.finished.emit(False, f"File not found: {self.wav_path}")
return
# Windows path: prefer winsound (native, safe)
if sys.platform.startswith('win'):
ws_error = "unknown"
try:
import winsound as _ws # type: ignore
# Resolve flags safely even if some attributes are missing
SND_FILENAME = getattr(_ws, 'SND_FILENAME', 0x00020000)
SND_SYNC = getattr(_ws, 'SND_SYNC', 0x0000) # 0 is synchronous by default
flags = SND_FILENAME | SND_SYNC
# Ensure PlaySound exists
play_fn = getattr(_ws, 'PlaySound', None)
if play_fn is None:
raise RuntimeError('winsound.PlaySound not available')
for _ in range(4):
if self._stop:
break
try:
play_fn(self.wav_path, flags)
except Exception as e:
# On failure, break to try alternative backends
ws_error = str(e)
break
time.sleep(0.002)
else:
# Completed all 4 plays
self.finished.emit(True, "Alert played")
return
# If here, winsound failed at some point; continue to fallbacks
except Exception as e:
ws_error = str(e)
# Try simpleaudio on Windows as fallback
if sa is not None:
try:
with wave.open(self.wav_path, 'rb') as wf:
n_channels = max(1, wf.getnchannels())
sampwidth = max(1, wf.getsampwidth())
framerate = max(8000, wf.getframerate() or 44100)
frames = wf.readframes(wf.getnframes())
for _ in range(4):
if self._stop:
break
self._play_obj = sa.play_buffer(frames, n_channels, sampwidth, framerate)
self._play_obj.wait_done()
time.sleep(0.002)
self.finished.emit(True, "Alert played")
return
except Exception as e2:
self.finished.emit(False, f"Playback error (winsound fallback -> simpleaudio): {e2}")
return
else:
self.finished.emit(False, f"Audio backend not available (winsound failed: {ws_error})")
return
# Non-Windows: try external players first
cmd, name = self._find_unix_player()
if cmd is not None:
for _ in range(4):
if self._stop:
break
try:
self._subproc = subprocess.Popen(cmd + [self.wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
# Poll until done or stop requested
while True:
if self._stop:
try:
self._subproc.terminate()
except Exception:
pass
break
ret = self._subproc.poll()
if ret is not None:
break
time.sleep(0.01)
except Exception as e:
# Try next backend
cmd = None
break
finally:
self._subproc = None
time.sleep(0.002)
if cmd is not None:
self.finished.emit(True, "Alert played")
return
# Fallback: simpleaudio if available
if sa is not None:
try:
with wave.open(self.wav_path, 'rb') as wf:
n_channels = max(1, wf.getnchannels())
sampwidth = max(1, wf.getsampwidth())
framerate = max(8000, wf.getframerate() or 44100)
frames = wf.readframes(wf.getnframes())
for _ in range(4):
if self._stop:
break
self._play_obj = sa.play_buffer(frames, n_channels, sampwidth, framerate)
self._play_obj.wait_done()
time.sleep(0.002)
self.finished.emit(True, "Alert played")
return
except Exception as e:
self.finished.emit(False, f"Playback error (simpleaudio): {e}")
return
self.finished.emit(False, "No audio backend available (afplay/paplay/aplay/ffplay/simpleaudio)")
except Exception as e:
try:
self.finished.emit(False, str(e))
except Exception:
pass
class MultiCamYOLODetector(QObject):
cameras_scanned = pyqtSignal(list, dict) # Emits (available_cameras, index_to_name)
def __init__(self, parent=None):
@@ -594,8 +769,8 @@ class MultiCamYOLODetector(QObject):
with open(classes_path, 'r') as f:
self.classes = f.read().strip().split('\n')
except FileNotFoundError:
exit(1)
pass
np.random.seed(42)
self.colors = np.random.randint(0, 255, size=(len(self.classes), 3), dtype='uint8')
return True
@@ -732,13 +907,26 @@ class MultiCamYOLODetector(QObject):
indices = cv2.dnn.NMSBoxes(boxes, confidences, self.confidence_threshold, 0.4)
person_detected = False
if len(indices) > 0:
for i in indices.flatten():
(x, y, w, h) = boxes[i]
color = [int(c) for c in self.colors[class_ids[i]]]
cv2.rectangle(frame, (x, y), (x + w, y + h), color, 2)
text = f"{self.classes[class_ids[i]]}: {confidences[i]:.2f}"
cls_name = self.classes[class_ids[i]] if 0 <= class_ids[i] < len(self.classes) else str(class_ids[i])
text = f"{cls_name}: {confidences[i]:.2f}"
cv2.putText(frame, text, (x, y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
if not person_detected and str(cls_name).lower() == 'person':
person_detected = True
# Auto-trigger alert if a person is detected on any camera and alerts are enabled
try:
if person_detected:
parent_window = self.parent()
if parent_window is not None:
# trigger_alert() has its own internal guards (enabled, cooldown, playing)
parent_window.trigger_alert()
except Exception:
pass
except Exception as e:
print(f"Detection error: {e}")
@@ -994,7 +1182,7 @@ class CameraDisplay(QLabel):
with open(self.get_camera_display_style,"r") as cdst:
self.setStyleSheet(cdst.read())
except FileNotFoundError:
exit(1)
pass
self.setMinimumSize(320, 240)
self.fullscreen_window = None
@@ -1230,7 +1418,7 @@ class AboutWindow(QDialog):
with open(toggle_btn_style,"r") as tgbstyle:
self.toggle_btn.setStyleSheet(tgbstyle.read())
except FileNotFoundError:
exit(1)
pass
# Debug shit
#print("i did shit")
@@ -1272,7 +1460,7 @@ class AboutWindow(QDialog):
with open(style_file,"r") as aboutstyle:
self.setStyleSheet(aboutstyle.read())
except FileNotFoundError:
exit(1)
pass
self.setLayout(layout)
@@ -1295,7 +1483,7 @@ class AboutWindow(QDialog):
pass
except FileNotFoundError:
print(f"Missing a Style File! => {todo_style_path}")
exit(1)
pass
# Create the labels for the fucking trodo ass shit ?
self.todo_archive_object = todo
@@ -1902,6 +2090,15 @@ class MainWindow(QMainWindow):
self.camera_settings = {}
self.detection_enabled = True # Add detection toggle flag
# Alert system state
self.alert_enabled = bool(self.config.load_setting('alert_enabled', True))
self._alert_playing = False
self._alert_cooldown = False
self._cooldown_timer = QTimer(self)
self._cooldown_timer.setSingleShot(True)
self._cooldown_timer.timeout.connect(self._on_cooldown_finished)
self._alert_worker = None
# Load saved settings first
self.load_saved_settings()
@@ -1918,8 +2115,8 @@ class MainWindow(QMainWindow):
with open(style_file, "r") as mainstyle:
self.setStyleSheet(mainstyle.read())
except FileNotFoundError:
exit(1)
pass
# Set palette for better dark mode support
palette = self.palette()
palette.setColor(palette.Window, QColor(45, 45, 45))
@@ -1977,6 +2174,12 @@ class MainWindow(QMainWindow):
if hasattr(self, 'model_label') and self.detector.model_dir:
self.model_label.setText(f"Model: {os.path.basename(self.detector.model_dir)}")
# Ensure alert UI reflects backend availability and state
try:
self._update_alert_ui()
except Exception:
pass
def create_menus(self):
menubar = self.menuBar()
@@ -2043,6 +2246,14 @@ class MainWindow(QMainWindow):
self.toggle_detection_action.setShortcut('Ctrl+D')
self.toggle_detection_action.triggered.connect(self.toggle_detection)
view_menu.addAction(self.toggle_detection_action)
# Alert toggle moved to View menu
self.toggle_alert_action = QAction('Enable Alert', self)
self.toggle_alert_action.setCheckable(True)
self.toggle_alert_action.setChecked(bool(self.alert_enabled))
self.toggle_alert_action.setStatusTip('Play an audible alert when a person is detected')
self.toggle_alert_action.triggered.connect(self.set_alert_enabled)
view_menu.addAction(self.toggle_alert_action)
# Camera menu
self.camera_menu = menubar.addMenu('Cameras')
@@ -2331,6 +2542,8 @@ class MainWindow(QMainWindow):
layout_layout.addWidget(self.layout_combo)
settings_layout.addLayout(layout_layout)
# Button enablement determined dynamically based on backend availability
# Add screenshot button to settings
screenshot_btn = QPushButton("Take Screenshot")
screenshot_btn.clicked.connect(self.take_screenshot)
@@ -2716,7 +2929,7 @@ class MainWindow(QMainWindow):
bar.setStyleSheet(u60_style.read())
except FileNotFoundError:
print("Styling for CPU U60 not found!")
exit(1)
pass
else:
u60seperate = getpath.resource_path("styling/bar/seperate/u60.qss")
try:
@@ -2735,7 +2948,7 @@ class MainWindow(QMainWindow):
bar.setStyleSheet(a85_styling.read())
except FileNotFoundError:
print("Styling for CPU u85 not found")
exit(1)
pass
else:
u85sep = getpath.resource_path("styling/bar/seperate/a85.qss")
try:
@@ -2754,7 +2967,7 @@ class MainWindow(QMainWindow):
bar.setStyleSheet(else_style.read())
except FileNotFoundError:
print("No ElseStyling found!")
exit(1)
pass
else:
else_file_seperate = getpath.resource_path("styling/bar/seperate/else.qss")
try:
@@ -2775,6 +2988,162 @@ class MainWindow(QMainWindow):
self.start_btn.setEnabled(False)
self.stop_btn.setEnabled(False)
# ===== Alert system methods =====
def set_alert_enabled(self, state: bool):
try:
self.alert_enabled = bool(state)
self.config.save_setting('alert_enabled', self.alert_enabled)
# Keep View menu action in sync
try:
if hasattr(self, 'toggle_alert_action'):
self.toggle_alert_action.setChecked(self.alert_enabled)
except Exception:
pass
except Exception:
pass
self._update_alert_ui()
def _has_audio_backend(self) -> bool:
"""Check if any audio backend is available for alerts."""
try:
if sys.platform.startswith('win'):
try:
import winsound as _ws # type: ignore
return hasattr(_ws, 'PlaySound')
except Exception:
return sa is not None
# macOS/Linux external players
if shutil.which('afplay') or shutil.which('paplay') or shutil.which('aplay') or shutil.which('ffplay'):
return True
return sa is not None
except Exception:
return sa is not None
def _update_alert_ui(self):
try:
backend_ok = self._has_audio_backend()
# Update status tip on the View->Enable Alert action
if hasattr(self, 'toggle_alert_action'):
tip = []
if not backend_ok:
tip.append("No audio backend found (winsound/afplay/paplay/aplay/ffplay/simpleaudio)")
if self._alert_cooldown:
tip.append("Alert cooldown active (30s)")
if self._alert_playing:
tip.append("Alert is playing")
if not self.alert_enabled:
tip.append("Alert disabled")
self.toggle_alert_action.setStatusTip("; ".join(tip) if tip else "Play an audible alert when a person is detected")
except Exception:
pass
def trigger_alert(self):
"""Trigger the alert sound if enabled and not in cooldown."""
try:
if not self.alert_enabled:
return
if self._alert_playing or self._alert_cooldown:
return
wav_path = getpath.resource_path("styling/sound/alert.wav")
if not os.path.exists(wav_path):
QMessageBox.warning(self, "Alert sound missing", f"Sound file not found:\n{wav_path}")
return
# Start worker (backend decided internally)
self._alert_worker = AlertWorker(wav_path, self)
self._alert_worker.finished.connect(self._on_alert_finished)
self._alert_playing = True
self._update_alert_ui()
self._alert_worker.start()
except Exception as e:
QMessageBox.critical(self, "Alert Error", f"Failed to trigger alert: {e}")
self._alert_playing = False
self._update_alert_ui()
def _on_alert_finished(self, success: bool, message: str):
# Clean state and start cooldown on success
try:
self._alert_playing = False
# Avoid holding a reference to finished thread
try:
if self._alert_worker:
self._alert_worker.deleteLater()
except Exception:
pass
self._alert_worker = None
if success:
self._alert_cooldown = True
# 30 seconds cooldown
self._cooldown_timer.start(30000)
else:
# Informative but non-fatal
try:
if message:
print(f"Alert playback failed: {message}")
except Exception:
pass
finally:
self._update_alert_ui()
def _on_cooldown_finished(self):
self._alert_cooldown = False
self._update_alert_ui()
def closeEvent(self, event):
"""Ensure background workers and timers are stopped to avoid crashes on exit."""
try:
# Stop periodic timers
try:
if hasattr(self, 'timer') and self.timer is not None:
self.timer.stop()
except Exception:
pass
try:
if hasattr(self, 'hw_timer') and self.hw_timer is not None:
self.hw_timer.stop()
except Exception:
pass
try:
if hasattr(self, '_cooldown_timer') and self._cooldown_timer is not None:
self._cooldown_timer.stop()
except Exception:
pass
# Stop alert worker if running
try:
if getattr(self, '_alert_worker', None) is not None:
try:
self._alert_worker.finished.disconnect(self._on_alert_finished)
except Exception:
pass
try:
if self._alert_worker.isRunning():
try:
self._alert_worker.stop()
except Exception:
pass
self._alert_worker.wait(2000)
except Exception:
pass
try:
self._alert_worker.deleteLater()
except Exception:
pass
self._alert_worker = None
except Exception:
pass
# Disconnect cameras and stop camera threads
try:
if hasattr(self, 'detector') and self.detector is not None:
self.detector.disconnect_cameras()
except Exception:
pass
finally:
try:
event.accept()
except Exception:
pass
class initQT:
"""
This is a QOL Change if you prefer to do it the hard way. Or you just like to get Fist Fucked then i suggest you remove the Function Calls in the
@@ -2800,7 +3169,7 @@ class initQT:
"echo $XDG_SESSION_TYPE"
"Run this command in bash!"
)
exit(1)
pass
def setenv(self):
# Set the Session Type to the one it got
@@ -2813,7 +3182,7 @@ class initQT:
f"export XDG_SESSION_TYPE={self.session_type}"
"run this command in bash"
)
exit(1)
pass
@staticmethod
def shutupCV():
# This needs some fixing as this only works before importing CV2 ; too much refactoring work tho!

BIN
mucapy/styling/sound/alert.wav Normal file
View File

Binary file not shown.