import sys import os import cv2 import numpy as np import threading import time import json import logging import telegram import face_recognition import pytesseract from datetime import datetime from PyQt5.QtWidgets import (QApplication, QMainWindow, QTabWidget, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QLineEdit, QComboBox, QTableWidget, QTableWidgetItem, QCheckBox, QFileDialog, QMessageBox, QGridLayout, QScrollArea, QFrame, QSplitter, QTextEdit, QGroupBox, QFormLayout, QSpinBox, QDateTimeEdit) from PyQt5.QtGui import QImage, QPixmap, QFont, QIcon from PyQt5.QtCore import Qt, QTimer, QThread, pyqtSignal, QDateTime, QSize # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler("surveillance.log"), logging.StreamHandler()]) logger = logging.getLogger(__name__) # Database simulation (in a real app, use SQLite or another DB) class Database: def __init__(self): self.cameras = [] self.persons = [] self.trucks = [] self.recordings = [] self.users = [{"username": "admin", "password": "admin", "role": "admin"}] self.settings = { "storage_path": "./recordings", "face_recognition_threshold": 0.6, "ocr_confidence": 70, "telegram_token": "", "telegram_chat_id": "" } # Create storage directory if it doesn't exist if not os.path.exists(self.settings["storage_path"]): os.makedirs(self.settings["storage_path"]) # Load data if exists self.load_data() def load_data(self): try: if os.path.exists("data/cameras.json"): with open("data/cameras.json", "r") as f: self.cameras = json.load(f) if os.path.exists("data/persons.json"): with open("data/persons.json", "r") as f: self.persons = json.load(f) if os.path.exists("data/trucks.json"): with open("data/trucks.json", "r") as f: self.trucks = json.load(f) if os.path.exists("data/recordings.json"): with open("data/recordings.json", "r") as f: self.recordings = json.load(f) if os.path.exists("data/users.json"): with open("data/users.json", "r") as f: self.users = json.load(f) if os.path.exists("data/settings.json"): with open("data/settings.json", "r") as f: self.settings = json.load(f) except Exception as e: logger.error(f"Error loading data: {e}") def save_data(self): try: # Create data directory if it doesn't exist if not os.path.exists("data"): os.makedirs("data") with open("data/cameras.json", "w") as f: json.dump(self.cameras, f) with open("data/persons.json", "w") as f: json.dump(self.persons, f) with open("data/trucks.json", "w") as f: json.dump(self.trucks, f) with open("data/recordings.json", "w") as f: json.dump(self.recordings, f) with open("data/users.json", "w") as f: json.dump(self.users, f) with open("data/settings.json", "w") as f: json.dump(self.settings, f) except Exception as e: logger.error(f"Error saving data: {e}") # Video capture thread class VideoThread(QThread): update_frame = pyqtSignal(np.ndarray, int) motion_detected = pyqtSignal(int) def __init__(self, camera_id, camera_url, parent=None): super().__init__(parent) self.camera_id = camera_id self.camera_url = camera_url self.running = False self.motion_detection = True self.face_recognition = False self.truck_recognition = False self.prev_frame = None self.motion_threshold = 1000 # Threshold for motion detection def run(self): self.running = True cap = cv2.VideoCapture(self.camera_url) if not cap.isOpened(): logger.error(f"Error opening camera {self.camera_id}: {self.camera_url}") return while self.running: ret, frame = cap.read() if not ret: logger.error(f"Error reading frame from camera {self.camera_id}") time.sleep(1) # Wait before trying again cap = cv2.VideoCapture(self.camera_url) # Try to reconnect continue # Motion detection if self.motion_detection and self.prev_frame is not None: # Convert frames to grayscale gray_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray_prev_frame = cv2.cvtColor(self.prev_frame, cv2.COLOR_BGR2GRAY) # Calculate absolute difference frame_diff = cv2.absdiff(gray_frame, gray_prev_frame) # Apply threshold _, thresh = cv2.threshold(frame_diff, 30, 255, cv2.THRESH_BINARY) # Calculate the number of changed pixels motion_pixels = np.sum(thresh) // 255 # If motion detected if motion_pixels > self.motion_threshold: self.motion_detected.emit(self.camera_id) self.prev_frame = frame.copy() self.update_frame.emit(frame, self.camera_id) # Control the frame rate time.sleep(0.03) # ~30 FPS cap.release() def stop(self): self.running = False self.wait() # Main application window class SurveillanceSystem(QMainWindow): def __init__(self): super().__init__() self.db = Database() self.video_threads = {} self.current_user = None self.init_ui() self.show_login() def init_ui(self): self.setWindowTitle("Smart Surveillance System") self.setGeometry(100, 100, 1200, 800) # Create tab widget self.tabs = QTabWidget() self.setCentralWidget(self.tabs) # Create tabs self.live_view_tab = QWidget() self.cameras_tab = QWidget() self.recordings_tab = QWidget() self.persons_tab = QWidget() self.trucks_tab = QWidget() self.telegram_tab = QWidget() self.reports_tab = QWidget() self.settings_tab = QWidget() self.users_tab = QWidget() # Add tabs to widget self.tabs.addTab(self.live_view_tab, "Live View") self.tabs.addTab(self.cameras_tab, "Cameras") self.tabs.addTab(self.recordings_tab, "Recordings") self.tabs.addTab(self.persons_tab, "Persons") self.tabs.addTab(self.trucks_tab, "Trucks") self.tabs.addTab(self.telegram_tab, "Telegram Bot") self.tabs.addTab(self.reports_tab, "Reports") self.tabs.addTab(self.settings_tab, "Settings") self.tabs.addTab(self.users_tab, "Users") # Initialize tab contents self.init_live_view_tab() self.init_cameras_tab() self.init_recordings_tab() self.init_persons_tab() self.init_trucks_tab() self.init_telegram_tab() self.init_reports_tab() self.init_settings_tab() self.init_users_tab() def show_login(self): login_dialog = QWidget(self) login_dialog.setWindowTitle("Login") login_dialog.setGeometry(400, 300, 300, 150) layout = QVBoxLayout() username_layout = QHBoxLayout() username_label = QLabel("Username:") self.username_input = QLineEdit() username_layout.addWidget(username_label) username_layout.addWidget(self.username_input) password_layout = QHBoxLayout() password_label = QLabel("Password:") self.password_input = QLineEdit() self.password_input.setEchoMode(QLineEdit.Password) password_layout.addWidget(password_label) password_layout.addWidget(self.password_input) login_button = QPushButton("Login") login_button.clicked.connect(self.login) layout.addLayout(username_layout) layout.addLayout(password_layout) layout.addWidget(login_button) login_dialog.setLayout(layout) login_dialog.show() self.login_dialog = login_dialog def login(self): username = self.username_input.text() password = self.password_input.text() for user in self.db.users: if user["username"] == username and user["password"] == password: self.current_user = user self.login_dialog.hide() self.apply_user_permissions() return QMessageBox.warning(self, "Login Failed", "Invalid username or password") def apply_user_permissions(self): # In a real app, implement proper RBAC here if self.current_user["role"] != "admin": # Disable admin-only tabs self.tabs.setTabEnabled(self.tabs.indexOf(self.settings_tab), False) self.tabs.setTabEnabled(self.tabs.indexOf(self.users_tab), False) def init_live_view_tab(self): layout = QVBoxLayout() # Controls controls_layout = QHBoxLayout() layout_selector = QComboBox() layout_selector.addItems(["Grid 2x2", "Grid 3x3", "Grid 4x4"]) layout_selector.currentIndexChanged.connect(self.change_grid_layout) start_all_button = QPushButton("Start All Cameras") start_all_button.clicked.connect(self.start_all_cameras) stop_all_button = QPushButton("Stop All Cameras") stop_all_button.clicked.connect(self.stop_all_cameras) controls_layout.addWidget(QLabel("Layout:")) controls_layout.addWidget(layout_selector) controls_layout.addWidget(start_all_button) controls_layout.addWidget(stop_all_button) controls_layout.addStretch() layout.addLayout(controls_layout) # Video grid self.video_grid = QGridLayout() self.video_grid.setSpacing(5) # Create placeholder for videos self.video_labels = {} self.current_grid_size = 2 # Default 2x2 grid for i in range(self.current_grid_size): for j in range(self.current_grid_size): index = i * self.current_grid_size + j label = QLabel() label.setAlignment(Qt.AlignCenter) label.setStyleSheet("background-color: black;") label.setMinimumSize(320, 240) self.video_grid.addWidget(label, i, j) self.video_labels[index] = label video_container = QWidget() video_container.setLayout(self.video_grid) scroll_area = QScrollArea() scroll_area.setWidgetResizable(True) scroll_area.setWidget(video_container) layout.addWidget(scroll_area) self.live_view_tab.setLayout(layout) def change_grid_layout(self, index): # Clear current grid for i in reversed(range(self.video_grid.count())): self.video_grid.itemAt(i).widget().setParent(None) # Determine new grid size if index == 0: self.current_grid_size = 2 # 2x2 elif index == 1: self.current_grid_size = 3 # 3x3 else: self.current_grid_size = 4 # 4x4 # Create new grid self.video_labels = {} for i in range(self.current_grid_size): for j in range(self.current_grid_size): index = i * self.current_grid_size + j label = QLabel() label.setAlignment(Qt.AlignCenter) label.setStyleSheet("background-color: black;") label.setMinimumSize(320, 240) self.video_grid.addWidget(label, i, j) self.video_labels[index] = label # Restart cameras if needed self.start_cameras_for_grid() def start_cameras_for_grid(self): # Stop all current threads self.stop_all_cameras() # Start cameras for current grid max_cameras = self.current_grid_size * self.current_grid_size cameras_to_show = self.db.cameras[:max_cameras] for i, camera in enumerate(cameras_to_show): if i < len(self.video_labels): self.start_camera(camera, i) def start_all_cameras(self): self.start_cameras_for_grid() def stop_all_cameras(self): for thread in self.video_threads.values(): thread.stop() self.video_threads.clear() # Reset video labels for label in self.video_labels.values(): label.clear() label.setText("Camera Off") def start_camera(self, camera, position): camera_id = camera["id"] camera_url = camera["url"] # Create and start video thread thread = VideoThread(camera_id, camera_url) thread.update_frame.connect(lambda frame, cam_id: self.update_video_frame(frame, cam_id, position)) thread.motion_detected.connect(self.handle_motion_detection) # Set recognition flags based on camera settings thread.face_recognition = camera.get("face_recognition", False) thread.truck_recognition = camera.get("truck_recognition", False) thread.start() self.video_threads[camera_id] = thread def update_video_frame(self, frame, camera_id, position): if position not in self.video_labels: return # Convert frame to QImage rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) h, w, ch = rgb_frame.shape bytes_per_line = ch * w q_img = QImage(rgb_frame.data, w, h, bytes_per_line, QImage.Format_RGB888) # Scale image to fit label while maintaining aspect ratio pixmap = QPixmap.fromImage(q_img) label = self.video_labels[position] scaled_pixmap = pixmap.scaled(label.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation) # Update label label.setPixmap(scaled_pixmap) def handle_motion_detection(self, camera_id): # Find camera in database camera = next((c for c in self.db.cameras if c["id"] == camera_id), None) if not camera: return logger.info(f"Motion detected on camera {camera['name']} (ID: {camera_id})") # If recording on motion is enabled, start recording if camera.get("record_on_motion", False): self.start_recording(camera_id) # If face or truck recognition is enabled, process the frame thread = self.video_threads.get(camera_id) if thread: if thread.face_recognition: # In a real app, this would process the current frame for face recognition pass if thread.truck_recognition: # In a real app, this would process the current frame for truck OCR pass def start_recording(self, camera_id): # In a real app, this would start recording the video to disk timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") recording_path = f"{self.db.settings['storage_path']}/camera_{camera_id}_{timestamp}.mp4" # Add recording to database recording = { "id": len(self.db.recordings) + 1, "camera_id": camera_id, "path": recording_path, "start_time": datetime.now().isoformat(), "end_time": None, "metadata": { "motion_detected": True, "faces_detected": [], "trucks_detected": [] } } self.db.recordings.append(recording) self.db.save_data() logger.info(f"Started recording for camera {camera_id}: {recording_path}") def init_cameras_tab(self): layout = QVBoxLayout() # Camera list self.camera_table = QTableWidget() self.camera_table.setColumnCount(7) self.camera_table.setHorizontalHeaderLabels(["ID", "Name", "URL", "Face Recognition", "Truck Recognition", "Recording Mode", "Actions"]) self.camera_table.horizontalHeader().setStretchLastSection(True) # Add camera form form_layout = QFormLayout() self.camera_name_input = QLineEdit() self.camera_url_input = QLineEdit() self.camera_face_recog_checkbox = QCheckBox() self.camera_truck_recog_checkbox = QCheckBox() self.camera_recording_mode = QComboBox() self.camera_recording_mode.addItems(["Continuous", "On Motion"]) add_camera_button = QPushButton("Add Camera") add_camera_button.clicked.connect(self.add_camera) form_layout.addRow("Camera Name:", self.camera_name_input) form_layout.addRow("Camera URL:", self.camera_url_input) form_layout.addRow("Enable Face Recognition:", self.camera_face_recog_checkbox) form_layout.addRow("Enable Truck Recognition:", self.camera_truck_recog_checkbox) form_layout.addRow("Recording Mode:", self.camera_recording_mode) form_layout.addRow("", add_camera_button) form_group = QGroupBox("Add New Camera") form_group.setLayout(form_layout) layout.addWidget(form_group) layout.addWidget(self.camera_table) self.cameras_tab.setLayout(layout) # Load cameras self.load_cameras() def load_cameras(self): self.camera_table.setRowCount(0) for camera in self.db.cameras: row = self.camera_table.rowCount() self.camera_table.insertRow(row) self.camera_table.setItem(row, 0, QTableWidgetItem(str(camera["id"]))) self.camera_table.setItem(row, 1, QTableWidgetItem(camera["name"])) self.camera_table.setItem(row, 2, QTableWidgetItem(camera["url"])) face_recog = QTableWidgetItem() face_recog.setCheckState(Qt.Checked if camera.get("face_recognition", False) else Qt.Unchecked) self.camera_table.setItem(row, 3, face_recog) truck_recog = QTableWidgetItem() truck_recog.setCheckState(Qt.Checked if camera.get("truck_recognition", False) else Qt.Unchecked) self.camera_table.setItem(row, 4, truck_recog) recording_mode = "Continuous" if camera.get("continuous_recording", True) else "On Motion" self.camera_table.setItem(row, 5, QTableWidgetItem(recording_mode)) # Actions actions_layout = QHBoxLayout() actions_layout.setContentsMargins(0, 0, 0, 0) edit_button = QPushButton("Edit") edit_button.setProperty("camera_id", camera["id"]) edit_button.clicked.connect(lambda _, cid=camera["id"]: self.edit_camera(cid)) delete_button = QPushButton("Delete") delete_button.setProperty("camera_id", camera["id"]) delete_button.clicked.connect(lambda _, cid=camera["id"]: self.delete_camera(cid)) actions_layout.addWidget(edit_button) actions_layout.addWidget(delete_button) actions_widget = QWidget() actions_widget.setLayout(actions_layout) self.camera_table.setCellWidget(row, 6, actions_widget) def add_camera(self): name = self.camera_name_input.text() url = self.camera_url_input.text() face_recognition = self.camera_face_recog_checkbox.isChecked() truck_recognition = self.camera_truck_recog_checkbox.isChecked() continuous_recording = self.camera_recording_mode.currentText() == "Continuous" if not name or not url: QMessageBox.warning(self, "Input Error", "Camera name and URL are required") return # Create new camera camera = { "id": len(self.db.cameras) + 1, "name": name, "url": url, "face_recognition": face_recognition, "truck_recognition": truck_recognition, "continuous_recording": continuous_recording, "record_on_motion": not continuous_recording } self.db.cameras.append(camera) self.db.save_data() # Clear form self.camera_name_input.clear() self.camera_url_input.clear() self.camera_face_recog_checkbox.setChecked(False) self.camera_truck_recog_checkbox.setChecked(False) self.camera_recording_mode.setCurrentIndex(0) # Reload camera list self.load_cameras() QMessageBox.information(self, "Success", f"Camera '{name}' added successfully") def edit_camera(self, camera_id): # Find camera in database camera = next((c for c in self.db.cameras if c["id"] == camera_id), None) if not camera: return # Create edit dialog dialog = QWidget(self, Qt.Window) dialog.setWindowTitle(f"Edit Camera: {camera['name']}") dialog.setGeometry(300, 300, 400, 250) layout = QFormLayout() name_input = QLineEdit(camera["name"]) url_input = QLineEdit(camera["url"]) face_recog_checkbox = QCheckBox() face_recog_checkbox.setChecked(camera.get("face_recognition", False)) truck_recog_checkbox = QCheckBox() truck_recog_checkbox.setChecked(camera.get("truck_recognition", False)) recording_mode = QComboBox() recording_mode.addItems(["Continuous", "On Motion"]) recording_mode.setCurrentIndex(0 if camera.get("continuous_recording", True) else 1) save_button = QPushButton("Save Changes") save_button.clicked.connect(lambda: self.save_camera_changes( camera_id, name_input.text(), url_input.text(), face_recog_checkbox.isChecked(), truck_recog_checkbox.isChecked(), recording_mode.currentText() == "Continuous", dialog )) layout.addRow("Camera Name:", name_input) layout.addRow("Camera URL:", url_input) layout.addRow("Enable Face Recognition:", face_recog_checkbox) layout.addRow("Enable Truck Recognition:", truck_recog_checkbox) layout.addRow("Recording Mode:", recording_mode) layout.addRow("", save_button) dialog.setLayout(layout) dialog.show() def save_camera_changes(self, camera_id, name, url, face_recognition, truck_recognition, continuous_recording, dialog): if not name or not url: QMessageBox.warning(dialog, "Input Error", "Camera name and URL are required") return # Find camera in database for camera in self.db.cameras: if camera["id"] == camera_id: camera["name"] = name camera["url"] = url camera["face_recognition"] = face_recognition camera["truck_recognition"] = truck_recognition camera["continuous_recording"] = continuous_recording camera["record_on_motion"] = not continuous_recording break self.db.save_data() self.load_cameras() dialog.close() QMessageBox.information(self, "Success", f"Camera '{name}' updated successfully") def delete_camera(self, camera_id): reply = QMessageBox.question(self, "Confirm Deletion", "Are you sure you want to delete this camera?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: # Remove camera from database self.db.cameras = [c for c in self.db.cameras if c["id"] != camera_id] self.db.save_data() # Stop camera thread if running if camera_id in self.video_threads: self.video_threads[camera_id].stop() del self.video_threads[camera_id] # Reload camera list self.load_cameras() QMessageBox.information(self, "Success", "Camera deleted successfully") def init_recordings_tab(self): layout = QVBoxLayout() # Filters filters_layout = QHBoxLayout() date_from_label = QLabel("From:") self.date_from = QDateTimeEdit(QDateTime.currentDateTime().addDays(-7)) self.date_from.setCalendarPopup(True) date_to_label = QLabel("To:") self.date_to = QDateTimeEdit(QDateTime.currentDateTime()) self.date_to.setCalendarPopup(True) camera_filter_label = QLabel("Camera:") self.camera_filter = QComboBox() self.camera_filter.addItem("All Cameras", -1) content_filter_label = QLabel("Content:") self.content_filter = QComboBox() self.content_filter.addItems(["All", "Faces Detected", "Trucks Detected", "Motion Only"]) apply_filter_button = QPushButton("Apply Filters") apply_filter_button.clicked.connect(self.load_recordings) filters_layout.addWidget(date_from_label) filters_layout.addWidget(self.date_from) filters_layout.addWidget(date_to_label) filters_layout.addWidget(self.date_to) filters_layout.addWidget(camera_filter_label) filters_layout.addWidget(self.camera_filter) filters_layout.addWidget(content_filter_label) filters_layout.addWidget(self.content_filter) filters_layout.addWidget(apply_filter_button) # Recordings table self.recordings_table = QTableWidget() self.recordings_table.setColumnCount(6) self.recordings_table.setHorizontalHeaderLabels(["ID", "Camera", "Start Time", "Duration", "Content", "Actions"]) self.recordings_table.horizontalHeader().setStretchLastSection(True) layout.addLayout(filters_layout) layout.addWidget(self.recordings_table) self.recordings_tab.setLayout(layout) # Load camera filter options for camera in self.db.cameras: self.camera_filter.addItem(camera["name"], camera["id"]) # Load recordings self.load_recordings() def load_recordings(self): self.recordings_table.setRowCount(0) # Get filter values date_from = self.date_from.dateTime().toPyDateTime() date_to = self.date_to.dateTime().toPyDateTime() camera_id = self.camera_filter.currentData() content_filter = self.content_filter.currentText() for recording in self.db.recordings: # Apply filters start_time = datetime.fromisoformat(recording["start_time"]) if start_time < date_from or start_time > date_to: continue if camera_id != -1 and recording["camera_id"] != camera_id: continue metadata = recording.get("metadata", {}) if content_filter == "Faces Detected" and not metadata.get("faces_detected"): continue elif content_filter == "Trucks Detected" and not metadata.get("trucks_detected"): continue elif content_filter == "Motion Only" and not metadata.get("motion_detected"): continue # Add to table row = self.recordings_table.rowCount() self.recordings_table.insertRow(row) self.recordings_table.setItem(row, 0, QTableWidgetItem(str(recording["id"]))) # Get camera name camera_name = "Unknown" for camera in self.db.cameras: if camera["id"] == recording["camera_id"]: camera_name = camera["name"] break self.recordings_table.setItem(row, 1, QTableWidgetItem(camera_name)) # Format start time formatted_time = start_time.strftime("%Y-%m-%d %H:%M:%S") self.recordings_table.setItem(row, 2, QTableWidgetItem(formatted_time)) # Calculate duration duration = "Ongoing" if recording["end_time"]: end_time = datetime.fromisoformat(recording["end_time"]) duration_seconds = (end_time - start_time).total_seconds() minutes, seconds = divmod(duration_seconds, 60) duration = f"{int(minutes)}m {int(seconds)}s" self.recordings_table.setItem(row, 3, QTableWidgetItem(duration)) # Content summary content = [] if metadata.get("motion_detected"): content.append("Motion") if metadata.get("faces_detected"): content.append(f"{len(metadata['faces_detected'])} Faces") if metadata.get("trucks_detected"): content.append(f"{len(metadata['trucks_detected'])} Trucks") content_text = ", ".join(content) if content else "No events" self.recordings_table.setItem(row, 4, QTableWidgetItem(content_text)) # Actions actions_layout = QHBoxLayout() actions_layout.setContentsMargins(0, 0, 0, 0) view_button = QPushButton("View") view_button.setProperty("recording_id", recording["id"]) view_button.clicked.connect(lambda _, rid=recording["id"]: self.view_recording(rid)) export_button = QPushButton("Export") export_button.setProperty("recording_id", recording["id"]) export_button.clicked.connect(lambda _, rid=recording["id"]: self.export_recording(rid)) delete_button = QPushButton("Delete") delete_button.setProperty("recording_id", recording["id"]) delete_button.clicked.connect(lambda _, rid=recording["id"]: self.delete_recording(rid)) actions_layout.addWidget(view_button) actions_layout.addWidget(export_button) actions_layout.addWidget(delete_button) actions_widget = QWidget() actions_widget.setLayout(actions_layout) self.recordings_table.setCellWidget(row, 5, actions_widget) def view_recording(self, recording_id): # Find recording in database recording = next((r for r in self.db.recordings if r["id"] == recording_id), None) if not recording: return # In a real app, this would open a video player with the recording QMessageBox.information(self, "View Recording", f"Viewing recording {recording_id} from {recording['path']}") def export_recording(self, recording_id): # Find recording in database recording = next((r for r in self.db.recordings if r["id"] == recording_id), None) if not recording: return # In a real app, this would export the recording to a file file_path, _ = QFileDialog.getSaveFileName(self, "Export Recording", f"recording_{recording_id}.mp4", "Video Files (*.mp4)") if file_path: QMessageBox.information(self, "Export Recording", f"Recording exported to {file_path}") def delete_recording(self, recording_id): reply = QMessageBox.question(self, "Confirm Deletion", "Are you sure you want to delete this recording?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: # Remove recording from database self.db.recordings = [r for r in self.db.recordings if r["id"] != recording_id] self.db.save_data() # Reload recordings self.load_recordings() QMessageBox.information(self, "Success", "Recording deleted successfully") def init_persons_tab(self): layout = QVBoxLayout() # Person list self.persons_table = QTableWidget() self.persons_table.setColumnCount(4) self.persons_table.setHorizontalHeaderLabels(["ID", "Name", "Photo", "Actions"]) self.persons_table.horizontalHeader().setStretchLastSection(True) # Add person form form_layout = QFormLayout() self.person_name_input = QLineEdit() self.person_photo_path = QLineEdit() self.person_photo_path.setReadOnly(True) browse_button = QPushButton("Browse...") browse_button.clicked.connect(self.browse_person_photo) photo_layout = QHBoxLayout() photo_layout.addWidget(self.person_photo_path) photo_layout.addWidget(browse_button) add_person_button = QPushButton("Add Person") add_person_button.clicked.connect(self.add_person) form_layout.addRow("Person Name:", self.person_name_input) form_layout.addRow("Photo:", photo_layout) form_layout.addRow("", add_person_button) form_group = QGroupBox("Add New Person") form_group.setLayout(form_layout) layout.addWidget(form_group) layout.addWidget(self.persons_table) self.persons_tab.setLayout(layout) # Load persons self.load_persons() def browse_person_photo(self): file_path, _ = QFileDialog.getOpenFileName(self, "Select Person Photo", "", "Image Files (*.png *.jpg *.jpeg)") if file_path: self.person_photo_path.setText(file_path) def add_person(self): name = self.person_name_input.text() photo_path = self.person_photo_path.text() if not name: QMessageBox.warning(self, "Input Error", "Person name is required") return if not photo_path: QMessageBox.warning(self, "Input Error", "Person photo is required") return try: # In a real app, this would extract face encodings using face_recognition # For this demo, we'll just copy the image to a local directory # Create persons directory if it doesn't exist persons_dir = "data/persons" if not os.path.exists(persons_dir): os.makedirs(persons_dir) # Copy image to persons directory image_filename = f"person_{len(self.db.persons) + 1}_{name.replace(' ', '_')}.jpg" image_path = f"{persons_dir}/{image_filename}" # In a real app, we would resize and process the image # For this demo, we'll just simulate it # shutil.copy(photo_path, image_path) # Create new person person = { "id": len(self.db.persons) + 1, "name": name, "photo_path": image_path, "face_encoding": "simulated_encoding" # In a real app, this would be the actual encoding } self.db.persons.append(person) self.db.save_data() # Clear form self.person_name_input.clear() self.person_photo_path.clear() # Reload person list self.load_persons() QMessageBox.information(self, "Success", f"Person '{name}' added successfully") except Exception as e: QMessageBox.critical(self, "Error", f"Failed to add person: {str(e)}") def load_persons(self): self.persons_table.setRowCount(0) for person in self.db.persons: row = self.persons_table.rowCount() self.persons_table.insertRow(row) self.persons_table.setItem(row, 0, QTableWidgetItem(str(person["id"]))) self.persons_table.setItem(row, 1, QTableWidgetItem(person["name"])) # Photo thumbnail (in a real app, this would display the actual image) photo_label = QLabel("Photo") photo_label.setAlignment(Qt.AlignCenter) self.persons_table.setCellWidget(row, 2, photo_label) # Actions actions_layout = QHBoxLayout() actions_layout.setContentsMargins(0, 0, 0, 0) edit_button = QPushButton("Edit") edit_button.setProperty("person_id", person["id"]) edit_button.clicked.connect(lambda _, pid=person["id"]: self.edit_person(pid)) delete_button = QPushButton("Delete") delete_button.setProperty("person_id", person["id"]) delete_button.clicked.connect(lambda _, pid=person["id"]: self.delete_person(pid)) actions_layout.addWidget(edit_button) actions_layout.addWidget(delete_button) actions_widget = QWidget() actions_widget.setLayout(actions_layout) self.persons_table.setCellWidget(row, 3, actions_widget) def edit_person(self, person_id): # Find person in database person = next((p for p in self.db.persons if p["id"] == person_id), None) if not person: return # Create edit dialog dialog = QWidget(self, Qt.Window) dialog.setWindowTitle(f"Edit Person: {person['name']}") dialog.setGeometry(300, 300, 400, 200) layout = QFormLayout() name_input = QLineEdit(person["name"]) photo_path = QLineEdit(person["photo_path"]) photo_path.setReadOnly(True) browse_button = QPushButton("Browse...") browse_button.clicked.connect(lambda: self.browse_photo_for_edit(photo_path)) photo_layout = QHBoxLayout() photo_layout.addWidget(photo_path) photo_layout.addWidget(browse_button) save_button = QPushButton("Save Changes") save_button.clicked.connect(lambda: self.save_person_changes( person_id, name_input.text(), photo_path.text(), dialog )) layout.addRow("Person Name:", name_input) layout.addRow("Photo:", photo_layout) layout.addRow("", save_button) dialog.setLayout(layout) dialog.show() def browse_photo_for_edit(self, photo_path_input): file_path, _ = QFileDialog.getOpenFileName(self, "Select Person Photo", "", "Image Files (*.png *.jpg *.jpeg)") if file_path: photo_path_input.setText(file_path) def save_person_changes(self, person_id, name, photo_path, dialog): if not name: QMessageBox.warning(dialog, "Input Error", "Person name is required") return # Find person in database for person in self.db.persons: if person["id"] == person_id: person["name"] = name # If photo path changed, update it if photo_path != person["photo_path"]: # In a real app, this would extract face encodings and update the image person["photo_path"] = photo_path person["face_encoding"] = "updated_simulated_encoding" break self.db.save_data() self.load_persons() dialog.close() QMessageBox.information(self, "Success", f"Person '{name}' updated successfully") def delete_person(self, person_id): reply = QMessageBox.question(self, "Confirm Deletion", "Are you sure you want to delete this person?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: # Remove person from database self.db.persons = [p for p in self.db.persons if p["id"] != person_id] self.db.save_data() # Reload person list self.load_persons() QMessageBox.information(self, "Success", "Person deleted successfully") def init_trucks_tab(self): layout = QVBoxLayout() # Truck list self.trucks_table = QTableWidget() self.trucks_table.setColumnCount(5) self.trucks_table.setHorizontalHeaderLabels(["ID", "License Plate", "Driver", "Type", "Actions"]) self.trucks_table.horizontalHeader().setStretchLastSection(True) # Add truck form form_layout = QFormLayout() self.truck_license_input = QLineEdit() self.truck_driver_input = QLineEdit() self.truck_type_input = QComboBox() self.truck_type_input.addItems(["Delivery", "Transport", "Construction", "Other"]) add_truck_button = QPushButton("Add Truck") add_truck_button.clicked.connect(self.add_truck) form_layout.addRow("License Plate:", self.truck_license_input) form_layout.addRow("Driver Name:", self.truck_driver_input) form_layout.addRow("Truck Type:", self.truck_type_input) form_layout.addRow("", add_truck_button) form_group = QGroupBox("Add New Truck") form_group.setLayout(form_layout) layout.addWidget(form_group) layout.addWidget(self.trucks_table) self.trucks_tab.setLayout(layout) # Load trucks self.load_trucks() def add_truck(self): license_plate = self.truck_license_input.text() driver = self.truck_driver_input.text() truck_type = self.truck_type_input.currentText() if not license_plate: QMessageBox.warning(self, "Input Error", "License plate is required") return # Create new truck truck = { "id": len(self.db.trucks) + 1, "license_plate": license_plate, "driver": driver, "type": truck_type } self.db.trucks.append(truck) self.db.save_data() # Clear form self.truck_license_input.clear() self.truck_driver_input.clear() self.truck_type_input.setCurrentIndex(0) # Reload truck list self.load_trucks() QMessageBox.information(self, "Success", f"Truck '{license_plate}' added successfully") def load_trucks(self): self.trucks_table.setRowCount(0) for truck in self.db.trucks: row = self.trucks_table.rowCount() self.trucks_table.insertRow(row) self.trucks_table.setItem(row, 0, QTableWidgetItem(str(truck["id"]))) self.trucks_table.setItem(row, 1, QTableWidgetItem(truck["license_plate"])) self.trucks_table.setItem(row, 2, QTableWidgetItem(truck["driver"])) self.trucks_table.setItem(row, 3, QTableWidgetItem(truck["type"])) # Actions actions_layout = QHBoxLayout() actions_layout.setContentsMargins(0, 0, 0, 0) edit_button = QPushButton("Edit") edit_button.setProperty("truck_id", truck["id"]) edit_button.clicked.connect(lambda _, tid=truck["id"]: self.edit_truck(tid)) delete_button = QPushButton("Delete") delete_button.setProperty("truck_id", truck["id"]) delete_button.clicked.connect(lambda _, tid=truck["id"]: self.delete_truck(tid)) actions_layout.addWidget(edit_button) actions_layout.addWidget(delete_button) actions_widget = QWidget() actions_widget.setLayout(actions_layout) self.trucks_table.setCellWidget(row, 4, actions_widget) def edit_truck(self, truck_id): # Find truck in database truck = next((t for t in self.db.trucks if t["id"] == truck_id), None) if not truck: return # Create edit dialog dialog = QWidget(self, Qt.Window) dialog.setWindowTitle(f"Edit Truck: {truck['license_plate']}") dialog.setGeometry(300, 300, 400, 200) layout = QFormLayout() license_input = QLineEdit(truck["license_plate"]) driver_input = QLineEdit(truck["driver"]) type_input = QComboBox() type_input.addItems(["Delivery", "Transport", "Construction", "Other"]) type_input.setCurrentText(truck["type"]) save_button = QPushButton("Save Changes") save_button.clicked.connect(lambda: self.save_truck_changes( truck_id, license_input.text(), driver_input.text(), type_input.currentText(), dialog )) layout.addRow("License Plate:", license_input) layout.addRow("Driver Name:", driver_input) layout.addRow("Truck Type:", type_input) layout.addRow("", save_button) dialog.setLayout(layout) dialog.show() def save_truck_changes(self, truck_id, license_plate, driver, truck_type, dialog): if not license_plate: QMessageBox.warning(dialog, "Input Error", "License plate is required") return # Find truck in database for truck in self.db.trucks: if truck["id"] == truck_id: truck["license_plate"] = license_plate truck["driver"] = driver truck["type"] = truck_type break self.db.save_data() self.load_trucks() dialog.close() QMessageBox.information(self, "Success", f"Truck '{license_plate}' updated successfully") def delete_truck(self, truck_id): reply = QMessageBox.question(self, "Confirm Deletion", "Are you sure you want to delete this truck?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: # Remove truck from database self.db.trucks = [t for t in self.db.trucks if t["id"] != truck_id] self.db.save_data() # Reload truck list self.load_trucks() QMessageBox.information(self, "Success", "Truck deleted successfully") def init_telegram_tab(self): layout = QVBoxLayout() # Telegram bot settings settings_layout = QFormLayout() self.telegram_token_input = QLineEdit(self.db.settings.get("telegram_token", "")) self.telegram_chat_id_input = QLineEdit(self.db.settings.get("telegram_chat_id", "")) test_button = QPushButton("Test Connection") test_button.clicked.connect(self.test_telegram_connection) save_button = QPushButton("Save Settings") save_button.clicked.connect(self.save_telegram_settings) settings_layout.addRow("Bot Token:", self.telegram_token_input) settings_layout.addRow("Chat ID:", self.telegram_chat_id_input) buttons_layout = QHBoxLayout() buttons_layout.addWidget(test_button) buttons_layout.addWidget(save_button) settings_layout.addRow("", buttons_layout) settings_group = QGroupBox("Telegram Bot Settings") settings_group.setLayout(settings_layout) # Notification settings notification_layout = QVBoxLayout() self.notify_face_checkbox = QCheckBox("Send notification when unknown face is detected") self.notify_face_checkbox.setChecked(True) self.notify_truck_checkbox = QCheckBox("Send notification when unknown truck is detected") self.notify_truck_checkbox.setChecked(True) self.notify_motion_checkbox = QCheckBox("Send notification on motion detection") self.notify_motion_checkbox.setChecked(False) notification_layout.addWidget(self.notify_face_checkbox) notification_layout.addWidget(self.notify_truck_checkbox) notification_layout.addWidget(self.notify_motion_checkbox) notification_group = QGroupBox("Notification Settings") notification_group.setLayout(notification_layout) # Log log_layout = QVBoxLayout() self.telegram_log = QTextEdit() self.telegram_log.setReadOnly(True) clear_log_button = QPushButton("Clear Log") clear_log_button.clicked.connect(self.clear_telegram_log) log_layout.addWidget(self.telegram_log) log_layout.addWidget(clear_log_button) log_group = QGroupBox("Telegram Bot Log") log_group.setLayout(log_layout) layout.addWidget(settings_group) layout.addWidget(notification_group) layout.addWidget(log_group) self.telegram_tab.setLayout(layout) def test_telegram_connection(self): token = self.telegram_token_input.text() chat_id = self.telegram_chat_id_input.text() if not token or not chat_id: QMessageBox.warning(self, "Input Error", "Bot token and chat ID are required") return try: # In a real app, this would test the connection to Telegram # For this demo, we'll just simulate it self.telegram_log.append(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Testing connection...") # Simulate delay QApplication.processEvents() time.sleep(1) self.telegram_log.append(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Connection successful!") QMessageBox.information(self, "Success", "Telegram bot connection successful") except Exception as e: self.telegram_log.append(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Connection failed: {str(e)}") QMessageBox.critical(self, "Error", f"Failed to connect to Telegram: {str(e)}") def save_telegram_settings(self): token = self.telegram_token_input.text() chat_id = self.telegram_chat_id_input.text() if not token or not chat_id: QMessageBox.warning(self, "Input Error", "Bot token and chat ID are required") return # Save settings self.db.settings["telegram_token"] = token self.db.settings["telegram_chat_id"] = chat_id self.db.settings["notify_face"] = self.notify_face_checkbox.isChecked() self.db.settings["notify_truck"] = self.notify_truck_checkbox.isChecked() self.db.settings["notify_motion"] = self.notify_motion_checkbox.isChecked() self.db.save_data() self.telegram_log.append(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Settings saved") QMessageBox.information(self, "Success", "Telegram settings saved successfully") def clear_telegram_log(self): self.telegram_log.clear() def init_reports_tab(self): layout = QVBoxLayout() # Report filters filters_layout = QFormLayout() self.report_date_from = QDateTimeEdit(QDateTime.currentDateTime().addDays(-7)) self.report_date_from.setCalendarPopup(True) self.report_date_to = QDateTimeEdit(QDateTime.currentDateTime()) self.report_date_to.setCalendarPopup(True) self.report_camera_filter = QComboBox() self.report_camera_filter.addItem("All Cameras", -1) for camera in self.db.cameras: self.report_camera_filter.addItem(camera["name"], camera["id"]) self.report_content_filter = QComboBox() self.report_content_filter.addItems(["All Events", "Faces Only", "Known Faces", "Unknown Faces", "Trucks Only", "Known Trucks", "Unknown Trucks"]) generate_button = QPushButton("Generate Report") generate_button.clicked.connect(self.generate_report) export_pdf_button = QPushButton("Export to PDF") export_pdf_button.clicked.connect(self.export_report_to_pdf) send_telegram_button = QPushButton("Send via Telegram") send_telegram_button.clicked.connect(self.send_report_via_telegram) filters_layout.addRow("From:", self.report_date_from) filters_layout.addRow("To:", self.report_date_to) filters_layout.addRow("Camera:", self.report_camera_filter) filters_layout.addRow("Content:", self.report_content_filter) buttons_layout = QHBoxLayout() buttons_layout.addWidget(generate_button) buttons_layout.addWidget(export_pdf_button) buttons_layout.addWidget(send_telegram_button) filters_layout.addRow("", buttons_layout) filters_group = QGroupBox("Report Filters") filters_group.setLayout(filters_layout) # Report preview self.report_preview = QTextEdit() self.report_preview.setReadOnly(True) preview_group = QGroupBox("Report Preview") preview_layout = QVBoxLayout() preview_layout.addWidget(self.report_preview) preview_group.setLayout(preview_layout) layout.addWidget(filters_group) layout.addWidget(preview_group) self.reports_tab.setLayout(layout) def generate_report(self): date_from = self.report_date_from.dateTime().toPyDateTime() date_to = self.report_date_to.dateTime().toPyDateTime() camera_id = self.report_camera_filter.currentData() content_filter = self.report_content_filter.currentText() # In a real app, this would generate a report based on the database # For this demo, we'll just generate a sample report report = f"

