# ================================================================ # services/data/app/external/madrid_opendata.py # ================================================================ """Madrid Open Data API client for traffic and events""" import math from typing import List, Dict, Any, Optional from datetime import datetime, timedelta import structlog from app.external.base_client import BaseAPIClient from app.core.config import settings logger = structlog.get_logger() class MadridOpenDataClient(BaseAPIClient): def __init__(self): super().__init__( base_url="https://datos.madrid.es/egob/catalogo", api_key=settings.MADRID_OPENDATA_API_KEY ) async def get_current_traffic(self, latitude: float, longitude: float) -> Optional[Dict[str, Any]]: """Get current traffic data for location""" try: # In production, this would call real Madrid Open Data API # For now, generate realistic synthetic data return await self._generate_synthetic_traffic(latitude, longitude) except Exception as e: logger.error("Failed to get current traffic", error=str(e)) return None async def get_historical_traffic(self, latitude: float, longitude: float, start_date: datetime, end_date: datetime) -> List[Dict[str, Any]]: """Get historical traffic data""" try: # Generate synthetic historical traffic data return await self._generate_historical_traffic(latitude, longitude, start_date, end_date) except Exception as e: logger.error("Failed to get historical traffic", error=str(e)) return [] async def get_events(self, latitude: float, longitude: float, radius_km: float = 5.0) -> List[Dict[str, Any]]: """Get events near location""" try: # In production, would fetch real events from Madrid Open Data return await self._generate_synthetic_events(latitude, longitude) except Exception as e: logger.error("Failed to get events", error=str(e)) return [] async def _generate_synthetic_traffic(self, latitude: float, longitude: float) -> Dict[str, Any]: """Generate realistic Madrid traffic data""" now = datetime.now() hour = now.hour is_weekend = now.weekday() >= 5 # Base traffic volume base_traffic = 100 # Madrid traffic patterns if not is_weekend: # Weekdays if 7 <= hour <= 9: # Morning rush traffic_multiplier = 2.2 congestion = "high" elif 18 <= hour <= 20: # Evening rush traffic_multiplier = 2.5 congestion = "high" elif 12 <= hour <= 14: # Lunch time traffic_multiplier = 1.6 congestion = "medium" elif 6 <= hour <= 22: # Daytime traffic_multiplier = 1.2 congestion = "medium" else: # Night traffic_multiplier = 0.4 congestion = "low" else: # Weekends if 11 <= hour <= 14: # Weekend shopping traffic_multiplier = 1.4 congestion = "medium" elif 19 <= hour <= 22: # Weekend evening traffic_multiplier = 1.6 congestion = "medium" else: traffic_multiplier = 0.8 congestion = "low" # Calculate pedestrian traffic (higher during meal times and school hours) pedestrian_base = 150 if 13 <= hour <= 15: # Lunch time pedestrian_multiplier = 2.8 elif hour == 14: # School pickup time pedestrian_multiplier = 3.5 elif 20 <= hour <= 22: # Dinner time pedestrian_multiplier = 2.2 elif 8 <= hour <= 9: # Morning commute pedestrian_multiplier = 2.0 else: pedestrian_multiplier = 1.0 traffic_volume = int(base_traffic * traffic_multiplier) pedestrian_count = int(pedestrian_base * pedestrian_multiplier) # Average speed based on congestion speed_map = {"low": 45, "medium": 25, "high": 15} average_speed = speed_map[congestion] + (hash(f"{latitude}{longitude}") % 10 - 5) return { "date": now, "traffic_volume": traffic_volume, "pedestrian_count": pedestrian_count, "congestion_level": congestion, "average_speed": max(10, average_speed), # Minimum 10 km/h "source": "madrid_opendata" } async def _generate_historical_traffic(self, latitude: float, longitude: float, start_date: datetime, end_date: datetime) -> List[Dict[str, Any]]: """Generate synthetic historical traffic data""" historical_data = [] current_date = start_date while current_date <= end_date: hour = current_date.hour is_weekend = current_date.weekday() >= 5 # Base patterns similar to current traffic base_traffic = 100 if not is_weekend: if 7 <= hour <= 9 or 18 <= hour <= 20: traffic_multiplier = 2.0 + (current_date.day % 5) * 0.1 elif 12 <= hour <= 14: traffic_multiplier = 1.5 else: traffic_multiplier = 1.0 else: traffic_multiplier = 0.7 + (current_date.day % 3) * 0.2 # Add seasonal variations month = current_date.month seasonal_factor = 1.0 if month in [12, 1]: # Holiday season seasonal_factor = 0.8 elif month in [7, 8]: # Summer vacation seasonal_factor = 0.9 traffic_volume = int(base_traffic * traffic_multiplier * seasonal_factor) # Determine congestion level if traffic_volume > 160: congestion_level = "high" avg_speed = 15 elif traffic_volume > 120: congestion_level = "medium" avg_speed = 25 else: congestion_level = "low" avg_speed = 40 # Pedestrian count pedestrian_base = 150 if 13 <= hour <= 15: pedestrian_multiplier = 2.5 elif hour == 14: pedestrian_multiplier = 3.0 else: pedestrian_multiplier = 1.0 historical_data.append({ "date": current_date, "traffic_volume": traffic_volume, "pedestrian_count": int(pedestrian_base * pedestrian_multiplier), "congestion_level": congestion_level, "average_speed": avg_speed + (current_date.day % 10 - 5), "source": "madrid_opendata" }) current_date += timedelta(hours=1) return historical_data async def _generate_synthetic_events(self, latitude: float, longitude: float) -> List[Dict[str, Any]]: """Generate synthetic Madrid events""" events = [] base_date = datetime.now().date() # Generate some sample events sample_events = [ { "name": "Mercado de San Miguel", "type": "market", "impact_level": "medium", "distance_km": 1.2 }, { "name": "Concierto en el Retiro", "type": "concert", "impact_level": "high", "distance_km": 2.5 }, { "name": "Partido Real Madrid", "type": "sports", "impact_level": "high", "distance_km": 8.0 } ] for i, event in enumerate(sample_events): event_date = base_date + timedelta(days=i + 1) events.append({ "id": f"event_{i+1}", "name": event["name"], "date": datetime.combine(event_date, datetime.min.time()), "type": event["type"], "impact_level": event["impact_level"], "distance_km": event["distance_km"], "latitude": latitude + (hash(event["name"]) % 100 - 50) / 1000, "longitude": longitude + (hash(event["name"]) % 100 - 50) / 1000, "source": "madrid_opendata" }) return events