Land cover & change — ESA WorldCover v200 (2021)¶
ESA WorldCover 2021 (10 m, 11 classes) — a one-image land-cover map; the default reducer is mosaic because the collection is tiled rather than time-stepped.
Setup¶
First the imports. pyramids provides Dataset (reading + plotting); earthlens provides the unified EarthLens entry point and the GEE Catalog.
import os
from pathlib import Path
import matplotlib.pyplot as plt
import numpy as np
from pyramids.dataset import Dataset as PyramidsDataset
from earthlens import EarthLens
from earthlens.gee import Catalog
Output directory¶
Each notebook writes its GeoTIFFs into a per-notebook out/ directory (which is .gitignored).
OUT_DIR = Path('out') / 'land-cover-change'
OUT_DIR.mkdir(parents=True, exist_ok=True)
print(f'output directory: {OUT_DIR.resolve()}')
Credentials¶
The notebook reads the GEE service-account credentials from the GEE_SERVICE_ACCOUNT / GEE_SERVICE_KEY environment variables. Both must be set before running this cell.
SERVICE_ACCOUNT = os.environ['GEE_SERVICE_ACCOUNT']
SERVICE_KEY = os.environ['GEE_SERVICE_KEY']
Inspect the catalog entry¶
Before downloading anything, look at what the bundled catalog knows about the asset — bands, cadence, license, provider.
cat = Catalog()
ds = cat.get_dataset('ESA/WorldCover/v200')
print(f'title: {ds.title}')
print(f'ee_type: {ds.ee_type}')
print(f'spatial_resolution: {ds.spatial_resolution} m')
print(f'extent.start_date: {ds.extent.start_date}')
print(f'extent.end_date: {ds.extent.end_date}')
print(f'default_reducer: {ds.default_reducer}')
print(f'license: {ds.license}')
print(f'provider: {ds.provider}')
print(f'#bands: {len(ds.bands)}')
print(f'band ids (first 5): {list(ds.bands)[:5]}')
Download¶
Tiny AOI ([29.9, 30.1] lat, [31.1, 31.3] lon) at 30.0 m, raw cadence — keeps the synchronous download under EE's 32768-px per-axis cap. We build the request first, then authenticate as a separate step so each is easy to read and re-run.
gee = EarthLens(
data_source="gee",
start='2021-01-01',
end='2021-12-31',
dataset='ESA/WorldCover/v200',
variables=['Map'],
aoi=[31.1, 29.9, 31.3, 30.1],
cadence='raw',
path=str(OUT_DIR),
scale=30.0,
reducer='mosaic',
)
gee.authenticate(service_account=SERVICE_ACCOUNT, service_key=SERVICE_KEY)
With the request authenticated, download() writes the GeoTIFF(s) to disk and returns their paths.
paths = gee.download(progress_bar=False)
download_ok = True
print(f'wrote {len(paths)} GeoTIFF(s):')
for p in paths:
print(f' {p} ({p.stat().st_size / 1024:.1f} KB)')
Quick preview¶
Load the first written GeoTIFF through pyramids and pull out the single band, masking the dataset's nodata so the colormap doesn't get pinned to it. (pyramids.dataset.Dataset is the project's GeoTIFF/NetCDF wrapper.)
if not download_ok or not paths:
arr = None
print('Skipping preview — no GeoTIFF was written.')
else:
pds = PyramidsDataset.read_file(str(paths[0]))
arr = pds.read_array()
if arr.ndim == 3:
arr = arr[0]
# Mask the dataset's nodata so the colormap doesn't get pinned to it.
nodata = pds.no_data_value
if nodata is not None:
try:
arr = np.ma.masked_equal(
arr, float(nodata[0] if isinstance(nodata, (list, tuple)) else nodata)
)
except (TypeError, ValueError):
pass
Render the land-cover band¶
Show the single band with a viridis colormap and print the value range.
if arr is not None:
fig, ax = plt.subplots(figsize=(6, 5))
im = ax.imshow(arr, cmap='viridis')
ax.set_title(f'{'ESA/WorldCover/v200'} / {'Map'}')
ax.set_xlabel('x (px)')
ax.set_ylabel('y (px)')
fig.colorbar(im, ax=ax, shrink=0.8)
plt.tight_layout()
plt.show()
print(f'value range: [{float(np.nanmin(arr)):.4g}, {float(np.nanmax(arr)):.4g}]')
Tracking submitted jobs (asynchronous export)¶
The download above uses export_via="url" — a synchronous getDownloadURL round-trip. Nothing was queued, so there's no Earth Engine job to track.
To track an export instead, switch to an asynchronous sink (drive / gcs / asset) and pass wait_for_export=False so .download() returns a TaskInfo at submission time rather than blocking until completion. The cells below submit the same (asset_id, band, AOI, scale) request as an export_via="asset" task into the service account's own asset folder, then walk the four jobs-API calls (list_recent_tasks → wait_for_task_id → ee.data.getAsset → ee.data.deleteAsset) to make the job finish and tidy up. See track-batch-exports.ipynb for a deeper worked example.
Imports and the demo asset folder¶
The async helpers live in earthlens.gee. The export writes into a Folder asset that we own — GEE._export_via_batch writes the actual image at <asset_id>/<prefix>, so asset_id here is the parent folder, not the final image path. Both must be cleaned up.
import ee
from earthlens.gee import cancel_task, list_recent_tasks, wait_for_task_id
_proj = ee.data._get_projects_path().removeprefix('projects/')
DEMO_FOLDER = f'projects/{_proj}/assets/earthlens-demo-land-cover-change'
print(f'demo folder: {DEMO_FOLDER}')
Best-effort cleanup of leftover children from a previous run, then (re)create the parent folder — EE requires it to exist before a child write.
# Best-effort cleanup of leftover children from a previous run (so the
# folder is empty before we try to delete it below).
try:
for child in ee.data.listAssets({'parent': DEMO_FOLDER}).get('assets', []):
ee.data.deleteAsset(child['name'])
print(f'cleared leftover child: {child["name"]}')
ee.data.deleteAsset(DEMO_FOLDER)
print(f'cleared leftover folder: {DEMO_FOLDER}')
except Exception:
pass
# Create the parent folder — EE requires it to exist before a child write.
ee.data.createAsset({'type': 'Folder'}, DEMO_FOLDER)
print(f'created folder: {DEMO_FOLDER}')
Submit¶
Same (asset_id, band, AOI, scale) request as the sync download above, just routed through export_via="asset" + wait_for_export=False. We build the request and authenticate on separate lines first.
async_gee = EarthLens(
data_source="gee",
start='2021-01-01',
end='2021-12-31',
dataset='ESA/WorldCover/v200',
variables=['Map'],
aoi=[31.1, 29.9, 31.3, 30.1],
cadence='raw',
path=str(OUT_DIR),
scale=30.0,
reducer='mosaic',
export_via='asset',
asset_id=DEMO_FOLDER,
wait_for_export=False,
)
async_gee.authenticate(service_account=SERVICE_ACCOUNT, service_key=SERVICE_KEY)
download() returns a TaskInfo per submitted bucket at the moment the task is queued — no blocking.
submitted = async_gee.download(progress_bar=False)
task_info = submitted[0]
submitted_ok = True
print(f'submitted: id={task_info.id} state={task_info.state}')
print(f' description={task_info.description}')
List + wait¶
list_recent_tasks(description_prefix=...) returns every matching task across the current project; wait_for_task_id blocks until the one we care about reaches a terminal state. A real workflow would just poll later from a separate process — the wait here exists so the notebook shows the full success path end-to-end.
if submitted_ok and task_info is not None:
recent = list_recent_tasks(
description_prefix=task_info.description,
max_age_min=10,
)
print(f'list_recent_tasks matched {len(recent)} task(s):')
for t in recent:
print(f' {t.id} {t.state:<12} {t.description}')
else:
recent = []
print('Skipping list — async submission did not succeed.')
Now block on the task we submitted. If it ends in FAILED / CANCELLED, wait_for_task_id raises — we cancel any still-running task so we don't leak an in-flight job on notebook restart.
if submitted_ok and task_info is not None:
try:
final = wait_for_task_id(
task_info.id,
poll_seconds=10,
progress_bar=False,
)
print(f'final state: {final.state}')
except RuntimeError as exc:
# Raised on FAILED / CANCELLED — cancel-if-still-running
# so we don't leak an in-flight task on notebook restart.
print(f'wait_for_task_id raised: {exc}')
try:
cancel_task(task_info.id)
except Exception:
pass
else:
print('Skipping wait — async submission did not succeed.')
Verify + clean up¶
Confirm the produced asset exists on Earth Engine, then delete it (and the surrounding demo folder) so we don't leak storage between notebook runs. The backend wrote the image at <DEMO_FOLDER>/<task description>.
if submitted_ok and task_info is not None:
produced = f'{DEMO_FOLDER}/{task_info.description}'
try:
meta = ee.data.getAsset(produced)
print(f'asset exists: type={meta.get("type")} ' f'name={meta.get("name")}')
ee.data.deleteAsset(produced)
print('asset deleted')
except Exception as exc:
print(f'verify/delete skipped: {exc}')
# Always try to tear down the parent folder.
try:
ee.data.deleteAsset(DEMO_FOLDER)
print(f'folder deleted: {DEMO_FOLDER}')
except Exception as exc:
print(f'folder delete skipped: {exc}')
What's on disk¶
The GeoTIFF is left under the per-notebook out/ directory for you to inspect. That directory is .gitignored — re-running the notebook overwrites it.
for p in sorted(OUT_DIR.iterdir()) if OUT_DIR.exists() else []:
print(f'{p} ({p.stat().st_size / 1024:.1f} KB)')