#!/usr/bin/env python3 """ Test OSM processing with a small sample to verify it works. """ import json import asyncio import xml.etree.ElementTree as ET from pathlib import Path from datetime import datetime import sys import math # Add the app directory to Python path sys.path.append(str(Path(__file__).parent.parent)) from app.services.street_tree_service import StreetTreeService from app.services.berlin_data_service import BerlinDataService async def test_processing(): """Test the processing with a small sample.""" print("๐Ÿงช Testing OSM processing with sample data...") # Initialize services tree_service = StreetTreeService() berlin_data = BerlinDataService() # Parse OSM file and get first 5 green spaces as test osm_file = Path("app/data/osm-raw/berlin_green_spaces.osm") if not osm_file.exists(): print("โŒ OSM file not found") return tree = ET.parse(osm_file) root = tree.getroot() ways = root.findall('.//way') print(f"๐Ÿ“Š Found {len(ways)} total ways in OSM file") # Process first 5 green spaces as test sample_spaces = [] processed_count = 0 for way in ways: if processed_count >= 5: break # Get tags tags = {} for tag in way.findall('tag'): tags[tag.get('k')] = tag.get('v') # Check if it's a green space green_space_type = None leisure = tags.get('leisure', '') landuse = tags.get('landuse', '') natural = tags.get('natural', '') if leisure in ['park', 'garden', 'nature_reserve']: green_space_type = leisure elif landuse in ['forest', 'grass', 'park']: green_space_type = landuse elif natural in ['forest', 'wood']: green_space_type = natural if not green_space_type: continue # Get coordinates from first and last node to estimate center nd_refs = [nd.get('ref') for nd in way.findall('nd')] if len(nd_refs) < 3: continue # Find first node coordinates first_node = root.find(f".//node[@id='{nd_refs[0]}']") if first_node is None: continue lat = float(first_node.get('lat')) lng = float(first_node.get('lon')) # Simple space data space_data = { 'id': f"test_{way.get('id')}", 'name': tags.get('name', f"Test {green_space_type} {processed_count + 1}"), 'fclass': green_space_type, 'lat': lat, 'lng': lng, 'area_sqm': 5000, # Default for test 'district': 'Test District' } sample_spaces.append(space_data) processed_count += 1 print(f"๐ŸŒณ Testing with {len(sample_spaces)} sample green spaces...") # Test enhancement with real data enhanced_spaces = [] for i, space_data in enumerate(sample_spaces, 1): print(f"\n[{i}/{len(sample_spaces)}] Testing {space_data['name']}...") try: # Get real tree data tree_response = await tree_service.get_trees_near_location( space_data['lat'], space_data['lng'], radius_m=200 ) # Get real toilet data nearby_toilets = await berlin_data.get_toilets_near_point( space_data['lat'], space_data['lng'], 500 ) # Create enhanced data enhanced_space = { "id": space_data['id'], "name": space_data['name'], "type": "PARK", "coordinates": { "lat": space_data['lat'], "lng": space_data['lng'] }, "tree_data": { "total_trees": tree_response.metrics.total_trees, "species_count": len(tree_response.metrics.dominant_species), "dominant_species": tree_response.metrics.dominant_species }, "toilet_accessibility": { "nearby_toilets_count": len(nearby_toilets), "nearest_distance_m": nearby_toilets[0]['distance_meters'] if nearby_toilets else None } } enhanced_spaces.append(enhanced_space) trees = tree_response.metrics.total_trees toilets = len(nearby_toilets) print(f"โœ… Success: {trees} trees, {toilets} toilets nearby") except Exception as e: print(f"โŒ Error: {e}") # Save test results output_file = Path("app/data/processed/test_green_spaces.json") test_data = { "test_results": enhanced_spaces, "total_tested": len(enhanced_spaces), "osm_ways_available": len(ways), "processing_successful": True, "timestamp": datetime.now().isoformat() } with open(output_file, 'w') as f: json.dump(test_data, f, indent=2) print(f"\n๐ŸŽ‰ Test completed successfully!") print(f"๐Ÿ“ Test results saved: {output_file}") print(f"๐Ÿ“Š Enhanced {len(enhanced_spaces)} sample spaces") print(f"๐Ÿ’ก Ready to process all {len(ways)} green spaces!") return True if __name__ == "__main__": asyncio.run(test_processing())