import streamlit as st import folium from streamlit_folium import folium_static from folium.plugins import Draw import requests import json import os import urllib.request import csv from osgeo import gdal import numpy as np import math import shutil import sys import glob # Configuration de la page Streamlit st.set_page_config(page_title="SwissTopo Downloader", layout="wide") # Constantes 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', } # Fonctions utilitaires def osm_to_decimal(tile_x, tile_y, zoom): """Convertit les coordonnées OSM en coordonnées décimales.""" n = 2.0 ** zoom lon_deg = tile_x / n * 360.0 - 180.0 lat_rad = math.atan(math.sinh(math.pi * (1 - 2 * tile_y / n))) lat_deg = math.degrees(lat_rad) return lat_deg, lon_deg def get_items(productname, LLlon, LLlat, URlon, URlat, first100=False): """Récupère les éléments pour une zone d'intérêt donnée.""" url = f"https://data.geo.admin.ch/api/stac/v0.9/collections/{productname}/items?bbox={LLlon},{LLlat},{URlon},{URlat}" response = requests.get(url) items_result = json.loads(response.content) assets = items_result.get('features', []) items_files = [] for asset in assets: asset_url = asset.get('assets', {}).get('data', {}).get('href') if asset_url: items_files.append(asset_url) # Filtrage des fichiers selon les cas spécifiques if "_krel_" in productname: items_files = [i for i in items_files if "_krel_" in i] if "0.1" in productname: items_files = [i for i in items_files if "_0.1_" in i] if productname == 'ch.swisstopo.swissalti3d': items_files = [i for i in items_files if i.endswith(".tif")] items_files = [i for i in items_files if "_0.5_" in i] return items_files def create_csv(productname, LLlon, LLlat, URlon, URlat): """Crée un fichier CSV contenant les URLs des données à télécharger.""" coords = f"{LLlon}_{LLlat}_{URlon}_{URlat}" csv_filepath = os.path.join(os.getcwd(), f"{productname}{coords}.csv") items_files = get_items(productname, LLlon, LLlat, URlon, URlat) with open(csv_filepath, 'w', newline='') as f: writer = csv.writer(f) for item in items_files: writer.writerow([item]) return csv_filepath def merge_rasters(input_files, output_file, compress_method='JPEG'): """Fusionne plusieurs rasters en un seul.""" gdal.UseExceptions() # Ouvrir le premier fichier pour obtenir les métadonnées src_ds = gdal.Open(input_files[0]) driver = gdal.GetDriverByName("GTiff") # Créer le fichier de sortie dst_ds = driver.CreateCopy(output_file, src_ds, 0, options=[f"COMPRESS={compress_method}", "TILED=YES", "BIGTIFF=YES"]) # Fermer le premier fichier src_ds = None # Fusionner les autres fichiers for file in input_files[1:]: gdal.Warp(output_file, file, format="GTiff", options=[f"COMPRESS={compress_method}"]) # Fermer le fichier de sortie dst_ds = None def process_csv(csv_filepath, no_merge=False): """Traite le fichier CSV pour télécharger et fusionner les données.""" download_dir = os.path.dirname(csv_filepath) order_name = os.path.basename(csv_filepath).split('.')[0] with open(csv_filepath, 'r') as file: urls = file.readlines() downloaded_files = [] for i, url in enumerate(urls): url = url.strip() filename = os.path.join(download_dir, url.split('/')[-1]) if not os.path.isfile(filename): st.text(f"Téléchargement du fichier : {i+1} sur {len(urls)}") urllib.request.urlretrieve(url, filename) downloaded_files.append(filename) if not no_merge: output_file = os.path.join(download_dir, f"{order_name}_merged.tif") st.text("Fusion des fichiers téléchargés...") merge_rasters(downloaded_files, output_file) st.text(f"Résultat fusionné dans : {output_file}") else: st.text(f"Fichiers téléchargés dans : {download_dir}") return output_file if not no_merge else download_dir # Interface utilisateur Streamlit def main(): st.title("SwissTopo Downloader") # Sélection du produit selected_product = st.selectbox("Sélectionnez un produit", list(PRODUCTS.keys())) # Carte interactive pour dessiner la bbox st.subheader("Dessinez une boîte englobante sur la carte") m = folium.Map(location=[46.8182, 8.2275], zoom_start=8) draw = Draw( draw_options={ 'rectangle': True, 'polygon': False, 'polyline': False, 'circle': False, 'marker': False, 'circlemarker': False, }, edit_options={'edit': False} ) draw.add_to(m) output = folium_static(m) # Récupération des coordonnées de la bbox bbox_coords = None if 'last_active_drawing' in st.session_state: bbox = st.session_state.last_active_drawing['geometry']['coordinates'][0] lons = [coord[0] for coord in bbox] lats = [coord[1] for coord in bbox] bbox_coords = [min(lons), min(lats), max(lons), max(lats)] st.write(f"Coordonnées de la bbox : {bbox_coords}") # Options no_merge = st.checkbox("Ne pas fusionner les fichiers téléchargés") if st.button("Télécharger") and bbox_coords: product = PRODUCTS[selected_product] csv_filepath = create_csv(product, *bbox_coords) result = process_csv(csv_filepath, no_merge) st.success(f"Téléchargement terminé ! Résultat : {result}") elif st.button("Télécharger") and not bbox_coords: st.error("Veuillez dessiner une boîte englobante sur la carte avant de télécharger.") if __name__ == "__main__": main()