Import Heat Vulnerability¶

def load_data() -> geopandas.GeoDataFrame:
    """Open the geopackage for vulnerabilty.

    Returns:
        geopandas.GeoDataFrame: The loaded geopackage as a GeoDataFrame.

    Raises:
        FileNotFoundError: If no folder with "vulnerability" in the name is found or no .gpkg file is found in the folder.
    """
    vulnerability_path = "file_data/vulnerability"
    if not os.path.isdir(vulnerability_path):
        raise FileNotFoundError(
            "No folder for 'vulnerability' found in 'file_data/' directory."
        )
    gpkg_file = None
    for file in os.listdir(vulnerability_path):
        if file.lower().endswith(".gpkg"):
            gpkg_file = file
            break
    if not gpkg_file:
        raise FileNotFoundError(
            f"No geopackage file found in the folder '{vulnerability_path}'."
        )

    gpkg_path = os.path.join(vulnerability_path, gpkg_file)
    gdf = geopandas.read_file(gpkg_path, layer="vulnérabilité_fortes_chaleurs")
    gdf.to_crs(TARGET_PROJ, inplace=True)
    gdf_details = geopandas.read_file(
        gpkg_path, layer="Vulnerabilite_fortes_chaleurs_détail"
    )
    gdf_details.to_crs(TARGET_PROJ, inplace=True)

    merged = gdf.merge(
        gdf_details, on="ID_RSU", suffixes=("", "_details"), validate="one_to_one"
    )
    merged.drop(
        columns=[
            "id",
            "commune",
            "ID_RSU",
            "Surface_RSU",
            "NOTE_EXPO_JOUR",
            "NOTE_EXPO_NUIT",
            "NOTE_SENSI_JOUR",
            "NOTE_SENSI_NUIT",
            "NOTE_CAPAF_JOUR",
            "NOTE_CAPAF_NUIT",
        ],
        inplace=True,
    )
    duplicate_columns = [
        col
        for col in merged.columns
        if col.endswith("_details") and col[:-8] in merged.columns
    ]

    for col in duplicate_columns:
        original_col = col[:-8]
        if not merged[original_col].equals(merged[col]):
            raise ValueError(f"Mismatch found in column '{original_col}' and '{col}'.")

    merged.drop(columns=duplicate_columns, inplace=True)

    def make_valid(geometry):
        """Fix minor topology errors, like Polygon not closed."""
        if geometry and not geometry.is_valid:
            return geometry.buffer(0)
        return geometry

    merged["geometry"] = merged["geometry"].apply(make_valid)
    merged["map_geometry"] = merged.geometry.to_crs(TARGET_MAP_PROJ)
    merged["map_geometry"] = merged["map_geometry"].apply(make_valid)

    merged.fillna(
        0, inplace=True
    )  # Columns ['majic_Log_av1949', 'majic_Log_1949_1989',
    # 'majic_Log_ap1990', 'majic_nlogh', 'majic_stoth', 'majic_slocal',
    #        'majic_nloghmais', 'majic_nloghappt', 'siret_densite_emploi'] contains NaN because
    # these data (economical indices) are missing in water, forest, etc.
    # For all this values 0 means absence.

    return merged

def save_geometries(vulnerability_datas: geopandas.GeoDataFrame) -> None:
    """Save vulnerability data to the database.

    Args:
        vulnerability_datas (GeoDataFrame): GeoDataFrame to save to the database.

    Returns:
        None
    """
    batch_size = 10000
    indices = [
        "geometry",
        "map_geometry",
        "VULNERABILITE_JOUR",
        "VULNERABILITE_NUIT",
        "EXPO_JOUR",
        "EXPO_NUIT",
        "CAPAF_JOUR",
        "CAPAF_NUIT",
        "SENSI_JOUR",
        "SENSI_NUIT",
    ]
    for start in tqdm(range(0, len(vulnerability_datas), batch_size)):
        end = start + batch_size
        batch = vulnerability_datas.loc[start:end]
        Vulnerability.objects.bulk_create(
            [
                Vulnerability(
                    geometry=GEOSGeometry(data["geometry"].wkt),
                    map_geometry=GEOSGeometry(data["map_geometry"].wkt),
                    vulnerability_index_day=data["VULNERABILITE_JOUR"],
                    vulnerability_index_night=data["VULNERABILITE_NUIT"],
                    expo_index_day=data["EXPO_JOUR"],
                    expo_index_night=data["EXPO_NUIT"],
                    capaf_index_day=data["CAPAF_JOUR"],
                    capaf_index_night=data["CAPAF_NUIT"],
                    sensibilty_index_day=data["SENSI_JOUR"],
                    sensibilty_index_night=data["SENSI_NUIT"],
                    details={
                        col: data[col] for col in data.index if col not in indices
                    },
                )
                for _, data in batch.iterrows()
            ]
        )