import tempfile from flask import Flask, jsonify, render_template, request import os import numpy as np import tensorflow as tf from tensorflow.keras.models import load_model import joblib from PIL import Image from io import BytesIO import matplotlib.pyplot as plt import matplotlib matplotlib.use('Agg') import tifffile import requests from datetime import datetime, timedelta from huggingface_hub import hf_hub_download UPLOAD_FOLDER = "static/uploads" VIS_FOLDER = "static/visualizations" os.makedirs(UPLOAD_FOLDER, exist_ok=True) os.makedirs(VIS_FOLDER, exist_ok=True) app = Flask(__name__, static_folder='static', template_folder='templates') app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER model_tif_path = hf_hub_download(repo_id= "AlshimaaAhmed/landclassification-models" , filename = "TIF_classification.h5") model_rgb_path = hf_hub_download(repo_id= "AlshimaaAhmed/landclassification-models" , filename = "land_restnet_RGB.h5") pca_model_path = hf_hub_download(repo_id= "AlshimaaAhmed/landclassification-models" , filename = "pca_model.pkl") model_tif = load_model(model_tif_path) model_rgb = load_model(model_rgb_path) pca_model = joblib.load(pca_model_path) LABELS = [ 'AnnualCrop', 'Forest', 'HerbaceousVegetation', 'Highway', 'Industrial', 'Pasture', 'PermanentCrop', 'Residential', 'River', 'SeaLake' ] CLASS_COLORS = { 'AnnualCrop': (255, 205, 86), 'Forest': (34, 139, 34), 'HerbaceousVegetation': (144, 238, 144), 'Highway': (169, 169, 169), 'Industrial': (128, 0, 128), 'Pasture': (255, 165, 0), 'PermanentCrop': (60, 179, 113), 'Residential': (220, 20, 60), 'River': (30, 144, 255), 'SeaLake': (25, 25, 112) } CLIENT_ID = os.environ.get("CLIENT_ID") CLIENT_SECRET = os.environ.get("CLIENT_SECRET") def RGB_preprocessing(file_path): img = tf.keras.preprocessing.image.load_img(file_path, target_size=(224, 224)) arr = tf.keras.preprocessing.image.img_to_array(img).astype(np.float32) arr = arr / 255.0 arr = np.expand_dims(arr, axis=0) return arr def preprocess_tif_inference(file_path, target_size=(64,64)): img = tifffile.imread(file_path).astype(np.float32) if img.ndim == 2: img = np.expand_dims(img, axis=-1) img = tf.image.resize(img, target_size).numpy() img = img / 10000.0 img = np.expand_dims(img, axis=0) return img def save_pil(image: Image.Image, path: str, quality=90): image.save(path, format='PNG', optimize=True) def make_overlay_image(base_img_path, class_name, alpha=0.45, out_path=None): base = Image.open(base_img_path).convert("RGBA").resize((512,512)) color = CLASS_COLORS.get(class_name, (255, 0, 0)) overlay = Image.new("RGBA", base.size, color + (int(alpha*255),)) combined = Image.alpha_composite(base, overlay) if out_path: combined.save(out_path) return combined def make_bar_chart(probs, labels, out_path): plt.figure(figsize=(6,3)) plt.barh(labels, probs, color=[np.array(CLASS_COLORS[l])/255.0 for l in labels]) plt.xlim(0,1) plt.xlabel("Probability") plt.tight_layout() plt.savefig(out_path, bbox_inches='tight') plt.close() def build_urls(saved, overlay, chart): return { "image_url": f"/static/uploads/{os.path.basename(saved)}", "overlay_url": f"/static/visualizations/{os.path.basename(overlay)}", "chart_url": f"/static/visualizations/{os.path.basename(chart)}", } def get_access_token(): if not CLIENT_ID or not CLIENT_SECRET: return None token_url = "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token" data = { "grant_type": "client_credentials", "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET } r = requests.post(token_url, data=data, timeout=30) if r.status_code == 200: return r.json().get("access_token") return None def download_sentinel_rgb(lat, lon, out_path): token = get_access_token() if token is None: return None, "Missing CLIENT_ID/CLIENT_SECRET in environment." evalscript = """ //VERSION=3 function setup() { return { input: [{ bands: ["B04","B03","B02","SCL"], units: "DN" }], output: { bands: 3, sampleType: "AUTO" } }; } function evaluatePixel(sample) { if (sample.SCL == 3 || sample.SCL == 8 || sample.SCL == 9 || sample.SCL == 10) { return [0,0,0]; } let gain = 2.5; return [Math.max(0,Math.min(1,gain*sample.B04/10000)), Math.max(0,Math.min(1,gain*sample.B03/10000)), Math.max(0,Math.min(1,gain*sample.B02/10000))]; } """ buffer = 0.008 bbox = [lon-buffer, lat-buffer, lon+buffer, lat+buffer] request_body = { "input": {"bounds":{"bbox":bbox,"properties":{"crs":"http://www.opengis.net/def/crs/EPSG/0/4326"}}, "data":[{"type":"sentinel-2-l2a", "dataFilter":{"timeRange":{"from":(datetime.now()-timedelta(days=60)).strftime("%Y-%m-%dT00:00:00Z"), "to":datetime.now().strftime("%Y-%m-%dT23:59:59Z")}, "maxCloudCoverage":30,"mosaickingOrder":"leastCC"}}]}, "output":{"width":512,"height":512,"responses":[{"identifier":"default","format":{"type":"image/png"}}]}, "evalscript": evalscript } headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} try: r = requests.post("https://sh.dataspace.copernicus.eu/api/v1/process", json=request_body, headers=headers, timeout=90) if r.status_code == 200 and 'image' in r.headers.get('content-type',''): img = Image.open(BytesIO(r.content)).convert("RGB").resize((512,512)) img.save(out_path, "PNG") return out_path, None return None, f"Process API error: {r.status_code}, {r.text[:200]}" except Exception as e: return None, str(e) @app.route('/') def index(): return render_template('index.html') @app.route('/predict', methods=['POST']) def predict(): input_method = request.form.get('input_method') timestamp = int(datetime.now().timestamp()) if input_method == 'map': try: lat = float(request.form.get('latitude')) lon = float(request.form.get('longitude')) except Exception: return jsonify({"success": False, "message": "Invalid coordinates"}), 400 out_image_path = os.path.join(UPLOAD_FOLDER, f"sent_{lat:.6f}_{lon:.6f}_{timestamp}.png") downloaded, err = download_sentinel_rgb(lat, lon, out_image_path) if not downloaded: return jsonify({"success": False, "message": f"Could not download image: {err}"}), 500 arr = RGB_preprocessing(downloaded) pred = model_rgb.predict(arr)[0] idx = int(np.argmax(pred)) label = LABELS[idx] confidence = float(pred[idx]) * 100.0 overlay_path = os.path.join(VIS_FOLDER, f"overlay_{timestamp}.png") chart_path = os.path.join(VIS_FOLDER, f"chart_{timestamp}.png") make_overlay_image(downloaded, label, alpha=0.45, out_path=overlay_path) make_bar_chart(pred, LABELS, chart_path) return jsonify({ "success": True, "prediction": label, "confidence": f"{confidence:.1f}%", "location": f"({lat:.6f}, {lon:.6f})", "image_url": f"/static/uploads/{os.path.basename(downloaded)}", "overlay_url": f"/static/visualizations/{os.path.basename(overlay_path)}", "chart_url": f"/static/visualizations/{os.path.basename(chart_path)}" }) elif input_method == 'upload': if 'file' not in request.files: return jsonify({"success": False, "message": "No file uploaded"}), 400 f = request.files['file'] secure_name = f"{timestamp}_{f.filename.replace(' ', '_')}" saved_path = os.path.join(UPLOAD_FOLDER, secure_name) f.save(saved_path) file_type = request.form.get('file_type', 'rgb') if file_type == 'rgb': arr = RGB_preprocessing(saved_path) pred = model_rgb.predict(arr)[0] overlay_path = os.path.join(VIS_FOLDER, f"overlay_{timestamp}.png") make_overlay_image(saved_path, LABELS[int(np.argmax(pred))], alpha=0.45, out_path=overlay_path) else: arr = preprocess_tif_inference(saved_path) pred = model_tif.predict(arr)[0] overlay_path = None idx = int(np.argmax(pred)) label = LABELS[idx] confidence = float(pred[idx]) * 100.0 chart_path = os.path.join(VIS_FOLDER, f"chart_{timestamp}.png") make_bar_chart(pred, LABELS, chart_path) result = { "success": True, "prediction": label, "confidence": f"{confidence:.1f}%", "image_url": f"/static/uploads/{os.path.basename(saved_path)}", "chart_url": f"/static/visualizations/{os.path.basename(chart_path)}" } if overlay_path: result["overlay_url"] = f"/static/visualizations/{os.path.basename(overlay_path)}" return jsonify(result) return jsonify({"success": False, "message": "Invalid input method"}), 400 if __name__ == "__main__": app.run(host="0.0.0.0", port=7860)