Blame for app.py

import gradio as gr
from fpdf import FPDF
from datetime import datetime
import zipfile
import os
import requests
import subprocess
from PIL import Image
from recognition import FaceDetection
import re
from report import ReportGenerator
from dotenv import set_key, load_dotenv
import cv2
import time 
base_dir = './temp'  ###เชื่อมตรงนี้จ้าๆๆๆๆๆๆๆ
from camera import ObjectTracker

# def run_external_command(command):
#     try:
#         # Execute the command
#         result = subprocess.run(command, capture_output=True, text=True, shell=True)
#         return result.stdout if result.returncode == 0 else result.stderr
#     except Exception as e:
#         return f"Error: {str(e)}"


def is_invalid_date(date_str):
    # Define a regex pattern for YYYY-MM-DD format
    pattern = r'^\d{4}-(0[1-9]|1[0-2])$'
    return re.match(pattern, date_str) is None


# Function to generate a PDF with selected date range or single date
def generate_pdf(start_date, end_date):
    try:
        start_valid_date = datetime.strptime(start_date, "%Y-%m")
        if end_date.strip() == "":
            end_valid_date = start_valid_date
        else:
            end_valid_date = datetime.strptime(end_date, "%Y-%m")
        
        if start_valid_date >= end_valid_date:
            return None, "Error: Start date must be before end date."
        
        if is_invalid_date(start_date) or is_invalid_date(end_date):
            return None, "Invalid date format. Please use YYYY-MM."
        

        formatted_start_date = start_valid_date.strftime("%b %Y")
        formatted_end_date = end_valid_date.strftime("%b %Y")
        
        report = ReportGenerator(formatted_start_date, formatted_end_date)
        file_name = report.generate()

        return file_name, None  # Return filename and no error
    except ValueError:
        return None, "Invalid date format. Please use YYYY-MM."

#register
def convert_named_string_to_bytes(named_string):
    """Convert NamedString object to bytes."""
    if hasattr(named_string, 'data'):
        return named_string.data  # Return the data directly if it has a data attribute
    return bytes(named_string, 'utf-8')  # Convert to bytes if it's not a NamedString


def convert_named_string_to_bytes(named_string):
    # สมมติว่ามีวิธีการในการแปลง NamedString เป็น bytes
    if hasattr(named_string, 'read'):
        return named_string.read()  # หากเป็นไฟล์ที่สามารถอ่านได้
    else:
        return bytes(named_string, 'utf-8')  # หากเป็น string ธรรมดา

def register_student(name, student_id, image_file=None, zip_file=None):
    try:
        # Convert name to string and student_id to integer
        name = str(name)
        student_id = int(student_id)

        # Validate that both name and student ID are provided
        if not name or not student_id:
            return "Error: Please provide both student name and student ID."

        # Clean the name for safe directory naming
        safe_name = name
        safe_student_id = str(student_id)  # Keep student_id as a string for directory naming

        # Create a directory to store student images
        student_dir = os.path.join(base_dir, f"{safe_student_id}_{safe_name}")
        os.makedirs(student_dir, exist_ok=True)

        # Handle single image upload (JPEG)
        if image_file is not None:
            img = Image.open(image_file)
            # Create the path for the uploaded image
            image_path = os.path.join(student_dir, f"{safe_student_id}_{safe_name}.jpg")
            img.save(image_path)

            process = FaceDetection(capture=img)
            process.manual_regis(student_dir)

        # Handle ZIP file upload
        if zip_file is not None:
            zip_path = os.path.join(student_dir, f"{safe_student_id}_{safe_name}_images.zip")
            with open(zip_path, "wb") as f:
                f.write(convert_named_string_to_bytes(zip_file))  # Convert to bytes before writing

            # Extract ZIP file to the student's directory
            with zipfile.ZipFile(zip_path, 'r') as zip_ref:
                zip_ref.extractall(student_dir)

        return f"Student {name} (ID: {student_id}) registered successfully!"

    except zipfile.BadZipFile:
        return "Error: The uploaded file is not a valid ZIP file."
    except Exception as e:
        return f"Error during registration: {str(e)}"

# การจัดการผลลัพธ์จาก generate_pdf
def handle_pdf(start_date, end_date):
    file_name, error = generate_pdf(start_date, end_date)
    if error:
        return None, error  # Return None and the error message
    return file_name, None  # Return the filename and None for error

