SceneSense/camera-client/camera_monitor.py

import cv2
import time
import requests
import numpy as np
from datetime import datetime
import json
import logging as lg
import logging.handlers as lg_handlers
import argparse

class CameraMonitor:
    def __init__(self, config, logger):
        self.config = config
        self.logger = logger
        self.last_detection = 0
        self.capture = self.initialize_camera()
        
    def initialize_camera(self):
        """
        Initialize the camera and return the VideoCapture object.
        """
        cap = cv2.VideoCapture(
            self.config['camera_address'],
            cv2.CAP_V4L2
        )
        if not cap.isOpened():
            raise Exception(f"Failed to open camera at {self.config['camera_address']}")
        self.logger.info("Camera initialized")
        print("Camera initialized")
        return cap

    def log_scene_change(self):
        """
        Log a scene change notification to the Scenesense server.
        """
        self.logger.info(f'Scene change detected')
        # print("Scene change detected")
        current_time = int(time.time())
        human_readable_time = datetime.utcfromtimestamp(
            current_time
        ).strftime('%Y-%m-%d %H:%M:%S')
        description = f'Scene change detected at {human_readable_time}'
        payload = {
            'timestamp': current_time,
            'description': description
        }
        headers = {
            'Authorization': self.config['api_key']
        }
        if (
            current_time - self.last_detection >= self.config['detection_delay']
        ):
            try:
                response = requests.post(
                    self.config['api_url'],
                    json=payload,
                    headers=headers
                )
                response.raise_for_status()
                self.logger.info(
                    f'Scene change logged to server db: {description}'
                )
                # print(f'Scene change logged: {description}')
                self.last_detection = current_time


            except requests.exceptions.RequestException as e:
                self.logger.error(f'Error connecting to Scenesense server: {e}')
                # print(f'Error connecting to Scenesense server: {e}')
            except requests.exceptions.HTTPError as e:
                self.logger.error(f'HTTP error: {e}')
                # print(f'HTTP error: {e}')


    def detect_scene_change(self, previous_frame, current_frame):
        """
        Detect scene change based on frame differences.
        """
        frame_diff = cv2.absdiff(previous_frame, current_frame)
        mean_diff = np.mean(frame_diff)
        threshold = self.config['sensitivity_threshold']
        # self.logger.info(
        #     f'Frame Diff: {frame_diff}\nMean Diff: {mean_diff}\nSensitivity Threshold: {threshold}'
        # )
        # print(
        #     f'Frame Diff: {frame_diff}\nMean Diff: {mean_diff}\nSensitivity Threshold: {threshold}'
        # )
        return mean_diff > threshold

    def run(self):
        self.logger.info("Camera Monitor started")
        print("Camera Monitor started")
        try:
            previous_frame = None

            while True:
                ret, current_frame = self.capture.read()

                if not ret:
                    self.logger.error("Failed to read frame")
                    delay = self.config['failure_delay']
                    self.logger.info(f"Will retry after {delay} seconds")
                    self.capture.release()
                    time.sleep(delay)
                    self.logger.info("Retrying")
                    self.capture = self.initialize_camera()
                    continue


                if previous_frame is not None:
                    if self.detect_scene_change(previous_frame, current_frame):
                        self.log_scene_change()

            
                previous_frame = current_frame
            
        except KeyboardInterrupt:
            self.logger.info("Camera monitoring stopped by user.")
            print("Camera monitoring stopped by user.")
        except Exception as e:
            self.logger.error(f"An error occurred: {e}")
            # print(f"An error occurred: {e}")
        finally:
            self.capture.release()

def parse_command_line_args():
    parser = argparse.ArgumentParser(description='Camera Monitor for Scenesense')
    parser.add_argument('--config', type=str, default='config.json', help='Path to configuration file')
    return parser.parse_args()

def load_config(config_path):
    try:
        with open(config_path, 'r') as config_file:
            return json.load(config_file)
    except FileNotFoundError:
        raise Exception(f"Config file not found: {config_path}")
    except json.JSONDecodeError:
        raise Exception(f"Invalid JSON format in config file: {config_path}")

def setup_logging(log_file='logs/camera_monitor.log'):
    # Configure the logger
    logger = lg.getLogger('CameraMonitor')
    logger.setLevel(lg.INFO)

    # Create a TimedRotatingFileHandler for log rotation
    file_handler = lg_handlers.TimedRotatingFileHandler(
        log_file, when='midnight', backupCount=7
    )  # Keeps logs for 7 days
    file_handler.suffix = '%Y-%m-%d'  # Append date to log file names

    # Define the log message format
    formatter = lg.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    file_handler.setFormatter(formatter)

    # Add the file handler to the logger
    logger.addHandler(file_handler)

    return logger

def main():
    args = parse_command_line_args()
    config = load_config(args.config)
    logger = setup_logging()

    monitor = CameraMonitor(config, logger)
    monitor.run()

if __name__ == "__main__":
    main()