from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool import datetime import requests import pytz import yaml from tools.final_answer import FinalAnswerTool import json from Gradio_UI import GradioUI from typing import Optional, Dict, Any, List import requests from pydantic import BaseModel, Field, field_validator from typing import Optional, Union import functools class PDOKLocationSearchInput(BaseModel): postal_code: Optional[str] = Field(None, description="Postal code in the format '1234 AA'.") house_number: Optional[str] = Field(None, description="House number.") street_name: Optional[str] = Field(None, description="Street name.") city: Optional[str] = Field(None, description="City name.") @field_validator('postal_code') def validate_postal_code(cls, v): if v is not None and (len(v) != 7 or not v[0:4].isdigit() or v[4] != " " or not v[5:7].isalpha()): raise ValueError("Invalid postal code format. It must be '1234 AA'.") return v def construct_query(self) -> str: """Constructs the query string based on provided inputs.""" if self.postal_code and self.house_number: return f"{self.postal_code} {self.house_number}" elif self.street_name and self.city and self.house_number: return f"{self.street_name} {self.house_number}, {self.city}" else: return "" # Create a requests session for reuse session = requests.Session() @functools.lru_cache(maxsize=128) def _fetch_cbs_data(postal_code: str, house_number: str, year: str): """Helper function to fetch CBS data for a specific year (with caching).""" url = f"https://service.pdok.nl/cbs/wijkenbuurten/2021/wfs?request=GetFeature&service=WFS&version=1.1.0&typeName=cbs_buurten_2021&outputFormat=json&srsName=EPSG:4326&CQL_FILTER=(postcode='{postal_code}')AND(huisnummer={house_number})" try: response = session.get(url) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) data = response.json() # Extract relevant data - adapt field names as needed if data["features"]: return data["features"][0]["properties"] else: return None # Or raise a custom exception if no features are found except requests.exceptions.RequestException as e: print(f"Error fetching CBS data for {year}: {e}") # Log the error return None except (ValueError, KeyError) as e: print(f"Error parsing CBS data for {year}: {e}") return None @tool def get_cbs_data(postal_code: str, house_number: str) -> str: """ Retrieves CBS (Statistics Netherlands) demographic and facility data for a given postal code and house number. Fetches data for 2021. Args: postal_code: The postal code (e.g., "1234AB"). house_number: The house number (e.g., "12"). Returns: A JSON string containing the CBS data, or an error message if the data cannot be retrieved. """ # Format inputs formatted_postal_code = postal_code.replace(" ", "").upper() formatted_house_number = house_number.strip() data_2021 = _fetch_cbs_data(formatted_postal_code, formatted_house_number, "2021") if data_2021: # Create the dictionary to match the desired output format, including *all* fields, handling potential missing data. cbs_data = { "p2021_amount_of_inhabitants": data_2021.get("aantal_inwoners"), "p2021_amount_of_men": data_2021.get("aantal_mannen"), "p2021_amount_of_women": data_2021.get("aantal_vrouwen"), "p2021_amount_of_inhabitants_0_to_15_year": data_2021.get("aantal_inwoners_0_tot_15_jaar"), "p2021_amount_of_inhabitants_15_to_25_year": data_2021.get("aantal_inwoners_15_tot_25_jaar"), "p2021_amount_of_inhabitants_25_to_45_year": data_2021.get("aantal_inwoners_25_tot_45_jaar"), "p2021_amount_of_inhabitants_45_to_65_year": data_2021.get("aantal_inwoners_45_tot_65_jaar"), "p2021_amount_of_inhabitants_65_years_and_older": data_2021.get("aantal_inwoners_65_jaar_en_ouder"), "p2021_nearest_supermarket_distance": data_2021.get("dichtstbijzijnde_grote_supermarkt_afstand_in_km"), "p2021_supermarkets_within_1km": data_2021.get("grote_supermarkt_aantal_binnen_1_km"), "p2021_supermarkets_within_3km": data_2021.get("grote_supermarkt_aantal_binnen_3_km"), "p2021_supermarkets_within_5km": data_2021.get("grote_supermarkt_aantal_binnen_5_km"), "p2021_nearest_daily_grocery_store_distance": data_2021.get("dichtstbijzijnde_winkels_ov_dagel_levensm_afst_in_km"), "p2021_daily_grocery_stores_within_1km": data_2021.get("winkels_ov_dagel_levensm_aantal_binnen_1_km"), "p2021_daily_grocery_stores_within_3km": data_2021.get("winkels_ov_dagel_levensm_aantal_binnen_3_km"), "p2021_daily_grocery_stores_within_5km": data_2021.get("winkels_ov_dagel_levensm_aantal_binnen_5_km"), "p2021_nearest_department_store_distance": data_2021.get("dichtstbijzijnde_warenhuis_afstand_in_km"), "p2021_department_stores_within_5km": data_2021.get("warenhuis_aantal_binnen_5_km"), "p2021_department_stores_within_10km": data_2021.get("warenhuis_aantal_binnen_10_km"), "p2021_department_stores_within_20km": data_2021.get("warenhuis_aantal_binnen_20_km"), "p2021_nearest_pub_distance": data_2021.get("dichtstbijzijnde_cafe_afstand_in_km"), "p2021_pubs_within_1km": data_2021.get("cafe_aantal_binnen_1_km"), "p2021_pubs_within_3km": data_2021.get("cafe_aantal_binnen_3_km"), "p2021_pubs_within_5km": data_2021.get("cafe_aantal_binnen_5_km"), "p2021_nearest_cafeteria_distance": data_2021.get("dichtstbijzijnde_cafetaria_afstand_in_km"), "p2021_cafeterias_within_1km": data_2021.get("cafetaria_aantal_binnen_1_km"), "p2021_cafeterias_within_3km": data_2021.get("cafetaria_aantal_binnen_3_km"), "p2021_cafeterias_within_5km": data_2021.get("cafetaria_aantal_binnen_5_km"), "p2021_nearest_hotel_distance": data_2021.get("dichtstbijzijnde_hotel_afstand_in_km"), "p2021_hotels_within_5km": data_2021.get("hotel_aantal_binnen_5_km"), "p2021_hotels_within_10km": data_2021.get("hotel_aantal_binnen_10_km"), "p2021_hotels_within_20km": data_2021.get("hotel_aantal_binnen_20_km"), "p2021_nearest_restaurant_distance": data_2021.get("dichtstbijzijnde_restaurant_afstand_in_km"), "p2021_restaurants_within_1km": data_2021.get("restaurant_aantal_binnen_1_km"), "p2021_restaurants_within_3km": data_2021.get("restaurant_aantal_binnen_3_km"), "p2021_restaurants_within_5km": data_2021.get("restaurant_aantal_binnen_5_km"), "p2021_nearest_after_school_care_distance": data_2021.get("dichtstbijzijnde_buitenschoolse_opvang_afstand_in_km"), "p2021_after_school_care_within_1km": data_2021.get("buitenschoolse_opvang_aantal_binnen_1_km"), "p2021_after_school_care_within_3km": data_2021.get("buitenschoolse_opvang_aantal_binnen_3_km"), "p2021_after_school_care_within_5km": data_2021.get("buitenschoolse_opvang_aantal_binnen_5_km"), "p2021_nearest_day_care_distance": data_2021.get("dichtstbijzijnde_kinderdagverblijf_afstand_in_km"), "p2021_day_care_within_1km": data_2021.get("kinderdagverblijf_aantal_binnen_1_km"), "p2021_day_care_within_3km": data_2021.get("kinderdagverblijf_aantal_binnen_3_km"), "p2021_day_care_within_5km": data_2021.get("kinderdagverblijf_aantal_binnen_5_km"), "p2021_nearest_fire_station_distance": data_2021.get("dichtstbijzijnde_brandweerkazerne_afstand_in_km"), "p2021_nearest_highway_access_distance": data_2021.get("dichtstbijzijnde_oprit_hoofdverkeersweg_afstand_in_km"), "p2021_nearest_transfer_station_distance": data_2021.get("dichtstbijzijnde_overstapstation_afstand_in_km"), "p2021_nearest_train_station_distance": data_2021.get("dichtstbijzijnde_treinstation_afstand_in_km"), "p2021_nearest_theme_park_distance": data_2021.get("dichtstbijzijnde_attractiepark_afstand_in_km"), "p2021_theme_parks_within_10km": data_2021.get("attractiepark_aantal_binnen_10_km"), "p2021_theme_parks_within_20km": data_2021.get("attractiepark_aantal_binnen_20_km"), "p2021_theme_parks_within_50km": data_2021.get("attractiepark_aantal_binnen_50_km"), "p2021_nearest_cinema_distance": data_2021.get("dichtstbijzijnde_bioscoop_afstand_in_km"), "p2021_cinemas_within_5km": data_2021.get("bioscoop_aantal_binnen_5_km"), "p2021_cinemas_within_10km": data_2021.get("bioscoop_aantal_binnen_10_km"), "p2021_cinemas_within_20km": data_2021.get("bioscoop_aantal_binnen_20_km"), "p2021_nearest_museum_distance": data_2021.get("dichtstbijzijnde_museum_afstand_in_km"), "p2021_museums_within_5km": data_2021.get("museum_aantal_binnen_5_km"), "p2021_museums_within_10km": data_2021.get("museum_aantal_binnen_10_km"), "p2021_museums_within_20km": data_2021.get("museum_aantal_binnen_20_km"), "p2021_nearest_theater_distance": data_2021.get("dichtstbijzijnde_theater_afstand_in_km"), "p2021_theaters_within_5km": data_2021.get("theater_aantal_binnen_5_km"), "p2021_theaters_within_10km": data_2021.get("theater_aantal_binnen_10_km"), "p2021_theaters_within_20km": data_2021.get("theater_aantal_binnen_20_km"), "p2021_nearest_library_distance": data_2021.get("dichtstbijzijnde_bibliotheek_afstand_in_km"), "p2021_nearest_ice_rink_distance": data_2021.get("dichtstbijzijnde_kunstijsbaan_afstand_in_km"), "p2021_nearest_music_venue_distance": data_2021.get("dichtstbijzijnde_poppodium_afstand_in_km"), "p2021_nearest_sauna_distance": data_2021.get("dichtstbijzijnde_sauna_afstand_in_km"), "p2021_nearest_tanning_salon_distance": data_2021.get("dichtstbijzijnde_zonnebank_afstand_in_km"), "p2021_nearest_swimming_pool_distance": data_2021.get("dichtstbijzijnde_zwembad_afstand_in_km"), "p2021_nearest_primary_school_distance": data_2021.get("dichtstbijzijnde_basisonderwijs_afstand_in_km"), "p2021_primary_schools_within_1km": data_2021.get("basisonderwijs_aantal_binnen_1_km"), "p2021_primary_schools_within_3km": data_2021.get("basisonderwijs_aantal_binnen_3_km"), "p2021_primary_schools_within_5km": data_2021.get("basisonderwijs_aantal_binnen_5_km"), "p2021_nearest_havovwo_distance": data_2021.get("dichtstbijzijnde_havo_vwo_afstand_in_km"), "p2021_havovwo_within_3km": data_2021.get("havo_vwo_aantal_binnen_3_km"), "p2021_havovwo_within_5km": data_2021.get("havo_vwo_aantal_binnen_5_km"), "p2021_havovwo_within_10km": data_2021.get("havo_vwo_aantal_binnen_10_km"), "p2021_nearest_vmbo_distance": data_2021.get("dichtstbijzijnde_vmbo_afstand_in_km"), "p2021_vmbo_within_3km": data_2021.get("vmbo_aantal_binnen_3_km"), "p2021_vmbo_within_5km": data_2021.get("vmbo_aantal_binnen_5_km"), "p2021_vmbo_within_10km": data_2021.get("vmbo_aantal_binnen_10_km"), "p2021_nearest_secondary_school_distance": data_2021.get("dichtstbijzijnde_voortgezet_onderwijs_afstand_in_km"), "p2021_secondary_schools_within_3km": data_2021.get("voortgezet_onderwijs_aantal_binnen_3_km"), "p2021_secondary_schools_within_5km": data_2021.get("voortgezet_onderwijs_aantal_binnen_5_km"), "p2021_secondary_schools_within_10km": data_2021.get("voortgezet_onderwijs_aantal_binnen_10_km"), "p2021_nearest_gp_distance": data_2021.get("dichtstbijzijnde_huisartsenpraktijk_afstand_in_km"), "p2021_gps_within_1km": data_2021.get("huisartsenpraktijk_aantal_binnen_1_km"), "p2021_gps_within_3km": data_2021.get("huisartsenpraktijk_aantal_binnen_3_km"), "p2021_gps_within_5km": data_2021.get("huisartsenpraktijk_aantal_binnen_5_km"), "p2021_nearest_hospital_excl_outpatient_distance": data_2021.get("dichtstbijzijnde_ziekenh_excl_buitenpoli_afst_in_km"), "p2021_hospitals_excl_outpatient_within_5km": data_2021.get("ziekenhuis_excl_buitenpoli_aantal_binnen_5_km"), "p2021_hospitals_excl_outpatient_within_10km": data_2021.get("ziekenhuis_excl_buitenpoli_aantal_binnen_10_km"), "p2021_hospitals_excl_outpatient_within_20km": data_2021.get("ziekenhuis_excl_buitenpoli_aantal_binnen_20_km"), "p2021_nearest_hospital_incl_outpatient_distance": data_2021.get("dichtstbijzijnde_ziekenh_incl_buitenpoli_afst_in_km"), "p2021_hospitals_incl_outpatient_within_5km": data_2021.get("ziekenhuis_incl_buitenpoli_aantal_binnen_5_km"), "p2021_hospitals_incl_outpatient_within_10km": data_2021.get("ziekenhuis_incl_buitenpoli_aantal_binnen_10_km"), "p2021_hospitals_incl_outpatient_within_20km": data_2021.get("ziekenhuis_incl_buitenpoli_aantal_binnen_20_km"), "p2021_nearest_pharmacy_distance": data_2021.get("dichtstbijzijnde_apotheek_afstand_in_km"), "p2021_nearest_gp_post_distance": data_2021.get("dichtstbijzijnde_huisartsenpost_afstand_in_km"), } return json.dumps(cbs_data) # Return as JSON string else: return json.dumps({"error": "Could not retrieve CBS data."}) @functools.lru_cache(maxsize=128) # Add caching @tool def pdok_location_info(postal_code: Optional[str] = None, house_number: Optional[str] = None, street_name: Optional[str] = None, city: Optional[str] = None) -> str: """Provides information about a Dutch address or postal code, including a Google Maps link and CBS data. Args: postal_code: Postal code in the format '1234 AA'. house_number: House number of the address. street_name: Name of the street. city: Name of the city. Returns: str: JSON string containing the location information, including a Google Maps link and CBS data, or an error message. """ debug_info = [] debug_info.append(f"Input values:\n" f" Postal code: {postal_code}\n" f" House number: {house_number}\n" f" Street name: {street_name}\n" f" City: {city}") base_url = "https://api.pdok.nl/bzk/locatieserver/search/v3_1/free" headers = {"accept": "application/json"} input_data = PDOKLocationSearchInput(postal_code=postal_code, house_number=house_number, street_name=street_name, city=city) query_string = input_data.construct_query() debug_info.append(f"Constructed query string: {query_string}") if not query_string: return f"Error: Empty query string\nDebug info:\n" + "\n".join(debug_info) params = { "q": query_string, "fl": "*", "fq": "type:(gemeente OR woonplaats OR weg OR postcode OR adres)", "df": "tekst", "bq": "type:provincie^1.5", "bq": "type:gemeente^1.5", "bq": "type:woonplaats^1.5", "bq": "type:weg^1.5", "bq": "type:postcode^0.5", "bq": "type:adres^1", "start": 0, "rows": 1, # Reduced to 1 "sort": "score desc,sortering asc,weergavenaam asc", "wt": "json", } try: response = session.get(base_url, params=params, headers=headers) # Use the session debug_info.append(f"Response status code: {response.status_code}") response.raise_for_status() data = response.json() docs = data.get("response", {}).get("docs", []) debug_info.append(f"Number of docs found: {len(docs)}") if not docs: return f"Error: No results found\nDebug info:\n" + "\n".join(debug_info) first_result = docs[0] debug_info.append(f"First result data: {first_result}") location_info = { "straatnaam": first_result.get("straatnaam", ""), "huisnummer": first_result.get("huisnummer", ""), "postcode": first_result.get("postcode", ""), "woonplaatsnaam": first_result.get("woonplaatsnaam", ""), "gemeentenaam": first_result.get("gemeentenaam", ""), "provincienaam": first_result.get("provincienaam", ""), "buurtnaam": first_result.get("buurtnaam", ""), "wijknaam": first_result.get("wijknaam", ""), "centroide_ll": first_result.get("centroide_ll", ""), "centroide_rd": first_result.get("centroide_rd", "") } centroide_ll = location_info.get('centroide_ll') if centroide_ll: lon, lat = centroide_ll.replace("POINT(", "").replace(")", "").split() location_info["google_maps_url"] = f"https://www.google.com/maps/search/?api=1&query={lat},{lon}" # --- Get CBS Data --- if location_info["postcode"] and location_info["huisnummer"]: cbs_data_str = get_cbs_data(location_info["postcode"], str(location_info["huisnummer"])) try: cbs_data = json.loads(cbs_data_str) # Check for error in CBS data if "error" in cbs_data: location_info["cbs_data_error"] = cbs_data["error"] # Add error message. else: location_info.update(cbs_data) # Merge CBS data except json.JSONDecodeError: location_info["cbs_data_error"] = "Failed to parse CBS data." else: location_info["cbs_data_error"] = "Postcode or house number missing, cannot fetch CBS data." return json.dumps(location_info) except requests.exceptions.RequestException as e: return f"Error during API request: {e}\nDebug info:\n" + "\n".join(debug_info) except (ValueError, KeyError) as e: return f"Error processing API response: {e}\nDebug info:\n" + "\n".join(debug_info) except Exception as e: return f"An unexpected error occurred: {e}\nDebug info:\n" + "\n".join(debug_info) @tool def my_custom_tool(arg1:str, arg2:int)-> str: """A tool that does nothing yet Args: arg1: the first argument arg2: the second argument """ return "What magic will you build ?" @tool def get_current_time_in_timezone(timezone: str) -> str: """A tool that fetches the current local time in a specified timezone. Args: timezone: A string representing a valid timezone (e.g., 'America/New_York'). """ try: tz = pytz.timezone(timezone) local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S") return f"The current local time in {timezone} is: {local_time}" except Exception as e: return f"Error fetching time for timezone '{timezone}': {str(e)}" final_answer = FinalAnswerTool() model = HfApiModel( max_tokens=2096, temperature=0.5, model_id='Qwen/Qwen2.5-Coder-32B-Instruct', custom_role_conversions=None, ) image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True) with open("prompts.yaml", 'r') as stream: prompt_templates = yaml.safe_load(stream) agent = CodeAgent( model=model, tools=[final_answer, get_current_time_in_timezone, pdok_location_info, get_cbs_data], # Include get_cbs_data max_steps=5, # Tune this value! Start low. verbosity_level=0, # Set to 0 for production grammar=None, planning_interval=None, name=None, description=None, prompt_templates=prompt_templates ) GradioUI(agent).launch()