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 = " " * (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"
{indent}📁 {item}{nested_contents}
") 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} path: {item_path}') # Add file with icon return "
".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 = """

Or upload video

""" 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)