Update pages/5_📍_VertXtractor.py

pages/5_📍_VertXtractor.py

@@ -1,157 +1,105 @@
-import os
-import csv
-import json
-import logging
+import streamlit as st
 import requests
-import urllib.request
-from typing import Tuple, List, Optional
-from osgeo import gdal
+import folium
+from streamlit_folium import folium_static
 import geopandas as gpd
 import rasterio
-import numpy as np
-from pyproj import Transformer
-
-# Configuration du logging
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
-logger = logging.getLogger(__name__)
+from rasterio.merge import merge
+import os
+import tempfile
+import zipfile
+from io import BytesIO
 
-# Constants
-SUPPORTED_FORMATS = ['.tif', '.png', '.tiff', '.TIFF']
-NON_IMAGE_LAYERS = 'swissalti'
+# Configurations et constantes
 STAC_API_URL = "https://data.geo.admin.ch/api/stac/v0.9/collections/"
+PRODUCTS = {
+    'Luftbild 10cm': 'ch.swisstopo.swissimage-dop10',
+    'Landeskarte 1:10': 'ch.swisstopo.landeskarte-farbe-10',
+    'Landeskarte 1:25': 'ch.swisstopo.pixelkarte-farbe-pk25.noscale',
+    'Landeskarte 1:50': 'ch.swisstopo.pixelkarte-farbe-pk50.noscale',
+    'Landeskarte 1:100': 'ch.swisstopo.pixelkarte-farbe-pk100.noscale',
+    'Landeskarte 1:200': 'ch.swisstopo.pixelkarte-farbe-pk200.noscale',
+    'Höhenmodell': 'ch.swisstopo.swissalti3d',
+}
 
-def get_items(product_name: str, ll_lon: str, ll_lat: str, ur_lon: str, ur_lat: str, first_100: bool = False) -> Tuple[List[str], bool]:
-    """
-    Retrieve items for an Area of Interest (AOI).
-    
-    Args:
-        product_name (str): Name of the product.
-        ll_lon (str): Lower-left longitude.
-        ll_lat (str): Lower-left latitude.
-        ur_lon (str): Upper-right longitude.
-        ur_lat (str): Upper-right latitude.
-        first_100 (bool): If True, only retrieve the first 100 items.
-    
-    Returns:
-        Tuple[List[str], bool]: List of item URLs and a boolean indicating if there are more than 100 items.
-    """
-    try:
-        url = f"{STAC_API_URL}{product_name}/items?bbox={ll_lon},{ll_lat},{ur_lon},{ur_lat}"
-        response = requests.get(url)
-        response.raise_for_status()
-        items_result = response.json()
-        
-        assets = items_result.get('features', [])
-        item_files = [asset['assets']['data']['href'] for asset in assets if 'data' in asset['assets']]
-        
-        more_than_100 = len(items_result.get('links', [])) >= 6
-        
-        if more_than_100 and not first_100:
-            while 'next' in [link['rel'] for link in items_result.get('links', [])]:
-                next_url = next(link['href'] for link in items_result['links'] if link['rel'] == 'next')
-                response = requests.get(next_url)
-                response.raise_for_status()
-                items_result = response.json()
-                assets = items_result.get('features', [])
-                item_files.extend([asset['assets']['data']['href'] for asset in assets if 'data' in asset['assets']])
-        
-        # Filter items based on specific conditions
-        item_files = filter_items(item_files, product_name)
-        
-        return item_files, more_than_100
-    except requests.RequestException as e:
-        logger.error(f"Error retrieving items: {e}")
-        return [], False
-
-def filter_items(items: List[str], product_name: str) -> List[str]:
-    """
-    Filter items based on specific conditions.
-    
-    Args:
-        items (List[str]): List of item URLs.
-        product_name (str): Name of the product.
-    
-    Returns:
-        List[str]: Filtered list of item URLs.
-    """
-    if "_krel_" in items[0]:
-        items = [i for i in items if "_krel_" in i]
-    
-    if "_0.1_" in items[0]:
-        items = [i for i in items if "_0.1_" in i]
-    
-    if product_name == 'ch.swisstopo.swissalti3d':
-        items = [i for i in items if ".tif" in i]
-        items = [i for i in items if "_0.5_" in i]
-    
-    return items
+@st.cache_data
+def get_items(product_name, bbox):
+    """Récupère les éléments pour une zone d'intérêt (AOI)."""
+    url = f"{STAC_API_URL}{product_name}/items?bbox={bbox}"
+    response = requests.get(url)
+    if response.status_code == 200:
+        items = response.json().get('features', [])
+        return [item['assets']['data']['href'] for item in items if 'data' in item['assets']]
+    else:
+        st.error(f"Erreur lors de la récupération des éléments: {response.status_code}")
+        return []
 
