Update app.py

app.py

@@ -42,6 +42,8 @@ try:
     from services.plantation.plant_count import process_plants
     from services.plantation.plant_health import process_plant_health
     from services.plantation.missing_patch_check import process_missing_patches
+    # General object detection
+    from services.object_detection import detect_objects
 except ImportError as e:
     print(f"Failed to import service modules: {str(e)}")
     logging.error(f"Import error: {str(e)}")
@@ -64,7 +66,7 @@ last_frame: Optional[np.ndarray] = None
 last_metrics: Dict[str, Any] = {}
 last_timestamp: str = ""
 detected_plants: List[str] = []  # For plants and missing patches
-detected_issues: List[str] = []  # For cracks, holes, and missing patches
+detected_issues: List[str] = []  # For cracks, holes, and other issues
 gps_coordinates: List[List[float]] = []
 media_loaded: bool = False
 active_service: Optional[str] = None
@@ -167,30 +169,28 @@ def set_active_service(
     pl_val: bool
 ) -> Tuple[Optional[str], str]:
     global active_service
-    toggles = {
-        "under_construction": uc_val,
-        "operations_maintenance": om_val,
-        "road_safety": rs_val,
-        "plantation": pl_val
-    }
-    
-    active_count = sum(toggles.values())
-    if active_count > 1:
-        log_entries.append("Error: Only one service category can be active at a time.")
-        logging.error("Multiple service categories enabled simultaneously.")
-        return None, "Error: Please enable only one service category at a time."
-
-    for service, enabled in toggles.items():
-        if enabled:
-            active_service = service
-            log_entries.append(f"{service.replace('_', ' ').title()} Services Enabled")
-            logging.info(f"{service} services enabled")
-            return active_service, f"{service.replace('_', ' ').title()} Services: Enabled"
-    
-    active_service = None
-    log_entries.append("No service category enabled.")
-    logging.info("No service category enabled.")
-    return None, "No Service Category Enabled"
+    # Enable all requested services
+    enabled_services = []
+    if uc_val:
+        enabled_services.append("under_construction")
+    if om_val:
+        enabled_services.append("operations_maintenance")
+    if rs_val:
+        enabled_services.append("road_safety")
+    if pl_val:
+        enabled_services.append("plantation")
+
+    if not enabled_services:
+        active_service = None
+        log_entries.append("No service category enabled.")
+        logging.info("No service category enabled.")
+        return None, "No Service Category Enabled"
+
+    # Since multiple services are requested, we'll process all enabled services
+    active_service = "all_enabled"  # Custom state to process all enabled services
+    log_entries.append(f"Enabled services: {', '.join(enabled_services)}")
+    logging.info(f"Enabled services: {', '.join(enabled_services)}")
+    return active_service, f"Enabled: {', '.join([s.replace('_', ' ').title() for s in enabled_services])}"
 
 def generate_line_chart() -> Optional[str]:
     if not detected_counts:
