Spaces:
Sleeping
Sleeping
| import os | |
| import sys | |
| import logging | |
| import cv2 | |
| import numpy as np | |
| from pathlib import Path | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| logger = logging.getLogger(__name__) | |
| # Add the app directory to the path so we can import modules | |
| sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from app.services.image_processing import ( | |
| detect_beach_scene, detect_water_scene, detect_plastic_bottles, | |
| detect_plastic_bottles_in_beach, detect_ships, | |
| check_for_plastic_bottle, check_for_ship, check_for_plastic_waste, | |
| detect_general_waste | |
| ) | |
| def analyze_image_regions(image_path): | |
| """ | |
| Analyze different regions of the image for plastic bottles and waste | |
| This is helpful to verify that our detection functions work on the pixel level | |
| """ | |
| logger.info(f"Analyzing regions in: {image_path}") | |
| # Read the image | |
| img = cv2.imread(image_path) | |
| if img is None: | |
| logger.error(f"Could not read image: {image_path}") | |
| return False | |
| height, width = img.shape[:2] | |
| logger.info(f"Image dimensions: {width}x{height}") | |
| # Convert to HSV for color-based detection | |
| hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV) | |
| # Create a copy for drawing results | |
| img_result = img.copy() | |
| # Detect scene type | |
| is_beach = detect_beach_scene(img, hsv) | |
| is_water = detect_water_scene(img, hsv) | |
| scene_type = "unknown" | |
| if is_beach and is_water: | |
| scene_type = "coastal" | |
| elif is_beach: | |
| scene_type = "beach" | |
| elif is_water: | |
| scene_type = "water" | |
| logger.info(f"Scene type: {scene_type}") | |
| # Add scene type text to image | |
| cv2.putText(img_result, f"Scene: {scene_type}", (10, 30), | |
| cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 2) | |
| # Divide the image into a grid of regions for analysis | |
| grid_size = 3 # 3x3 grid | |
| region_width = width // grid_size | |
| region_height = height // grid_size | |
| region_results = [] | |
| # Analyze each region | |
| for i in range(grid_size): | |
| for j in range(grid_size): | |
| # Define region coordinates | |
| x1 = j * region_width | |
| y1 = i * region_height | |
| x2 = min(width, (j + 1) * region_width) | |
| y2 = min(height, (i + 1) * region_height) | |
| # Extract region | |
| region = img[y1:y2, x1:x2] | |
| region_hsv = hsv[y1:y2, x1:x2] | |
| # Check for plastic bottle | |
| is_bottle = check_for_plastic_bottle(region, region_hsv) | |
| # Check for plastic waste | |
| is_waste = check_for_plastic_waste(region, region_hsv) | |
| # Check for ship | |
| is_ship = check_for_ship(region, region_hsv) | |
| # Run general waste detection | |
| has_waste, waste_type, waste_confidence = detect_general_waste(region, region_hsv) | |
| # Color for the grid cell based on results | |
| grid_color = (100, 100, 100) # Default gray | |
| if is_bottle: | |
| grid_color = (0, 0, 255) # Red for bottle | |
| logger.info(f"Region ({i},{j}) contains plastic bottle") | |
| elif is_waste: | |
| grid_color = (0, 165, 255) # Orange for waste | |
| logger.info(f"Region ({i},{j}) contains plastic waste") | |
| elif is_ship: | |
| grid_color = (255, 0, 0) # Blue for ship | |
| logger.info(f"Region ({i},{j}) contains ship") | |
| elif has_waste: | |
| grid_color = (0, 255, 255) # Yellow for general waste | |
| logger.info(f"Region ({i},{j}) contains {waste_type} ({waste_confidence:.2f})") | |
| # Draw grid cell | |
| cv2.rectangle(img_result, (x1, y1), (x2, y2), grid_color, 2) | |
| # Add text with detection results | |
| detection_text = "" | |
| if is_bottle: | |
| detection_text = "Bottle" | |
| elif is_waste: | |
| detection_text = "Waste" | |
| elif is_ship: | |
| detection_text = "Ship" | |
| elif has_waste: | |
| detection_text = waste_type | |
| if detection_text: | |
| cv2.putText(img_result, detection_text, (x1 + 5, y1 + 15), | |
| cv2.FONT_HERSHEY_SIMPLEX, 0.4, grid_color, 1) | |
| # Store region result | |
| region_results.append({ | |
| "region": (i, j), | |
| "is_bottle": is_bottle, | |
| "is_waste": is_waste, | |
| "is_ship": is_ship, | |
| "general_waste": has_waste, | |
| "waste_type": waste_type if has_waste else None, | |
| "waste_confidence": waste_confidence if has_waste else 0.0 | |
| }) | |
| # Save the result | |
| output_dir = Path("test_output/region_analysis") | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| base_name = os.path.basename(image_path) | |
| output_path = output_dir / f"region_{base_name}" | |
| cv2.imwrite(str(output_path), img_result) | |
| logger.info(f"Region analysis result saved to: {output_path}") | |
| return { | |
| "scene_type": scene_type, | |
| "region_results": region_results, | |
| "output_path": str(output_path) | |
| } | |
| def main(): | |
| """Main function to test region-level detection""" | |
| # Test directory | |
| test_dir = "test_files" | |
| # Check if test directory exists | |
| if not os.path.isdir(test_dir): | |
| logger.error(f"Test directory not found: {test_dir}") | |
| return | |
| # Get all image files in the test directory | |
| image_files = [f for f in os.listdir(test_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))] | |
| if not image_files: | |
| logger.error(f"No image files found in {test_dir}") | |
| return | |
| results = {} | |
| # Process each image | |
| for img_file in image_files: | |
| img_path = os.path.join(test_dir, img_file) | |
| results[img_file] = analyze_image_regions(img_path) | |
| # Print summary | |
| logger.info("\n\n--- Region Analysis Summary ---") | |
| for img_file, result in results.items(): | |
| if not result: | |
| continue | |
| logger.info(f"{img_file}:") | |
| logger.info(f" Scene type: {result['scene_type']}") | |
| # Count detections by type | |
| bottles = sum(1 for r in result['region_results'] if r['is_bottle']) | |
| waste = sum(1 for r in result['region_results'] if r['is_waste']) | |
| ships = sum(1 for r in result['region_results'] if r['is_ship']) | |
| general = sum(1 for r in result['region_results'] if r['general_waste']) | |
| logger.info(f" Regions with bottles: {bottles}") | |
| logger.info(f" Regions with plastic waste: {waste}") | |
| logger.info(f" Regions with ships: {ships}") | |
| logger.info(f" Regions with general waste: {general}") | |
| logger.info(f" Output: {result['output_path']}") | |
| logger.info("---") | |
| if __name__ == "__main__": | |
| main() |