path: root/app/scripts/monitor.py

import argparse
import os
import time
import smtplib
import json
from email.mime.text import MIMEText
from tensorflow.keras.models import load_model
from tensorflow.keras.preprocessing import image
import numpy as np
import requests
from io import BytesIO
from PIL import Image
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler

# Load configuration
with open("config.json", "r") as config_file:
    config = json.load(config_file)

# Load model
MODEL_PATH = "./model/garage_door_classifier.h5"
model = load_model(MODEL_PATH)

# Constants from config
STREAM_URL = config["stream_url"]
ALERT_INTERVAL = config["alert_interval"]
RETRY_LIMIT = config["retry_limit"]
RETRY_DELAY = config["retry_delay"]
STATS_INTERVAL = config["stats_interval"]
MIN_OPEN_FRAMES = config["min_open_frames"]
FRAME_WINDOW = config["frame_window"]
MAX_BUFFER_SIZE_MB = config["max_buffer_size_mb"]

# SMTP configuration
SMTP_SERVER = config["smtp_server"]
SMTP_PORT = config["smtp_port"]

EMAIL_FROM = config["email_from"]
EMAIL_TO = config["email_to"]
EMAIL_SUBJECT = config["subject"]
EMAIL_BODY = config["body"]

# Variables
last_alert_time = 0
frames_processed = 0
emails_sent = 0
confidence_scores = []  # Rolling list of confidence scores
retry_count = 0
frame_status = []  # Track frame statuses for the rolling window
buffer_size = 0
is_healthy = False

# Helper function to preprocess image
def preprocess_image(img):
    img = img.resize((224, 224))  # Resize to model's input size
    img_array = np.array(img) / 255.0  # Normalize
    return np.expand_dims(img_array, axis=0)


# Helper function to send email
def send_email():
    global emails_sent
    try:
        # Split multiple email addresses by semicolon and strip any whitespace
        recipient_list = [email.strip() for email in EMAIL_TO.split(';') if email.strip()]

        # Create email message
        msg = MIMEText(EMAIL_BODY)
        msg["From"] = EMAIL_FROM
        msg["To"] = ", ".join(recipient_list)
        msg["Subject"] = EMAIL_SUBJECT

        # Connect to SMTP server with authentication
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.sendmail(EMAIL_FROM, recipient_list, msg.as_string())

        emails_sent += 1
        print(f"Alert email sent to: {', '.join(recipient_list)}")
    except Exception as e:
        print(f"Failed to send email: {e}")


# Helper function to emit statistics
def emit_statistics():
    global frames_processed, emails_sent, confidence_scores, buffer_size
    if frames_processed > 0:
        avg_conf = sum(confidence_scores) / len(confidence_scores)
        print(f"Stats: Frames Processed: {frames_processed}, Emails Sent: {emails_sent}, "
              f"Avg Confidence: {avg_conf:.2f}, Min Confidence: {min(confidence_scores):.2f}, "
              f"Max Confidence: {max(confidence_scores):.2f}, Buffer Size: {buffer_size / (1024 * 1024):.2f} MB")
    else:
        print(f"No frames processed in the last {STATS_INTERVAL} seconds.")
    # Reset stats for the next interval
    frames_processed = 0
    confidence_scores.clear()


