app.py

import os
# Set a writable Matplotlib configuration directory to avoid permission errors.
os.environ["MPLCONFIGDIR"] = "/tmp/matplotlib"

from flask import Flask, render_template, Response, jsonify
import cv2
import time
import numpy as np
import threading
import requests
from twilio.rest import Client
from datetime import datetime
from inference_sdk import InferenceHTTPClient
import pygame  # Ensure pygame is installed

app = Flask(__name__)

# Attempt to initialize the webcam.
# Note: In many Hugging Face Spaces, a physical camera may not be available.
camera = cv2.VideoCapture(0)
if not camera.isOpened():
    print("Warning: Camera not available. Video processing will be disabled or use a fallback image.")

# Initialize the Roboflow Inference Client
CLIENT = InferenceHTTPClient(
    api_url="https://detect.roboflow.com",
    api_key="IkQtIl5NGRTc0llwyIMo"
)

# Detection settings
ALERT_INTERVAL = 300  # seconds between alerts
last_alert_time = 0

# Define the classes for this project
PROJECT_CLASSES = [
    "Balls", "Bird", "Cat", "Dog", "Elephant", "Pig", "Tikus",
    "apple", "bean", "bunny", "cattle", "cute", "leopard", "lion",
    "rat", "standpig", "tiger"
]

# Set your location (address or a Google Maps URL)
SITE_LOCATION = "1234 Main St, City, Country"

def make_call():
    """Initiate a call using Twilio."""
    try:
        account_sid = "AC3988de38b87b0de231ee7704d9e6dafb"
        auth_token = "2a282eeb0a72c2a2bec9a1331d3cc803"
        client = Client(account_sid, auth_token)
        call = client.calls.create(
            url="http://demo.twilio.com/docs/voice.xml",
            to="+918999094929",
            from_="+19046820459"
        )
        print("Call initiated. Call SID:", call.sid)
    except Exception as e:
        print("Error making call:", str(e))

def send_telegram_message(image, caption):
    """Send an alert image with caption via Telegram."""
    try:
        TOKEN = "7289300782:AAF0qzc38BQ1S5a4kyXj7F02kUjIswb1YDY"
        CHAT_ID = "6186075118"
        send_photo_url = f"https://api.telegram.org/bot{TOKEN}/sendPhoto"
        
        ret, buffer = cv2.imencode('.jpg', image)
        if not ret:
            print("Failed to encode image.")
            return
        
        files = {"photo": ("alert.jpg", buffer.tobytes(), "image/jpeg")}
        data = {"chat_id": CHAT_ID, "caption": caption}
        response = requests.post(send_photo_url, data=data, files=files)
        
        if response.status_code == 200:
            print("Telegram alert sent.")
        else:
            print("Failed to send Telegram alert. Status code:", response.status_code)
    except Exception as e:
        print("Error sending Telegram message:", str(e))

def play_siren():
    """Play a siren sound alert using pygame."""
    try:
        pygame.mixer.init()
        pygame.mixer.music.load("alarm_tune.mp3")
        pygame.mixer.music.play()
    except Exception as e:
        print("Error playing siren:", str(e))

def gen_frames():
    """Generate video frames with object detection overlays."""
    global last_alert_time

    while True:
        success, frame = camera.read()
        if not success:
            print("Error: Unable to read from camera.")
            break

        # Save frame for inference
        image_path = "temp_frame.jpg"
        cv2.imwrite(image_path, frame)

        # Perform object detection using Roboflow
        result = CLIENT.infer(image_path, model_id="animal-detection-yolov8/1")
        predictions = result.get('predictions', [])

        # Check for objects in our project classes
        detected_object = any(obj['class'] in PROJECT_CLASSES for obj in predictions)

        # Draw detections on the frame
        for obj in predictions:
            x, y, w, h = int(obj['x']), int(obj['y']), int(obj['width']), int(obj['height'])
            cv2.rectangle(frame, (x - w // 2, y - h // 2), (x + w // 2, y + h // 2), (0, 255, 0), 2)
            cv2.putText(frame, obj['class'], (x, y - 10),
                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)

        # Trigger alert if an object is detected and sufficient time has passed
        current_time = time.time()
        if detected_object and (current_time - last_alert_time >= ALERT_INTERVAL):
            detected_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            detected_labels = {obj['class'] for obj in predictions if obj['class'] in PROJECT_CLASSES}
            caption = (
                f"Alert! Detected: {', '.join(detected_labels)}\n"
                f"Time: {detected_time}\n"
                f"Location: {SITE_LOCATION}"
            )
            threading.Thread(target=make_call).start()
            threading.Thread(target=send_telegram_message, args=(frame.copy(), caption)).start()
            threading.Thread(target=play_siren).start()
            last_alert_time = current_time

        ret, buffer = cv2.imencode('.jpg', frame)
        if not ret:
            continue
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n')

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/video_feed')
def video_feed():
    return Response(gen_frames(), mimetype='multipart/x-mixed-replace; boundary=frame')

if __name__ == '__main__':
    # Run on all interfaces to be accessible outside the container
    app.run(debug=True, host='0.0.0.0')