# Help function to explain the available commands with examples
def help_function():
    """
    **Available Commands:**
    1. **upload zip**: Upload a ZIP file containing images for face recognition.
       - Example: upload zip my_images.zip
    2. **request zip**: Enter a date range to generate a ZIP file of images of people not wearing helmets.
       - Example: request zip 2023-01 2023-01
    3. **upload video**: Upload a video for object detection (provide the date of the video).
       - Example: upload video my_video.mp4 2023-01
    4. **camera**: Enter the stream URL to configure the camera.
       - Example: camera http://100.86.155.229:8080/video
    5. **pdf**: Generate a PDF for the selected date range.
       - Example: pdf 2023-01 2023-01
    """
    help_text = """
    **Available Commands:**
    1. **upload video**: Upload a video for object detection (provide the date of the video).
       - Example: upload video my_video.mp4 2023-01
    2. **camera**: Enter the stream URL to configure the camera.
       - Example: camera http://100.86.155.229:8080/video
    """
    return help_text


# Function to upload zip file for face recognition
def upload_zip(file):
    try:
        with zipfile.ZipFile(file, 'r') as zip_ref:
            zip_ref.extractall('C:/zoom/AI/gradio/my_images.zip')  # Change the path to your desired directory
        return "ZIP file uploaded and extracted successfully!"
    except Exception as e:
        return str(e)

# Function to generate zip based on date range
def generate_zip_by_date(start_date, end_date):
    try:
        start_valid_date = datetime.strptime(start_date, "%Y-%m-%d")
        end_valid_date = datetime.strptime(end_date, "%Y-%m-%d")
        
        zip_file_name = f"no_helmet_images_{start_date}_to_{end_date}.zip"
        with zipfile.ZipFile(zip_file_name, 'w') as zipf:
            for root, dirs, files in os.walk('/path/to/images'):  # Change this path to your image directory
                for file in files:
                    file_date = datetime.strptime(file.split('_')[0], '%Y-%m-%d')  # Assuming the file has the date
                    if start_valid_date <= file_date <= end_valid_date:
                        zipf.write(os.path.join(root, file))
        return zip_file_name
    except Exception as e:
        return str(e)

# Function to upload video and detect objects
def upload_video(file, date):
    return f"Video from {date} processed successfully!"

# Function to set IP for the camera
def set_camera_ip(ip_address):
    env_file = ".env"
    set_key(env_file, "CAMERA_URL", ip_address)
    load_dotenv(override=True)
    print(ip_address)
    return f"Camera IP set to {os.getenv('CAMERA_URL')}"


def chatbot_response(user_input, chat_history):
    if user_input.lower().startswith("help"):
        # แยกคำสั่งและพารามิเตอร์
        parts = user_input.split()
        response = help_function()
    elif user_input.lower().startswith("camera"):
        # แยกคำสั่งและพารามิเตอร์
        parts = user_input.split()
        if len(parts) == 2:
            ip_address = parts[1]
            response = set_camera_ip(ip_address)
        else:
            response = "Error: Please provide camera url."
    
        
    else:
        response = "Unknown command. Type 'help' to see the available commands."

    # อัปเดตประวัติการสนทนา
    chat_history.append((user_input, response))
    
    return chat_history, chat_history

streaming = False

def get_frame():
    """Generator function to yield frames from the IP camera."""
    global streaming
    cap = cv2.VideoCapture(str(os.getenv('CAMERA_URL')))  # Open the video stream

    while streaming:
        ret, frame = cap.read()  # Read a frame
        if not ret:
            break  # Stop if no more frames

        # Resize the frame (optional)
        frame = cv2.resize(frame, (640, 480))

        # Convert the frame from BGR to RGB
        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

        yield frame  # Yield the frame for Gradio to display

        time.sleep(0.03)
        
    cap.release()  # Release the video stream


tracker = ObjectTracker()

# Function to explore a directory and list its contents recursively with indentation
def explore_directory(path, depth=0):
    try:
        # Get the list of files and directories
        contents = os.listdir(path)
        items = []

        for item in contents:
            item_path = os.path.join(path, item)
            indent = "&nbsp;" * (depth * 4)  # Indentation for each depth level using non-breaking spaces
            
            # Append directories with expandable details
            if os.path.isdir(item_path):
                nested_contents = explore_directory(item_path, depth + 1)  # Recursively explore subdirectories
                items.append(f"<details><summary>{indent}📁 {item}</summary>{nested_contents}</details>")
            else:
                if "./faces/__faces__/" in item_path:
                    item_path = item_path[18:]
                elif "./faces/__full_pics__/" in item_path:
                    item_path = item_path[22:]
                items.append(f'{indent}📄 {item} <span style="opacity: 0.5; margin-left:30px">path: {item_path}</span>')  # Add file with icon

        return "<br>".join(items)
    except Exception as e:
        return f"Error: {str(e)}"  # Return error message if any

# Main function to handle path input and explore
def on_explore(path="./faces"):
    return explore_directory(path)