# Stream handler
def process_stream():
    global is_healthy, frames_processed, last_alert_time, retry_count, frame_status, buffer_size
    retry_count = 0  # Reset retry count on successful stream
    session = requests.Session()
    start_time = time.time()
    try:
        with session.get(STREAM_URL, stream=True, timeout=5) as response:
            response.raise_for_status()
            boundary = b'--videoboundary'
            buffer = b''
            last_stats_time = time.time()

            for chunk in response.iter_content(chunk_size=1024):
                buffer += chunk
                buffer_size = len(buffer)

                # Check buffer size and frame lag
                if buffer_size > MAX_BUFFER_SIZE_MB * 1024 * 1024:
                    print("Buffer exceeded maximum size. Clearing buffer.")
                    buffer = b''
                    buffer_size = 0
                    continue

                while boundary in buffer:
                    # Find the start of a new frame
                    parts = buffer.split(boundary, 1)
                    if len(parts) < 2:
                        break

                    # Extract headers and image data
                    headers_end = parts[1].find(b'\r\n\r\n')
                    if headers_end == -1:
                        break  # Wait for more data if headers aren't complete

                    headers = parts[1][:headers_end]
                    remaining = parts[1][headers_end + 4:]  # Skip past '\r\n\r\n'

                    # Parse headers to find Content-length
                    content_length = None
                    for line in headers.split(b'\r\n'):
                        if line.lower().startswith(b'content-length:'):
                            content_length = int(line.split(b':')[1].strip())
                            break

                    if content_length is None:
                        print("Content-length not found; skipping frame.")
                        buffer = remaining
                        continue

                    # Extract the image data
                    if len(remaining) < content_length:
                        # Wait for more data if the full image isn't in the buffer yet
                        break

                    frame_data = remaining[:content_length]
                    buffer = remaining[content_length:]

                    # Decode and process the frame
                    try:
                        with Image.open(BytesIO(frame_data)) as img:
                            preprocessed_img = preprocess_image(img)
                            prediction = model.predict(preprocessed_img)
                            confidence = prediction[0][0]
                            is_open = confidence > 0.5

                            frames_processed += 1
                            confidence_scores.append(confidence)
                            frame_status.append((is_open, time.time()))

                            # Maintain rolling window of frame statuses
                            frame_status = [
                                (status, ts)
                                for status, ts in frame_status
                                if time.time() - ts <= FRAME_WINDOW
                            ]

                            # Check if the door is open based on the rolling window
                            open_frames = sum(1 for status, ts in frame_status if status)
                            if open_frames >= min(MIN_OPEN_FRAMES, len(frame_status)):
                                print(f"Garage door is OPEN (Confidence: {confidence:.2f}).")
                                current_time = time.time()
                                if current_time - last_alert_time >= ALERT_INTERVAL:
                                    send_email()
                                    last_alert_time = current_time
                            elif len(frame_status) >= MIN_OPEN_FRAMES:
                                print(f"Garage door is CLOSED with sufficient frames (Confidence: {confidence:.2f}).")
                            else:
                                print(f"Garage door is CLOSED with insufficient frames (Confidence: {confidence:.2f}).")
                            
                        is_healthy = True
                    except Exception as e:
                        is_healthy = False
                        print(f"Error processing frame: {e}")

                # Emit statistics every STATS_INTERVAL seconds
                if time.time() - last_stats_time >= STATS_INTERVAL:
                    emit_statistics()
                    last_stats_time = time.time()
                    
                # Refresh stream for stability
                if time.time() - start_time >= 120:
                    print("Refreshing stream")
                    break;

    except Exception as e:
        print(f"Stream error: {e}")
        is_healthy = False
        raise RuntimeError("Stream failed.")
    finally:
        session.close()


def test_mode():
    """
    Perform a lightweight test to ensure the script runs correctly.
    """
    print("Running in test mode: Verifying setup...")
    print(f"Model path: {MODEL_PATH}")
    print(f"SMTP server: {SMTP_SERVER}, Port: {SMTP_PORT}")
    print(f"Stream URL: {STREAM_URL}")
    print("Test mode completed successfully!")


def monitor():
    global is_healthy, retry_count
    start_time = time.time()
    while time.time() - start_time <= 1800:
        try:
            process_stream()
        except RuntimeError:
            is_healthy = False
            retry_count += 1
            if retry_count >= RETRY_LIMIT:
                print(f"Max retries reached ({RETRY_LIMIT}). Exiting.")
                break
            print(f"Retrying... ({retry_count}/{RETRY_LIMIT})")
            time.sleep(RETRY_DELAY)


# Health Check HTTP Handler
class HealthCheckHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        global is_healthy
        if is_healthy:
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"OK")
        else:
            self.send_response(500)
            self.end_headers()
            self.wfile.write(b"UNHEALTHY")


def start_health_check_server():
    server = HTTPServer(('0.0.0.0', 8080), HealthCheckHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Garage Door Monitoring Script")
    parser.add_argument("--test", action="store_true", help="Run in test mode")
    args = parser.parse_args()

    if args.test:
        test_mode()
    else:
        start_health_check_server()
        monitor()