Cloud detection
This Jupyter notebook demonstrates a complete workflow for cloud detection on satellite imagery using a pre-trained segmentation model in PyTorch. It guides you through downloading and preparing the data, loading the model, running inference on a mosaic, processing the resulting mask into vector format, and uploading the results back to the Geovio platform. The notebook serves as an example of automated cloud analysis in the context of remote sensing.
Files for download:
- Entrypoint file: Download Jupyter Notebook
cloud_detection.ipynb
- Datazip file: Download
cloud_detection_datazip.zip
Video walkthrough:
Import Libraries¶
import os
import json
import uuid
import torch
import zipfile
import requests
import rasterio
import numpy as np
import geopandas as gpd
import segmentation_models_pytorch as smp
from rasterio.features import shapes
from shapely.geometry import shape, Polygon, MultiPolygon
Authorization¶
Before making API requests, we need to include an authorization header.
This header ensures that the server can verify our identity and grant us access to the requested resources.
# Load environment variables
access_token = os.getenv("JWT_TOKEN")
geovio_api_url = os.getenv("GEOVIO_API_URL")
transfer_api_url = os.getenv("TRANSFER_API_URL")
input_parameters_raw = os.getenv("INPUT_PARAMETERS")
# Validate inputs
if not access_token:
raise ValueError("JWT_TOKEN environment variable is not set")
if not geovio_api_url:
raise ValueError("GEOVIO_API_URL environment variable is not set")
if not transfer_api_url:
raise ValueError("TRANSFER_API_URL environment variable is not set")
if not input_parameters_raw:
raise ValueError("INPUT_PARAMETERS environment variable is not set")
# Parse input
try:
input_parameters = json.loads(input_parameters_raw)
mosaic_ids = input_parameters["mosaic"]
mosaic_id = mosaic_ids[0]
except Exception as e:
raise ValueError(f"Invalid INPUT_PARAMETERS format: {e}")
headers = {
"accept": "application/json",
"Authorization": f"Bearer {access_token}"
}
Extract Pre-trained Model from Datazip File¶
This section extracts the pre-trained machine learning model from a ZIP archive and verifies its presence.
with zipfile.ZipFile("/workdir/cloud_detection_datazip.zip", 'r') as zip_ref:
zip_ref.extractall("/workdir")
trained_model = "/workdir/example_data/cloud_model.pth"
if not os.path.exists(trained_model):
raise FileNotFoundError("ML model was not extracted corectly")
Download Mosaic¶
This section downloads the mosaic file from the Geovio transfer API.
download_url = f"{transfer_api_url}/Mosaic/{mosaic_id}/download"
params = {
"fileName": f"{mosaic_id}.tif"
}
response = requests.get(download_url, headers=headers, params=params, stream=True)
if response.status_code != 200:
raise Exception(f"Failed to download mosaic for ID {mosaic_id}")
output_file = "/tmp/file.tif"
if response.status_code == 200:
with open(output_file, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
print(f"File saved as: {output_file}")
else:
print("Download error:", response.status_code)
raise Exception(response.text)
Load Pre-trained PyTorch Model¶
This section initializes and loads the pre-trained segmentation model for inference.
- device – Sets the computation device to GPU (
cuda
) if available, otherwise CPU (cpu
). - model – Creates a
Unet
model fromsegmentation_models_pytorch
with:resnet34
encoder,- no pre-trained weights for the encoder,
- 3 input channels (RGB),
- 1 output class (cloud mask).
- load_state_dict – Loads the weights from the previously extracted
.pth
file onto the selected device. - model.eval() – Puts the model in evaluation mode, disabling training-specific operations like dropout and batch normalization updates.
device = "cuda" if torch.cuda.is_available() else "cpu"
model = smp.Unet(encoder_name="resnet34", encoder_weights=None, in_channels=3, classes=1).to(device)
model.load_state_dict(torch.load(trained_model, map_location=device))
model.eval()
Load Mosaic and Prepare for Inference¶
This section reads the downloaded mosaic and prepares a blank mask for predictions.
- patch_size – Defines the size of the image patches for model inference (256×256 pixels).
- rasterio.open(output_file) – Opens the downloaded mosaic file.
- src.read([4, 3, 2]) – Reads bands 4, 3, 2 (red, green, blue) from the Sentinel-2 raster to create an RGB image.
- astype(np.float32) / 10000.0 – Converts raster values to float32 and normalizes them to [0, 1] range.
- profile – Stores the raster metadata for later use (e.g., saving predicted masks).
- H, W – Image height and width extracted from the raster shape.
- cloud_mask – Initializes an empty 2D array of zeros to store the predicted cloud mask.
patch_size = 256
with rasterio.open(output_file) as src:
image = src.read([4, 3, 2]).astype(np.float32) / 10000.0
profile = src.profile
H, W = image.shape[1], image.shape[2]
cloud_mask = np.zeros((H, W), dtype=np.uint8)
Run Model Inference on Patches¶
This section performs cloud detection on the mosaic using the pre-trained model, processing the image in smaller patches.
- torch.no_grad() – Disables gradient computation for inference, reducing memory usage and speeding up processing.
- Patch iteration – Loops over the image in steps of
patch_size
to handle large rasters in manageable chunks. - patch shape check – Skips edge patches smaller than
patch_size
to ensure consistent input dimensions. - patch_tensor – Converts the patch to a PyTorch tensor, adds a batch dimension (
unsqueeze(0)
), and moves it to the selected device (cpu
orcuda
). - out = torch.sigmoid(model(...)) – Performs a forward pass through the model and applies the sigmoid function to obtain probabilities.
- Thresholding – Converts probabilities to binary cloud mask using a 0.5 threshold.
- cloud_mask[i:i+patch_size, j:j+patch_size] – Stores the predicted patch results in the full-size mask array.
with torch.no_grad():
for i in range(0, H, patch_size):
for j in range(0, W, patch_size):
patch = image[:, i:min(i+patch_size, H), j:min(j+patch_size, W)]
padded_patch = np.zeros((3, patch_size, patch_size), dtype=np.float32)
padded_patch[:, :patch.shape[1], :patch.shape[2]] = patch
patch_tensor = torch.tensor(padded_patch, dtype=torch.float32).unsqueeze(0).to(device)
out = torch.sigmoid(model(patch_tensor)).cpu().numpy()[0, 0]
out_cropped = out[:patch.shape[1], :patch.shape[2]]
cloud_mask[i:i+patch.shape[1], j:j+patch.shape[2]] = (out_cropped > 0.5).astype(np.uint8)
Save Predicted Cloud Mask¶
This section writes the predicted cloud mask to a new raster file.
- profile.update(dtype=rasterio.uint8, count=1) – Updates the raster metadata to use a single band of type
uint8
for the binary mask. - rasterio.open(..., "w", **profile) – Opens a new file for writing using the updated profile.
- dst.write(cloud_mask, 1) – Writes the predicted cloud mask to the first band of the output raster.
- Output path – The mask is saved as
/tmp/detected_clouds.tif
. - Print confirmation – Notifies that the inference is complete and the mask has been saved.
profile.update(dtype=rasterio.uint8, count=1)
detected_clouds = "/tmp/detected_clouds"
with rasterio.open(f"{detected_clouds}.tif", "w", **profile) as dst:
dst.write(cloud_mask, 1)
print(f"Cloud mask saved as {detected_clouds}.tif")
Remove Small Holes from Polygons¶
This function removes interior rings (holes) in polygons that are smaller than a specified area threshold.
- poly – Input
Polygon
object to process. - min_area – Minimum area threshold for interior rings (default: 10,000 square units).
- poly.interiors – Checks if the polygon has any holes.
- new_interiors – Keeps only interior rings whose area is greater than or equal to
min_area
. - return Polygon(...) – Returns a new
Polygon
with the original exterior and filtered interiors. - If the polygon has no interiors, it is returned unchanged.
def remove_small_holes(poly, min_area=10000):
if poly.interiors:
new_interiors = [ring for ring in poly.interiors if Polygon(ring).area >= min_area]
return Polygon(shell=poly.exterior.coords, holes=[ring.coords for ring in new_interiors])
else:
return poly
Convert Cloud Mask Raster to Simplified Vector (GeoJSON)¶
This section converts the predicted cloud mask raster into a vector format and performs post-processing.
- rasterio.open – Opens the raster mask and reads the first band.
- shapes(mask_data, transform=transform) – Converts raster pixels with value
1
(clouds) into polygon geometries. - gpd.GeoDataFrame – Stores the polygons in a GeoDataFrame with the original CRS.
- CRS transformation – Converts geographic coordinates to a projected CRS (EPSG:3857) for accurate area calculations.
- Filter by area – Removes polygons smaller than 10,000 square units to exclude noise.
- Remove small holes – Applies
remove_small_holes
to clean up small interior rings. - Simplify geometries – Reduces vertex count using a
tolerance
of 50 units to make the GeoJSON lighter and faster to render in tools like OpenLayers. - Save as GeoJSON – Writes the cleaned and simplified polygons to a GeoJSON file.
- Print confirmation – Notifies that the vector mask has been successfully saved.
with rasterio.open(f"{detected_clouds}.tif") as src:
mask_data = src.read(1)
transform = src.transform
crs = src.crs
polygons = []
for geom, value in shapes(mask_data, mask=None, transform=transform):
if value == 1:
polygons.append(shape(geom))
gdf = gpd.GeoDataFrame(geometry=polygons, crs=crs)
if gdf.crs.is_geographic:
gdf = gdf.to_crs(epsg=3857)
gdf["area"] = gdf.geometry.area
gdf = gdf[gdf["area"] > 10000].copy()
gdf["geometry"] = gdf["geometry"].apply(
lambda geom: remove_small_holes(geom) if geom.geom_type == "Polygon" else
MultiPolygon([remove_small_holes(p) for p in geom.geoms])
)
gdf["geometry"] = gdf["geometry"].simplify(tolerance=50)
gdf.to_file(f"{detected_clouds}.geojson", layer=detected_clouds, driver="GeoJSON")
print(f"Vector mask saved as {detected_clouds}.geojson")
Create Analysis from Mosaic¶
This code retrieves mosaic details and creates a new analysis (project) in Geovio.
- mosaic_detail_response – Request to get details of the mosaic.
- workspace_id, workspace_name – Extracted from the mosaic details.
- map_id – A new UUID generated for the map.
- payload – Configuration for creating the analysis, including mosaic ID, number of classes, project type, workspace info, and permissions.
- create_analysis_response – Sends the POST request to create the analysis.
- analysis_detail – JSON response with details of the created analysis.
- project_id – ID of the newly created project.
mosaic_detail_response = requests.get(f"{geovio_api_url}/Mosaic/{mosaic_id}", headers=headers)
mosaic_detail = mosaic_detail_response.json()
workspace_id: int = mosaic_detail["workspace"]
workspace_name: str = mosaic_detail["workspaceName"]
map_id: str = str(uuid.uuid4())
# create analysis
payload = {
"config": [
{
"type": 1,
"mosaic": mosaic_id,
"classesCount": 30,
"mapId": map_id
}
],
"title": "Cloud detection",
"projectType": 1,
"workspace": workspace_id,
"workspaceName": workspace_name,
"type": "PROJECT",
"permissions": [
"gvdb:read",
"gvdb:write",
"gvdb:manage",
"gvdb:owner"
]
}
create_analysis_response = requests.post(f"{geovio_api_url}/Session/project", headers=headers, json=payload)
if create_analysis_response.status_code != 200:
raise Exception(create_analysis_response.text)
analysis_detail = create_analysis_response.json()
project_id: int = analysis_detail["project"]
Upload Cloud Detection Results¶
This code uploads the generated cloud detection results (GeoJSON + style file) to Geovio as a demo dataset.
- data – Metadata for the upload, including project ID, map ID, title, and additional info.
- files – Contains the detected clouds GeoJSON and the corresponding Mapbox style file, boh opened in binary mode.
- upload_url – API endpoint for uploading files.
- response – Executes a POST request with metadata and files.
- status check – If the response status is not 200, an exception is raised with the error message.
data: dict = {
"projectId": project_id,
"mapId": map_id,
"title": "Cloud detection",
"indexPlatformId": 0,
"copyrightValueId": None
}
files: dict = {
"geoJsonFile": (f"{detected_clouds}.geojson", open(f"{detected_clouds}.geojson", "rb"), "application/json"),
"styleFile": ("mapbox_style_cloud_detection.json", open("/workdir/example_data/mapbox_style_cloud_detection.json", "rb"), "application/json"),
}
upload_url: str = f"{geovio_api_url}/Dataset/from-files"
response = requests.post(upload_url, headers=headers, data=data, files=files)
if response.status_code != 200:
raise Exception(response.text)