-def create_csv(product_name: str, ll_lon: str, ll_lat: str, ur_lon: str, ur_lat: str) -> str:
-    """
-    Create a CSV file for import.
-    
-    Args:
-        product_name (str): Name of the product.
-        ll_lon (str): Lower-left longitude.
-        ll_lat (str): Lower-left latitude.
-        ur_lon (str): Upper-right longitude.
-        ur_lat (str): Upper-right latitude.
-    
-    Returns:
-        str: Path to the created CSV file.
-    """
-    try:
-        coords = f"{ll_lon}_{ll_lat}_{ur_lon}_{ur_lat}"
-        csv_filepath = os.path.join(os.getcwd(), f"{product_name}{coords}.csv")
-        
-        item_files, _ = get_items(product_name, ll_lon, ll_lat, ur_lon, ur_lat, first_100=False)
-        
-        with open(csv_filepath, 'w', newline='') as f:
-            writer = csv.writer(f)
-            writer.writerows([[item] for item in item_files])
+def download_and_merge(urls):
+    """Télécharge et fusionne les tuiles raster."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        files = []
+        for url in urls:
+            response = requests.get(url)
+            if response.status_code == 200:
+                file_path = os.path.join(tmpdir, os.path.basename(url))
+                with open(file_path, 'wb') as f:
+                    f.write(response.content)
+                files.append(file_path)
         
-        return csv_filepath
-    except IOError as e:
-        logger.error(f"Error creating CSV file: {e}")
-        return ""
-
-def process_csv(csv_filepath: str, bbox: Optional[Tuple[float, float, float, float]] = None) -> None:
-    """
-    Process the CSV file containing download URLs.
-    
-    Args:
-        csv_filepath (str): Path to the CSV file.
-        bbox (Optional[Tuple[float, float, float, float]]): Bounding box for cropping (minx, miny, maxx, maxy).
-    """
-    try:
-        download_dir, order_name = os.path.split(os.path.abspath(csv_filepath))
-        os.chdir(download_dir)
-        
-        with open(csv_filepath, 'r') as file:
-            urls = file.readlines()
-        
-        for i, url in enumerate(urls):
-            url = url.strip()
-            filename = os.path.join(download_dir, url.rsplit('/', 1)[-1])
+        if files:
+            src_files_to_mosaic = [rasterio.open(fp) for fp in files]
+            mosaic, out_trans = merge(src_files_to_mosaic)
             
-            if not os.path.isfile(filename):
-                logger.info(f"Downloading file {i+1} of {len(urls)}: {filename}")
-                urllib.request.urlretrieve(url, filename)
+            out_meta = src_files_to_mosaic[0].meta.copy()
+            out_meta.update({
+                "driver": "GTiff",
+                "height": mosaic.shape[1],
+                "width": mosaic.shape[2],
+                "transform": out_trans
+            })
             
-            if i == 0:
-                check_disk_space(filename, len(urls))
+            merged_path = os.path.join(tmpdir, "merged.tif")
+            with rasterio.open(merged_path, "w", **out_meta) as dest:
+                dest.write(mosaic)
             
-            if not args.noMERGE:
-                merge_raster(i, filename, order_name, download_dir)
-        
-        if bbox and not args.noCROP and not args.noMERGE:
-            logger.info("Cropping final raster...")
-            crop_raster(os.path.join(download_dir, f"{order_name}_temp_merged.tif"),
-                        os.path.join(download_dir, f"{order_name}_merged.tif"),
-                        bbox)
-        
-        logger.info(f"Result saved in {download_dir}")
-    except Exception as e:
-        logger.error(f"Error processing CSV: {e}")
+            return merged_path
+    return None
+
+def main():
+    st.title("SwissTopo Map Downloader")
+
+    # Sélection du produit
+    product = st.selectbox("Choisissez un produit", list(PRODUCTS.keys()))
+
+    # Carte pour sélectionner la bbox
+    m = folium.Map(location=[46.8182, 8.2275], zoom_start=8)
+    draw = folium.plugins.Draw(export=True)
+    draw.add_to(m)
+    folium_static(m)
+
+    # Récupération de la bbox dessinée
+    if 'last_active_drawing' in st.session_state:
+        bbox = st.session_state.last_active_drawing['geometry']['coordinates'][0]
+        bbox = f"{bbox[0][0]},{bbox[0][1]},{bbox[2][0]},{bbox[2][1]}"
+        st.write(f"BBox sélectionnée : {bbox}")
+
+        if st.button("Télécharger"):
+            with st.spinner("Téléchargement en cours..."):
+                urls = get_items(PRODUCTS[product], bbox)
+                if urls:
+                    merged_file = download_and_merge(urls)
+                    if merged_file:
+                        with open(merged_file, "rb") as file:
+                            btn = st.download_button(
+                                label="Télécharger le fichier fusionné",
+                                data=file,
+                                file_name="merged_map.tif",
+                                mime="image/tiff"
+                            )
+                    else:
+                        st.error("Erreur lors de la fusion des fichiers.")
+                else:
+                    st.warning("Aucune donnée trouvée pour cette zone.")
 
-# Other functions (merge_raster, crop_raster, etc.) should be similarly improved...
+if __name__ == "__main__":
+    main()