import os from dataclasses import dataclass @dataclass class AppConfig: STORAGE_PATH: str = "./storage" DATABASE_URL: str = "sqlite:///./surveillance.db" TELEGRAM_TOKEN: str = "" TELEGRAM_CHAT_ID: str = "" MOTION_THRESHOLD: float = 0.5 FACE_RECOGNITION_THRESHOLD: float = 0.6 OCR_CONFIDENCE: float = 0.7 @classmethod def load(cls): return cls( STORAGE_PATH=os.getenv("STORAGE_PATH", cls.STORAGE_PATH), DATABASE_URL=os.getenv("DATABASE_URL", cls.DATABASE_URL), TELEGRAM_TOKEN=os.getenv("TELEGRAM_TOKEN", cls.TELEGRAM_TOKEN), TELEGRAM_CHAT_ID=os.getenv("TELEGRAM_CHAT_ID", cls.TELEGRAM_CHAT_ID), MOTION_THRESHOLD=float(os.getenv("MOTION_THRESHOLD", cls.MOTION_THRESHOLD)), FACE_RECOGNITION_THRESHOLD=float(os.getenv("FACE_RECOGNITION_THRESHOLD", cls.FACE_RECOGNITION_THRESHOLD)), OCR_CONFIDENCE=float(os.getenv("OCR_CONFIDENCE", cls.OCR_CONFIDENCE)) ) from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Boolean, ForeignKey from sqlalchemy.orm import declarative_base, relationship, sessionmaker from datetime import datetime Base = declarative_base() class Camera(Base): __tablename__ = 'cameras' id = Column(Integer, primary_key=True) name = Column(String(100), nullable=False) stream_url = Column(String(255), nullable=False) username = Column(String(50)) password = Column(String(50)) location = Column(String(100)) is_active = Column(Boolean, default=True) record_continuously = Column(Boolean, default=False) detect_faces = Column(Boolean, default=False) detect_trucks = Column(Boolean, default=False) created_at = Column(DateTime, default=datetime.utcnow) recordings = relationship("Recording", back_populates="camera") class Person(Base): __tablename__ = 'persons' id = Column(Integer, primary_key=True) name = Column(String(100), nullable=False) photo_path = Column(String(255), nullable=False) face_encoding = Column(String(2000), nullable=False) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class Truck(Base): __tablename__ = 'trucks' id = Column(Integer, primary_key=True) license_plate = Column(String(20), nullable=False, unique=True) driver_name = Column(String(100)) truck_type = Column(String(50)) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class Recording(Base): __tablename__ = 'recordings' id = Column(Integer, primary_key=True) camera_id = Column(Integer, ForeignKey('cameras.id')) file_path = Column(String(255), nullable=False) start_time = Column(DateTime, nullable=False) end_time = Column(DateTime, nullable=False) has_motion = Column(Boolean, default=False) has_faces = Column(Boolean, default=False) has_trucks = Column(Boolean, default=False) camera = relationship("Camera", back_populates="recordings") events = relationship("Event", back_populates="recording") class Event(Base): __tablename__ = 'events' id = Column(Integer, primary_key=True) recording_id = Column(Integer, ForeignKey('recordings.id')) event_type = Column(String(20), nullable=False) # 'motion', 'face', 'truck' timestamp = Column(DateTime, nullable=False) confidence = Column(Float) details = Column(String(255)) recording = relationship("Recording", back_populates="events") class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) username = Column(String(50), nullable=False, unique=True) password_hash = Column(String(255), nullable=False) full_name = Column(String(100)) is_admin = Column(Boolean, default=False) created_at = Column(DateTime, default=datetime.utcnow) last_login = Column(DateTime) from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models import Base from config import AppConfig config = AppConfig.load() engine = create_engine(config.DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) def init_db(): Base.metadata.create_all(bind=engine) def get_db(): db = SessionLocal() try: yield db finally: db.close() import cv2 import numpy as np from datetime import datetime import os from config import AppConfig from models import Recording, Event from database import get_db config = AppConfig.load() class VideoProcessor: def __init__(self, camera_id): self.camera_id = camera_id self.background_subtractor = cv2.createBackgroundSubtractorMOG2() self.is_recording = False self.video_writer = None self.current_recording = None self.frame_count = 0 self.motion_frames = 0 self.recording_start = None def process_frame(self, frame): self.frame_count += 1 # Motion detection motion_detected = self._detect_motion(frame) # Start/stop recording based on motion if motion_detected: self.motion_frames += 1 if not self.is_recording: self._start_recording(frame) else: if self.is_recording and self.motion_frames > 0: self.motion_frames -= 1 if self.motion_frames == 0: self._stop_recording() # Write frame if recording if self.is_recording: self.video_writer.write(frame) return motion_detected def _detect_motion(self, frame): gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) fg_mask = self.background_subtractor.apply(gray) _, thresh = cv2.threshold(fg_mask, 127, 255, cv2.THRESH_BINARY) contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) motion_detected = False for contour in contours: if cv2.contourArea(contour) > 500: # Minimum area threshold motion_detected = True break return motion_detected def _start_recording(self, first_frame): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"recording_{self.camera_id}_{timestamp}.avi" filepath = os.path.join(config.STORAGE_PATH, filename) # Create storage directory if not exists os.makedirs(config.STORAGE_PATH, exist_ok=True) # Initialize video writer height, width = first_frame.shape[:2] fourcc = cv2.VideoWriter_fourcc(*'XVID') self.video_writer = cv2.VideoWriter(filepath, fourcc, 20.0, (width, height)) # Create recording record in database db = next(get_db()) self.current_recording = Recording( camera_id=self.camera_id, file_path=filepath, start_time=datetime.now(), end_time=datetime.now(), # Will be updated when stopped has_motion=True ) db.add(self.current_recording) db.commit() self.is_recording = True self.recording_start = datetime.now() def _stop_recording(self): if self.video_writer is not None: self.video_writer.release() self.video_writer = None if self.current_recording is not None: db = next(get_db()) self.current_recording.end_time = datetime.now() db.commit() self.current_recording = None self.is_recording = False self.recording_start = None import face_recognition import numpy as np from models import Person from database import get_db from config import AppConfig config = AppConfig.load() class FaceRecognizer: def __init__(self): self.known_faces = [] self.known_names = [] self._load_known_faces() def _load_known_faces(self): db = next(get_db()) persons = db.query(Person).all() for person in persons: encoding = np.fromstring(person.face_encoding[1:-1], sep=',') self.known_faces.append(encoding) self.known_names.append(person.name) def recognize_faces(self, frame): # Convert the image from BGR color (OpenCV) to RGB color rgb_frame = frame[:, :, ::-1] # Find all face locations and encodings in the current frame face_locations = face_recognition.face_locations(rgb_frame) face_encodings = face_recognition.face_encodings(rgb_frame, face_locations) recognized_faces = [] for face_encoding, face_location in zip(face_encodings, face_locations): # Compare the face with known faces matches = face_recognition.compare_faces( self.known_faces, face_encoding, tolerance=config.FACE_RECOGNITION_THRESHOLD ) name = "Unknown" # Find the best match face_distances = face_recognition.face_distance(self.known_faces, face_encoding) best_match_index = np.argmin(face_distances) if matches[best_match_index]: name = self.known_names[best_match_index] recognized_faces.append({ 'name': name, 'location': face_location, 'confidence': 1 - face_distances[best_match_index] }) return recognized_faces import cv2 import pytesseract import re from config import AppConfig from models import Truck from database import get_db config = AppConfig.load() class TruckRecognizer: def __init__(self): self.known_trucks = self._load_known_trucks() def _load_known_trucks(self): db = next(get_db()) return {truck.license_plate: truck for truck in db.query(Truck).all()} def recognize_trucks(self, frame): # Preprocess the image for better OCR gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) blurred = cv2.GaussianBlur(gray, (5, 5), 0) edged = cv2.Canny(blurred, 30, 200) # Find contours in the edged image contours, _ = cv2.findContours(edged.copy(), cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE) contours = sorted(contours, key=cv2.contourArea, reverse=True)[:5] recognized_trucks = [] for contour in contours: # Approximate the contour perimeter = cv2.arcLength(contour, True) approx = cv2.approxPolyDP(contour, 0.02 * perimeter, True) # If the contour has 4 points, we assume it's a license plate if len(approx) == 4: x, y, w, h = cv2.boundingRect(contour) plate_roi = gray[y:y+h, x:x+w] # Apply OCR to the license plate region plate_text = pytesseract.image_to_string(plate_roi, config='--psm 8') plate_text = self._clean_plate_text(plate_text) if plate_text and len(plate_text) >= 3: # Minimum length for a plate truck_info = self.known_trucks.get(plate_text, None) recognized_trucks.append({ 'license_plate': plate_text, 'location': (x, y, w, h), 'known': truck_info is not None, 'truck_info': truck_info }) return recognized_trucks def _clean_plate_text(self, text): # Remove all non-alphanumeric characters cleaned = re.sub(r'[^a-zA-Z0-9]', '', text) return cleaned.upper() import telegram from telegram.ext import Updater, CommandHandler, MessageHandler, Filters from config import AppConfig from models import Recording from database import get_db from datetime import datetime, timedelta config = AppConfig.load() class TelegramBot: def __init__(self): self.bot = telegram.Bot(token=config.TELEGRAM_TOKEN) self.updater = Updater(token=config.TELEGRAM_TOKEN, use_context=True) # Register handlers dp = self.updater.dispatcher dp.add_handler(CommandHandler("start", self.start)) dp.add_handler(CommandHandler("latest", self.send_latest_recording)) dp.add_handler(CommandHandler("alerts", self.toggle_alerts)) self.alerts_enabled = True def start(self, update, context): update.message.reply_text( "Surveillance System Bot\n\n" "Available commands:\n" "/start - Show this message\n" "/latest - Get the latest recording\n" "/alerts - Toggle motion detection alerts" ) def send_latest_recording(self, update, context): db = next(get_db()) recording = db.query(Recording).order_by(Recording.start_time.desc()).first() if recording: try: with open(recording.file_path, 'rb') as video_file: context.bot.send_video( chat_id=update.effective_chat.id, video=video_file, caption=f"Recording from {recording.start_time}" ) except Exception as e: update.message.reply_text(f"Error sending recording: {str(e)}") else: update.message.reply_text("No recordings available") def toggle_alerts(self, update, context): self.alerts_enabled = not self.alerts_enabled status = "enabled" if self.alerts_enabled else "disabled" update.message.reply_text(f"Motion detection alerts are now {status}") def send_alert(self, message, video_path=None): if not self.alerts_enabled: return try: if video_path: with open(video_path, 'rb') as video_file: self.bot.send_video( chat_id=config.TELEGRAM_CHAT_ID, video=video_file, caption=message ) else: self.bot.send_message( chat_id=config.TELEGRAM_CHAT_ID, text=message ) except Exception as e: print(f"Error sending Telegram alert: {str(e)}") def start_polling(self): self.updater.start_polling() print("Telegram bot started polling") def stop_polling(self): self.updater.stop() from fpdf import FPDF from datetime import datetime from models import Recording, Event from database import get_db class ReportGenerator: def __init__(self): self.pdf = FPDF() self.pdf.set_auto_page_break(auto=True, margin=15) def generate_report(self, start_date, end_date, report_type="all"): db = next(get_db()) # Query recordings in the date range recordings = db.query(Recording).filter( Recording.start_time >= start_date, Recording.end_time <= end_date ).all() # Create PDF self.pdf.add_page() self._add_header(start_date, end_date) self._add_summary(recordings) if report_type in ["all", "details"]: self._add_detailed_events(recordings) return self.pdf def _add_header(self, start_date, end_date): self.pdf.set_font("Arial", 'B', 16) self.pdf.cell(0, 10, "Surveillance System Report", 0, 1, 'C') self.pdf.set_font("Arial", '', 12) self.pdf.cell(0, 10, f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}", 0, 1, 'C') self.pdf.ln(10) def _add_summary(self, recordings): self.pdf.set_font("Arial", 'B', 14) self.pdf.cell(0, 10, "Summary", 0, 1) total_recordings = len(recordings) motion_events = sum(1 for r in recordings if r.has_motion) face_events = sum(1 for r in recordings if r.has_faces) truck_events = sum(1 for r in recordings if r.has_trucks) self.pdf.set_font("Arial", '', 12) self.pdf.cell(0, 10, f"Total recordings: {total_recordings}", 0, 1) self.pdf.cell(0, 10, f"Recordings with motion: {motion_events}", 0, 1) self.pdf.cell(0, 10, f"Recordings with faces: {face_events}", 0, 1) self.pdf.cell(0, 10, f"Recordings with trucks: {truck_events}", 0, 1) self.pdf.ln(10) def _add_detailed_events(self, recordings): self.pdf.set_font("Arial", 'B', 14) self.pdf.cell(0, 10, "Detailed Events", 0, 1) self.pdf.ln(5) self.pdf.set_font("Arial", 'B', 12) self.pdf.cell(60, 10, "Timestamp", 1, 0, 'C') self.pdf.cell(50, 10, "Camera", 1, 0, 'C') self.pdf.cell(40, 10, "Event Type", 1, 0, 'C') self.pdf.cell(40, 10, "Details", 1, 1, 'C') self.pdf.set_font("Arial", '', 10) for recording in recordings: for event in recording.events: self.pdf.cell(60, 10, event.timestamp.strftime("%Y-%m-%d %H:%M"), 1, 0) self.pdf.cell(50, 10, recording.camera.name, 1, 0) self.pdf.cell(40, 10, event.event_type.capitalize(), 1, 0) self.pdf.cell(40, 10, event.details[:35] if event.details else "", 1, 1) import cv2 import time from threading import Thread from config import AppConfig from video_processing import VideoProcessor from face_recognition import FaceRecognizer from truck_recognition import TruckRecognizer from telegram_bot import TelegramBot from models import Camera, Event from database import get_db, init_db config = AppConfig.load() class SurveillanceSystem: def __init__(self): init_db() self.face_recognizer = FaceRecognizer() self.truck_recognizer = TruckRecognizer() self.telegram_bot = TelegramBot() self.running = False self.processors = {} # Start Telegram bot in a separate thread self.bot_thread = Thread(target=self.telegram_bot.start_polling) self.bot_thread.daemon = True self.bot_thread.start() def start(self): self.running = True self._load_cameras() while self.running: try: self._process_cameras() time.sleep(0.1) # Small delay to prevent high CPU usage except KeyboardInterrupt: self.stop() except Exception as e: print(f"Error in main loop: {str(e)}") time.sleep(5) # Wait before retrying def stop(self): self.running = False self.telegram_bot.stop_polling() # Stop all video processors for processor in self.processors.values(): if processor.is_recording: processor._stop_recording() def _load_cameras(self): db = next(get_db()) cameras = db.query(Camera).filter(Camera.is_active == True).all() for camera in cameras: if camera.id not in self.processors: self.processors[camera.id] = VideoProcessor(camera.id) def _process_cameras(self): db = next(get_db()) cameras = db.query(Camera).filter(Camera.is_active == True).all() for camera in cameras: if camera.id not in self.processors: self.processors[camera.id] = VideoProcessor(camera.id) processor = self.processors[camera.id] try: cap = cv2.VideoCapture(camera.stream_url) if not cap.isOpened(): print(f"Failed to open camera {camera.id}") continue ret, frame = cap.read() if not ret: print(f"Failed to read frame from camera {camera.id}") cap.release() continue # Process frame for motion detection motion_detected = processor.process_frame(frame) if motion_detected and processor.current_recording: # Perform face recognition if enabled if camera.detect_faces: faces = self.face_recognizer.recognize_faces(frame) for face in faces: event = Event( recording_id=processor.current_recording.id, event_type="face", timestamp=datetime.now(), confidence=face['confidence'], details=f"Person: {face['name']}" ) db.add(event) # Perform truck recognition if enabled if camera.detect_trucks: trucks = self.truck_recognizer.recognize_trucks(frame) for truck in trucks: details = f"Truck: {truck['license_plate']}" if truck['known'] and truck['truck_info']: details += f" (Driver: {truck['truck_info'].driver_name})" event = Event( recording_id=processor.current_recording.id, event_type="truck", timestamp=datetime.now(), confidence=0.9 if truck['known'] else 0.5, details=details ) db.add(event) db.commit() cap.release() except Exception as e: print(f"Error processing camera {camera.id}: {str(e)}") if 'cap' in locals(): cap.release() if __name__ == "__main__": system = SurveillanceSystem() try: system.start() except KeyboardInterrupt: system.stop() numpy==1.24.3 opencv-python==4.7.0.72 face-recognition==1.3.0 pytesseract==0.3.10 python-telegram-bot==20.3 fpdf2==2.7.4 sqlalchemy==2.0.15 python-dotenv==1.0.0