236 lines
8.9 KiB
Python
236 lines
8.9 KiB
Python
# ================================================================
|
|
# 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
|