liss4_cloud_removal/

├── assets/
├── data/
│ ├── processed/
│ │ ├── patches_clear/
│ │ ├── patches_cloudy/
│ │ └── sentinel/
│ │ └── train/
│ │ ├── input/
│ │ └── target/
│ └── raw/
│ ├── cloud_free/
│ ├── cloudy/
│ └── sentinel/
├── docs/
├── final_outputs/
├── inference_results/
├── inference_sentinel_results/
├── inferencepics/
├── liss4_env/
│ ├── Include/
│ ├── Lib/
│ ├── Scripts/
│ ├── share/
│ └── [Link]
├── models/
├── sentinel_gan_samples/
├── sentinel_samples/
├── src/
│ ├── __pycache__/
│ ├── __init__.py
│ ├── [Link]
│ └── [Link]
├── create_patches.py
├── download_sentinel.py
├── evaluate_metrics.py
├── final_cloud_free_map.png
├── folder_stack.py
├── folder_stack_clear.py
├── generate_full_map.py
├── [Link]
├── inference_result1.[Link]
├── inference_sentinel.py
├── inspect_data.py
├── live_demo.py
├── live_generation_result.png
├── [Link]
├── prepare_sentinel_data.py
├── [Link]
├── reconstructed_full_map.[Link]
├── reconstructed_full_map11.tif
├── stack_bands.py
├── stack_bands_ncl.py
├── stitch_results.py
├── [Link]
├── train_sentinel.py
└── train_sentinel_gan.py
[Link] ->​
import os
import torch
from [Link] import Dataset
import numpy as np
from PIL import Image
from torchvision import transforms

# ==========================================
# TOOL 1: The Original LISS-4 Dataset Loader
# ==========================================
class SatellitePatchDataset(Dataset):
def __init__(self, cloudy_dir, clear_dir):
self.cloudy_dir = cloudy_dir
self.clear_dir = clear_dir
self.image_filenames = [Link](cloudy_dir)

