Spaces:
Runtime error
Runtime error
Download app.py from KnockKnockProject/README: direct link, hf CLI and curl.
- Browser
- Download file 15.6 kB
-
https://huggingface.co/spaces/KnockKnockProject/README/resolve/main/app.py
- Command line
-
hf download hf://spaces/KnockKnockProject/README/app.py
-
curl -L -o app.py https://huggingface.co/spaces/KnockKnockProject/README/resolve/main/app.py
15.6 kB
| 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"<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) | |