Surveillance Report

" report += f"

Period: {date_from.strftime('%Y-%m-%d %H:%M')} to {date_to.strftime('%Y-%m-%d %H:%M')}

" if camera_id == -1: report += "

Cameras: All Cameras

" else: camera_name = "Unknown" for camera in self.db.cameras: if camera["id"] == camera_id: camera_name = camera["name"] break report += f"

Camera: {camera_name}

" report += f"

Content Filter: {content_filter}

" report += "

Events Summary

" report += "" report += "" # Generate sample data for i in range(10): event_time = date_from + (date_to - date_from) * (i / 10) camera_name = "Camera " + str(i % len(self.db.cameras) + 1) if content_filter in ["All Events", "Faces Only", "Known Faces"] and i % 3 == 0: event_type = "Face Detected" details = "John Doe (Known Person)" elif content_filter in ["All Events", "Faces Only", "Unknown Faces"] and i % 3 == 1: event_type = "Face Detected" details = "Unknown Person" elif content_filter in ["All Events", "Trucks Only", "Known Trucks"] and i % 3 == 2: event_type = "Truck Detected" details = "ABC123 (Delivery Truck)" else: event_type = "Motion Detected" details = "Movement in scene" report += f"" report += "

Date/Time	Camera	Event Type	Details
{event_time.strftime('%Y-%m-%d %H:%M:%S')}	{camera_name}	{event_type}	{details}

