| import os |
| import requests |
|
|
| def download_hotosm_data(iso3: str, feature_type: str = "populated_places", geometry: str = "points", save_dir: str = "./downloads"): |
| """ |
| Download HOTOSM data (e.g., populated_places, roads, etc.) as SHP zip. |
| |
| Args: |
| iso3 (str): ISO3 country code (e.g., 'SSD'). |
| feature_type (str): HOTOSM feature type (e.g., 'populated_places', 'roads'). |
| geometry (str): Geometry type ('points', 'lines', 'polygons'). |
| save_dir (str): Directory to save the downloaded file. |
| |
| Returns: |
| str or None: Path to saved file, or None if failed. |
| """ |
| iso3_upper = iso3.upper() |
| iso3_lower = iso3.lower() |
|
|
| |
| filename = f"hotosm_{iso3_lower}_{feature_type}_{geometry}_shp.zip" |
| url = f"https://s3.dualstack.us-east-1.amazonaws.com/production-raw-data-api/ISO3/{iso3_upper}/{feature_type}/{geometry}/{filename}" |
| local_filename = os.path.join(save_dir, filename) |
|
|
| try: |
| response = requests.get(url, stream=True) |
| if response.status_code != 200: |
| print(f"[{iso3_upper}] Failed to download HOTOSM {feature_type}/{geometry}. HTTP {response.status_code}") |
| return None |
|
|
| os.makedirs(save_dir, exist_ok=True) |
| with open(local_filename, 'wb') as f: |
| for chunk in response.iter_content(chunk_size=8192): |
| f.write(chunk) |
|
|
| print(f"[{iso3_upper}] Downloaded: {local_filename}") |
| return local_filename |
|
|
| except Exception as e: |
| print(f"[{iso3_upper}] Error: {e}") |
| return None |
|
|
