AI employed

.gitignore

@@ -0,0 +1,2 @@
+.cursor
+.venv

app.py

@@ -1,27 +1,215 @@
 import gradio as gr
+from typing import Any, List, Dict, Optional
+import httpx
+import math
+from pydantic import BaseModel, Field, ValidationError
+import datetime
 
-def letter_counter(word, letter):
-    """
-    Count the number of occurrences of a letter in a word or text.
+# 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:
-        word (str): The input text to search through
-        letter (str): The letter to search for
+        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)
 
-    Returns:
-        str: A message indicating how many times the letter appears
+def get_bounding_box(lat: float, lng: float) -> tuple[float, float, float, float]:
     """
-    word = word.lower()
-    letter = letter.lower()
-    count = word.count(letter)
-    return count
+    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=letter_counter,
-    inputs=[gr.Textbox("strawberry"), gr.Textbox("r")],
-    outputs=[gr.Number()],
-    title="Letter Counter",
-    description="Enter text and a letter to count how many times the letter appears in the text."
+    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__":

requirements.txt

@@ -0,0 +1,3 @@
+gradio[mcp]
+httpx
+pydantic