" self.report_preview.setHtml(report) def export_report_to_pdf(self): if self.report_preview.toPlainText() == "": QMessageBox.warning(self, "No Report", "Please generate a report first") return file_path, _ = QFileDialog.getSaveFileName(self, "Export Report", f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf", "PDF Files (*.pdf)") if file_path: # In a real app, this would export the report to a PDF file # For this demo, we'll just simulate it QMessageBox.information(self, "Export Report", f"Report exported to {file_path}") def send_report_via_telegram(self): if self.report_preview.toPlainText() == "": QMessageBox.warning(self, "No Report", "Please generate a report first") return if not self.db.settings.get("telegram_token") or not self.db.settings.get("telegram_chat_id"): QMessageBox.warning(self, "Telegram Not Configured", "Please configure Telegram bot settings first") return # In a real app, this would send the report via Telegram # For this demo, we'll just simulate it QMessageBox.information(self, "Send Report", "Report sent via Telegram successfully") def init_settings_tab(self): layout = QVBoxLayout() # Storage settings storage_layout = QFormLayout() self.storage_path_input = QLineEdit(self.db.settings.get("storage_path", "./recordings")) browse_button = QPushButton("Browse...") browse_button.clicked.connect(self.browse_storage_path) path_layout = QHBoxLayout() path_layout.addWidget(self.storage_path_input) path_layout.addWidget(browse_button) storage_layout.addRow("Recording Storage Path:", path_layout) storage_group = QGroupBox("Storage Settings") storage_group.setLayout(storage_layout) # Recognition settings recognition_layout = QFormLayout() self.face_threshold_input = QDoubleSpinBox() self.face_threshold_input.setRange(0.1, 1.0) self.face_threshold_input.setSingleStep(0.05) self.face_threshold_input.setValue(self.db.settings.get("face_recognition_threshold", 0.6)) self.ocr_confidence_input = QSpinBox() self.ocr_confidence_input.setRange(0, 100) self.ocr_confidence_input.setSingleStep(5) self.ocr_confidence_input.setValue(self.db.settings.get("ocr_confidence", 70)) recognition_layout.addRow("Face Recognition Threshold:", self.face_threshold_input) recognition_layout.addRow("OCR Confidence (%):", self.ocr_confidence_input) recognition_group = QGroupBox("Recognition Settings") recognition_group.setLayout(recognition_layout) # Backup settings backup_layout = QVBoxLayout() backup_button = QPushButton("Create Backup") backup_button.clicked.connect(self.create_backup) restore_button = QPushButton("Restore from Backup") restore_button.clicked.connect(self.restore_from_backup) backup_layout.addWidget(backup_button) backup_layout.addWidget(restore_button) backup_group = QGroupBox("Backup and Restore") backup_group.setLayout(backup_layout) # Save settings save_button = QPushButton("Save Settings") save_button.clicked.connect(self.save_settings) layout.addWidget(storage_group) layout.addWidget(recognition_group) layout.addWidget(backup_group) layout.addWidget(save_button) layout.addStretch() self.settings_tab.setLayout(layout) def browse_storage_path(self): dir_path = QFileDialog.getExistingDirectory(self, "Select Storage Directory", self.storage_path_input.text()) if dir_path: self.storage_path_input.setText(dir_path) def create_backup(self): backup_dir = QFileDialog.getExistingDirectory(self, "Select Backup Directory", "") if not backup_dir: return try: # In a real app, this would create a backup of the database # For this demo, we'll just simulate it backup_file = f"{backup_dir}/surveillance_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip" # Simulate backup process QApplication.processEvents() time.sleep(1) QMessageBox.information(self, "Backup Created", f"Backup created successfully at {backup_file}") except Exception as e: QMessageBox.critical(self, "Backup Failed", f"Failed to create backup: {str(e)}") def restore_from_backup(self): backup_file, _ = QFileDialog.getOpenFileName(self, "Select Backup File", "", "Backup Files (*.zip)") if not backup_file: return reply = QMessageBox.question(self, "Confirm Restore", "Restoring from backup will overwrite all current data. Continue?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: try: # In a real app, this would restore from the backup # For this demo, we'll just simulate it # Simulate restore process QApplication.processEvents() time.sleep(1) QMessageBox.information(self, "Restore Completed", "Data restored successfully from backup") except Exception as e: QMessageBox.critical(self, "Restore Failed", f"Failed to restore from backup: {str(e)}") def save_settings(self): # Save settings self.db.settings["storage_path"] = self.storage_path_input.text() self.db.settings["face_recognition_threshold"] = self.face_threshold_input.value() self.db.settings["ocr_confidence"] = self.ocr_confidence_input.value() self.db.save_data() QMessageBox.information(self, "Success", "Settings saved successfully") def init_users_tab(self): layout = QVBoxLayout() # User list self.users_table = QTableWidget() self.users_table.setColumnCount(4) self.users_table.setHorizontalHeaderLabels(["Username", "Role", "Last Login", "Actions"]) self.users_table.horizontalHeader().setStretchLastSection(True) # Add user form form_layout = QFormLayout() self.user_username_input = QLineEdit() self.user_password_input = QLineEdit() self.user_password_input.setEchoMode(QLineEdit.Password) self.user_role_input = QComboBox() self.user_role_input.addItems(["admin", "operator", "viewer"]) add_user_button = QPushButton("Add User") add_user_button.clicked.connect(self.add_user) form_layout.addRow("Username:", self.user_username_input) form_layout.addRow("Password:", self.user_password_input) form_layout.addRow("Role:", self.user_role_input) form_layout.addRow("", add_user_button) form_group = QGroupBox("Add New User") form_group.setLayout(form_layout) layout.addWidget(form_group) layout.addWidget(self.users_table) self.users_tab.setLayout(layout) # Load users self.load_users() def add_user(self): username = self.user_username_input.text() password = self.user_password_input.text() role = self.user_role_input.currentText() if not username or not password: QMessageBox.warning(self, "Input Error", "Username and password are required") return # Check if username already exists if any(user["username"] == username for user in self.db.users): QMessageBox.warning(self, "Input Error", "Username already exists") return # Create new user user = { "username": username, "password": password, "role": role, "last_login": None } self.db.users.append(user) self.db.save_data() # Clear form self.user_username_input.clear() self.user_password_input.clear() self.user_role_input.setCurrentIndex(0) # Reload user list self.load_users() QMessageBox.information(self, "Success", f"User '{username}' added successfully") def load_users(self): self.users_table.setRowCount(0) for user in self.db.users: row = self.users_table.rowCount() self.users_table.insertRow(row) self.users_table.setItem(row, 0, QTableWidgetItem(user["username"])) self.users_table.setItem(row, 1, QTableWidgetItem(user["role"])) last_login = user.get("last_login", "Never") self.users_table.setItem(row, 2, QTableWidgetItem(last_login)) # Actions actions_layout = QHBoxLayout() actions_layout.setContentsMargins(0, 0, 0, 0) edit_button = QPushButton("Edit") edit_button.setProperty("username", user["username"]) edit_button.clicked.connect(lambda _, u=user["username"]: self.edit_user(u)) delete_button = QPushButton("Delete") delete_button.setProperty("username", user["username"]) delete_button.clicked.connect(lambda _, u=user["username"]: self.delete_user(u)) actions_layout.addWidget(edit_button) actions_layout.addWidget(delete_button) actions_widget = QWidget() actions_widget.setLayout(actions_layout) self.users_table.setCellWidget(row, 3, actions_widget) def edit_user(self, username): # Find user in database user = next((u for u in self.db.users if u["username"] == username), None) if not user: return # Create edit dialog dialog = QWidget(self, Qt.Window) dialog.setWindowTitle(f"Edit User: {username}") dialog.setGeometry(300, 300, 400, 200) layout = QFormLayout() password_input = QLineEdit() password_input.setEchoMode(QLineEdit.Password) password_input.setPlaceholderText("Leave empty to keep current password") role_input = QComboBox() role_input.addItems(["admin", "operator", "viewer"]) role_input.setCurrentText(user["role"]) save_button = QPushButton("Save Changes") save_button.clicked.connect(lambda: self.save_user_changes( username, password_input.text(), role_input.currentText(), dialog )) layout.addRow("New Password:", password_input) layout.addRow("Role:", role_input) layout.addRow("", save_button) dialog.setLayout(layout) dialog.show() def save_user_changes(self, username, password, role, dialog): # Find user in database for user in self.db.users: if user["username"] == username: if password: user["password"] = password user["role"] = role break self.db.save_data() self.load_users() dialog.close() QMessageBox.information(self, "Success", f"User '{username}' updated successfully") def delete_user(self, username): # Prevent deleting the last admin user admin_count = sum(1 for user in self.db.users if user["role"] == "admin") is_admin = next((user["role"] == "admin" for user in self.db.users if user["username"] == username), False) if admin_count <= 1 and is_admin: QMessageBox.warning(self, "Cannot Delete", "Cannot delete the last admin user") return reply = QMessageBox.question(self, "Confirm Deletion", f"Are you sure you want to delete user '{username}'?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: # Remove user from database self.db.users = [u for u in self.db.users if u["username"] != username] self.db.save_data() # Reload user list self.load_users() QMessageBox.information(self, "Success", f"User '{username}' deleted successfully") def main(): app = QApplication(sys.argv) window = SurveillanceSystem() window.show() sys.exit(app.exec_()) if __name__ == "__main__": main() البدء في تنفيذ واجهه ويب npm tsx والبد على العمل على النحو التالي ## المقدمة والنظرة العامة يهدف النظام إلى تقديم حل مراقبة ذكي ومتكامل يُمكّن المؤسسة من إدارة كاميرات المراقبة، تسجيل الفيديوهات، وتحليل المحتوى باستخدام تقنيات التعرف على الوجوه وتقنيات التعرف على الشاحنات عبر OCR. ويجمع النظام بين عدة واجهات إدارة وكل وحدة تؤدي وظيفة محددة، مثل إدارة الكاميرات والتسجيلات، إدارة الأشخاص والشاحنات، إدارة المستخدمين والصلاحيات، إدارة التنبيهات عبر Telegram، بالإضافة إلى وحدة إنشاء التقارير واستخراجها بصيغة PDF. كما يتميز النظام بمرونته في اختيار نمط التسجيل (استمرار التسجيل أو التسجيل عند اكتشاف حركة) مع تفعيل عمليات التحليل الذكي (التعرف على الوجوه أو الشاحنات) فقط عند وجود حركة في المشهد، مما يساهم في توفير موارد المعالجة وتحسين أداء النظام. ## 1. نافذة البث المباشر والعرض (Live View) ### الهدف والوظيفة - **عرض البث الحي:** تُعرض جميع البثوث الحية للقنوات المستلمة من الكاميرات المُدارة في النظام على واجهة واحدة موحدة، سواء كان ذلك في شبكة (Grid) أو في نوافذ منفصلة، بحيث يجد المراقب كل كاميرا بصورة فورية. - **تفاعل ذكي:** لا يتم تفعيل عمليات التحليل الذكي (مثل التعرف على الوجوه أو الشاحنات) في البث المباشر إلا عند اكتشاف حركة، لتقليل استهلاك الموارد وبالتالي ضمان أداءً مستقرًا. ### آلية العمل يتم فتح اتصال دائم مع كل كاميرا اعتمادًا على بيانات الاتصال المُدخلة مسبقاً في وحدة إدارة الكاميرات. وفي وقت التشغيل، تُقرأ الإطارات في الوقت الحقيقي؛ وخلال الأوقات الهادئة تُعرض لقطات الفيديو دون تنفيذ عمليات تحليل إضافية. أما عند الكشف عن حركة في أحد المشاهد، فينتقل النظام تلقائيًا إلى تفعيل خوارزميات التحليل، فيتم استخراج المعلومات الخاصة بالوجوه أو الشاحنات، مع تظليلها أو عرض البيانات المرافقة (مثل اسم الشخص أو الرقم التعريفي للشاحنة). ## 2. نافذة إدارة الكاميرات ### الهدف والوظيفة تُعد واجهة إدارة الكاميرات القلب النابض للنظام، حيث يتم من خلالها: - **إضافة وتحديث بيانات الكاميرات:** إدخال عنوان البث (URL) أو بيانات كاميرات IP (بما في ذلك اسم المستخدم وكلمة المرور) مع إمكانية تعيين اسم مخصص لكل كاميرا. - **خيارات تفعيل التحليل الذكي:** - **تفعيل التعرف على الوجوه:** يقوم المستخدم بتحديد هذا الخيار لتمكين النظام من إرسال البث الوارد إلى وحدة تحليل الوجوه، بحيث يتم استخراج ملامح الوجه ومقارنتها مع قاعدة بيانات الأشخاص. - **تفعيل التعرف على الشاحنات:** عند اختيار هذا الخيار، يُوجه النظام الفيديو إلى وحدة OCR المتخصصة في قراءة الأرقام التعريفية من لوحات الشاحنات، فيتم بعد ذلك مقارنة هذه البيانات مع سجلات الشاحنات. - **اختيار نمط التسجيل:** يتيح النظام للمستخدم تحديد ما إذا كانت الكاميرا ستسجل الفيديو وفقًا لنمط: - **التسجيل عند كشف الحركة:** تُسجل اللقطات فقط عند وجود حركة ملحوظة داخل نطاق الكاميرا. - **التسجيل المستمر:** يُسجل الفيديو دائمًا، لكن مع ضرورة تفعيل عمليات التحليل (مثل التعرف على الوجوه والشاحنات) فقط عند ظهور حركة، لتقليل الضغط على النظام. ### الفوائد يساعد هذا التصميم على تخصيص الموارد بحيث تُستَخدم عمليات التحليل الذكي فقط عند الضرورة، مما يوفر قدرة معالجة أعلى وكفاءة في التخزين. كما يتيح للمسؤول إدارة وإعداد كل كاميرا وفقًا لمتطلبات البيئة المُراقبة دون الحاجة لإعادة ضبط النظام بأكمله. اضافة نافذة نافذة إدارة التسجيلات ### الهدف والوظيفة تُمكّن واجهة إدارة التسجيلات المستخدم من مراجعة التسجيلات المخزنة بناءً على معايير تصفية متقدمة. تشمل الوظائف: - **فرز التسجيلات حسب الزمن والمكان:** إمكانية استرجاع التسجيلات باستخدام فلترة الوقت والتاريخ، وكذلك بناءً على مناطق الكشف المُعدّة مسبقًا. - **التصفية حسب نتائج التحليل:** يمكن استخراج التسجيلات التي تحتوي على ظهور وجوه (سواء معروف أم غير معروف) أو تسجيلات تعرض شاحنات مع بيانات محددة. - **إدراج بيانات وصفية (Metadata):** تُرفق كافة التسجيلات بعلامات زمنية وبيانات توضيحية تم استخراجها أثناء عملية اكتشاف الحركة؛ وهذا يُسهل عمليات البحث والمراجعة اللاحقة. ### آلية العمل يتم تلقائيًا تسجيل كافة اللقطات مع بيانات وصفية تُحدِّد توقيت ظهور الحركة ونقاط التحليل التي تمت (مثل نتائج التعرف على الوجوه والأرقام التعريفية للشاحنات)، مما يسهل عملية الفرز والتصفية عبر معايير متعددة. وبذلك يستطيع المستخدم مراجعة الأحداث الهامة دون الحاجة للاطلاع على التسجيل بأكمله. ## 4. نافذة إدارة الأشخاص ### الهدف والوظيفة تعمل هذه الوحدة على تخزين وإدارة بيانات الأشخاص المعروفين داخل النظام: - **إضافة بيانات الأشخاص:** يقوم المسؤول بإدخال اسم الشخص وتحميل صورة واضحة يتم استخراج معالم الوجه منها باستخدام خوارزميات التعرف. - **تخزين بيانات المقارنة:** تُستخلص مميزات الوجه (face encoding) ويتم تخزينها في قاعدة البيانات، لتستخدم لاحقًا في مقارنة الوجوه المكتشفة أثناء البث أو التسجيل. - **التفاعل مع عملية التعرف:** عند ظهور وجه في الفيديو، يُقارن النظام الملامح المستخلصة مع البيانات المسجلة، وإذا تم التطابق يُعرض الاسم مع رسم إطار حول الوجه على الشاشة. ### الفوائد تتيح هذه الوظيفة تحديد هوية الأشخاص الذين تظهر على كاميرات المراقبة، مما يُعزز من مستوى الأمان ويساعد في تتبع النشاطات والمراقبة الدقيقة للأفراد داخل البيئة المُراقبة. ## 5. نافذة إدارة الشاحنات ### الهدف والوظيفة تهدف إلى إدارة بيانات الشاحنات وتسهيل التعرف عليها تلقائيًا أثناء الرصد: - **إضافة بيانات الشاحنات:** يشمل ذلك إدخال بيانات مثل اسم السائق، نوع الشاحنة، والرقم التعريفي الخاص بكل شاحنة. - **استخدام تقنيات OCR:** عند ظهور شاحنة في الفيديو، يقوم النظام باستخدام خوارزميات التعرف على الأحرف من الصور لتحليل اللوحة التعريفية واستخراج الرقم. - **المطابقة والتعرف:** تُقارن البيانات المستخلصة من تقنية OCR مع سجلات الشاحنات المخزنة، وعند التطابق يتم تمييز الشاحنة وإظهار البيانات (مثل تفاصيل السائق ونوع الشاحنة). ### الفوائد يساعد ذلك في تتبع الشاحنات داخل مواقع المراقبة ومنع الدخول غير المصرح به، كما يوفر سجلًا دقيقًا لكل المركبات التي تم رصدها داخل النظام. اضافة النوافذ التالية ## 6. نافذة إدارة بوت Telegram ### الهدف والوظيفة تهدف هذه الوحدة إلى دمج النظام مع منصة Telegram من أجل: - **إعداد البوت وإدخال بياناته:** يتيح المسؤول إدخال بيانات البوت مثل الاسم، التوكن، والمعرّف الخاص به. - **التفاعل الفوري:** يتم إرسال تنبيهات لحظية عند حدوث أحداث محددة (مثل ظهور شخص غير معروف أو شاحنة مخالفة)، كما يمكن للمستخدم إرسال أوامر عبر Telegram لاسترجاع تسجيلات أو استخراج تقارير. - **التكامل مع باقي النظام:** يربط نظام Telegram الأحداث التي يتم اكتشافها عبر واجهة المراقبة وإدارة التسجيلات، ليتيح للمستخدم إدارة نظام المراقبة عن بُعد. ### الفوائد يُمكن المسؤول من تلقي تنبيهات فورية حول الأحداث الحرجة، والتفاعل مع النظام دون الحاجة للدخول إلى الواجهة الخاصة به، مما يُعزز من سرعة الاستجابة ودقة إدارة النظام. ## 7. نافذة التقارير ### الهدف والوظيفة تُعد وحدة التقارير أداة تحليلية تُتيح استخراج تقارير شاملة عن بيانات التسجيلات والأحداث الأمنية: - **تحديد معايير التقارير:** يمكن تصفية التقارير بالاعتماد على ظهور الأشخاص (الكامل أو المعروف أو غير المعروف) والشاحنات (الكامل أو المعروفة أو غير المعروفة)، بالإضافة إلى تحديد الفترات الزمنية المطلوبة. - **تنظيم البيانات في جدول:** يُعرض التقرير في شكل جدول متكامل يحتوي على بيانات مثل توقيت ظهور الأحداث، الموقع، والنتائج المُستخلصة. - **تصدير التقارير:** يوفر النظام إمكانية تصدير التقرير إلى ملف PDF منسق يُمكن مشاركته أو استخدامه في تقارير رسمية. - **التكامل مع Telegram:** يمكن إرسال التقارير عند الطلب عبر Telegram، بحيث يستلم المستخدم الملف مباشرة بعد الانتهاء من المعالجة. ### الفوائد يُسهم التقارير في توفير نظرة تحليلية دقيقة، تساعد الجهات الأمنية والإدارية على اتخاذ قرار مبني على بيانات موثوقة دون الحاجة لمراجعة التسجيلات بأكملها. اضافة النوافذ التالية ## 8. نافذة إعدادات النظام ### الهدف والوظيفة تحتوي هذه الوحدة على كافة الإعدادات التشغيلية والتأمينية التي تتيح: - **إدارة النسخ الاحتياطي والاستعادة:** إمكانية إجراء نسخ احتياطية منتظمة لقاعدة البيانات والإعدادات، مع آلية لاستعادة البيانات عند الحاجة. - **ضبط حساسية التحليل:** توفير إعدادات لضبط معايير التعرف على الوجوه (على سبيل المثال قيمة العتبة) وكذلك ضبط حساسية OCR لتقليل الإنذارات الكاذبة وزيادة الدقة. - **تحديد مسارات التخزين:** يتيح اختيار المسار المناسب لتخزين ملفات التسجيل سواء محلياً أو عبر شبكة تخزين. - **تخصيص الميزات:** إمكانية تفعيل أو تعطيل بعض الميزات ضمن النظام، لتلبية متطلبات كل بيئة عمل. ### الفوائد يضمن ذلك استقرار النظام واستمراريته، كما يسمح بتخصيص نظام المراقبة بحسب الاحتياجات التشغيلية والأمنية لكل مؤسسة. ## 9. نافذة إدارة المستخدمين والصلاحيات ### الهدف والوظيفة يُعدّ هذا الجانب من النظام حجر الأساس لتأمين الوصول وإدارة الحسابات: - **إعداد المستخدم الافتراضي ذي الصلاحيات الكاملة:** عند تشغيل النظام لأول مرة، يُنشأ حساب افتراضي (مثلاً admin) بكلمة مرور مُؤمنة يتم تغييرها فور بدء الاستخدام. - **إضافة وتعديل المستخدمين:** يمكن للمسؤول إنشاء حسابات جديدة وتحديد الصلاحيات المخصصة لكل حساب باستخدام آلية تحكم وصول تعتمد على الأدوار (RBAC). - **تحديد النوافذ والميزات المسموحة:** عند إدارة المستخدم، يتم تحديد أي النوافذ (كنافذة العرض المباشر، إدارة الكاميرات، إدارة التقارير، إلخ) يمكن أن تظهر له والتي يستطيع التفاعل معها، مما يقلل من تعرض البيانات أو الميزات الحساسة للمستخدمين ذوي الصلاحيات المحدودة. - **تسجيل نشاط المستخدم:** تُسجل كافة الأنشطة والتعديلات الخاصة بالمستخدمين لأغراض التدقيق والأمان. ### الفوائد تضمن هذه الآلية أن يبقى النظام آمناً عن طريق توزيع صلاحيات الوصول ومتابعة نشاط الحسابات، ويتيح للمسؤول ضبط تجربة المستخدم بما يتناسب مع أدواره ومسؤولياته داخل المؤسسة. ## 10. آلية تشغيل النظام مع الكشف عن الحركة والتفعيل الديناميكي العمل على ان النظام يعمل على النحو التالي ### المفهوم العام يعد الكشف عن الحركة عنصراً أساسياً في النظام لضمان فعالية استخدام الموارد. لذا يتم تصميم النظام بحيث: - تُراقب كل كاميرا الإطارات الواردة باستمرار باستخدام خوارزميات متخصصة (مثل مقارنة الإطارات أو background subtraction). - **عند اكتشاف حركة:** - يبدأ التسجيل في نمط "التسجيل عند كشف الحركة" إذا كان هذا الخيار مُفعّل، أو يستمر التسجيل (في حالة التسجيل المستمر). - تُفعّل خوارزميات التعرف على الوجوه والشاحنات، بحيث تتم المعالجات الذكية فقط عند حدوث تغييرات ملحوظة في المحتوى. - تُضاف بيانات وصفية (Timestamp، موقع الكشف، نتائج التحليل) إلى التسجيل لتسهيل عملية الفرز والمراجعة لاحقًا. ### الفوائد العملية - تقليل استهلاك وحدة المعالجة والذاكرة عند غياب الحركة. - تحسين زمن الاستجابة عند حدوث أحداث حاسمة. - توفير بيانات دقيقة تساعد في تحليل المشاهد الأمنية دون إهدار موارد النظام. ## الخلاصة والتوجه العام يجمع هذا التصميم بين الاتساق والمرونة في التعامل مع البيانات والتسجيلات الحية، بحيث يتكون النظام من وحدات متخصصة تعمل بتناغم لتوفير منصة مراقبة شاملة على غرار أنظمة DVR/NVR الحديثة. ويأتي التصميم بعدة مستويات من إدارة الوظائف: - عرض البث الحي بفعالية مع تفعيل التحليل الذكي عند الحدث. - إدارة الكاميرات مع خيارات تفعيل التحليل (التعرف على الوجوه والشاحنات) وتحديد نمط التسجيل المناسب لتوفير الموارد. - إدارة التسجيلات وتصفية ومراجعتها باستخدام بيانات وصفية دقيقة. - إدارة الكيانات (الأشخاص والشاحنات) لتوفير آلية مراقبة دقيقة، بالإضافة إلى التواصل الفوري عبر Telegram Bot لاستقبال التنبيهات وتنفيذ الأوامر عن بُعد. - توفير تقارير تحليلية شاملة ودقيقة تُصدر بصيغة PDF وتكون متاحة حسب معايير محددة. - نظام إعدادات متكامل يشمل النسخ الاحتياطي، وضبط الحساسية، وتحديد مسارات التخزين. - نظام إدارة للمستخدمين مع آلية RBAC لتوزيع الصلاحيات وتحديد النوافذ التي يمكن لكل فئة الوصول إليها. بهذا التوجه، يوفر النظام بيئة تشغيل آمنة وفعالة وقابلة للتوسع تلبي متطلبات المؤسسات التي تحتاج إلى مراقبة شاملة وتحليل دقيق للأحداث في الوقت الحقيقي مع الحفاظ على كفاءة الموارد والتفاعل السريع مع التنبيهات الأمنية. عمل ملف bat للتشغيل المشروع كاملا على وندوز وملف sh للتشغيل كاملا على لينكس وثناء التنفيذ شرح ما تم تنفيذة ما هو المتبقي