Spaces:
Sleeping
Sleeping
Add real KNN pipeline: train on your own polygon labels + spectral bands, evaluate against ground truth
32e60fc verified | import gradio as gr | |
| import numpy as np | |
| import matplotlib | |
| matplotlib.use('Agg') | |
| import matplotlib.pyplot as plt | |
| import rasterio | |
| import json | |
| import os | |
| from sklearn.neighbors import KNeighborsClassifier | |
| from sklearn.preprocessing import StandardScaler | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Constants | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| N_POLYGONS = 23 | |
| K_MIN, K_MAX, K_DEFAULT = 1, 15, 5 | |
| MAP_MAX_DIM = 500 # downsample target for map visualizations (speed) | |
| CLASSES = { | |
| 1: "Eau", | |
| 2: "Vergers", | |
| 3: "Cultures dans le Delta", | |
| 4: "Zones bâties", | |
| 5: "Cultures irriguées dans le désert", | |
| 6: "Cultures non irriguées en zone sèche", | |
| 7: "Zones sableuses", | |
| } | |
| CLASS_CHOICES = [f"{k} - {v}" for k, v in CLASSES.items()] | |
| # index 0 = fond / non classé, 1..7 = classes ci-dessus | |
| COLORS_RGB = np.array([ | |
| [20, 20, 20], [0, 100, 220], [0, 160, 60], [120, 220, 100], | |
| [220, 50, 50], [255, 165, 0], [160, 90, 30], [240, 230, 140], | |
| ], dtype=np.uint8) | |
| BAND_FILES = [ | |
| 'band_1_uv.tif', 'band_2_blue.tif', 'band_3_green.tif', 'band_4_red.tif', | |
| 'band_5_nir.tif', 'band_6_swir1.tif', 'band_7_swir2.tif', | |
| ] | |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Data loading (once at startup) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def load_data(): | |
| def read_tif(name): | |
| path = os.path.join(BASE_DIR, 'data', name) | |
| with rasterio.open(path) as src: | |
| return src.read(1) | |
| ground_truth = read_tif('ground_truth.tif') | |
| knn_result = read_tif('knn_result.tif') | |
| training_ids = read_tif('training_polygons.tif') | |
| bands = np.stack( | |
| [read_tif(os.path.join('bands', f)) for f in BAND_FILES], axis=-1 | |
| ).astype(np.float32) | |
| with open(os.path.join(BASE_DIR, 'data', 'polygon_teacher_classes.json')) as f: | |
| polygon_teacher = {int(k): v for k, v in json.load(f).items()} | |
| # Pre-compute reference KNN confusion matrix (teacher labels, pixels where GT > 0) | |
| gt_flat = ground_truth.flatten() | |
| knn_flat = knn_result.flatten() | |
| valid = gt_flat > 0 | |
| knn_matrix = np.zeros((7, 7), dtype=np.int64) | |
| np.add.at(knn_matrix, (gt_flat[valid] - 1, knn_flat[valid] - 1), 1) | |
| knn_oa = knn_matrix.diagonal().sum() / knn_matrix.sum() | |
| h, w = ground_truth.shape | |
| stride = max(1, max(h, w) // MAP_MAX_DIM) | |
| return dict( | |
| polygon_teacher= polygon_teacher, | |
| ground_truth = ground_truth, | |
| knn_result = knn_result, | |
| training_ids = training_ids, | |
| bands = bands, | |
| knn_matrix = knn_matrix, | |
| knn_oa = knn_oa, | |
| stride = stride, | |
| ) | |
| DATA = load_data() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Model training (from the student's own labels + real spectral bands) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def train_and_predict(student_labels, k): | |
| """Train a KNN classifier on the student's 23 labeled polygons (using the | |
| real spectral bands), then evaluate it against Ground Truth and produce a | |
| downsampled classification map for display.""" | |
| bands = DATA['bands'] | |
| training_ids = DATA['training_ids'] | |
| ground_truth = DATA['ground_truth'] | |
| knn_result = DATA['knn_result'] | |
| stride = DATA['stride'] | |
| train_mask = training_ids > 0 | |
| X_train = bands[train_mask] | |
| y_train = np.array([int(student_labels[pid - 1]) for pid in training_ids[train_mask]]) | |
| scaler = StandardScaler().fit(X_train) | |
| clf = KNeighborsClassifier(n_neighbors=int(k), algorithm='kd_tree', n_jobs=-1) | |
| clf.fit(scaler.transform(X_train), y_train) | |
| # Accuracy / confusion matrix on the real Ground Truth pixels | |
| gt_mask = ground_truth > 0 | |
| X_gt = bands[gt_mask] | |
| pred_gt = clf.predict(scaler.transform(X_gt)) | |
| y_gt = ground_truth[gt_mask] | |
| model_matrix = np.zeros((7, 7), dtype=np.int64) | |
| np.add.at(model_matrix, (y_gt - 1, pred_gt - 1), 1) | |
| model_oa = model_matrix.diagonal().sum() / model_matrix.sum() | |
| # Downsampled full-image map for visualization (fast: predict only on the | |
| # decimated grid, not the full 4.6M pixels) | |
| bands_small = bands[::stride, ::stride] | |
| hs, ws, _ = bands_small.shape | |
| pred_small = clf.predict(scaler.transform(bands_small.reshape(-1, 7))) | |
| student_map = pred_small.reshape(hs, ws) | |
| gt_small = ground_truth[::stride, ::stride] | |
| knn_small = knn_result[::stride, ::stride] | |
| # zero out predictions outside the study area so the map matches GT/KNN extent | |
| student_map = np.where(knn_small > 0, student_map, 0) | |
| return model_oa, model_matrix, student_map, knn_small, gt_small | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Visualization helpers | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def fig_confusion_matrix(matrix, oa, title, cmap='Blues'): | |
| """Generic 7×7 confusion matrix plot: predicted classes vs Ground Truth.""" | |
| short = ["Eau", "Vergers", "Δ-Cult.", "Bâti", "Irr.-Dés.", "Non-Irr.", "Sable"] | |
| fig, ax = plt.subplots(figsize=(9, 7)) | |
| row_tot = matrix.sum(axis=1, keepdims=True) | |
| pct = np.where(row_tot > 0, matrix / row_tot * 100, 0) | |
| im = ax.imshow(pct, cmap=cmap, vmin=0, vmax=100) | |
| plt.colorbar(im, ax=ax, label="% de la classe réelle", shrink=0.8) | |
| ax.set_xticks(range(7)); ax.set_yticks(range(7)) | |
| ax.set_xticklabels([f"C{i+1}\n{short[i]}" for i in range(7)], fontsize=8) | |
| ax.set_yticklabels([f"C{i+1}\n{short[i]}" for i in range(7)], fontsize=8) | |
| ax.set_xlabel("Classe prédite (KNN)", fontsize=11, labelpad=8) | |
| ax.set_ylabel("Classe réelle (vérité terrain)", fontsize=11, labelpad=8) | |
| ax.set_title(f"{title}\nPrécision globale = {oa*100:.1f}%", | |
| fontsize=12, fontweight='bold', pad=12) | |
| for r in range(7): | |
| for c in range(7): | |
| v, p = matrix[r, c], pct[r, c] | |
| color = 'white' if p > 50 else 'black' | |
| ax.text(c, r, f"{v}\n({p:.0f}%)", ha='center', va='center', | |
| fontsize=7, color=color) | |
| plt.tight_layout() | |
| return fig | |
| def fig_knn_matrix(): | |
| """Reference matrix: pre-computed KNN (trained on teacher labels) vs Ground Truth.""" | |
| return fig_confusion_matrix( | |
| DATA['knn_matrix'], DATA['knn_oa'], | |
| "Matrice de confusion – Modèle de référence (enseignant) vs Vérité terrain", | |
| cmap='Blues', | |
| ) | |
| def fig_model_matrix(model_matrix, model_oa): | |
| """Student's own trained KNN model vs Ground Truth.""" | |
| return fig_confusion_matrix( | |
| model_matrix, model_oa, | |
| "Matrice de confusion – VOTRE modèle KNN vs Vérité terrain", | |
| cmap='Purples', | |
| ) | |
| def fig_three_panel_map(student_map, knn_map, gt_map): | |
| """Side-by-side classification maps: student's trained model / teacher | |
| reference KNN / Ground Truth.""" | |
| fig, axes = plt.subplots(1, 3, figsize=(15, 5)) | |
| for ax, rast, title in zip( | |
| axes, | |
| [student_map, knn_map, gt_map], | |
| ["Votre modèle KNN", "Modèle de référence (enseignant)", "Vérité terrain"], | |
| ): | |
| rgb = COLORS_RGB[rast] | |
| ax.imshow(rgb, interpolation='nearest') | |
| ax.set_title(title, fontsize=11, fontweight='bold') | |
| ax.axis('off') | |
| plt.tight_layout() | |
| return fig | |
| def fig_student_matrix(student_labels): | |
| """7×7 confusion matrix: student labels vs teacher labels.""" | |
| matrix = np.zeros((7, 7), dtype=int) | |
| for pid in range(1, 24): | |
| s = student_labels[pid - 1] | |
| t = DATA['polygon_teacher'].get(pid, 0) | |
| if s is not None and t > 0: | |
| matrix[t - 1][int(s) - 1] += 1 | |
| short = ["Eau", "Vergers", "Δ-Cult.", "Bâti", "Irr.-Dés.", "Non-Irr.", "Sable"] | |
| fig, ax = plt.subplots(figsize=(8, 6)) | |
| row_tot = matrix.sum(axis=1, keepdims=True) | |
| pct = np.where(row_tot > 0, matrix / row_tot * 100, 0) | |
| im = ax.imshow(pct, cmap='Greens', vmin=0, vmax=100) | |
| plt.colorbar(im, ax=ax, label="% de la classe enseignant", shrink=0.8) | |
| ax.set_xticks(range(7)); ax.set_yticks(range(7)) | |
| ax.set_xticklabels([f"C{i+1}\n{short[i]}" for i in range(7)], fontsize=8) | |
| ax.set_yticklabels([f"C{i+1}\n{short[i]}" for i in range(7)], fontsize=8) | |
| ax.set_xlabel("Votre réponse", fontsize=11, labelpad=8) | |
| ax.set_ylabel("Réponse de l'enseignant", fontsize=11, labelpad=8) | |
| ax.set_title("Votre interprétation vs Réponse de l'enseignant\n(polygones d'entraînement)", | |
| fontsize=11, fontweight='bold', pad=12) | |
| for r in range(7): | |
| for c in range(7): | |
| v, p = matrix[r, c], pct[r, c] | |
| color = 'white' if p > 50 else 'black' | |
| if v > 0: | |
| ax.text(c, r, f"{v}\n({p:.0f}%)", ha='center', va='center', | |
| fontsize=8, color=color) | |
| plt.tight_layout() | |
| return fig | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Gradio helpers | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def polygon_image_path(idx): | |
| return os.path.join(BASE_DIR, 'polygon_images', f'polygon_{idx+1:02d}.png') | |
| def count_labeled(labels): | |
| return sum(1 for l in labels if l is not None) | |
| def build_results_table(student_labels): | |
| rows = [] | |
| for pid in range(1, 24): | |
| s = student_labels[pid - 1] | |
| t = DATA['polygon_teacher'].get(pid, 0) | |
| s_str = f"{s} – {CLASSES.get(int(s), '?')}" if s is not None else "—" | |
| t_str = f"{t} – {CLASSES.get(t, '?')}" | |
| match = "✅" if (s is not None and int(s) == t) else ("❌" if s is not None else "—") | |
| rows.append([pid, s_str, t_str, match]) | |
| return rows | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Gradio Interface | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| with gr.Blocks(title="Interprétation de polygones – Vallée du Nil") as demo: | |
| # ── State ────────────────────────────────────────────────────────────── | |
| cur_idx = gr.State(value=0) | |
| labels = gr.State(value=[None] * N_POLYGONS) | |
| # ── Header ───────────────────────────────────────────────────────────── | |
| gr.Markdown(""" | |
| # 🛰️ Interprétation de polygones d'entraînement – Vallée du Nil | |
| **Objectif :** Interprétez visuellement chacun des 23 polygones d'entraînement, | |
| puis soumettez vos réponses pour générer la carte et la matrice de confusion. | |
| > Cette application s'inscrit dans un TD sur la classification d'occupation du sol par IA (algorithme KNN). | |
| """) | |
| with gr.Tabs() as tabs: | |
| # ── Tab 1: Légende des classes ──────────────────────────────────── | |
| with gr.Tab("📋 Classes d'occupation du sol"): | |
| gr.Markdown(""" | |
| ## Classes d'occupation du sol | |
| | N° | Classe | Couleur | | |
| |----|--------|---------| | |
| | 1 | Eau | 🔵 Bleu | | |
| | 2 | Vergers | 🟢 Vert foncé | | |
| | 3 | Cultures dans le Delta | 💚 Vert clair | | |
| | 4 | Zones bâties | 🔴 Rouge | | |
| | 5 | Cultures irriguées dans le désert | 🟠 Orange | | |
| | 6 | Cultures non irriguées en zone sèche | 🟤 Marron | | |
| | 7 | Zones sableuses | 🟡 Jaune clair | | |
| --- | |
| **Conseils d'interprétation :** | |
| - L'image montre une composition colorée de l'image Landsat | |
| - Les zones bleues/sombres correspondent à l'eau | |
| - La végétation dense apparaît en vert (vergers, cultures dans le delta) | |
| - Les zones bâties apparaissent en teintes rosées ou grises | |
| - Les zones cultivées irriguées dans le désert forment des parcelles géométriques | |
| - Les zones sableuses apparaissent en jaune/beige | |
| """) | |
| # ── Tab 2: Interprétation ───────────────────────────────────────── | |
| with gr.Tab("🖊️ Interprétation des polygones"): | |
| progress_md = gr.Markdown("**0 / 23 polygones étiquetés**") | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| polygon_title = gr.Markdown("### Polygone 1 / 23") | |
| polygon_img = gr.Image( | |
| value=polygon_image_path(0), | |
| label="Image satellite", | |
| show_label=False, | |
| height=500, | |
| ) | |
| with gr.Row(): | |
| btn_prev = gr.Button("◀ Précédent", size="sm", variant="secondary") | |
| btn_next = gr.Button("Suivant ▶", size="sm", variant="primary") | |
| with gr.Column(scale=1): | |
| gr.Markdown("### Classe du polygone") | |
| gr.Markdown("*(Cochez la classe qui correspond le mieux à l'occupation du sol visible dans le polygone)*") | |
| class_radio = gr.Radio( | |
| choices=CLASS_CHOICES, | |
| label="Classe", | |
| value=None, | |
| ) | |
| gr.Markdown("---") | |
| progress_bar = gr.Markdown("**Progression : 0 / 23**") | |
| gr.Markdown("---") | |
| gr.Markdown("### Entraînement du modèle KNN") | |
| gr.Markdown( | |
| "*(Une fois les 23 polygones étiquetés, ce modèle sera " | |
| "entraîné sur VOS réponses + les vraies valeurs spectrales " | |
| "des 7 bandes satellite, puis évalué sur la vérité terrain)*" | |
| ) | |
| k_slider = gr.Slider( | |
| minimum=K_MIN, maximum=K_MAX, value=K_DEFAULT, step=1, | |
| label="k (nombre de voisins)", | |
| ) | |
| btn_submit = gr.Button( | |
| "🚀 Entraîner mon modèle KNN et voir les résultats", | |
| variant="primary", | |
| visible=False, | |
| size="lg", | |
| ) | |
| # ── Tab 3: Résultats ─────────────────────────────────────────────── | |
| with gr.Tab("📊 Résultats", id="tab_results") as tab_results: | |
| results_placeholder = gr.Markdown( | |
| "*(Les résultats apparaîtront ici après soumission)*" | |
| ) | |
| gr.Markdown("## 1️⃣ Qualité de votre interprétation visuelle") | |
| accuracy_md = gr.Markdown(visible=False) | |
| results_tbl = gr.Dataframe( | |
| headers=["Polygone", "Votre réponse", "Réponse enseignant", "Résultat"], | |
| visible=False, | |
| wrap=True, | |
| ) | |
| with gr.Row(visible=False) as row_plots_student: | |
| student_matrix_plot = gr.Plot(label="Votre interprétation vs Enseignant") | |
| gr.Markdown("## 2️⃣ Votre modèle KNN entraîné sur vos étiquettes") | |
| model_accuracy_md = gr.Markdown(visible=False) | |
| with gr.Row(visible=False) as row_plots_model: | |
| model_matrix_plot = gr.Plot(label="Votre modèle KNN vs Vérité terrain") | |
| knn_plot = gr.Plot(label="Modèle de référence (enseignant) vs Vérité terrain") | |
| three_panel_plot = gr.Plot( | |
| label="Cartes de classification", visible=False, | |
| ) | |
| # ───────────────────────────────────────────────────────────────────── | |
| # Event handlers | |
| # ───────────────────────────────────────────────────────────────────── | |
| def nav_to(idx, current_labels): | |
| """Render a polygon slide: image, title, current radio value.""" | |
| img = polygon_image_path(idx) | |
| title = f"### Polygone {idx + 1} / {N_POLYGONS}" | |
| saved = current_labels[idx] | |
| radio_val = None | |
| if saved is not None: | |
| radio_val = CLASS_CHOICES[int(saved) - 1] | |
| n_done = count_labeled(current_labels) | |
| prog = f"**Progression : {n_done} / {N_POLYGONS}**" | |
| return img, title, radio_val, prog | |
| def on_prev(idx, current_labels): | |
| new_idx = max(0, idx - 1) | |
| img, title, radio_val, prog = nav_to(new_idx, current_labels) | |
| return new_idx, img, title, radio_val, prog | |
| def on_next(idx, current_labels): | |
| new_idx = min(N_POLYGONS - 1, idx + 1) | |
| img, title, radio_val, prog = nav_to(new_idx, current_labels) | |
| return new_idx, img, title, radio_val, prog | |
| def on_class_select(choice, idx, current_labels): | |
| """Save selected class for current polygon.""" | |
| if choice is None: | |
| return current_labels, gr.update(), gr.update(), gr.update() | |
| cls_num = int(choice.split(" - ")[0]) | |
| new_labels = list(current_labels) | |
| new_labels[idx] = cls_num | |
| n_done = count_labeled(new_labels) | |
| prog = f"**Progression : {n_done} / {N_POLYGONS}**" | |
| overall= f"**{n_done} / {N_POLYGONS} polygones étiquetés**" | |
| show_submit = (n_done == N_POLYGONS) | |
| return (new_labels, | |
| gr.update(value=prog), | |
| gr.update(value=overall), | |
| gr.update(visible=show_submit)) | |
| def on_submit(current_labels, k): | |
| """Compute and display all results: labeling accuracy vs teacher, then | |
| train a real KNN model on the student's labels + spectral bands and | |
| evaluate it against Ground Truth.""" | |
| # 1) Accuracy of the visual interpretation vs teacher's answer key | |
| correct = sum( | |
| 1 for pid in range(1, 24) | |
| if current_labels[pid-1] is not None | |
| and int(current_labels[pid-1]) == DATA['polygon_teacher'].get(pid, 0) | |
| ) | |
| total = count_labeled(current_labels) | |
| pct = correct / total * 100 if total > 0 else 0 | |
| acc_text = ( | |
| f"**{correct} / {total} polygones correctement identifiés ({pct:.0f}%)**\n\n" | |
| f"*(Comparaison avec la légende fournie par l'enseignant)*" | |
| ) | |
| table = build_results_table(current_labels) | |
| student_fig = fig_student_matrix(current_labels) | |
| # 2) Train a KNN model on the student's own labels + real spectral bands, | |
| # then evaluate it against Ground Truth | |
| model_oa, model_matrix, student_map, knn_map, gt_map = train_and_predict( | |
| current_labels, k | |
| ) | |
| model_acc_text = ( | |
| f"**Précision globale = {model_oa*100:.1f}%** " | |
| f"(évaluée sur les {int((DATA['ground_truth'] > 0).sum())} pixels de vérité terrain, k={int(k)})\n\n" | |
| f"*(C'est la précision réelle d'un modèle KNN entraîné uniquement sur VOS 23 polygones étiquetés — " | |
| f"comparez-la à celle du modèle de référence de l'enseignant ci-dessous)*" | |
| ) | |
| model_fig = fig_model_matrix(model_matrix, model_oa) | |
| knn_fig = fig_knn_matrix() | |
| map_fig = fig_three_panel_map(student_map, knn_map, gt_map) | |
| return ( | |
| gr.update(value=""), | |
| gr.update(value=acc_text, visible=True), | |
| gr.update(value=table, visible=True), | |
| gr.update(visible=True), | |
| gr.update(value=student_fig), | |
| gr.update(value=model_acc_text, visible=True), | |
| gr.update(visible=True), | |
| gr.update(value=model_fig), | |
| gr.update(value=knn_fig), | |
| gr.update(value=map_fig, visible=True), | |
| ) | |
| # Wire up navigation | |
| btn_prev.click( | |
| on_prev, | |
| inputs=[cur_idx, labels], | |
| outputs=[cur_idx, polygon_img, polygon_title, class_radio, progress_bar], | |
| ) | |
| btn_next.click( | |
| on_next, | |
| inputs=[cur_idx, labels], | |
| outputs=[cur_idx, polygon_img, polygon_title, class_radio, progress_bar], | |
| ) | |
| # Wire up class selection | |
| class_radio.change( | |
| on_class_select, | |
| inputs=[class_radio, cur_idx, labels], | |
| outputs=[labels, progress_bar, progress_md, btn_submit], | |
| ) | |
| # Wire up submit (labeling accuracy + model training/evaluation) | |
| btn_submit.click( | |
| on_submit, | |
| inputs=[labels, k_slider], | |
| outputs=[ | |
| results_placeholder, | |
| accuracy_md, | |
| results_tbl, | |
| row_plots_student, | |
| student_matrix_plot, | |
| model_accuracy_md, | |
| row_plots_model, | |
| model_matrix_plot, | |
| knn_plot, | |
| three_panel_plot, | |
| ], | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=7860) | |