Update app.py

app.py

@@ -15,6 +15,7 @@ import ultralytics
 import time
 import piexif
 import zipfile
+from fpdf import FPDF  # PDF generation
 
 # Set YOLO config directory
 os.environ["YOLO_CONFIG_DIR"] = "/tmp/Ultralytics"
@@ -56,13 +57,58 @@ print(f"Ultralytics version: {ultralytics.__version__}")
 print(f"CUDA available: {torch.cuda.is_available()}")
 
 # Load custom YOLO model
-device = "cuda" if torch.cuda.is_available() else "cpu"
+device = "cpu"  # Force the use of CPU
 print(f"Using device: {device}")
-model = YOLO('./data/best.pt').to(device)
-if device == "cuda":
-    model.half()
+model = YOLO('./data/best.pt').to(device)  # Model is now moved to the CPU
+if device == "cpu":  # Optional check, not strictly needed but can help to ensure
+    model.float()  # Ensure model is running on the CPU
 print(f"Model classes: {model.names}")
 
+# Generate PDF report
+def generate_pdf_report(log_entries, detected_issues, chart_path, map_path, metrics):
+    pdf = FPDF()
+    pdf.set_auto_page_break(auto=True, margin=15)
+    pdf.add_page()
+
+    # Add Title
+    pdf.set_font("Arial", "B", 16)
+    pdf.cell(200, 10, txt="Road Defect Detection Report", ln=True, align="C")
+    pdf.ln(10)
+
+    # Add Log Entries
+    pdf.set_font("Arial", size=12)
+    pdf.cell(200, 10, txt="Log Entries:", ln=True)
+    pdf.multi_cell(0, 10, txt="\n".join(log_entries))
+    pdf.ln(5)
+
+    # Add Detected Issues
+    pdf.cell(200, 10, txt="Detected Issues:", ln=True)
+    for issue in detected_issues:
+        pdf.cell(200, 10, txt=issue, ln=True)
+    pdf.ln(5)
+
+    # Add Metrics
+    pdf.cell(200, 10, txt="Detection Metrics:", ln=True)
+    pdf.multi_cell(0, 10, txt=json.dumps(metrics, indent=2))
+    pdf.ln(5)
+
+    # Add Chart Image
+    if chart_path:
+        pdf.cell(200, 10, txt="Detection Trend Chart:", ln=True)
+        pdf.image(chart_path, x=10, y=pdf.get_y(), w=180)
+        pdf.ln(80)
+
+    # Add Map Image
+    if map_path:
+        pdf.cell(200, 10, txt="Issue Locations Map:", ln=True)
+        pdf.image(map_path, x=10, y=pdf.get_y(), w=180)
+        pdf.ln(80)
+
+    # Save PDF
+    pdf_output_path = os.path.join(OUTPUT_DIR, "detection_report.pdf")
+    pdf.output(pdf_output_path)
+    return pdf_output_path
+
 def zip_directory(folder_path: str, zip_path: str) -> str:
     """Zip all files in a directory."""
     try:
@@ -90,57 +136,6 @@ def generate_map(gps_coords: List[List[float]], items: List[Dict[str, Any]]) ->
     plt.close()
     return map_path
 
