import json import math from pathlib import Path from PIL import Image, ImageFilter import requests import numpy as np from scipy.interpolate import griddata # Multi-state approximate bounds (MH, Goa, KA) LAT_MIN, LAT_MAX = 11.5, 22.0 LON_MIN, LON_MAX = 73.0, 80.9 # For the PNG, we want high resolution PIXELS_PER_DEGREE = 50 WIDTH = int((LON_MAX - LON_MIN) * PIXELS_PER_DEGREE) HEIGHT = int((LAT_MAX - LAT_MIN) * PIXELS_PER_DEGREE) def interpolate_color(val, stops): if np.isnan(val): return stops[0][1] if val <= stops[0][0]: return stops[0][1] if val >= stops[-1][0]: return stops[-1][1] for i in range(len(stops) - 1): v1, c1 = stops[i] v2, c2 = stops[i+1] if v1 <= val <= v2: t = (val - v1) / (v2 - v1) return ( int(c1[0] + (c2[0] - c1[0]) * t), int(c1[1] + (c2[1] - c1[1]) * t), int(c1[2] + (c2[2] - c1[2]) * t), int(c1[3] + (c2[3] - c1[3]) * t) ) return stops[-1][1] def get_temp_color(temp): stops = [ (20, (254, 217, 118, 0)), (25, (254, 217, 118, 150)), (30, (254, 178, 76, 180)), (35, (253, 141, 60, 200)), (40, (252, 78, 42, 220)), (43, (227, 26, 28, 240)), (45, (177, 0, 38, 255)) ] return interpolate_color(temp, stops) def get_rain_color(rain): stops = [ (0, (198, 219, 239, 0)), (10, (198, 219, 239, 100)), (40, (158, 202, 225, 150)), (80, (107, 174, 214, 180)), (120, (66, 146, 198, 200)), (160, (33, 113, 181, 220)), (200, (8, 69, 148, 255)) ] return interpolate_color(rain, stops) def get_wind_color(speed): stops = [ (0, (100, 200, 255, 0)), (5, (100, 200, 255, 150)), (10, (255, 200, 50, 180)), (15, (255, 50, 50, 220)), (25, (180, 0, 0, 255)) ] return interpolate_color(speed, stops) def fetch_and_generate(out_dir): print("Generating sample points for Open-Meteo...") # Increase sample density to avoid huge interpolation gaps sample_lats = np.linspace(LAT_MIN - 0.5, LAT_MAX + 0.5, 10) sample_lons = np.linspace(LON_MIN - 0.5, LON_MAX + 0.5, 8) query_lats = [] query_lons = [] for lat in sample_lats: for lon in sample_lons: query_lats.append(round(lat, 2)) query_lons.append(round(lon, 2)) lat_str = ",".join(map(str, query_lats)) lon_str = ",".join(map(str, query_lons)) url = f"https://api.open-meteo.com/v1/forecast?latitude={lat_str}&longitude={lon_str}¤t=temperature_2m,precipitation,wind_speed_10m,wind_direction_10m&timezone=Asia/Kolkata" print("Fetching live data from Open-Meteo...") try: response = requests.get(url, timeout=10) response.raise_for_status() data = response.json() except Exception as e: print(f"Failed to fetch from Open-Meteo: {e}") return points = [] temp_values = [] rain_values = [] for i in range(len(query_lats)): loc_data = data[i] if isinstance(data, list) else data current = loc_data.get('current', {}) t = current.get('temperature_2m', 25.0) r = current.get('precipitation', 0.0) # Wind processing (speed in km/h, direction in degrees) wind_speed = current.get('wind_speed_10m', 0.0) wind_dir = current.get('wind_direction_10m', 0.0) # Convert to U and V components (m/s for standard wind layers) wind_speed_ms = wind_speed * (1000 / 3600) # Math: direction is meteorological (where wind comes FROM) # U = -speed * sin(dir) # V = -speed * cos(dir) wind_u = -wind_speed_ms * math.sin(math.radians(wind_dir)) wind_v = -wind_speed_ms * math.cos(math.radians(wind_dir)) points.append({ "lon": query_lons[i], "lat": query_lats[i], "temp": t, "rain": r, "wind_u": wind_u, "wind_v": wind_v }) print("Saving raw grid data to JSON...") out_file = Path(out_dir) / "weather_grid.json" with open(out_file, "w") as f: json.dump(points, f) print(f"Saved {len(points)} raw data points to {out_file} for WebGL interpolation.") # Extract values for scipy interpolation points_xy = [(p["lon"], p["lat"]) for p in points] temp_values = [p["temp"] for p in points] rain_values = [p["rain"] for p in points] wind_values = [math.sqrt(p["wind_u"]**2 + p["wind_v"]**2) for p in points] # Also generate high-resolution PNGs for smooth WebGL BitmapLayer interpolation print("Interpolating data to a high-resolution grid using scipy...") # Increase resolution to 100 pixels per degree for perfect smoothness PIXELS_PER_DEGREE = 100 WIDTH = int((LON_MAX - LON_MIN) * PIXELS_PER_DEGREE) HEIGHT = int((LAT_MAX - LAT_MIN) * PIXELS_PER_DEGREE) grid_x, grid_y = np.mgrid[LON_MIN:LON_MAX:complex(0, WIDTH), LAT_MIN:LAT_MAX:complex(0, HEIGHT)] grid_temp = griddata(points_xy, temp_values, (grid_x, grid_y), method='cubic') grid_rain = griddata(points_xy, rain_values, (grid_x, grid_y), method='cubic') grid_wind = griddata(points_xy, wind_values, (grid_x, grid_y), method='cubic') print("Generating Raster PNGs...") temp_img = Image.new('RGBA', (WIDTH, HEIGHT), (0, 0, 0, 0)) rain_img = Image.new('RGBA', (WIDTH, HEIGHT), (0, 0, 0, 0)) wind_img = Image.new('RGBA', (WIDTH, HEIGHT), (0, 0, 0, 0)) temp_pixels = temp_img.load() rain_pixels = rain_img.load() wind_pixels = wind_img.load() for y in range(HEIGHT): npy = HEIGHT - 1 - y for x in range(WIDTH): t = grid_temp[x, npy] r = grid_rain[x, npy] w = grid_wind[x, npy] temp_pixels[x, y] = get_temp_color(t) rain_pixels[x, y] = get_rain_color(r) wind_pixels[x, y] = get_wind_color(w) # Apply a slight blur to smooth out interpolation artifacts temp_img = temp_img.filter(ImageFilter.GaussianBlur(radius=2)) rain_img = rain_img.filter(ImageFilter.GaussianBlur(radius=2)) wind_img = wind_img.filter(ImageFilter.GaussianBlur(radius=2)) # Apply geographical mask from state_boundaries.json print("Applying geographical mask...") try: from PIL import ImageDraw with open("frontend/public/data/state_boundaries.json", "r") as f: bounds_data = json.load(f) mask_img = Image.new('L', (WIDTH, HEIGHT), 0) draw = ImageDraw.Draw(mask_img) geom = bounds_data["features"][0]["geometry"] def draw_poly(coords): pixels = [] for lon, lat in coords: px = int((lon - LON_MIN) / (LON_MAX - LON_MIN) * WIDTH) py = HEIGHT - 1 - int((lat - LAT_MIN) / (LAT_MAX - LAT_MIN) * HEIGHT) pixels.append((px, py)) draw.polygon(pixels, fill=255) if geom["type"] == "Polygon": draw_poly(geom["coordinates"][0]) elif geom["type"] == "MultiPolygon": for poly in geom["coordinates"]: draw_poly(poly[0]) # Apply the mask to both images temp_img.putalpha(mask_img) rain_img.putalpha(mask_img) wind_img.putalpha(mask_img) except Exception as e: print(f"Warning: Failed to apply geographic mask: {e}") # Save the PNGs temp_out = Path(out_dir) / "temp_grid.png" rain_out = Path(out_dir) / "rain_grid.png" wind_out = Path(out_dir) / "wind_grid.png" temp_img.save(temp_out) rain_img.save(rain_out) wind_img.save(wind_out) print(f"Saved raster overlays to {temp_out}, {rain_out}, and {wind_out}") # Generate a dense wind vector grid (e.g. every 10th pixel) print("Generating dense wind vector field...") grid_u = griddata(points_xy, [p["wind_u"] for p in points], (grid_x, grid_y), method='cubic') grid_v = griddata(points_xy, [p["wind_v"] for p in points], (grid_x, grid_y), method='cubic') dense_wind = [] # Sample every 25th pixel for arrows so it's not too crowded STEP = 25 for y in range(0, HEIGHT, STEP): npy = HEIGHT - 1 - y for x in range(0, WIDTH, STEP): lon = LON_MIN + (x / PIXELS_PER_DEGREE) lat = LAT_MIN + (npy / PIXELS_PER_DEGREE) u = grid_u[x, npy] v = grid_v[x, npy] if not np.isnan(u) and not np.isnan(v): dense_wind.append({ "lon": float(lon), "lat": float(lat), "wind_u": float(u), "wind_v": float(v) }) wind_out = Path(out_dir) / "wind_vectors.json" with open(wind_out, "w") as f: json.dump(dense_wind, f) print(f"Saved {len(dense_wind)} interpolated wind vectors to {wind_out}") if __name__ == "__main__": out_dir = Path("frontend/public/data") out_dir.mkdir(parents=True, exist_ok=True) fetch_and_generate(out_dir)