@@ -291,31 +291,35 @@ def monitor_feed() -> Tuple[
         thermal_flag = False
 
         try:
-            # Process frame based on active service
-            if active_service == "under_construction":
-                earthwork_dets, detection_frame = process_earthwork(detection_frame)
-                culvert_dets, detection_frame = process_culverts(detection_frame)
-                bridge_pier_dets, detection_frame = process_bridge_piers(detection_frame)
-                all_detected_items.extend(earthwork_dets + culvert_dets + bridge_pier_dets)
-            elif active_service == "operations_maintenance":
-                crack_hole_dets, detection_frame = detect_cracks_and_holes(detection_frame)
-                pothole_dets, detection_frame = process_potholes(detection_frame)
-                signage_dets, detection_frame = process_signages(detection_frame)
-                all_detected_items.extend(crack_hole_dets + pothole_dets + signage_dets)
-            elif active_service == "road_safety":
-                barrier_dets, detection_frame = process_barriers(detection_frame)
-                lighting_dets, detection_frame = process_lighting(detection_frame)
-                accident_dets, detection_frame = process_accident_spots(detection_frame)
-                pothole_crack_dets, detection_frame = detect_potholes_and_cracks(detection_frame)
-                all_detected_items.extend(barrier_dets + lighting_dets + accident_dets + pothole_crack_dets)
-            elif active_service == "plantation":
-                plant_dets, detection_frame = process_plants(detection_frame)
-                health_dets, detection_frame = process_plant_health(detection_frame)
-                missing_dets, detection_frame = process_missing_patches(detection_frame)
-                all_detected_items.extend(plant_dets + health_dets + missing_dets)
-            else:
-                generic_dets, detection_frame = process_generic(detection_frame)
-                all_detected_items.extend(generic_dets)
+            # Process all enabled services
+            # Under Construction Services
+            earthwork_dets, detection_frame = process_earthwork(detection_frame)
+            culvert_dets, detection_frame = process_culverts(detection_frame)
+            bridge_pier_dets, detection_frame = process_bridge_piers(detection_frame)
+            all_detected_items.extend(earthwork_dets + culvert_dets + bridge_pier_dets)
+
+            # Operations Maintenance Services
+            crack_hole_dets, detection_frame = detect_cracks_and_holes(detection_frame)
+            pothole_dets, detection_frame = process_potholes(detection_frame)
+            signage_dets, detection_frame = process_signages(detection_frame)
+            all_detected_items.extend(crack_hole_dets + pothole_dets + signage_dets)
+
+            # Road Safety Services
+            barrier_dets, detection_frame = process_barriers(detection_frame)
+            lighting_dets, detection_frame = process_lighting(detection_frame)
+            accident_dets, detection_frame = process_accident_spots(detection_frame)
+            pothole_crack_dets, detection_frame = detect_potholes_and_cracks(detection_frame)
+            all_detected_items.extend(barrier_dets + lighting_dets + accident_dets + pothole_crack_dets)
+
+            # Plantation Services
+            plant_dets, detection_frame = process_plants(detection_frame)
+            health_dets, detection_frame = process_plant_health(detection_frame)
+            missing_dets, detection_frame = process_missing_patches(detection_frame)
+            all_detected_items.extend(plant_dets + health_dets + missing_dets)
+
+            # General Object Detection (cars, bikes, humans, dogs, etc.)
+            object_dets, detection_frame = detect_objects(detection_frame)
+            all_detected_items.extend(object_dets)
 
             # Apply shadow detection
             try:
@@ -348,7 +352,7 @@ def monitor_feed() -> Tuple[
                         int(box[3] * scale_y)
                     ]
 
-            # Overlay detections on the original frame
+            # Overlay detections on the original frame with specified colors
             for item in all_detected_items:
                 box = item.get("box", [])
                 if not box:
@@ -356,17 +360,38 @@ def monitor_feed() -> Tuple[
                 x_min, y_min, x_max, y_max = box
                 label = item.get("label", "")
                 dtype = item.get("type", "")
-                # Assign colors based on detection type
+                health = item.get("health", "")  # For plant health
+
+                # Assign colors based on detection type as per requirements
                 if dtype == "plant":
-                    color = (0, 255, 0)  # Green for plants
-                elif dtype == "crack":
-                    color = (255, 0, 0)  # Red for cracks
-                elif dtype == "hole":
-                    color = (0, 0, 255)  # Blue for holes
+                    color = (255, 0, 0)  # Blue mark for plant count
+                    if health == "healthy":
+                        color = (255, 165, 0)  # Orange mark for healthy plants
                 elif dtype == "missing_patch":
-                    color = (255, 165, 0)  # Orange for missing patches
+                    color = (0, 0, 255)  # Red mark for missing patches
+                elif dtype == "earthwork":
+                    color = (255, 105, 180)  # Pink for earthwork
+                elif dtype == "culvert":
+                    color = (0, 165, 255)  # Blue and orange mix (approximated)
+                elif dtype == "bridge_pier":
+                    color = (255, 99, 71)  # Light red for bridge piers
+                elif dtype == "pothole" or dtype == "hole":
+                    color = (255, 0, 0)  # Red for potholes (from pothole_detection and pothole_crack_detection)
+                elif dtype == "crack":
+                    color = (255, 105, 180)  # Pink for cracks
+                elif dtype == "signage":
+                    color = (255, 255, 0)  # Yellow for signage
+                elif dtype == "car":
+                    color = (128, 0, 128)  # Purple for cars
+                elif dtype == "bike":
+                    color = (0, 255, 255)  # Cyan for bikes
+                elif dtype == "person":
+                    color = (0, 255, 0)  # Green for humans
+                elif dtype == "dog":
+                    color = (139, 69, 19)  # Brown for dogs
                 else:
-                    color = (255, 255, 0)  # Yellow for others
+                    color = (255, 255, 255)  # White for other objects
+
                 cv2.rectangle(frame, (x_min, y_min), (x_max, y_max), color, 2)
                 cv2.putText(frame, label, (x_min, y_min - 10),
                             cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
@@ -380,7 +405,7 @@ def monitor_feed() -> Tuple[
 
         except Exception as e:
             log_entries.append(f"Processing Error: {str(e)}")
-            logging.error(f"Processing error in {active_service}: {str(e)}")
+            logging.error(f"Processing error: {str(e)}")
             all_detected_items = []
 
         # Update detection metrics
@@ -408,7 +433,7 @@ def monitor_feed() -> Tuple[
                         detected_plants.append(captured_frame_path)
                         if len(detected_plants) > 100:
                             detected_plants.pop(0)
-                    elif dtype in ["crack", "hole", "missing_patch"]:
+                    else:
                         detected_issues.append(captured_frame_path)
                         if len(detected_issues) > 100:
                             detected_issues.pop(0)
@@ -446,16 +471,21 @@ def monitor_feed() -> Tuple[
 
         # Update global variables
         frame_count += 1
-        last_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        last_timestamp = datetime.now().strftime("%Y-%m-d %H:%M:%S")
         last_frame = frame.copy()
         last_metrics = metrics
 
         # Track detections for metrics
         plant_detected = len([item for item in all_detected_items if item.get("type") == "plant"])
         crack_detected = len([item for item in all_detected_items if item.get("type") == "crack"])
-        hole_detected = len([item for item in all_detected_items if item.get("type") == "hole"])
+        hole_detected = len([item for item in all_detected_items if item.get("type") == "hole" or item.get("type") == "pothole"])
         missing_detected = len([item for item in all_detected_items if item.get("type") == "missing_patch"])
-        detected_counts.append(plant_detected + crack_detected + hole_detected + missing_detected)
+        car_detected = len([item for item in all_detected_items if item.get("type") == "car"])
+        bike_detected = len([item for item in all_detected_items if item.get("type") == "bike"])
+        person_detected = len([item for item in all_detected_items if item.get("type") == "person"])
+        dog_detected = len([item for item in all_detected_items if item.get("type") == "dog"])
+        detected_counts.append(plant_detected + crack_detected + hole_detected + missing_detected +
+                              car_detected + bike_detected + person_detected + dog_detected)
 
         # Log frame processing details in the requested format
         processing_time = time.time() - start_time
@@ -466,6 +496,10 @@ def monitor_feed() -> Tuple[
             "cracks": crack_detected,
             "holes": hole_detected,
             "missing_patches": missing_detected,
+            "cars": car_detected,
+            "bikes": bike_detected,
+            "persons": person_detected,
+            "dogs": dog_detected,
             "gps": gps_coord,
             "processing_time_ms": processing_time * 1000
         }
@@ -485,7 +519,7 @@ def monitor_feed() -> Tuple[
     cv2.putText(frame, f"{last_timestamp}", (10, 50), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
 
     # Generate map
-    map_items = [item for item in last_metrics.get("items", []) if item.get("type") in ["crack", "hole", "missing_patch"]]
+    map_items = [item for item in last_metrics.get("items", []) if item.get("type") in ["crack", "hole", "pothole", "missing_patch"]]
     map_path = generate_map(gps_coordinates[-5:], map_items)
 
     return (
@@ -503,7 +537,7 @@ with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="green"))
     gr.Markdown(
         """
         # 🛡️ NHAI Drone Road Inspection Dashboard
-        Monitor highway conditions in real-time using drone footage or static images. Select a service category to analyze specific aspects of the road or plantation.
+        Monitor highway conditions in real-time using drone footage or static images. All services are enabled as requested.
         """
     )
 
@@ -520,17 +554,17 @@ with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="green"))
 
     with gr.Row():
         with gr.Column():
-            uc_toggle = gr.Checkbox(label="Enable Under Construction Services", value=False)
-            uc_status = gr.Textbox(label="Under Construction Status", value="Disabled", interactive=False)
+            uc_toggle = gr.Checkbox(label="Enable Under Construction Services", value=True)
+            uc_status = gr.Textbox(label="Under Construction Status", value="Enabled", interactive=False)
         with gr.Column():
-            om_toggle = gr.Checkbox(label="Enable Operations Maintenance Services", value=False)
-            om_status = gr.Textbox(label="Operations Maintenance Status", value="Disabled", interactive=False)
+            om_toggle = gr.Checkbox(label="Enable Operations Maintenance Services", value=True)
+            om_status = gr.Textbox(label="Operations Maintenance Status", value="Enabled", interactive=False)
         with gr.Column():
-            rs_toggle = gr.Checkbox(label="Enable Road Safety Services", value=False)
-            rs_status = gr.Textbox(label="Road Safety Status", value="Disabled", interactive=False)
+            rs_toggle = gr.Checkbox(label="Enable Road Safety Services", value=True)
+            rs_status = gr.Textbox(label="Road Safety Status", value="Enabled", interactive=False)
         with gr.Column():
-            pl_toggle = gr.Checkbox(label="Enable Plantation Services", value=False)
-            pl_status = gr.Textbox(label="Plantation Status", value="Disabled", interactive=False)
+            pl_toggle = gr.Checkbox(label="Enable Plantation Services", value=True)
+            pl_status = gr.Textbox(label="Plantation Status", value="Enabled", interactive=False)
 
     status_text = gr.Markdown("**Status:** 🟢 Ready (Upload a media file to start)")
 
@@ -600,10 +634,10 @@ with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="green"))
 
     def update_toggles(uc_val: bool, om_val: bool, rs_val: bool, pl_val: bool) -> Tuple[str, str, str, str, str]:
         active, status_message = set_active_service("toggle", uc_val, om_val, rs_val, pl_val)
-        uc_status_val = "Enabled" if active == "under_construction" else "Disabled"
-        om_status_val = "Enabled" if active == "operations_maintenance" else "Disabled"
-        rs_status_val = "Enabled" if active == "road_safety" else "Disabled"
-        pl_status_val = "Enabled" if active == "plantation" else "Disabled"
+        uc_status_val = "Enabled" if "under_construction" in status_message.lower() else "Disabled"
+        om_status_val = "Enabled" if "operations_maintenance" in status_message.lower() else "Disabled"
+        rs_status_val = "Enabled" if "road_safety" in status_message.lower() else "Disabled"
+        pl_status_val = "Enabled" if "plantation" in status_message.lower() else "Disabled"
         return (
             uc_status_val, om_status_val, rs_status_val, pl_status_val, status_message
         )