import gradio as gr from typing import Any, List, Dict, Optional import httpx import math from pydantic import BaseModel, Field, ValidationError import datetime # Constants DPO_API_BASE = "https://api.data.gov.hk" USER_AGENT = "carpark-app/1.0" class VehicleVacancy(BaseModel): vacancy: Optional[int] = None vacancyEV: Optional[int] = None vacancyDIS: Optional[int] = None lastupdate: Optional[str] = None class CarPark(BaseModel): park_Id: str name: Optional[str] = None displayAddress: Optional[str] = None vacancy: Dict[str, List[VehicleVacancy]] = Field(default_factory=dict) class CarParkApiResponse(BaseModel): results: List[CarPark] async def make_dpo_request(url: str) -> dict[str, Any] | None: """Make a request to the DPO API with proper error handling.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/json" } async with httpx.AsyncClient() as client: try: response = await client.get(url, headers=headers, timeout=30.0) response.raise_for_status() return response.json() except Exception: return None async def get_carpark_vacancy(lat: str, lng: str, vehicle_type: str = 'privateCar') -> str: """Get car park vacancy nearby a location. Args: lat: Minimum latitude of the location lng: Maximum latitude of the location """ extent = get_bounding_box(float(lat), float(lng)) carparks = await get_carparks_by_extent_and_vehicle_type(extent, vehicle_type) markdown_str = format_carparks_markdown(carparks) return markdown_str async def get_carparks_by_extent_and_vehicle_type(extent: tuple[float, float, float, float], vehicle_type: str) -> List[CarPark]: """ Query car parks within a bounding box (extent) and for a specific vehicle type. Returns a list of CarPark objects with validated data. extent: (min_lng, min_lat, max_lng, max_lat) vehicle_type: e.g. 'privateCar', 'LGV', etc. """ min_lng, min_lat, max_lng, max_lat = extent # 1. First call: get vacancies vacancy_url = ( f"{DPO_API_BASE}/v1/carpark-info-vacancy?data=vacancy&lang=zh_TW" f"&extent={min_lng},{min_lat},{max_lng},{max_lat}" f"&vehicleTypes={vehicle_type}" ) vacancy_data = await make_dpo_request(vacancy_url) if not vacancy_data or not vacancy_data.get('results'): return [] # 2. Extract car park IDs and map id -> vacancy info id_to_vacancy = {} for info in vacancy_data['results']: park_id = info.get('park_Id') if not park_id: continue id_to_vacancy[park_id] = {} for vt in ['privateCar', 'LGV', 'HGV', 'coach', 'motorCycle']: if vt in info: vt_data = info[vt] if isinstance(vt_data, dict): vt_data = [vt_data] id_to_vacancy[park_id][vt] = vt_data park_ids = list(id_to_vacancy.keys()) if not park_ids: return [] # 3. Second call: get details ids_str = ','.join(park_ids) info_url = ( f"{DPO_API_BASE}/v1/carpark-info-vacancy?data=info&lang=zh_TW" f"&vehicleTypes={vehicle_type}" f"&carparkIds={ids_str}" ) info_data = await make_dpo_request(info_url) if not info_data or not info_data.get('results'): return [] # 4. Merge vacancy info into details merged_results = [] for info in info_data['results']: park_id = info.get('park_Id') if not park_id or park_id not in id_to_vacancy: continue info['vacancy'] = id_to_vacancy[park_id] merged_results.append(info) # 5. Validate with Pydantic try: validated = CarParkApiResponse(results=merged_results) carparks = validated.results # Filter out car parks with last update > 15 mins ago for the selected vehicle type now = datetime.datetime.now() def is_recent(cp): vlist = cp.vacancy.get(vehicle_type, []) if vlist and hasattr(vlist[0], 'lastupdate') and vlist[0].lastupdate: try: lastupdate = datetime.datetime.strptime(vlist[0].lastupdate, "%Y-%m-%d %H:%M:%S") return (now - lastupdate).total_seconds() <= 15 * 60 except Exception: return False return False carparks = [cp for cp in carparks if is_recent(cp)] # Filter out car parks with no vacancy for the selected vehicle type def has_vacancy(cp): vlist = cp.vacancy.get(vehicle_type, []) if vlist and hasattr(vlist[0], 'vacancy') and vlist[0].vacancy is not None: return vlist[0].vacancy > 0 return False carparks = [cp for cp in carparks if has_vacancy(cp)] if not carparks: return [] # Sort by vacancy for the selected vehicle_type, descending def get_vacancy(cp): vlist = cp.vacancy.get(vehicle_type, []) if vlist and hasattr(vlist[0], 'vacancy') and vlist[0].vacancy is not None: return vlist[0].vacancy return -1 # Treat missing/None as lowest carparks_sorted = sorted(carparks, key=get_vacancy, reverse=True) return carparks_sorted[:5] except ValidationError as e: print(f"Validation error: {e}") return [] def format_carparks_markdown(carparks: List[CarPark]) -> str: """ Format a list of CarPark objects into a markdown string. Each car park will show its name, address, last update, and a table of vehicle type vacancies. """ if not carparks: return "目前沒有可用車位。" lines = [] for carpark in carparks: lines.append(f"### {carpark.name or '未知停車場'}") lines.append(f"**地址**: {carpark.displayAddress or '無資料'}") # Show last update for the main vehicle type (if available) lastupdate = None for vlist in carpark.vacancy.values(): if vlist and hasattr(vlist[0], 'lastupdate') and vlist[0].lastupdate: lastupdate = vlist[0].lastupdate break if lastupdate: lines.append(f"**最後更新**: {lastupdate}") lines.append("") lines.append("| 車輛類型 | 空位數量 |") lines.append("|---|---|") for vt, vacancies in carpark.vacancy.items(): if vacancies and isinstance(vacancies, list): vacancy = vacancies[0].vacancy if vacancies[0].vacancy is not None else '無資料' else: vacancy = '無資料' lines.append(f"| {vt} | {vacancy} |") lines.append("\n---\n") return "\n".join(lines) def get_bounding_box(lat: float, lng: float) -> tuple[float, float, float, float]: """ Given a central point (lat, lng), return the bounding box (min_lng, min_lat, max_lng, max_lat) that fully contains a circle of 2km radius around the point. """ earth_radius_km = 6371.0 distance_km = 2.0 # Latitude: 1 deg ≈ 111 km delta_lat = math.degrees(distance_km / earth_radius_km) min_lat = lat - delta_lat max_lat = lat + delta_lat # Longitude delta varies with latitude delta_lng = math.degrees(distance_km / (earth_radius_km * math.cos(math.radians(lat)))) min_lng = lng - delta_lng max_lng = lng + delta_lng return (min_lng, min_lat, max_lng, max_lat) demo = gr.Interface( fn=get_carpark_vacancy, inputs=[ gr.Textbox("22.3742", label="Latitude"), gr.Textbox("114.185", label="Longitude"), gr.Dropdown( choices=[ ("私家車 (privateCar)", "privateCar"), ("輕型貨車 (LGV)", "LGV"), ("重型貨車 (HGV)", "HGV"), ("旅遊巴士 (coach)", "coach"), ("電單車 (motorCycle)", "motorCycle") ], value="privateCar", label="車輛類型 (Vehicle Type)" ) ], outputs=[gr.Textbox()], title="Car park vacancy in Hong Kong", description="Enter lat, lng of the location and select vehicle type" ) if __name__ == "__main__": demo.launch(mcp_server=True)