|
|
from dataclasses import dataclass |
|
|
from typing import Optional, List |
|
|
import gradio as gr |
|
|
import logging |
|
|
from camptocamp_api import CamptocampAPI |
|
|
from typing import Optional |
|
|
|
|
|
|
|
|
logging.basicConfig( |
|
|
level=logging.INFO, |
|
|
format="%(asctime)s [%(levelname)s] %(message)s" |
|
|
) |
|
|
logger = logging.getLogger("CamptocampApp") |
|
|
|
|
|
|
|
|
c2c = CamptocampAPI(language="en") |
|
|
|
|
|
ACTIVITIES = [ |
|
|
"hiking", "snowshoeing", "skitouring", "snow_ice_mixed", |
|
|
"rock_climbing", "ice_climbing", "mountaineering", "via_ferrata", |
|
|
"paragliding", "bouldering", "multi_pitch", |
|
|
"bike", "mountain_bike", "trail_running", "alpine_climbing" |
|
|
] |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class SimplifiedOuting: |
|
|
title: Optional[str] |
|
|
condition_rating: Optional[str] |
|
|
date_start: Optional[str] |
|
|
date_end: Optional[str] |
|
|
elevation_max: Optional[int] |
|
|
global_rating: Optional[str] |
|
|
equipment_rating: Optional[str] |
|
|
rock_free_rating: Optional[str] |
|
|
area_titles: List[str] |
|
|
link: str |
|
|
|
|
|
|
|
|
def simplify_outings_response( |
|
|
response: dict, |
|
|
elevation_max_threshold: Optional[int] = None |
|
|
) -> List[SimplifiedOuting]: |
|
|
results = [] |
|
|
for doc in response.get("documents", []): |
|
|
elevation = doc.get("elevation_max") |
|
|
if elevation is None: |
|
|
continue |
|
|
elif elevation_max_threshold is not None and elevation is not None: |
|
|
if elevation < elevation_max_threshold: |
|
|
continue |
|
|
|
|
|
title = None |
|
|
if doc.get("locales"): |
|
|
title = doc["locales"][0].get("title") |
|
|
|
|
|
area_titles = [] |
|
|
for area in doc.get("areas", []): |
|
|
for loc in area.get("locales", []): |
|
|
t = loc.get("title") |
|
|
if t: |
|
|
area_titles.append(t) |
|
|
break |
|
|
|
|
|
results.append(SimplifiedOuting( |
|
|
title=title, |
|
|
condition_rating=doc.get("condition_rating"), |
|
|
date_start=doc.get("date_start"), |
|
|
date_end=doc.get("date_end"), |
|
|
elevation_max=elevation, |
|
|
global_rating=doc.get("global_rating"), |
|
|
equipment_rating=doc.get("equipment_rating"), |
|
|
rock_free_rating=doc.get("rock_free_rating"), |
|
|
area_titles=area_titles, |
|
|
link=f"https://www.camptocamp.org/outings/{doc.get('document_id')}" |
|
|
)) |
|
|
return results |
|
|
|
|
|
|
|
|
def get_outings_by_location( |
|
|
location: str, |
|
|
start_date: Optional[str] = None, |
|
|
end_date: Optional[str] = None, |
|
|
activity: Optional[str] = None, |
|
|
limit: int = 10, |
|
|
elevation_max_threshold: Optional[int] = 4000 |
|
|
) -> List[SimplifiedOuting]: |
|
|
logger.info(f"[Outings] Resolving location: {location}") |
|
|
bbox = c2c.get_bbox_from_location(location) |
|
|
if not bbox: |
|
|
logger.warning(f"No bounding box found for: {location}") |
|
|
return {"error": f"Could not resolve bounding box for location: {location}"} |
|
|
logger.info(f"BBox for '{location}': {bbox}") |
|
|
date_range = (start_date, end_date) if start_date and end_date else None |
|
|
result = c2c.get_outings(bbox, date_range, activity, limit) |
|
|
logger.info(f"Returned {len(result.get('documents', []))} outings.") |
|
|
return simplify_outings_response(result, elevation_max_threshold=elevation_max_threshold) |
|
|
|
|
|
def search_routes_by_location(location: str, activity: str, limit: int = 10) -> dict: |
|
|
logger.info(f"[Routes] Resolving location: {location}") |
|
|
bbox = c2c.get_bbox_from_location(location) |
|
|
if not bbox: |
|
|
logger.warning(f"No bounding box found for: {location}") |
|
|
return {"error": f"Could not resolve bounding box for location: {location}"} |
|
|
logger.info(f"BBox for '{location}': {bbox}") |
|
|
result = c2c.search_routes_by_activity(bbox, activity, limit) |
|
|
logger.info(f"Returned {len(result.get('documents', []))} routes.") |
|
|
return result |
|
|
|
|
|
def get_route_details(route_id: int) -> dict: |
|
|
logger.info(f"[Details] Fetching route ID: {route_id}") |
|
|
return c2c.get_route_details(route_id) |
|
|
|
|
|
def search_waypoints_by_location(location: str, limit: int = 10) -> dict: |
|
|
logger.info(f"[Waypoints] Resolving location: {location}") |
|
|
bbox = c2c.get_bbox_from_location(location) |
|
|
if not bbox: |
|
|
logger.warning(f"No bounding box found for: {location}") |
|
|
return {"error": f"Could not resolve bounding box for location: {location}"} |
|
|
logger.info(f"BBox for '{location}': {bbox}") |
|
|
result = c2c.search_waypoints(bbox, limit) |
|
|
logger.info(f"Returned {len(result.get('documents', []))} waypoints.") |
|
|
return result |
|
|
|
|
|
def lookup_bbox_from_location(location_name: str) -> Optional[dict]: |
|
|
logger.info(f"[Lookup] Resolving location: {location_name}") |
|
|
bbox = c2c.get_bbox_from_location(location_name) |
|
|
if not bbox: |
|
|
logger.warning(f"No bounding box found for: {location_name}") |
|
|
return {"error": f"No bounding box found for: {location_name}"} |
|
|
return { |
|
|
"west": bbox[0], |
|
|
"south": bbox[1], |
|
|
"east": bbox[2], |
|
|
"north": bbox[3] |
|
|
} |
|
|
|
|
|
|
|
|
with gr.Blocks(title="Camptocamp MCP Server") as demo: |
|
|
gr.Markdown("# 🏔️ Camptocamp API MCP") |
|
|
|
|
|
with gr.Tab("📍 Recent Outings"): |
|
|
loc = gr.Textbox(label="Location (e.g. Chamonix, La Grave)") |
|
|
start = gr.Textbox(label="Start Date (YYYY-MM-DD)") |
|
|
end = gr.Textbox(label="End Date (YYYY-MM-DD)") |
|
|
act = gr.Dropdown(label="Activity", choices=ACTIVITIES, value="alpine_climbing") |
|
|
limit = gr.Number(label="Result Limit", value=30) |
|
|
elev_slider = gr.Slider(label="Minimum maximal elevation (m)", minimum=0, maximum=4000, value=4000, step=100) |
|
|
|
|
|
out = gr.JSON() |
|
|
|
|
|
gr.Button("Get Outings").click( |
|
|
get_outings_by_location, |
|
|
inputs=[loc, start, end, act, limit, elev_slider], |
|
|
outputs=out |
|
|
) |
|
|
|
|
|
with gr.Tab("🧗 Search Routes"): |
|
|
rloc = gr.Textbox(label="Location (e.g. Alps)") |
|
|
ract = gr.Dropdown(label="Activity", choices=ACTIVITIES, value="alpine_climbing") |
|
|
rlim = gr.Number(label="Result Limit", value=5) |
|
|
rout = gr.JSON() |
|
|
gr.Button("Search Routes").click(search_routes_by_location, |
|
|
inputs=[rloc, ract, rlim], |
|
|
outputs=rout) |
|
|
|
|
|
with gr.Tab("📄 Route Details"): |
|
|
rid = gr.Number(label="Route ID") |
|
|
rdet = gr.JSON() |
|
|
gr.Button("Get Route Details").click(get_route_details, |
|
|
inputs=[rid], |
|
|
outputs=rdet) |
|
|
|
|
|
with gr.Tab("⛰ Waypoints"): |
|
|
wloc = gr.Textbox(label="Location (e.g. Mont Blanc)") |
|
|
wlim = gr.Number(label="Result Limit", value=5) |
|
|
wout = gr.JSON() |
|
|
gr.Button("Get Waypoints").click(search_waypoints_by_location, |
|
|
inputs=[wloc, wlim], |
|
|
outputs=wout) |
|
|
|
|
|
with gr.Tab("🌍 Location → BBox"): |
|
|
lstr = gr.Textbox(label="Location Name") |
|
|
lout = gr.JSON() |
|
|
gr.Button("Lookup BBox").click(lookup_bbox_from_location, |
|
|
inputs=[lstr], |
|
|
outputs=lout) |
|
|
|
|
|
demo.launch(mcp_server=True) |
|
|
|