-def write_geotag(image_path: str, gps_coord: List[float]) -> bool:
-    try:
-        lat = abs(gps_coord[0])
-        lon = abs(gps_coord[1])
-        lat_ref = "N" if gps_coord[0] >= 0 else "S"
-        lon_ref = "E" if gps_coord[1] >= 0 else "W"
-        exif_dict = piexif.load(image_path) if os.path.exists(image_path) else {"GPS": {}}
-        exif_dict["GPS"] = {
-            piexif.GPSIFD.GPSLatitudeRef: lat_ref,
-            piexif.GPSIFD.GPSLatitude: ((int(lat), 1), (0, 1), (0, 1)),
-            piexif.GPSIFD.GPSLongitudeRef: lon_ref,
-            piexif.GPSIFD.GPSLongitude: ((int(lon), 1), (0, 1), (0, 1))
-        }
-        piexif.insert(piexif.dump(exif_dict), image_path)
-        return True
-    except Exception as e:
-        logging.error(f"Failed to geotag {image_path}: {str(e)}")
-        log_entries.append(f"Error: Failed to geotag {image_path}: {str(e)}")
-        return False
-
-def write_flight_log(frame_count: int, gps_coord: List[float], timestamp: str) -> str:
-    log_path = os.path.join(FLIGHT_LOG_DIR, f"flight_log_{frame_count:06d}.csv")
-    try:
-        with open(log_path, 'w', newline='') as csvfile:
-            writer = csv.writer(csvfile)
-            writer.writerow(["Frame", "Timestamp", "Latitude", "Longitude", "Speed_ms", "Satellites", "Altitude_m"])
-            writer.writerow([frame_count, timestamp, gps_coord[0], gps_coord[1], 5.0, 12, 60])
-        return log_path
-    except Exception as e:
-        logging.error(f"Failed to write flight log {log_path}: {str(e)}")
-        log_entries.append(f"Error: Failed to write flight log {log_path}: {str(e)}")
-        return ""
-
-def check_image_quality(frame: np.ndarray, input_resolution: int) -> bool:
-    height, width, _ = frame.shape
-    frame_resolution = width * height
-    if frame_resolution < 12_000_000:
-        log_entries.append(f"Frame {frame_count}: Resolution {width}x{height} ({frame_resolution/1e6:.2f}MP) below 12MP, non-compliant")
-    if frame_resolution < input_resolution:
-        log_entries.append(f"Frame {frame_count}: Output resolution {width}x{height} below input resolution")
-        return False
-    return True
-
-def update_metrics(detections: List[Dict[str, Any]]) -> Dict[str, Any]:
-    counts = Counter([det["label"] for det in detections])
-    return {
-        "items": [{"type": k, "count": v} for k, v in counts.items()],
-        "total_detections": len(detections),
-        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    }
-
 def generate_line_chart() -> Optional[str]:
     if not detected_counts:
         return None
@@ -361,6 +356,9 @@ def process_video(video, resize_width=4000, resize_height=3000, frame_skip=5):
     images_zip = zip_directory(CAPTURED_FRAMES_DIR, os.path.join(OUTPUT_DIR, "captured_frames.zip"))
     logs_zip = zip_directory(FLIGHT_LOG_DIR, os.path.join(OUTPUT_DIR, "flight_logs.zip"))
 
+    # Generate PDF report
+    pdf_path = generate_pdf_report(log_entries, detected_issues, chart_path, map_path, last_metrics)
+
     return (
         output_path,
         json.dumps(last_metrics, indent=2),
@@ -368,56 +366,8 @@ def process_video(video, resize_width=4000, resize_height=3000, frame_skip=5):
         detected_issues,
         chart_path,
         map_path,
-        submission_json_path,
+        pdf_path,  # PDF report path
         images_zip,
         logs_zip,
         output_path  # For video download
     )
-
-# Gradio interface
-with gr.Blocks(theme=gr.themes.Soft(primary_hue="orange")) as iface:
-    gr.Markdown("# NHAI Road Defect Detection Dashboard")
-    with gr.Row():
-        with gr.Column(scale=3):
-            video_input = gr.Video(label="Upload Video (12MP recommended for NHAI compliance)")
-            width_slider = gr.Slider(320, 4000, value=4000, label="Output Width", step=1)
-            height_slider = gr.Slider(240, 3000, value=3000, label="Output Height", step=1)
-            skip_slider = gr.Slider(1, 10, value=5, label="Frame Skip", step=1)
-            process_btn = gr.Button("Process Video", variant="primary")
-        with gr.Column(scale=1):
-            metrics_output = gr.Textbox(label="Detection Metrics", lines=5, interactive=False)
-    with gr.Row():
-        video_output = gr.Video(label="Processed Video")
-        issue_gallery = gr.Gallery(label="Detected Issues", columns=4, height="auto", object_fit="contain")
-    with gr.Row():
-        chart_output = gr.Image(label="Detection Trend")
-        map_output = gr.Image(label="Issue Locations Map")
-    with gr.Row():
-        logs_output = gr.Textbox(label="Logs", lines=5, interactive=False)
-    with gr.Row():
-        gr.Markdown("## Download Results")
-    with gr.Row():
-        json_download = gr.File(label="Download Data Lake JSON")
-        images_zip_download = gr.File(label="Download Geotagged Images (ZIP)")
-        logs_zip_download = gr.File(label="Download Flight Logs (ZIP)")
-        video_download = gr.File(label="Download Processed Video")
-
-    process_btn.click(
-        fn=process_video,
-        inputs=[video_input, width_slider, height_slider, skip_slider],
-        outputs=[
-            video_output,
-            metrics_output,
-            logs_output,
-            issue_gallery,
-            chart_output,
-            map_output,
-            json_download,
-            images_zip_download,
-            logs_zip_download,
-            video_download
-        ]
-    )
-
-if __name__ == "__main__":
-    iface.launch()