[Link] = [Link]([
[Link](),
[Link]((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
])

def __len__(self):
return len(self.image_filenames)

def __getitem__(self, idx):

img_name = self.image_filenames[idx]
cloudy_path = [Link](self.cloudy_dir, img_name)
clear_path = [Link](self.clear_dir, img_name)

cloudy_img = [Link](cloudy_path).convert("RGB")
clear_img = [Link](clear_path).convert("RGB")

cloudy_tensor = [Link](cloudy_img)
clear_tensor = [Link](clear_img)

return {"cloudy": cloudy_tensor, "clear": clear_tensor}

# ==========================================
# TOOL 2: The New Sentinel-1 & 2 Dataset Loader
# ==========================================
class SentinelDataset(Dataset):
def __init__(self, data_dir, transform=None):
self.input_dir = [Link](data_dir, "input")
self.target_dir = [Link](data_dir, "target")
self.patch_files = sorted([f for f in [Link](self.input_dir)
if [Link](".npy")])
[Link] = transform

def __len__(self):
return len(self.patch_files)

def __getitem__(self, idx):

filename = self.patch_files[idx]

input_np = [Link]([Link](self.input_dir, filename))

target_np = [Link]([Link](self.target_dir, filename))

input_tensor = torch.from_numpy(input_np).float()
target_tensor = torch.from_numpy(target_np).float()

# Min-Max Normalization per channel

for c in range(input_tensor.shape[0]):
min_val = input_tensor[c].min()
max_val = input_tensor[c].max()
if max_val > min_val:
input_tensor[c] = (input_tensor[c] - min_val) / (max_val -
min_val)

for c in range(target_tensor.shape[0]):
min_val = target_tensor[c].min()
max_val = target_tensor[c].max()
if max_val > min_val:
target_tensor[c] = (target_tensor[c] - min_val) / (max_val
- min_val)

return input_tensor, target_tensor

​
[Link] _>
import torch
import [Link] as nn

# ==========================================
# 1. GENERATOR COMPONENTS (V2 ARCHITECTURE)
# ==========================================
class UNetDown([Link]):
def __init__(self, in_size, out_size, normalize=True, dropout=0.0):
super(UNetDown, self).__init__()
layers = [nn.Conv2d(in_size, out_size, kernel_size=4, stride=2,
padding=1, bias=False)]
if normalize:
[Link](nn.BatchNorm2d(out_size))
[Link]([Link](0.2))
if dropout:
[Link]([Link](dropout))
[Link] = [Link](*layers)

def forward(self, x):

return [Link](x)

class UNetUp([Link]):
def __init__(self, in_size, out_size, dropout=0.0):
super(UNetUp, self).__init__()

# --- THE V2 UPGRADE: Bilinear Upsampling ---

# This replaces ConvTranspose2d and permanently removes the
checkerboard grid.
layers = [
[Link](scale_factor=2, mode='bilinear',
align_corners=True),
nn.ReflectionPad2d(1),
nn.Conv2d(in_size, out_size, kernel_size=3, stride=1,
padding=0, bias=False), # The comma is safely here
nn.BatchNorm2d(out_size),
[Link](inplace=True)
]
if dropout:
[Link]([Link](dropout))
[Link] = [Link](*layers)
 def forward(self, x, skip_input):
x = [Link](x)
# Skip connection: concatenate along the channel axis
x = [Link]((x, skip_input), 1)
return x

class GeneratorUNet([Link]):
def __init__(self, in_channels=5, out_channels=3):
super(GeneratorUNet, self).__init__()

# Encoder (Downsampling)
self.down1 = UNetDown(in_channels, 64, normalize=False)
self.down2 = UNetDown(64, 128)
self.down3 = UNetDown(128, 256)
self.down4 = UNetDown(256, 512, dropout=0.5)
self.down5 = UNetDown(512, 512, dropout=0.5)
self.down6 = UNetDown(512, 512, dropout=0.5)
self.down7 = UNetDown(512, 512, dropout=0.5)
self.down8 = UNetDown(512, 512, normalize=False, dropout=0.5)

# Decoder (Upsampling with V2 Architecture)

self.up1 = UNetUp(512, 512, dropout=0.5)
self.up2 = UNetUp(1024, 512, dropout=0.5)
self.up3 = UNetUp(1024, 512, dropout=0.5)
self.up4 = UNetUp(1024, 512, dropout=0.0)
self.up5 = UNetUp(1024, 256, dropout=0.0)
self.up6 = UNetUp(512, 128, dropout=0.0)
self.up7 = UNetUp(256, 64, dropout=0.0)

# Final Layer (Maps back to RGB image)

[Link] = [Link](
[Link](scale_factor=2, mode='bilinear',
align_corners=True),
nn.ReflectionPad2d(1),
nn.Conv2d(128, out_channels, kernel_size=3, stride=1,
padding=0),
[Link]()
)

def forward(self, x, dataset_type=None):

 # Allow the dataset_type kwarg from your training script to pass
through safely
d1 = self.down1(x)
d2 = self.down2(d1)
d3 = self.down3(d2)
d4 = self.down4(d3)
d5 = self.down5(d4)
d6 = self.down6(d5)
d7 = self.down7(d6)
d8 = self.down8(d7)

u1 = self.up1(d8, d7)
u2 = self.up2(u1, d6)
u3 = self.up3(u2, d5)
u4 = self.up4(u3, d4)
u5 = self.up5(u4, d3)
u6 = self.up6(u5, d2)
u7 = self.up7(u6, d1)

return [Link](u7)

# ==========================================
# 2. DISCRIMINATOR COMPONENT
# ==========================================
class Discriminator([Link]):
def __init__(self, in_channels=5, target_channels=3):
super(Discriminator, self).__init__()

def discriminator_block(in_filters, out_filters,

normalization=True):
layers = [nn.Conv2d(in_filters, out_filters, kernel_size=4,
stride=2, padding=1)]
if normalization:
[Link](nn.BatchNorm2d(out_filters))
[Link]([Link](0.2, inplace=True))
return layers

# The Discriminator looks at BOTH the 5-channel input and the

3-channel generated/real image
total_in_channels = in_channels + target_channels
 [Link] = [Link](
*discriminator_block(total_in_channels, 64,
normalization=False),
*discriminator_block(64, 128),
*discriminator_block(128, 256),
*discriminator_block(256, 512),
nn.ZeroPad2d((1, 0, 1, 0)),
nn.Conv2d(512, 1, kernel_size=4, padding=1, bias=False)
)

def forward(self, img_target, img_input):

# Concatenate condition (input) and target/generated image
img_concat = [Link]((img_target, img_input), 1)
return [Link](img_concat)
​
Create_patches.py ->​
import os
import rasterio
from [Link] import Window
import numpy as np

def batch_generate_patches(stacked_dir, output_dir, patch_size=256):

"""
Finds all stacked .tif files in a folder and slices them into 256x256
patches.
"""
[Link](output_dir, exist_ok=True)

# Find all .tif files in the stacked directory

stacked_files = [f for f in [Link](stacked_dir) if
[Link]('.tif')]

if len(stacked_files) == 0:
print(f"No stacked .tif files found in {stacked_dir}")
return

print(f"Found {len(stacked_files)} stacked images in {stacked_dir}.

Chopping them up...\n")
 total_patches = 0

for file_name in stacked_files:

input_path = [Link](stacked_dir, file_name)
base_name = file_name.replace(".tif", "")

with [Link](input_path) as src:

width = [Link]
height = [Link]

# Slide a 256x256 window across the image

for row in range(0, height, patch_size):
for col in range(0, width, patch_size):
window = Window(col, row, patch_size, patch_size)
data = [Link](window=window)

# Skip patches that hit the edge (not exactly 256x256)

or are completely blank
if [Link][1:] != (patch_size, patch_size) or
[Link](data) == 0:
continue

# Copy metadata for the patch

kwargs = [Link]()
[Link]({
'height': patch_size,
'width': patch_size,
'transform': [Link](window,
[Link])
})

# Name the patch uniquely based on its parent image

and grid coordinates
patch_filename = f"{base_name}_patch_{row}_{col}.tif"
out_path = [Link](output_dir, patch_filename)

with [Link](out_path, 'w', **kwargs) as dst:

[Link](data)

total_patches += 1
 print(f"Finished slicing: {file_name}")

print(f"\nSuccess! Generated a total of {total_patches} patches in

{output_dir}")

if __name__ == "__main__":
# --- Cloudy Data ---
# Points to the output folder made by folder_stack.py
cloudy_stacked = r"data\raw\cloudy\stacked_output"
cloudy_patches = r"data\processed\patches_cloudy"

print("--- Processing Cloudy Images ---")

batch_generate_patches(cloudy_stacked, cloudy_patches)

# --- Clear Data ---

# Points to the output folder made by folder_stack_clear.py
clear_stacked = r"data\raw\cloud_free\stacked_output"
clear_patches = r"data\processed\patches_clear"

print("\n--- Processing Clear Images ---")

batch_generate_patches(clear_stacked, clear_patches)
​
​
download_sentinel .py ->​
import os
import planetary_computer as pc
from pystac_client import Client
import rasterio
from [Link] import from_bounds
from [Link] import transform_bounds
import datetime

def download_sentinel_triplet():
print("Initializing Microsoft Planetary Computer API...")

# 1. Connect to Microsoft STAC API

catalog = [Link](
"[Link]
modifier=pc.sign_inplace,
)
# 2. Define Area (Middle India - Lat/Lon Degrees)
bbox = [78.50, 20.50, 79.50, 21.50]

# 3. Create Output Folders

output_dir = [Link]("data", "raw", "sentinel")
[Link](output_dir, exist_ok=True)
print(f"Data will be saved to: {output_dir}\n")

# ==========================================
# STEP A: GET THE "ANSWER KEY" (Clear Season)
# ==========================================
print("--- Searching for Clear Target Image (Jan - May) ---")
clear_search = [Link](
collections=["sentinel-2-l2a"],
bbox=bbox,
datetime="2023-01-01/2023-05-30",
query={"eo:cloud_cover": {"lt": 5}} # Less than 5% clouds
)
clear_items = list(clear_search.items())
best_clear = clear_items[0]
print(f"[+] Found Clear S2 Image from: {best_clear.datetime}")

# ==========================================
# STEP B: GET THE "PROBLEM" (Monsoon Season)
# ==========================================
print("\n--- Searching for Cloudy Input Image (Jun - Aug) ---")
cloudy_search = [Link](
collections=["sentinel-2-l2a"],
bbox=bbox,
datetime="2023-06-01/2023-08-30",
query={"eo:cloud_cover": {"gt": 40, "lt": 80}} # 40% to 80% cloudy
)
cloudy_items = list(cloudy_search.items())
best_cloudy = cloudy_items[0]
cloudy_date_str = best_cloudy.[Link]("%Y-%m-%d")
print(f"[+] Found Cloudy S2 Image from: {best_cloudy.datetime}")

# ==========================================
# STEP C: GET THE "X-RAY" (Radar on the exact Cloudy Day)
 # ==========================================
print(f"\n--- Searching for Radar Image near {cloudy_date_str} ---")

cloudy_date = best_cloudy.datetime
start_date = (cloudy_date -
[Link](days=2)).strftime("%Y-%m-%d")
end_date = (cloudy_date +
[Link](days=2)).strftime("%Y-%m-%d")
radar_time_window = f"{start_date}/{end_date}"

radar_search = [Link](
collections=["sentinel-1-rtc"],
bbox=bbox,
datetime=radar_time_window
)
radar_items = list(radar_search.items())
best_radar = radar_items[0]
print(f"[+] Found matching S1 Radar Image from:
{best_radar.datetime}\n")

# ==========================================
# STEP D: DOWNLOAD PIXELS TO HARD DRIVE
# ==========================================
print("--- Starting Direct Cloud-to-Disk Download ---")

def download_band(asset_href, output_filename):

with [Link](asset_href) as src:
# FIX: Convert lat/lon degrees (EPSG:4326) into the native
image projection (UTM meters)
left, bottom, right, top = transform_bounds("EPSG:4326",
[Link], *bbox)

# Now calculate the pixel window using the correctly matching

projection units
window = from_bounds(left, bottom, right, top,
transform=[Link])

# Read only the pixels within our window

data = [Link](1, window=window)
 # Save the cropped band to our local folder
out_meta = [Link]()
out_meta.update({
"driver": "GTiff",
"height": [Link][0], # Pull height directly from read
data size
"width": [Link][1], # Pull width directly from read
data size
"transform": src.window_transform(window)
})

out_path = [Link](output_dir, output_filename)

with [Link](out_path, "w", **out_meta) as dest:
[Link](data, 1)
print(f"Saved: {output_filename} (Size:
{[Link][1]}x{[Link][0]})")

# 1. Download Clear Optical Bands

print("Downloading Clear Target...")
download_band(best_clear.assets["B04"].href, "clear_S2_Red.tif")
download_band(best_clear.assets["B03"].href, "clear_S2_Green.tif")
download_band(best_clear.assets["B02"].href, "clear_S2_Blue.tif")

# 2. Download Cloudy Optical Bands

print("Downloading Cloudy Input...")
download_band(best_cloudy.assets["B04"].href, "cloudy_S2_Red.tif")
download_band(best_cloudy.assets["B03"].href, "cloudy_S2_Green.tif")
download_band(best_cloudy.assets["B02"].href, "cloudy_S2_Blue.tif")

# 3. Download Radar Bands

print("Downloading Radar Input...")
download_band(best_radar.assets["vv"].href, "radar_S1_VV.tif")
download_band(best_radar.assets["vh"].href, "radar_S1_VH.tif")

print("\n[+] Download complete! All files saved directly to

data/raw/sentinel")

if __name__ == "__main__":
download_sentinel_triplet()
​
Evaluate_metrics.py ->​
import os
import torch
import numpy as np
from [Link] import peak_signal_noise_ratio as compute_psnr
from [Link] import structural_similarity as compute_ssim
from [Link] import GeneratorUNet

def evaluate():
print("[*] Starting V2 Model Evaluation...")
device = [Link]("cuda" if [Link].is_available() else "cpu")

# 1. Load the V2 Brain

generator = GeneratorUNet().to(device)
checkpoint_path = "models/checkpoint_latest.pth"

if not [Link](checkpoint_path):
print(f"[!] Error: No trained model found at {checkpoint_path}!")
return

checkpoint = [Link](checkpoint_path, map_location=device)

generator.load_state_dict([Link]('generator_state_dict',
checkpoint), strict=False)
[Link]()

input_dir = [Link]("data", "processed", "sentinel", "train",

"input")
target_dir = [Link]("data", "processed", "sentinel", "train",
"target")

files = [f for f in [Link](input_dir) if [Link]('.npy')]

total_psnr = 0.0
total_ssim = 0.0

print(f"[*] Evaluating {len(files)} patches...")

# 2. Evaluation Loop
with torch.no_grad():
 for idx, file_name in enumerate(files):
in_arr = [Link]([Link](input_dir,
file_name)).astype(np.float32)
tar_arr = [Link]([Link](target_dir,
file_name)).astype(np.float32)

# Normalize Input for the AI

in_min, in_max = [Link](in_arr), [Link](in_arr)
if in_max > in_min: in_arr = (in_arr - in_min) / (in_max -
in_min)
in_arr_norm = (in_arr - 0.5) / 0.5

# --- SHAPE SAFETY CHECK FOR INPUT ---

in_tensor = torch.from_numpy(in_arr_norm)
if in_tensor.shape[-1] == 5: # If [256, 256, 5] -> [5,
256, 256]
in_tensor = in_tensor.permute(2, 0, 1)
elif in_tensor.shape[1] == 5: # If [256, 5, 256] -> [5,
256, 256]
in_tensor = in_tensor.permute(1, 0, 2)
# If already [5, 256, 256], do nothing

in_tensor = in_tensor.unsqueeze(0).to(device)
in_tensor = torch.nan_to_num(in_tensor, nan=0.0)

# --- SHAPE SAFETY CHECK FOR TARGET ---

tar_min, tar_max = [Link](tar_arr), [Link](tar_arr)
if tar_max > tar_min: tar_arr = (tar_arr - tar_min) / (tar_max
- tar_min)

# skimage expects (Height, Width, Channels) [256, 256, 3]

if tar_arr.shape[0] == 3: # If [3, 256, 256]
tar_arr = [Link](tar_arr, (1, 2, 0))
elif tar_arr.shape[1] == 3: # If [256, 3, 256]
tar_arr = [Link](tar_arr, (0, 2, 1))

# Generate AI Prediction
ai_prediction = generator(in_tensor, dataset_type="sentinel")
ai_img = (ai_prediction.cpu().squeeze().numpy() + 1) / 2.0
ai_img = [Link]([Link](ai_img, (1, 2, 0)), 0, 1)
 # 3. Calculate Math Metrics
patch_psnr = compute_psnr(tar_arr, ai_img, data_range=1.0)
patch_ssim = compute_ssim(tar_arr, ai_img, data_range=1.0,
channel_axis=2)

total_psnr += patch_psnr
total_ssim += patch_ssim

if (idx + 1) % 50 == 0 or (idx + 1) == len(files):

print(f" -> Processed {idx + 1}/{len(files)}")

# 4. Averages
avg_psnr = total_psnr / len(files)
avg_ssim = total_ssim / len(files)

print("\n========================================")
print(" 🏆 FINAL V2 MODEL METRICS 🏆")
print("========================================")
print(f"Average PSNR: {avg_psnr:.2f} dB")
print(f"Average SSIM: {avg_ssim:.4f}")
print("========================================")

if __name__ == "__main__":
evaluate()
​
folder_stack_ [Link] ->
import os
import rasterio

def stack_clear_folders(main_raw_dir, output_stacked_dir):

"""
Looks inside every sub-folder in the cloud_free directory,
stacks the bands, and names the output after the folder.
"""
[Link](output_stacked_dir, exist_ok=True)

print(f"Scanning for clear scene folders in: {main_raw_dir}\n")

success_count = 0
 # Loop through everything inside the clear directory
for folder_name in [Link](main_raw_dir):
folder_path = [Link](main_raw_dir, folder_name)

# Make sure it's a directory and skip the output folder itself
if [Link](folder_path) and folder_name != "stacked_output":

# Match the band names inside the folder

b2_path = [Link](folder_path, "[Link]")
b3_path = [Link](folder_path, "[Link]")
b4_path = [Link](folder_path, "[Link]")

# Verify all bands exist

if [Link](b2_path) and [Link](b3_path) and
[Link](b4_path):
print(f"Stacking clear data inside folder: {folder_name}
...")

output_path = [Link](output_stacked_dir,
f"stacked_{folder_name}.tif")

# Read metadata from Band 2

with [Link](b2_path) as src_g:
meta = src_g.[Link]()
[Link](count=3)

# Write the 3 bands into a single stacked image

with [Link](b2_path) as src_g, \
[Link](b3_path) as src_r, \
[Link](b4_path) as src_nir:

with [Link](output_path, 'w', **meta) as dst:

[Link](src_g.read(1), 1) # Green
[Link](src_r.read(1), 2) # Red
[Link](src_nir.read(1), 3) # NIR

success_count += 1
else:
print(f"[Skipped] Clear folder '{folder_name}' is missing
band files.")
 print(f"\nOperation Complete! Successfully stacked {success_count}
clear scenes.")

if __name__ == "__main__":
# Target paths set specifically for cloud_free data
clear_folders = r"data\raw\cloud_free"
output_directory = r"data\raw\cloud_free\stacked_output"

stack_clear_folders(clear_folders, output_directory)
​
Folder_stack.py ->
import os
import rasterio

def stack_from_folders(main_raw_dir, output_stacked_dir):

"""
Looks inside every sub-folder for generically named band files,
stacks them, and names the output after the folder.
"""
[Link](output_stacked_dir, exist_ok=True)

print(f"Scanning for scene folders in: {main_raw_dir}\n")

success_count = 0

# Loop through everything inside your main directory

for folder_name in [Link](main_raw_dir):
folder_path = [Link](main_raw_dir, folder_name)

# Make sure it's actually a folder, and skip our output folder
if [Link](folder_path) and folder_name != "stacked_output":

# --- IMPORTANT: Update these names if your files are named

differently (e.g., "[Link]") ---
b2_path = [Link](folder_path, "[Link]")
b3_path = [Link](folder_path, "[Link]")
b4_path = [Link](folder_path, "[Link]")

# Check if all 3 files exist inside this specific folder

 if [Link](b2_path) and [Link](b3_path) and
[Link](b4_path):
print(f"Stacking data inside folder: {folder_name} ...")

output_path = [Link](output_stacked_dir,
f"stacked_{folder_name}.tif")

# Stack the bands

with [Link](b2_path) as src_g:
meta = src_g.[Link]()
[Link](count=3)

with [Link](b2_path) as src_g, \

[Link](b3_path) as src_r, \
[Link](b4_path) as src_nir:

with [Link](output_path, 'w', **meta) as dst:

[Link](src_g.read(1), 1)
[Link](src_r.read(1), 2)
[Link](src_nir.read(1), 3)

success_count += 1
else:
print(f"[Skipped] Folder '{folder_name}' is missing one or
more band files.")

print(f"\nOperation Complete! Successfully stacked {success_count}

scenes.")

if __name__ == "__main__":
# Point this to the directory containing all your extracted scene
folders
cloudy_folders = r"data\raw\cloudy"
output_directory = r"data\raw\cloudy\stacked_output"

stack_from_folders(cloudy_folders, output_directory)
​
Generate_full_map.py ->
import os
import torch
import numpy as np
from PIL import Image
from [Link] import GeneratorUNet
import math
import re

def generate_full_map(city_name="cherrapunji"):
print(f"[*] Initializing Full Map Reconstruction for
{city_name.upper()}...")
device = [Link]("cuda" if [Link].is_available() else "cpu")

# 1. Load the V2 Brain

generator = GeneratorUNet().to(device)
checkpoint_path = "models/checkpoint_latest.pth"

if not [Link](checkpoint_path):
print(f"[!] Error: No trained model found at {checkpoint_path}!")
return

checkpoint = [Link](checkpoint_path, map_location=device)

generator.load_state_dict([Link]('generator_state_dict',
checkpoint), strict=False)
[Link]()
print("[+] V2 Model loaded successfully.")

# 2. Grab all patches for the specific city

input_dir = [Link]("data", "processed", "sentinel", "train",
"input")
files = [f for f in [Link](input_dir) if [Link](city_name)
and [Link]('.npy')]

# Sort files numerically so they stitch together in the correct

geographical order
try:
[Link](key=lambda x: int([Link](r'patch_(\d+)',
x).group(1)))
except Exception as e:
print("[!] Error sorting patches. Make sure patch names contain
numbers.")
 if not files:
print(f"[!] No patches found for {city_name}.")
return

print(f"[*] Found {len(files)} patches. Processing and stitching...")

# 3. Setup the massive canvas

patch_size = 256
# Calculate grid dimensions based on total patches
grid_dim = [Link]([Link](len(files)))

full_clear_map = [Link]('RGB', (grid_dim * patch_size, grid_dim *

patch_size))
full_cloudy_map = [Link]('RGB', (grid_dim * patch_size, grid_dim *
patch_size))

# 4. Process and Stitch Loop

for idx, file_name in enumerate(files):
# Load and Normalize
in_arr = [Link]([Link](input_dir,
file_name)).astype(np.float32)

in_min, in_max = [Link](in_arr), [Link](in_arr)

if in_max > in_min: in_arr = (in_arr - in_min) / (in_max - in_min)
in_arr_norm = (in_arr - 0.5) / 0.5

in_tensor = torch.from_numpy(in_arr_norm).permute(2, 0,
1).unsqueeze(0).to(device)
in_tensor = torch.nan_to_num(in_tensor, nan=0.0)

# AI Prediction
with torch.no_grad():
ai_prediction = generator(in_tensor, dataset_type="sentinel")

# Convert AI Tensor to Image

ai_img = (ai_prediction.cpu().squeeze().numpy() + 1) / 2.0
ai_img = [Link]([Link](ai_img, (1, 2, 0)), 0, 1)
ai_img_uint8 = (ai_img * 255).astype(np.uint8)
ai_pil = [Link](ai_img_uint8)
 # Convert Cloudy Tensor to Image (Channels 2, 3, 4 are RGB)
cloudy_rgb = in_arr[:, :, 2:5]
cloudy_rgb = [Link](cloudy_rgb, 0, 1)
cloudy_uint8 = (cloudy_rgb * 255).astype(np.uint8)
cloudy_pil = [Link](cloudy_uint8)

# Calculate exact pixel coordinates on the giant canvas

row = idx // grid_dim
col = idx % grid_dim
x_offset = col * patch_size
y_offset = row * patch_size

# Paste the patches onto the canvas

full_clear_map.paste(ai_pil, (x_offset, y_offset))
full_cloudy_map.paste(cloudy_pil, (x_offset, y_offset))

# Print progress so you know it hasn't crashed

if (idx + 1) % 10 == 0 or (idx + 1) == len(files):
print(f" -> Processed patch {idx+1}/{len(files)}")

# 5. Save the final Masterpieces

[Link]("final_outputs", exist_ok=True)
clear_path = f"final_outputs/{city_name}_V2_Clear_Map.png"
cloudy_path = f"final_outputs/{city_name}_Cloudy_Input.png"

full_clear_map.save(clear_path)
full_cloudy_map.save(cloudy_path)

print(f"\n[+] MASSIVE SUCCESS! Maps saved to final_outputs/")

if __name__ == "__main__":
cities = ["cherrapunji", "mumbai", "bengaluru"]

for city in cities:

print(f"\n{'='*50}")
print(f" 🚀 STARTING MAPPING SEQUENCE FOR: {[Link]()}")
print(f"{'='*50}")
generate_full_map(city)

Inference_sentinel.py ->
import os
import torch
import numpy as np
import [Link] as plt
from [Link] import GeneratorUNet
import random

def run_inference():
print("[*] Initializing Hackathon Inference Engine...")
device = [Link]("cuda" if [Link].is_available() else "cpu")

# 1. Load the AI Brain

print("[*] Loading trained model weights...")
generator = GeneratorUNet().to(device)

# Make sure this points to the checkpoint you want to test!

# If you renamed your Epoch 181 backup, change this string to match
it.
checkpoint_path = "models/checkpoint_latest.pth"

if not [Link](checkpoint_path):
print(f"[!] Error: No trained model found at {checkpoint_path}!")
return

checkpoint = [Link](checkpoint_path, map_location=device)

generator.load_state_dict([Link]('generator_state_dict',
checkpoint), strict=False)
[Link]() # Set model to "test" mode
print("[+] Model loaded successfully.")

# 2. Grab a Random Test Image

input_dir = [Link]("data", "processed", "sentinel", "train",
"input")
target_dir = [Link]("data", "processed", "sentinel", "train",
"target")

all_files = [f for f in [Link](input_dir) if [Link]('.npy')]

test_file = [Link](all_files)
print(f"[*] Testing on patch: {test_file}")
 # 3. Load and Normalize (Explosion-Proofing)
in_arr = [Link]([Link](input_dir,
test_file)).astype(np.float32)
tar_arr = [Link]([Link](target_dir,
test_file)).astype(np.float32)

in_min, in_max = [Link](in_arr), [Link](in_arr)

if in_max > in_min: in_arr = (in_arr - in_min) / (in_max - in_min)
tar_min, tar_max = [Link](tar_arr), [Link](tar_arr)
if tar_max > tar_min: tar_arr = (tar_arr - tar_min) / (tar_max -
tar_min)

in_arr = (in_arr - 0.5) / 0.5

tar_arr = (tar_arr - 0.5) / 0.5

in_tensor = torch.from_numpy(in_arr).permute(2, 0,
1).unsqueeze(0).to(device)
in_tensor = torch.nan_to_num(in_tensor, nan=0.0)

# 4. Generate the Cloud-Free Image!

print("[*] AI is generating topography...")
with torch.no_grad():
ai_prediction = generator(in_tensor, dataset_type="sentinel")

# 5. Format for Visualization

ai_img = (ai_prediction.cpu().squeeze().numpy() + 1) / 2.0
ai_img = [Link]([Link](ai_img, (1, 2, 0)), 0, 1)

tar_img = (tar_arr + 1) / 2.0

# Extract RGB from the 5-channel input

cloudy_rgb = (in_arr[:, :, 2:5] + 1) / 2.0
cloudy_rgb = [Link](cloudy_rgb, 0, 1)

# 6. Plot the Side-by-Side Dashboard

# Saving directly to your existing inference_sentinel_results folder
[Link]("inference_sentinel_results", exist_ok=True)

[Link](figsize=(15, 5))
 [Link](1, 3, 1)
[Link]("1. Input (Cloudy Optical + SAR)")
[Link](cloudy_rgb)
[Link]('off')

[Link](1, 3, 2)
[Link]("2. Ground Truth (Clear Optical)")
[Link](tar_img)
[Link]('off')

[Link](1, 3, 3)
[Link]("3. AI Generation (V1 Output)")
[Link](ai_img)
[Link]('off')

output_path = f"inference_sentinel_results/{test_file.replace('.npy',
'.png')}"
plt.tight_layout()
[Link](output_path, dpi=300, bbox_inches='tight')
print(f"\n[+] SUCCESS! Check the folder for: {output_path}")

if __name__ == "__main__":
run_inference()
​
[Link]->
import os
import torch
import rasterio
import numpy as np
from [Link] import save_image

from [Link] import GeneratorUNet

def test_ai_model():
device = [Link]("cuda" if [Link].is_available() else "cpu")
print(f"Preparing AI on: {device}")

# Initialize model
generator = GeneratorUNet().to(device)
 # ==========================================
# 1. LOAD THE BRAIN (UNPACKING CHECKPOINT)
# ==========================================
weights_path = [Link]("models", "checkpoint_latest.pth")

if [Link](weights_path):
# Load the master checkpoint dictionary
checkpoint = [Link](weights_path, map_location=device)

# Extract strictly the Generator's weights

if "generator_state_dict" in checkpoint:
# strict=False allows it to try loading even if the
architecture shifted slightly
generator.load_state_dict(checkpoint["generator_state_dict"],
strict=False)
print(f"[+] Successfully extracted Generator weights from
{weights_path}")
else:
generator.load_state_dict(checkpoint, strict=False)
print(f"[+] Loaded raw weights from {weights_path}")
else:
print(f"\n[!!!] STOPPING: Could not find '{weights_path}'\n")
return

[Link]()

# ==========================================
# 2. SETUP DIRECTORIES
# ==========================================
input_dir = [Link]("data", "processed", "patches_cloudy")
output_dir = "inference_results"
[Link](output_dir, exist_ok=True)

valid_extensions = (".png", ".jpg", ".jpeg", ".tif", ".tiff")

image_files = [f for f in [Link](input_dir) if
[Link]().endswith(valid_extensions)]

print(f"Found {len(image_files)} images total.")

# LIMIT TO 50 IMAGES FOR TESTING

 test_batch = image_files
print(f"Processing a test batch of {len(test_batch)} images...")

# ==========================================
# 3. RUN INFERENCE USING RASTERIO
# ==========================================
with torch.no_grad():
for img_name in test_batch:
img_path = [Link](input_dir, img_name)

try:
with [Link](img_path) as src:
img_array = [Link]()

if img_array.shape[0] >= 3:
img_array = img_array[:3, :, :]

img_array = (img_array.astype(np.float32) / 255.0 -

0.5) / 0.5
cloudy_tensor =
torch.from_numpy(img_array).unsqueeze(0).to(device)

# Pass through the model using the LISS adapter

fake_clear = generator(cloudy_tensor, dataset_type="liss")

output_path = [Link](output_dir,
f"clear_{img_name.replace('.tif', '.png')}")
save_image(fake_clear, output_path, normalize=True)
print(f"Saved: {output_path}")

except Exception as e:
print(f"[!] Skipping {img_name} due to error: {e}")

print("\n[+] Test inference finished! Check the 'inference_results'

folder.")

if __name__ == "__main__":
test_ai_model()

Inspect_data.py ->
import rasterio
import numpy as np
import os

def inspect_satellite_image(image_path):
"""
Opens a LISS-IV GeoTIFF file, reads its metadata,
and verifies its shape and spectral integrity.
"""
if not [Link](image_path):
raise FileNotFoundError(f"Target image not found at:
{image_path}")

print(f"--- Inspecting: {[Link](image_path)} ---")

with [Link](image_path) as src:

width = [Link]
height = [Link]
band_count = [Link]
crs = [Link]
transform = [Link]

print(f"Dimensions : {width} x {height} pixels")

print(f"Total Bands: {band_count}")
print(f"CRS : {crs}")
print(f"Transform :\n{transform}")

bands_data = []
for i in range(1, band_count + 1):
band = [Link](i)
bands_data.append(band)

print(f"\nBand {i} Stats:")

print(f" Data Type: {[Link]}")
print(f" Min Value: {[Link](band)}")
print(f" Max Value: {[Link](band)}")
print(f" Mean : {[Link](band):.2f}")

full_image_matrix = [Link](bands_data, axis=0)

print(f"\nFinal Data Array Shape: {full_image_matrix.shape}")
 return full_image_matrix, [Link]

if __name__ == "__main__":
# Placeholder path for execution
sample_path = r"data\raw\cloudy\sample_liss4.tif"

try:
matrix, metadata = inspect_satellite_image(sample_path)
except FileNotFoundError as e:
print(f"\n[Setup Status]: Setup complete. Waiting for raw data
input. Error caught successfully: {e}")

Live_demo.py ->
import torch
import numpy as np
from PIL import Image
from [Link] import GeneratorUNet

print(" 🚀 Waking up the V2 Neural Network...")

# 1. Load Your Custom AI Brain
device = [Link]("cuda" if [Link].is_available() else "cpu")
generator = GeneratorUNet().to(device)

# Load the weights you just trained

checkpoint = [Link]("models/checkpoint_latest.pth",
map_location=device)
generator.load_state_dict([Link]('generator_state_dict',
checkpoint), strict=False)
[Link]()
print("[+] Model loaded successfully.")

# 2. Grab a raw cloudy/radar data patch (Change this filename to any patch
you want)
patch_path = "data/processed/sentinel/train/input/mumbai_patch_10.npy"
in_arr = [Link](patch_path).astype(np.float32)

# Normalize the data exactly how the AI expects it

in_min, in_max = [Link](in_arr), [Link](in_arr)
if in_max > in_min: in_arr = (in_arr - in_min) / (in_max - in_min)
in_arr_norm = (in_arr - 0.5) / 0.5

# Convert to PyTorch Tensor and fix the shape

in_tensor = torch.from_numpy(in_arr_norm)
if in_tensor.shape[-1] == 5:
in_tensor = in_tensor.permute(2, 0, 1)
in_tensor = in_tensor.unsqueeze(0).to(device)
in_tensor = torch.nan_to_num(in_tensor, nan=0.0)

print("[*] Stripping clouds and reconstructing topography...")

# 3. Generate the Image using the GAN

with torch.no_grad():
prediction = generator(in_tensor, dataset_type="sentinel")

# 4. Convert the AI's mathematical output back into a visible picture

out_img = ([Link]().squeeze().numpy() + 1) / 2.0
out_img = [Link]([Link](out_img, (1, 2, 0)), 0, 1)
final_image = [Link]((out_img * 255).astype(np.uint8))

# Save it to your folder

save_path = "live_generation_result.png"
final_image.save(save_path)

print(f" ✅ SUCCESS! Image mathematically reconstructed and saved as:

{save_path}")

Prepare_sentinel_data.py ->
import os
import numpy as np
import rasterio
from [Link] import Resampling
from [Link] import Window

def reproject_to_match(master_src, slave_path):

"""Reads a slave image and resamples it to match the exact grid/shape
of the master image."""
with [Link](slave_path) as slave_src:
 # Read slave data and resample it dynamically to match master's
width and height
data = slave_src.read(
1,
out_shape=(master_src.height, master_src.width),
resampling=[Link]
)
return data

def process_sentinel_dataset():
raw_dir = [Link]("data", "raw", "sentinel")
processed_dir = [Link]("data", "processed", "sentinel")

train_input_dir = [Link](processed_dir, "train", "input")

train_target_dir = [Link](processed_dir, "train", "target")
[Link](train_input_dir, exist_ok=True)
[Link](train_target_dir, exist_ok=True)

print("Opening master alignment template (cloudy_S2_Red.tif)...")

master_path = [Link](raw_dir, "cloudy_S2_Red.tif")

with [Link](master_path) as master_src:

height = master_src.height
width = master_src.width
print(f"Master Scene Dimensions: {width}x{height} pixels.")

# --- Load and Align Input Bands (5 Channels total) ---

print("\nAligning and stacking 5-channel Input tensors...")
input_stack = [Link]((5, height, width), dtype=np.float32)
input_stack[0] = master_src.read(1) # Cloudy Red (already matches)
input_stack[1] = reproject_to_match(master_src,
[Link](raw_dir, "cloudy_S2_Green.tif"))
input_stack[2] = reproject_to_match(master_src,
[Link](raw_dir, "cloudy_S2_Blue.tif"))
input_stack[3] = reproject_to_match(master_src,
[Link](raw_dir, "radar_S1_VV.tif"))
input_stack[4] = reproject_to_match(master_src,
[Link](raw_dir, "radar_S1_VH.tif"))

# --- Load and Align Target Bands (3 Channels total) ---

 print("Aligning and stacking 3-channel Target tensors...")
target_stack = [Link]((3, height, width), dtype=np.float32)
target_stack[0] = reproject_to_match(master_src,
[Link](raw_dir, "clear_S2_Red.tif"))
target_stack[1] = reproject_to_match(master_src,
[Link](raw_dir, "clear_S2_Green.tif"))
target_stack[2] = reproject_to_match(master_src,
[Link](raw_dir, "clear_S2_Blue.tif"))

# --- Slice into 256x256 Patches ---

patch_size = 256
patch_count = 0
print(f"\nSlicing master scenes into {patch_size}x{patch_size}
patches...")

for y in range(0, height - patch_size, patch_size):

for x in range(0, width - patch_size, patch_size):
# Extract 256x256 blocks across all channels
input_patch = input_stack[:, y:y+patch_size,
x:x+patch_size]
target_patch = target_stack[:, y:y+patch_size,
x:x+patch_size]

# Filter out patches that contain invalid or dead edge

pixels (0.0 values)
if [Link](input_patch[0] == 0) or [Link](target_patch[0]
== 0):
continue

# Save patches as standard numpy files for lightning-fast

PyTorch loading
[Link]([Link](train_input_dir,
f"patch_{patch_count}.npy"), input_patch)
[Link]([Link](train_target_dir,
f"patch_{patch_count}.npy"), target_patch)
patch_count += 1

print(f"\n[+] Preprocessing Complete! Generated {patch_count}

clean training patch pairs.")
print(f"Patches saved successfully to: {processed_dir}")
if __name__ == "__main__":
process_sentinel_dataset()

[Link]->
import os
import glob
from PIL import Image

def stitch_patches():
# Increase PIL's pixel limit for massive satellite images
Image.MAX_IMAGE_PIXELS = None

input_dir = "inference_results"
output_file = "final_cloud_free_map.png"
patch_size = 256

print("Scanning inference results...")

images = [Link]([Link](input_dir, "clear_*_patch_*.png"))

if not images:
print("[!] No images found in inference_results folder.")
return

max_x = 0
max_y = 0

print("Calculating final canvas dimensions...")

# Read coordinates from the filenames to find the edges of the map
for img_path in images:
basename = [Link](img_path).replace(".png", "")
parts = [Link]("_")
try:
x = int(parts[-1])
y = int(parts[-2])
max_x = max(max_x, x)
max_y = max(max_y, y)
except ValueError:
continue
 canvas_width = max_x + patch_size
canvas_height = max_y + patch_size
print(f"Canvas size will be: {canvas_width} x {canvas_height}
pixels.")

# Create a massive blank canvas

canvas = [Link]("RGB", (canvas_width, canvas_height))

print(f"Stitching {len(images)} patches together. This may take a

moment...")

for img_path in images:

basename = [Link](img_path).replace(".png", "")
parts = [Link]("_")
try:
x = int(parts[-1])
y = int(parts[-2])
patch = [Link](img_path)
[Link](patch, (x, y))
except Exception as e:
print(f"Skipping {basename} due to error: {e}")

print("Saving the master file...")

[Link](output_file)
print(f"\n[+] Success! Master image saved as {output_file}")

if __name__ == "__main__":
stitch_patches()

Stack_band_ncl.py ->import rasterio

import os

def stack_satellite_bands(band_green_path, band_red_path, band_nir_path,

output_path):
"""
Reads 3 individual LISS-IV band files and stacks them into a single
multi-band GeoTIFF for AI processing.
"""
print("Reading individual bands...")
 # Read the spatial metadata from the first band
with [Link](band_green_path) as src_g:
meta = src_g.[Link]()

# Update the metadata to tell it we now have 3 bands instead of 1

[Link](count=3)

# Create the output folder if it doesn't exist

[Link]([Link](output_path), exist_ok=True)

# Open all three files and write them into the new stacked file
with [Link](band_green_path) as src_g, \
[Link](band_red_path) as src_r, \
[Link](band_nir_path) as src_nir:

with [Link](output_path, 'w', **meta) as dst:

print("Writing Band 1 (Green)...")
[Link](src_g.read(1), 1)

print("Writing Band 2 (Red)...")

[Link](src_r.read(1), 2)

print("Writing Band 3 (Near-Infrared)...")

[Link](src_nir.read(1), 3)

print(f"\nSuccess! Stacked image saved to: {output_path}")

if __name__ == "__main__":
# --- UPDATE THESE PATHS TO MATCH YOUR DOWNLOADED FILES ---
# For example, if your folder is named 'cloudy_scene' and contains the
3 bands:

green = r"data\raw\cloud_free\[Link]" # Replace with your actual

Green file name
red = r"data\raw\cloud_free\[Link]" # Replace with your actual
Red file name
nir = r"data\raw\cloud_free\[Link]" # Replace with your actual
NIR file name

# This is the single file we will feed to the AI later

 output_stacked = r"data\raw\cloud_free\stacked_cloudy_scene.tif"

try:
stack_satellite_bands(green, red, nir, output_stacked)
except FileNotFoundError as e:
print(f"\n[Error]: Could not find one of the files. Please check
your file paths! Details: {e}")

stack_band->import rasterio
import os

def stack_satellite_bands(band_green_path, band_red_path, band_nir_path,

output_path):
"""
Reads 3 individual LISS-IV band files and stacks them into a single
multi-band GeoTIFF for AI processing.
"""
print("Reading individual bands...")

# Read the spatial metadata from the first band

with [Link](band_green_path) as src_g:
meta = src_g.[Link]()

# Update the metadata to tell it we now have 3 bands instead of 1

[Link](count=3)

# Create the output folder if it doesn't exist

[Link]([Link](output_path), exist_ok=True)

# Open all three files and write them into the new stacked file
with [Link](band_green_path) as src_g, \
[Link](band_red_path) as src_r, \
[Link](band_nir_path) as src_nir:

with [Link](output_path, 'w', **meta) as dst:

print("Writing Band 1 (Green)...")
[Link](src_g.read(1), 1)

print("Writing Band 2 (Red)...")

[Link](src_r.read(1), 2)
 print("Writing Band 3 (Near-Infrared)...")
[Link](src_nir.read(1), 3)

print(f"\nSuccess! Stacked image saved to: {output_path}")

if __name__ == "__main__":
# --- UPDATE THESE PATHS TO MATCH YOUR DOWNLOADED FILES ---
# For example, if your folder is named 'cloudy_scene' and contains the
3 bands:

green = r"data\raw\cloudy\[Link]" # Replace with your actual Green

file name
red = r"data\raw\cloudy\[Link]" # Replace with your actual Red
file name
nir = r"data\raw\cloudy\[Link]" # Replace with your actual NIR
file name

# This is the single file we will feed to the AI later

output_stacked = r"data\raw\cloudy\stacked_cloudy_scene.tif"

try:
stack_satellite_bands(green, red, nir, output_stacked)
except FileNotFoundError as e:
print(f"\n[Error]: Could not find one of the files. Please check
your file paths! Details: {e}")

Stick_result ->
import os
import glob
from PIL import Image

def stitch_patches():
# Remove PIL pixel limit for large satellite imagery scale
Image.MAX_IMAGE_PIXELS = None

input_dir = "inference_results"
output_file = "final_cloud_free_map.png"
patch_size = 256
 print("Scanning inference results...")
images = [Link]([Link](input_dir, "clear_*_patch_*.png"))

if not images:
print("[!] No images found in inference_results folder.")
return

max_x = 0
max_y = 0

print("Calculating final canvas dimensions...")

for img_path in images:
basename = [Link](img_path).replace(".png", "")
parts = [Link]("_")
try:
x = int(parts[-1])
y = int(parts[-2])
max_x = max(max_x, x)
max_y = max(max_y, y)
except ValueError:
continue

canvas_width = max_x + patch_size

canvas_height = max_y + patch_size
print(f"Canvas size calculated: {canvas_width} x {canvas_height}
pixels.")

# Create blank canvas

canvas = [Link]("RGB", (canvas_width, canvas_height))

print(f"Stitching {len(images)} patches together...")

for img_path in images:

basename = [Link](img_path).replace(".png", "")
parts = [Link]("_")
try:
x = int(parts[-1])
y = int(parts[-2])
patch = [Link](img_path)
 [Link](patch, (x, y))
except Exception as e:
print(f"Skipping {basename} due to error: {e}")

print("Saving the final stitched map...")

[Link](output_file)
print(f"\n[+] Success! Master image saved as {output_file}")

if __name__ == "__main__":
stitch_patches()

Train_sentinel_gan.py ->
import os
import torch
import [Link] as nn
import [Link] as optim
from [Link] import Dataset, DataLoader
import numpy as np
from [Link] import save_image
import ee

from [Link] import GeneratorUNet, Discriminator

# =====================================================================
# 1. DATA ACQUISITION BLOCK
# =====================================================================
GEE_PROJECT = "norse-quest-474800-i5"

try:
[Link](project=GEE_PROJECT)
except Exception:
[Link]()
[Link](project=GEE_PROJECT)

CITIES = {
"mumbai": [72.75, 18.85, 73.10, 19.25],
"cherrapunji": [91.60, 25.15, 91.90, 25.45],
"bengaluru": [77.45, 12.85, 77.75, 13.15]
}
INPUT_DIR = [Link]("data", "processed", "sentinel", "train",
"input")
TARGET_DIR = [Link]("data", "processed", "sentinel", "train",
"target")
[Link](INPUT_DIR, exist_ok=True)
[Link](TARGET_DIR, exist_ok=True)

PATCH_SIZE = 256
TILE_STEP = 0.04

if len([Link](INPUT_DIR)) <= 10:

print(f"\n[!] Dataset incomplete. Starting extraction...")
for city_name, bbox in [Link]():
print(f"[*] Processing {city_name}...")
lon_steps = [Link](bbox[0], bbox[2], TILE_STEP)
lat_steps = [Link](bbox[1], bbox[3], TILE_STEP)

patch_count = 0
for x_start in lon_steps:
for y_start in lat_steps:
x_end = min(x_start + TILE_STEP, bbox[2])
y_end = min(y_start + TILE_STEP, bbox[3])
tile_roi = [Link]([x_start, y_start, x_end,
y_end])

try:
s1 =
[Link]('COPERNICUS/S1_GRD').filterBounds(tile_roi).filterDate(
'2025-06-01', '2025-09-01').median().select(['VV', 'VH'])
s2_cloudy =
[Link]('COPERNICUS/S2_SR_HARMONIZED').filterBounds(tile_roi).f
ilterDate('2025-06-01',
'2025-09-01').filter([Link]('CLOUDY_PIXEL_PERCENTAGE',
50)).median().select(['B4', 'B3', 'B2'])
s2_clear =
[Link]('COPERNICUS/S2_SR_HARMONIZED').filterBounds(tile_roi).f
ilterDate('2025-11-01',
'2026-02-01').filter([Link]('CLOUDY_PIXEL_PERCENTAGE',
5)).median().select(['B4', 'B3', 'B2'])
 input_stack = [Link]([s1,
s2_cloudy]).reproject(crs='EPSG:4326', scale=10)
target_stack = s2_clear.reproject(crs='EPSG:4326',
scale=10)

input_info =
input_stack.sampleRectangle(region=tile_roi, defaultValue=0).getInfo()
target_info =
target_stack.sampleRectangle(region=tile_roi, defaultValue=0).getInfo()

in_matrix =
[Link]([[Link](input_info['properties'][b]) for b in ['VV', 'VH',
'B4', 'B3', 'B2']], axis=-1)
tar_matrix =
[Link]([[Link](target_info['properties'][b]) for b in ['B4', 'B3',
'B2']], axis=-1)

h, w, _ = in_matrix.shape
for y in range(0, h - PATCH_SIZE, PATCH_SIZE):
for x in range(0, w - PATCH_SIZE, PATCH_SIZE):
in_patch = in_matrix[y:y+PATCH_SIZE,
x:x+PATCH_SIZE, :]
tar_patch = tar_matrix[y:y+PATCH_SIZE,
x:x+PATCH_SIZE, :]
if in_patch.shape == (PATCH_SIZE, PATCH_SIZE,
5) and tar_patch.shape == (PATCH_SIZE, PATCH_SIZE, 3):
patch_id =
f"{city_name}_patch_{patch_count}.npy"
[Link]([Link](INPUT_DIR, patch_id),
in_patch)
[Link]([Link](TARGET_DIR,
patch_id), tar_patch)
patch_count += 1
except Exception:
continue
else:
print(f"\n[+] Dataset found. Skipping GEE.")

# =====================================================================
# 2. DATASET DEFINITION
# =====================================================================
class SentinelDataset(Dataset):
def __init__(self, input_dir, target_dir):
self.input_dir = input_dir
self.target_dir = target_dir
self.file_names = [f for f in [Link](input_dir) if
[Link]('.npy')]

def __len__(self): return len(self.file_names)

def __getitem__(self, idx):

file_name = self.file_names[idx]
in_arr = [Link]([Link](self.input_dir,
file_name)).astype(np.float32)
tar_arr = [Link]([Link](self.target_dir,
file_name)).astype(np.float32)

# --- EXPLOSION-PROOF NORMALIZATION ---

in_min, in_max = [Link](in_arr), [Link](in_arr)
if in_max > in_min:
in_arr = (in_arr - in_min) / (in_max - in_min)

tar_min, tar_max = [Link](tar_arr), [Link](tar_arr)

if tar_max > tar_min:
tar_arr = (tar_arr - tar_min) / (tar_max - tar_min)

in_arr = (in_arr - 0.5) / 0.5

tar_arr = (tar_arr - 0.5) / 0.5

in_tensor = torch.from_numpy(in_arr)
tar_tensor = torch.from_numpy(tar_arr)

if in_tensor.shape[-1] == 5:
in_tensor = in_tensor.permute(2, 0, 1)
if tar_tensor.shape[-1] == 3:
tar_tensor = tar_tensor.permute(2, 0, 1)

in_tensor = torch.nan_to_num(in_tensor, nan=0.0)

tar_tensor = torch.nan_to_num(tar_tensor, nan=0.0)
 return in_tensor, tar_tensor

# =====================================================================
# 3. GAN TRAINING PIPELINE
# =====================================================================
def train_gan():
device = [Link]("cuda" if [Link].is_available() else "cpu")
print(f"[*] Using device: {device}")

generator = GeneratorUNet().to(device)
discriminator = Discriminator().to(device)

dataset = SentinelDataset(INPUT_DIR, TARGET_DIR)

dataloader = DataLoader(dataset, batch_size=4, shuffle=True,
drop_last=True)

# Conservatively lowered learning rate for stability

lr = 0.00005
optimizer_G = [Link]([Link](), lr=lr, betas=(0.5,
0.999))
optimizer_D = [Link]([Link](), lr=lr,
betas=(0.5, 0.999))

criterion_GAN = [Link]()
criterion_L1 = nn.L1Loss()

# Reduced pixel penalty to allow the generator to form organic

structures
lambda_pixel = 10

[Link]("models", exist_ok=True)
checkpoint_path = "models/checkpoint_latest.pth"

start_epoch = 0
if [Link](checkpoint_path):
print("[*] Found existing checkpoint. Loading weights...")
checkpoint = [Link](checkpoint_path, map_location=device)
generator.load_state_dict([Link]('generator_state_dict',
checkpoint), strict=False)
if 'discriminator_state_dict' in checkpoint:
discriminator.load_state_dict(checkpoint['discriminator_state_dict'],
strict=False)
if 'optimizer_G_state_dict' in checkpoint:

optimizer_G.load_state_dict(checkpoint['optimizer_G_state_dict'])

optimizer_D.load_state_dict(checkpoint['optimizer_D_state_dict'])
else:
print("[!] Optimizers not found. Starting fresh optimizers.")
start_epoch = [Link]('epoch', -1) + 1
print(f"[*] Resuming from Epoch {start_epoch}")

[Link]("sentinel_gan_samples", exist_ok=True)

print("\n[*] Starting Training Loop...")

# INCREASED MAX EPOCHS TO 500: You can safely run this script over and
over.
for epoch in range(start_epoch, 500):
for i, (imgs_in, imgs_tar) in enumerate(dataloader):
imgs_in, imgs_tar = imgs_in.to(device), imgs_tar.to(device)

with torch.no_grad():
dummy = discriminator(imgs_tar, imgs_in)
valid = torch.ones_like(dummy, device=device)
fake = torch.zeros_like(dummy, device=device)

# ------------------
# Train Generator
# ------------------
optimizer_G.zero_grad()
gen_imgs = generator(imgs_in, dataset_type="sentinel")
loss_G = criterion_GAN(discriminator(gen_imgs, imgs_in),
valid) + lambda_pixel * criterion_L1(gen_imgs, imgs_tar)
loss_G.backward()

# Gradient clipping (Prevents the white square explosion)

[Link].clip_grad_norm_([Link](),
max_norm=1.0)
 optimizer_G.step()

# ---------------------
# Train Discriminator
# ---------------------
optimizer_D.zero_grad()
loss_D = 0.5 * (criterion_GAN(discriminator(imgs_tar,
imgs_in), valid) +
criterion_GAN(discriminator(gen_imgs.detach(),
imgs_in), fake))
loss_D.backward()

# Gradient clipping (Prevents the Discriminator from getting

too strong)
[Link].clip_grad_norm_([Link](),
max_norm=1.0)

optimizer_D.step()

if i % 20 == 0:
print(f"[Epoch {epoch}/500] [Batch {i}] [D loss:
{loss_D.item():.4f}] [G loss: {loss_G.item():.4f}]")
save_image((gen_imgs.data[:1] + 1) / 2.0,
f"sentinel_gan_samples/ep{epoch}_b{i}.png")

[Link]({
'epoch': epoch,
'generator_state_dict': generator.state_dict(),
'discriminator_state_dict': discriminator.state_dict(),
'optimizer_G_state_dict': optimizer_G.state_dict(),
'optimizer_D_state_dict': optimizer_D.state_dict()
}, checkpoint_path)

if __name__ == "__main__":
train_gan()

train_sentinel.py->
import os
import torch
import [Link] as nn
import [Link] as optim
from [Link] import Dataset, DataLoader
import numpy as np
from [Link] import save_image

from [Link] import GeneratorUNet

# ==========================================
# 1. SENTINEL DATA LOADER
# ==========================================
class SentinelDataset(Dataset):
def __init__(self, input_dir, target_dir):
self.input_dir = input_dir
self.target_dir = target_dir
# Only load the .npy files
[Link] = [f for f in [Link](input_dir) if
[Link]('.npy')]

def __len__(self):
return len([Link])

def __getitem__(self, idx):

file_name = [Link][idx]

# Load 5-Channel Cloudy Input (Radar + Optical)

in_arr = [Link]([Link](self.input_dir,
file_name)).astype(np.float32)
if in_arr.shape[-1] == 5:
in_arr = [Link](in_arr, (2, 0, 1))

# Load 3-Channel Clear Target (Optical Only)

tar_arr = [Link]([Link](self.target_dir,
file_name)).astype(np.float32)
if tar_arr.shape[-1] == 3:
tar_arr = [Link](tar_arr, (2, 0, 1))

# ==========================================
# THE FIX: NORMALIZATION (Signal Step-Down)
# ==========================================
# Safely scale the Input array to [-1, 1]
 in_min, in_max = [Link](in_arr), [Link](in_arr)
if in_max - in_min > 0:
in_arr = (in_arr - in_min) / (in_max - in_min) # Scale to [0,
1]
in_arr = (in_arr - 0.5) / 0.5 # Scale to [-1,
1]

# Safely scale the Target array to [-1, 1]

tar_min, tar_max = [Link](tar_arr), [Link](tar_arr)
if tar_max - tar_min > 0:
tar_arr = (tar_arr - tar_min) / (tar_max - tar_min)
tar_arr = (tar_arr - 0.5) / 0.5

return torch.from_numpy(in_arr), torch.from_numpy(tar_arr)

# ==========================================
# 2. MAIN TRAINING LOOP
# ==========================================
def train_sentinel():
device = [Link]("cuda" if [Link].is_available() else "cpu")
print(f"Preparing Sentinel Training on: {device}")

# Initialize Model
generator = GeneratorUNet().to(device)

# CRITICAL: Load the existing master brain!

weights_path = [Link]("models", "checkpoint_latest.pth")
if [Link](weights_path):
checkpoint = [Link](weights_path, map_location=device)
if "generator_state_dict" in checkpoint:
generator.load_state_dict(checkpoint["generator_state_dict"],
strict=False)
else:
generator.load_state_dict(checkpoint, strict=False)
print("[+] Successfully loaded the master brain. Adding Sentinel
knowledge...")
else:
print("[!] CRITICAL ERROR: Could not find checkpoint_latest.pth.
Stop training!")
return
 optimizer_G = [Link]([Link](), lr=0.0002,
betas=(0.5, 0.999))
criterion_L1 = nn.L1Loss()

# Define Data Folders

input_dir = [Link]("data", "processed", "sentinel", "train",
"input")
target_dir = [Link]("data", "processed", "sentinel", "train",
"target")

dataset = SentinelDataset(input_dir, target_dir)

# Using batch_size=2 so your CPU doesn't get overwhelmed by the heavy
5-channel data
dataloader = DataLoader(dataset, batch_size=2, shuffle=True)

[Link]("sentinel_samples", exist_ok=True)

# We only need 5 epochs to teach the new adapter the basics

num_epochs = 150
print(f"\nStarting Sentinel training for {num_epochs} epochs...")

for epoch in range(num_epochs):

for i, (real_cloudy, real_clear) in enumerate(dataloader):
real_cloudy = real_cloudy.to(device)
real_clear = real_clear.to(device)

optimizer_G.zero_grad()

# THE FIX: Tell the AI to use the Sentinel Adapter

fake_clear = generator(real_cloudy, dataset_type="sentinel")

loss_G = criterion_L1(fake_clear, real_clear)

loss_G.backward()
optimizer_G.step()

# Print progress and save a sample image every 10 batches

if i % 10 == 0:
print(f"[Epoch {epoch}/{num_epochs}] [Batch
{i}/{len(dataloader)}] [Loss: {loss_G.item():.4f}]")
 save_image(fake_clear,
f"sentinel_samples/epoch_{epoch}_batch_{i}.png", normalize=True)

# Save the updated master brain after every epoch

[Link]({"generator_state_dict": generator.state_dict()},
weights_path)
print(f"[+] Master checkpoint updated for Epoch {epoch}")

print("\n[+] Sentinel Training Complete!")

if __name__ == "__main__":
train_sentinel()

[Link]->
import os
import time
import torch
import [Link] as nn
from [Link] import DataLoader

# Import the components

from [Link] import GeneratorUNet, Discriminator
from [Link] import SatellitePatchDataset

def train_gan():
# 1. Hyperparameters
epochs = 10
batch_size = 4
lr = 0.0002
lambda_pixel = 100.0
SAVE_INTERVAL_SECONDS = 300 # 300 seconds = 5 minutes

device = [Link]("cuda" if [Link].is_available() else "cpu")

print(f"Using processing device: {device}")

# 2. Setup Data Paths

cloudy_patches_dir = r"data\processed\patches_cloudy"
clear_patches_dir = r"data\processed\patches_clear"

dataset = SatellitePatchDataset(cloudy_patches_dir, clear_patches_dir)

 dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

# 3. Initialize Models
generator = GeneratorUNet(in_channels=3, out_channels=3).to(device)
discriminator = Discriminator(in_channels=3).to(device)

# 4. Define Loss Functions

criterion_GAN = [Link]()
criterion_pixel = nn.L1Loss()

# 5. Optimizers
optimizer_G = [Link]([Link](), lr=lr,
betas=(0.5, 0.999))
optimizer_D = [Link]([Link](), lr=lr,
betas=(0.5, 0.999))

# 6. Checkpoint Loading (The "Resume" Function with ZERO-WAIT FIX)

start_epoch = 1
checkpoint_path = "models/checkpoint_latest.pth"
[Link]("models", exist_ok=True)

if [Link](checkpoint_path):
print("\n[+] Found existing save file! Loading AI memories...")
checkpoint = [Link](checkpoint_path, map_location=device)

generator.load_state_dict(checkpoint['generator_state_dict'])

discriminator.load_state_dict(checkpoint['discriminator_state_dict'])
optimizer_G.load_state_dict(checkpoint['optimizer_G_state_dict'])
optimizer_D.load_state_dict(checkpoint['optimizer_D_state_dict'])

start_epoch = checkpoint['epoch']

# --- THE ZERO-WAIT FIX ---

# Instantly start at batch 0 with shuffled data instead of waiting
5 mins
start_batch = 0

print(f"[+] Resuming training from Epoch {start_epoch}, Batch

{start_batch} (Zero-Wait Enabled)\n")
else:
print("\n[!] No save file found. Starting fresh from Epoch 1...")

# Initialize our timers

print("Starting Training Pipeline Optimization...")
last_save_time = [Link]()

try:
for epoch in range(start_epoch, epochs + 1):
for i, batch in enumerate(dataloader):

real_cloudy = batch["cloudy"].to(device)
real_clear = batch["clear"].to(device)

# ---------------------------
# Train Generator
# ---------------------------
optimizer_G.zero_grad()

fake_clear = generator(real_cloudy)
pred_fake = discriminator(real_cloudy, fake_clear)

valid_label = torch.ones_like(pred_fake).to(device)
fake_label = torch.zeros_like(pred_fake).to(device)

loss_GAN = criterion_GAN(pred_fake, valid_label)

loss_pixel = criterion_pixel(fake_clear, real_clear)

loss_G = loss_GAN + (lambda_pixel * loss_pixel)

loss_G.backward()
optimizer_G.step()

# ---------------------------
# Train Discriminator
# ---------------------------
optimizer_D.zero_grad()

pred_real = discriminator(real_cloudy, real_clear)

loss_real = criterion_GAN(pred_real, valid_label)
 pred_fake_d = discriminator(real_cloudy,
fake_clear.detach())
loss_fake = criterion_GAN(pred_fake_d, fake_label)

loss_D = 0.5 * (loss_real + loss_fake)

loss_D.backward()
optimizer_D.step()

# Print updates
print(f"[Epoch {epoch}/{epochs}] [Batch
{i}/{len(dataloader)}] [D loss: {loss_D.item():.4f}] [G loss:
{loss_G.item():.4f}]")

#
---------------------------------------------------------
# 5-MINUTE AUTOMATIC TIMER CHECK
#
---------------------------------------------------------
current_time = [Link]()
if (current_time - last_save_time) >=
SAVE_INTERVAL_SECONDS:
print(f"\n[~] 5 minutes elapsed. Running background
auto-save for Batch {i}...")
[Link]({
'epoch': epoch,
'batch': i,
'generator_state_dict': generator.state_dict(),
'discriminator_state_dict':
discriminator.state_dict(),
'optimizer_G_state_dict':
optimizer_G.state_dict(),
'optimizer_D_state_dict':
optimizer_D.state_dict(),
}, checkpoint_path)
last_save_time = current_time
print("[+] Auto-save complete. Resuming training
metrics smoothly...\n")

# Standard End-of-Epoch checkpoint safety save

[Link]({
 'epoch': epoch + 1,
'batch': 0,
'generator_state_dict': generator.state_dict(),
'discriminator_state_dict': discriminator.state_dict(),
'optimizer_G_state_dict': optimizer_G.state_dict(),
'optimizer_D_state_dict': optimizer_D.state_dict(),
}, checkpoint_path)
print(f"\n[+] Finished Epoch {epoch}. Global checkpoint
updated.\n")

# Emergency Ctrl+C handler

except KeyboardInterrupt:
print("\n\n[!!!] EMERGENCY STOP DETECTED (Ctrl+C) [!!!]")
print("Saving exact batch state before shutting down...")

current_save_epoch = epoch if 'epoch' in locals() else start_epoch

current_save_batch = i if 'i' in locals() else 0

[Link]({
'epoch': current_save_epoch,
'batch': current_save_batch,
'generator_state_dict': generator.state_dict(),
'discriminator_state_dict': discriminator.state_dict(),
'optimizer_G_state_dict': optimizer_G.state_dict(),
'optimizer_D_state_dict': optimizer_D.state_dict(),
}, checkpoint_path)
print("[+] Emergency save complete. You can safely close your
terminal.")
return

# Final complete success save

[Link](generator.state_dict(), "models/generator_liss4_final.pth")
print("\nTraining completed successfully! Saved final weights to:
models/generator_liss4_final.pth")

if __name__ == "__main__":
train_gan()