def show_image(path):
    # Read the image from the file path
    img_Full = cv2.imread(f"./faces/__faces__/{path}")
    img_head = cv2.imread(f"./faces/__full_pics__/{path}")
    # Convert the image from BGR (OpenCV default) to RGB
    img_Full = cv2.cvtColor(img_Full, cv2.COLOR_BGR2RGB)
    img_head = cv2.cvtColor(img_head, cv2.COLOR_BGR2RGB)
    return img_Full,img_head


# Gradio Interface
with gr.Blocks() as demo:
    with gr.Tab("Home"):
        gr.Markdown("### Welcome to the Command Center")
        gr.Markdown("**Type 'help' to see the list of available commands.**")
        
        #chatbot_widget = gr.Chatbot(label="Chat with the bot")
        chatbot = gr.Chatbot()
        user_input = gr.Textbox(label="Type your command here", placeholder="Enter your command...")
        

    with gr.Tab("Generate PDF"):
        date_input_start = gr.Textbox(label="Start Date (YYYY-MM)", placeholder="YYYY-MM")
        date_input_end = gr.Textbox(label="End Date (Leave blank for same as start date)", placeholder="YYYY-MM")
        pdf_output = gr.File(label="Generated PDF")
        pdf_btn = gr.Button("Generate PDF")

        # Generate PDF when the button is clicked
        pdf_btn.click(fn=handle_pdf, 
                      inputs=[date_input_start, date_input_end], 
                      outputs=[pdf_output, gr.Textbox(label="Error Message", visible=True)])

    with gr.Tab("Live Stream"):
        #ip_camera_url = "http://100.86.155.229:8080/video"  # เปลี่ยน URL ตามที่อยู่ IP ของกล้อง
        #live_image_output = gr.Image(label="Live Stream", interactive=False)
        start_button = gr.Button("Start/Stop Stream")

        def toggle():
            global streaming
            streaming = not streaming

        start_button.click(toggle)
        stream = gr.Interface(
            fn=get_frame,  # Function to fetch the latest frame
            inputs=[], 
            outputs=gr.Image(type="numpy", label="Live IP Camera Stream"),
            live=True  # Enable continuous frame updates
        )

        markdown_content = """
        <div style="text-align: center;">
        <h2>Or upload video</h2>
        </div>
        """
        gr.Markdown(markdown_content)

        start_button = gr.Button("Stop Process")

        def toggle():
            global tracker
            try:
                tracker.start = False
            except: pass

        start_button.click(toggle)
        
        with gr.Row():
            #    video_input = gr.File(label="Upload your video", file_types=[".mp4", ".avi", ".mov"])
            
            #video_output = gr.Video(label="Processed Video Output")
            video_output = gr.Interface(
                fn=tracker.detect_and_track_2,  # Function to fetch the latest frame
                inputs="video",
                outputs=gr.Image(type="numpy", label="Process"),
                live=False,  # Enable continuous frame updates
                flagging_mode = "never"
            )

        #video_input.change(process_video, inputs=video_input, outputs=video_output)

        # กด Enter เพื่อส่งข้อความ
        user_input.submit(chatbot_response, 
                      inputs=[user_input, chatbot], 
                      outputs=[chatbot, chatbot])


    # Adding Register tab
    with gr.Tab("Register"):
        gr.Markdown("### Register student")
        name_input = gr.Textbox(label="Student Name", placeholder="Enter student name")
        student_id_input = gr.Textbox(label="Student ID", placeholder="Enter student ID")
        image_input = gr.File(label="Upload a image or a Zip file", file_types=["image", ".jpg", ".zip"], file_count="single")
        register_button = gr.Button("Register Student")
        register_output = gr.Textbox(label="Registration Output", placeholder="Result will be shown here")

        # Register student when button is clicked
        register_button.click(fn=register_student, 
                            inputs=[name_input, student_id_input, image_input], 
                            outputs=[register_output])


    # Gradio interface
    with gr.Tab("File explorer"):
        gr.Markdown("# File Explorer")
        # path_input = gr.Textbox(label="Enter Directory Path", placeholder="e.g., C:/Users/YourUsername/Documents")
        output_display = gr.HTML(label="Contents")
        explore_button = gr.Button("Explore")

        # Set up button clicks
        explore_button.click(fn=on_explore, outputs=output_display)
        # explore_button.click(fn=on_explore, inputs=path_input, outputs=output_display)

        interface = gr.Interface(
            fn=show_image,  # Function to display the image
            inputs=gr.Textbox(label="Enter Directory Path", placeholder="e.g., name_of_folder/name_of_image"),    
            outputs= [gr.Image(label="Head"), 
                      gr.Image(label="Full image")], # Output type is an image
            flagging_mode = "never"
        )
        

load_dotenv(override=True)

# Launch the Gradio app
demo.launch(share=True)