Update pages/5_📍_VertXtractor.py

pages/5_📍_VertXtractor.py

@@ -10,16 +10,15 @@ import geopandas as gpd
 import tempfile
 import os
 
-# Importing all the functions and constants from the original code
-from paste import (
-    wgs84_to_lv95, lv95_to_wgs84, detect_and_convert_bbox,
-    get_list_from_STAC_swisstopo, suppr_doublons_bati3D_v2,
-    suppr_doublons_bati3D_v3, suppr_doublons_list_ortho,
-    suppr_doublons_list_mnt, get_trees_points_from_bbox,
-    geojson_forest, CATEGORIES, MERGE_CATEGORIES
-)
-
-# Constants from the notebook
+# Constants
+CATEGORIES = {
+    'Gebueschwald': 'Forêt buissonnante',
+    'Wald': 'Forêt',
+    'Wald offen': 'Forêt claisemée',
+    'Gehoelzflaeche': 'Zone boisée',
+}
+MERGE_CATEGORIES = True
+
 URL_STAC_SWISSTOPO_BASE = 'https://data.geo.admin.ch/api/stac/v0.9/collections/'
 
 DIC_LAYERS = {
@@ -30,6 +29,189 @@ DIC_LAYERS = {
     'bati3D_v3': 'ch.swisstopo.swissbuildings3d_3_0',
 }
 
+# Functions from the original notebook
+def wgs84_to_lv95(lat, lon):
+    url = f'http://geodesy.geo.admin.ch/reframe/wgs84tolv95?easting={lat}&northing={lon}&format=json'
+    site = urllib.request.urlopen(url)
+    data = json.load(site)
+    return data['easting'], data['northing']
+
+def lv95_to_wgs84(x, y):
+    url = f'http://geodesy.geo.admin.ch/reframe/lv95towgs84?easting={x}&northing={y}&format=json'
+    f = urllib.request.urlopen(url)
+    txt = f.read().decode('utf-8')
+    json_res = json.loads(txt)
+    return json_res
+
+def detect_and_convert_bbox(bbox):
+    xmin, ymin, xmax, ymax = bbox
+    
+    wgs84_margin = 0.9
+    wgs84_bounds = {
+        'xmin': 5.96 - wgs84_margin,
+        'ymin': 45.82 - wgs84_margin,
+        'xmax': 10.49 + wgs84_margin,
+        'ymax': 47.81 + wgs84_margin
+    }
+    
+    lv95_margin = 100000
+    lv95_bounds = {
+        'xmin': 2485000 - lv95_margin,
+        'ymin': 1075000 - lv95_margin,
+        'xmax': 2834000 + lv95_margin,
+        'ymax': 1296000 + lv95_margin
+    }
+    
+    if (wgs84_bounds['xmin'] <= xmin <= wgs84_bounds['xmax'] and
+        wgs84_bounds['ymin'] <= ymin <= wgs84_bounds['ymax'] and
+        wgs84_bounds['xmin'] <= xmax <= wgs84_bounds['xmax'] and
+        wgs84_bounds['ymin'] <= ymax <= wgs84_bounds['ymax']):
+        
+        lv95_min = wgs84_to_lv95(xmin, ymin)
+        lv95_max = wgs84_to_lv95(xmax, ymax)
+        
+        bbox_lv95 = (lv95_min[0], lv95_min[1], lv95_max[0], lv95_max[1])
+        return (bbox, bbox_lv95)
+    
+    if (lv95_bounds['xmin'] <= xmin <= lv95_bounds['xmax'] and
+        lv95_bounds['ymin'] <= ymin <= lv95_bounds['ymax'] and
+        lv95_bounds['xmin'] <= xmax <= lv95_bounds['xmax'] and
+        lv95_bounds['ymin'] <= ymax <= lv95_bounds['ymax']):
+        
+        wgs84_min = lv95_to_wgs84(xmin, ymin)
+        wgs84_max = lv95_to_wgs84(xmax, ymax)
+        
+        bbox_wgs84 = (wgs84_min['easting'], wgs84_min['northing'], wgs84_max['easting'], wgs84_max['northing'])
+        return (bbox_wgs84, bbox)
+    
+    return None
+
+def get_list_from_STAC_swisstopo(url, est, sud, ouest, nord, gdb=False):
+    if gdb:
+        lst_indesirables = []
+    else:
+        lst_indesirables = ['.xyz.zip', '.gdb.zip']
+
+    sufixe_url = f"/items?bbox={est},{sud},{ouest},{nord}"
+    url += sufixe_url
+    res = []
+
+    while url:
+        f = urllib.request.urlopen(url)
+        txt = f.read().decode('utf-8')
+        json_res = json.loads(txt)
+        url = None
+        links = json_res.get('links', None)
+        if links:
+            for link in links:
+                if link['rel'] == 'next':
+                    url = link['href']   
+        
+        for item in json_res['features']:
+            for k, dic in item['assets'].items():
+                href = dic['href']
+                if gdb:
+                    if href[-8:] == '.gdb.zip':
+                        if len(dic['href'].split('/')[-1].split('_')) == 7:
+                            res.append(dic['href'])
+                else:
+                    if href[-8:] not in lst_indesirables:
+                        res.append(dic['href'])           
+    return res
+
+def suppr_doublons_bati3D_v2(lst_url):    
+    dico = {}
+    dxf_files = [url for url in lst_url if url[-8:] == '.dxf.zip']
+    for dxf in dxf_files:
+        *a, date, feuille = dxf.split('/')[-2].split('_')
+        dico.setdefault(feuille, []).append((date, dxf))
+    res = []
+    for k, liste in dico.items():
+        res.append(sorted(liste, reverse=True)[0][1])
+    return res
+
+def suppr_doublons_bati3D_v3(lst_url):    
+    dico = {}
+    gdb_files = [url for url in lst_url if url[-8:] == '.gdb.zip']
+    for gdb in gdb_files:
+        *a, date, feuille = gdb.split('/')[-2].split('_')
+        dico.setdefault(feuille, []).append((date, gdb))
+    res = []
+    for k, liste in dico.items():
+        res.append(sorted(liste, reverse=True)[0][1])
+    return res
+
+def suppr_doublons_list_ortho(lst):
+    dic = {}
+    for url in lst:
+        nom, an, noflle, taille_px, epsg = url.split('/')[-1][:-4].split('_')
+        dic.setdefault((noflle, float(taille_px)), []).append((an, url))
+    res = []
+    for noflle, lst in dic.items():
+        an, url = sorted(lst, reverse=True)[0]
+        res.append(url)
+    return res
+
+def suppr_doublons_list_mnt(lst):
+    dic = {}
+    for url in lst:
+        nom, an, noflle, taille_px, epsg, inconnu = url.split('/')[-1][:-4].split('_')
+        dic.setdefault((noflle, float(taille_px)), []).append((an, url))
+    res = []
+    for noflle, lst in dic.items():
+        an, url = sorted(lst, reverse=True)[0]
+        res.append(url)
+    return res
+
+def get_trees_points_from_bbox(bbox, fn_geojson):
+    xmin, ymin, xmax, ymax = bbox
+    url_base = 'https://hepiadata.hesge.ch/arcgis/rest/services/suisse/TLM_C4D_couverture_sol/FeatureServer/0/query?'
+    params = {
+        "geometry": f"{xmin},{ymin},{xmax},{ymax}",
+        "geometryType": "esriGeometryEnvelope",
+        "returnGeometry": "true",
+        "returnZ": "true",
+        "spatialRel": "esriSpatialRelIntersects",
+        "f": "json"
+    }
+    query_string = urllib.parse.urlencode(params)
+    url = url_base + query_string
+
+    with urllib.request.urlopen(url) as response:
+        response_data = response.read()
+        data = json.loads(response_data)
+
+        with open(fn_geojson, 'w') as f:
+            json.dump(data, f)
+
+def geojson_forest(bbox, fn_geojson):
+    xmin, ymin, xmax, ymax = bbox
+    url_base = 'https://hepiadata.hesge.ch/arcgis/rest/services/suisse/TLM_C4D_couverture_sol/FeatureServer/1/query?'
+
+    sql = ' OR '.join([f"OBJEKTART='{cat}'" for cat in CATEGORIES.keys()])
+
+    params = {
+        "geometry": f"{xmin},{ymin},{xmax},{ymax}",
+        "geometryType": "esriGeometryEnvelope",
+        "returnGeometry": "true",
+        "outFields": "OBJEKTART",
+        "orderByFields": "OBJEKTART",
+        "where": sql,
+        "returnZ": "true",
+        "outSR": '2056',
+        "spatialRel": "esriSpatialRelIntersects",
+        "f": "geojson"
+    }
+    query_string = urllib.parse.urlencode(params)
+    url = url_base + query_string
+    
+    with urllib.request.urlopen(url) as response:
+        response_data = response.read()
+        data = json.loads(response_data)
+        
+        with open(fn_geojson, 'w') as f:
+            json.dump(data, f)
+
 # Function to get URLs based on bbox and selected options
 def get_urls(bbox_wgs84, mnt=True, mns=True, bati3D_v2=True, bati3D_v3=True, ortho=True, mnt_resol=0.5, ortho_resol=0.1):
     est, sud, ouest, nord = bbox_wgs84