Skip to content

Google Earth Engine — API reference#

Google Earth Engine data source subpackage. Usage and authentication setup are covered under the other pages in this section; this page is the rendered API.

earthlens.gee.backend #

Google Earth Engine backend — :class:GEE, an :class:AbstractDataSource.

Downloads imagery from Google Earth Engine. A request is {asset_id: [band, ...], ...} (the addressable units of an EE dataset are bands, and one image carries many at once), plus a date range, a bbox (or a GeoDataFrame region), a temporal-compositing resolution ("raw"/"daily"/"monthly"/"yearly"), and an output pixel scale in metres. The asset ids and band metadata are resolved through :class:earthlens.gee.Catalog (loaded from the per-category YAMLs under src/earthlens/gee/catalog/).

Per (asset, band-set, time-bucket) the pipeline is:

  • :meth:_build_collection — ee.ImageCollection(asset_id) (or the single ee.Image wrapped in one), .filterDate(...), .filterBounds(region), then any constructor filters and the per-image cloud_mask (.map-applied), and finally .select(bands). Pure: no I/O.
  • :meth:_composite — split the request window into buckets at the requested cadence and collapse each with the dataset's default_reducer (or the constructor reducer override) — mean for continuous fields / rates, median for cloud-screened optical scenes, mosaic for tiled / annual static maps. Yields one ee.Image per bucket.
  • :meth:_api — export the bucket image via the configured export_via: "url" (the default) computes the request's pixel dimensions and refuses if either axis exceeds Earth Engine's 32768-px synchronous limit (a clear, actionable ValueError), else image.getDownloadURL({..., "format": "GEO_TIFF"}) → an HttpClient GET → a GeoTIFF under the output directory; multi-band responses (which Earth Engine returns as a zip of per-band tifs) are unpacked through pyramids.dataset.Dataset.from_archive into a single multi-band tif. "drive" / "gcs" queue an asynchronous ee.batch.Export.image.to{Drive,CloudStorage} task (maxPixels only, no 32768-px cap), poll it to completion, and return a "drive://…" / "gs://…" destination string (the file is left in the Drive folder / GCS bucket for the caller to pull).

Authentication is a one-time ee.Initialize against a registered Cloud project, performed by :meth:_initialize via :class:earthlens.gee.auth.EarthEngineAuth (service-account key) or, if no key is given, an interactive ee.Authenticate() against an explicit project. Credential / registration failures surface as :class:AuthenticationError.

AuthenticationError #

Bases: AuthenticationError

Raised when the Earth Engine connection cannot be established.

Wraps the underlying ee / Google credential errors with an actionable message — most commonly a missing or malformed service key, an unregistered Cloud project, or a service account that lacks an Earth Engine IAM role on the target project.

A subclass of the cross-backend :class:earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause; backward compatible with existing except earthlens.gee.AuthenticationError consumers.

Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class AuthenticationError(_BaseAuthenticationError):
    """Raised when the Earth Engine connection cannot be established.

    Wraps the underlying `ee` / Google credential errors with an
    actionable message — most commonly a missing or malformed service
    key, an unregistered Cloud project, or a service account that lacks
    an Earth Engine IAM role on the target project.

    A subclass of the cross-backend
    :class:`earthlens.base.AuthenticationError` so callers can catch
    every backend's auth failure with one `except` clause; backward
    compatible with existing `except earthlens.gee.AuthenticationError`
    consumers.
    """

EedaiPlan #

Bases: NamedTuple

How the EEDAI reader should serve one request, or why it should not.

Attributes:

Name Type Description
can_serve bool

Whether the reader takes this read at all.
tile_size int | None

Output pixels per tile side when the read is streamed, or None for a single pass (and when can_serve is False).
tiles int

How many tiles the streamed read is cut into; 1 for a single pass. Carried here so the exporter never re-derives it — a second derivation is free to disagree with the one that was routed on.
reason str

Why the reader declined, empty when it did not.
Source code in libs/providers/imagery/src/earthlens/gee/backend.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
class EedaiPlan(NamedTuple):
    """How the EEDAI reader should serve one request, or why it should not.

    Attributes:
        can_serve: Whether the reader takes this read at all.
        tile_size: Output pixels per tile side when the read is streamed, or
            `None` for a single pass (and when `can_serve` is `False`).
        tiles: How many tiles the streamed read is cut into; `1` for a single
            pass. Carried here so the exporter never re-derives it — a second
            derivation is free to disagree with the one that was routed on.
        reason: Why the reader declined, empty when it did not.
    """

    can_serve: bool
    tile_size: int | None
    tiles: int
    reason: str

GEE #

Bases: LazyClientMixin, AbstractDataSource

Google Earth Engine data source.

Parameters:

Name Type Description Default
start str

Inclusive start date string (parsed with fmt).

 required
end str

Inclusive end date string.

 required
variables dict[str, list[str]]

Mapping {asset_id: [band, ...]} — each asset_id must be a key of :attr:Catalog.datasets and each band a band of that dataset (see src/earthlens/gee/catalog/).

 required
lat_lim list[float]

[lat_min, lat_max] in degrees.

 required
lon_lim list[float]

[lon_min, lon_max] in degrees.

 required
temporal_resolution str

How to composite over time — "raw" (one image: reduce the whole window), "daily", "monthly", or "yearly". Defaults to "raw".

 'raw'
path Path | str | None

Output directory (created if absent). Defaults to the configured earthlens output directory (set_output_dir() / EARTHLENS_DATA_DIR); see earthlens.config.

 None
fmt str

strptime format for start / end. Defaults to "%Y-%m-%d".

 '%Y-%m-%d'
scale float | None

Output pixel size in metres. If omitted, each dataset's nominal spatial_resolution is used.

 None
crs str

Output CRS (EPSG code string). Defaults to "EPSG:4326".

 'EPSG:4326'
reducer str | None

Override the per-dataset default_reducer for the temporal composite (mean / median / min / max / mode / mosaic / sum). None (the default) uses each dataset's own default_reducer.

 None
export_via Literal['url', 'drive', 'gcs', 'asset']

How to get pixels out — "url" (synchronous getDownloadURL, capped at 32768 px per axis; the default), "drive" (asynchronous ee.batch.Export.image.toDrive; requires drive_folder), "gcs" (asynchronous ee.batch.Export.image.toCloudStorage; requires gcs_bucket), or "asset" (asynchronous ee.batch.Export.image.toAsset; requires asset_id).

 'url'
drive_folder str | None

Google Drive folder name for export_via="drive".

 None
gcs_bucket str | None

Cloud Storage bucket name for export_via="gcs" (the service account needs roles/storage.objectAdmin on it).

 None
asset_id str | None

Parent folder asset id for export_via="asset" (e.g. "projects/my-project/assets/my-folder"). Each export's asset is created at <asset_id>/<prefix>.

 None
region GeoDataFrame | None

Optional GeoDataFrame to clip to precisely; when given it supersedes the lat/lon bbox for the actual clip (the bbox is still used for the "url" size estimate).

 None
http_timeout float | None

Timeout in seconds for the synchronous getDownloadURL HTTP request (export_via="url"). Defaults to 300 s.

 None
auto_split bool

For export_via="url", when the estimated request exceeds Earth Engine's 32768-px per-axis cap, automatically split the AOI into tiles each within the cap, download each tile separately, and mosaic them back into a single GeoTIFF via pyramids.dataset.merge.merge_rasters. Defaults to False — the previous behaviour, which raises ValueError with an actionable message.

 False
discover_extent bool

When the catalog entry's extent.end_date (and/or start_date) is missing, fall back to an EE-side reduceColumns(minMax) over system:time_start to discover the collection's actual extent and clamp the request window to it. The discovered extent is cached per asset for the lifetime of the GEE instance. Defaults to False — the previous behaviour, which uses now() + 1 day as the upper bound for open-ended catalog entries.

 False
wait_for_export bool

For asynchronous sinks (export_via="drive" / "gcs" / "asset"), whether download() blocks until each task reaches a terminal state. Defaults to True (the historical behaviour — returns the destination string). When False, each task is started and download() returns a list of :class:TaskInfo objects so the caller can track them asynchronously via :mod:earthlens.gee.jobs. Ignored for export_via="url", which is always synchronous.

 True
cloud_mask CloudMask | None

Optional per-image mask .map-applied to every image in the stack before the reducer, so the composite is built from cloud-screened pixels — the usual way to get a clean optical mosaic. A callable ee.Image -> ee.Image; see :mod:earthlens.gee.cloud_masks (landsat_sr / sentinel2_scl). It runs before the band select, so it may read quality bands (QA_PIXEL / SCL) that are not listed in variables. Meant for image collections; on a static ee_type="image" dataset it is applied verbatim and a warning is logged (see :meth:_build_collection). Defaults to None (no masking).

 None
filters Iterable[CollectionFilter] | None

Optional iterable of ee.ImageCollection -> ee.ImageCollection filters applied to the stack after the spatial / temporal filters (filterBounds, and filterDate for image collections) and before the cloud_mask and reducer — e.g. a metadata cloud-cover cap. Each entry takes the collection and returns it; wrap the :mod:earthlens.gee.filters helpers (by_cloud_cover_lte / by_property_in / ...), whose first argument is the collection, with functools.partial or a lambda — partial(by_cloud_cover_lte, max_pct=60). Applied left to right, so pass an ordered iterable (a set would apply in arbitrary order); like cloud_mask, meant for image collections. Defaults to None (no extra filters).

 None
engine Literal['auto', 'ee', 'eedai']

Which layer materialises the pixels for export_via="url". "auto" (the default) uses the pyramids-eo EEDAI reader when the request is a raw read of a materialised asset — no reducer over a collection, no cloud_mask, no filters, and crs="EPSG:4326" — and the [eedai] extra is installed, falling back to Earth Engine's getDownloadURL otherwise. "ee" always uses getDownloadURL (the historical behaviour). "eedai" forces the reader and raises if the request is not eligible. The EEDAI path reads pixels straight from the asset, so Earth Engine's 32768-px synchronous cap does not apply and auto_split is unnecessary — a window too large to materialise is streamed to disk in tiles and mosaicked. It cannot run server-side compute, which is why composited requests stay on Earth Engine. Ignored for the asynchronous "drive" / "gcs" / "asset" sinks, which are Earth Engine-only.

The two engines do not produce byte-identical rasters. Earth Engine reads scale in a geographic CRS as a uniform degree-equivalent, while the EEDAI grid is sized for square metres on the ground, so away from the equator the column counts differ; and the reader downsamples locally (nearest by default) where Earth Engine aggregates server-side. The AOI, CRS and values agree — the sampling does not. "eedai" still needs the [gee] extra and Earth Engine credentials: the request is built through ee before the pixels are fetched.

 'auto'
cog bool

Write the EEDAI path's raster as a Cloud Optimized GeoTIFF (tiled, with overviews) via Dataset.cog.to_cog instead of a plain GeoTIFF. Applies only to the EEDAI path — the Earth Engine getDownloadURL and batch-export sinks are unaffected. Defaults to False.

 False
resample str

Resampling kernel the EEDAI reader warps the native grid with — "nearest" (the default), "average", "bilinear", … . This path always warps from the asset's native resolution to the requested scale, so for continuous fields (elevation, temperature, reflectance) being read coarser than native, "average" is closer to Earth Engine's server-side aggregation than the point-sampling default; keep "nearest" for categorical data such as land cover. Ignored on the Earth Engine path, which resamples server-side.

 'nearest'

Credentials are not constructor arguments — the constructor describes only what to fetch. Supply them at the authentication step: :meth:authenticate accepts service_account= / service_key= / project=, each falling back to the GEE_SERVICE_ACCOUNT / GEE_SERVICE_KEY / GEE_PROJECT environment variable when omitted. download() opens the connection lazily (resolving the same way) if authenticate() was never called.

Raises:

Type Description
AuthenticationError

If Earth Engine cannot be initialised (missing/invalid key, unregistered project, missing IAM role).
ValueError

At construction for a bad export_via (or "drive" without drive_folder / "gcs" without gcs_bucket / "asset" without asset_id); from the parent on a bad date range; from :meth:_check_input_dates on an unknown temporal_resolution; from :meth:_api on a missing scale or an oversized "url" request (unless auto_split=True); from :meth:_download_dataset on an unknown asset id or band.
TypeError

At construction when cloud_mask is not callable, or filters is a str / bytes-like / mapping / non-iterable or contains a non-callable entry.
NotImplementedError

From :meth:download when aggregate= is passed (not yet supported).
RuntimeError

From :meth:_api if a "drive" / "gcs" export task does not complete.

Examples:

  • Authenticate against a service account, then download SRTM over a small bbox: 
    >>> from earthlens.gee import GEE  # doctest: +SKIP
>>> gee = GEE(  # doctest: +SKIP
...     start="2000-02-11", end="2000-02-12",
...     variables={"USGS/SRTMGL1_003": ["elevation"]},
...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3],
...     path="data/gee",
... )
>>> paths = gee.authenticate(  # doctest: +SKIP
...     service_account="sa@my-project.iam.gserviceaccount.com",
...     service_key="/path/to/key.json",
... ).download()
  • Read the same raw asset through the pyramids-eo EEDAI reader and write a Cloud Optimized GeoTIFF (no 32768-px cap, no auto_split): 
    >>> from earthlens.gee import GEE  # doctest: +SKIP
>>> gee = GEE(  # doctest: +SKIP
...     start="2000-02-11", end="2000-02-12",
...     variables={"USGS/SRTMGL1_003": ["elevation"]},
...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3],
...     path="data/gee", scale=90,
...     engine="eedai", cog=True,
... )
>>> paths = gee.authenticate().download()  # doctest: +SKIP
>>> paths[0].name  # doctest: +SKIP
'USGS_SRTMGL1_003_elevation_20000211.tif'
Source code in libs/providers/imagery/src/earthlens/gee/backend.py 
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
2116
2117
class GEE(LazyClientMixin, AbstractDataSource):
    """Google Earth Engine data source.

    Args:
        start: Inclusive start date string (parsed with `fmt`).
        end: Inclusive end date string.
        variables: Mapping `{asset_id: [band, ...]}` — each `asset_id`
            must be a key of :attr:`Catalog.datasets` and each band a
            band of that dataset (see `src/earthlens/gee/catalog/`).
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: How to composite over time — `"raw"` (one
            image: reduce the whole window), `"daily"`, `"monthly"`, or
            `"yearly"`. Defaults to `"raw"`.
        path: Output directory (created if absent). Defaults to the configured
            earthlens output directory (`set_output_dir()` /
            `EARTHLENS_DATA_DIR`); see `earthlens.config`.
        fmt: `strptime` format for `start` / `end`. Defaults to `"%Y-%m-%d"`.
        scale: Output pixel size in metres. If omitted, each dataset's
            nominal `spatial_resolution` is used.
        crs: Output CRS (EPSG code string). Defaults to `"EPSG:4326"`.
        reducer: Override the per-dataset `default_reducer` for the
            temporal composite (`mean` / `median` / `min` / `max` /
            `mode` / `mosaic` / `sum`). `None` (the default) uses each
            dataset's own `default_reducer`.
        export_via: How to get pixels out — `"url"` (synchronous
            `getDownloadURL`, capped at 32768 px per axis; the default),
            `"drive"` (asynchronous `ee.batch.Export.image.toDrive`;
            requires `drive_folder`), `"gcs"` (asynchronous
            `ee.batch.Export.image.toCloudStorage`; requires `gcs_bucket`),
            or `"asset"` (asynchronous `ee.batch.Export.image.toAsset`;
            requires `asset_id`).
        drive_folder: Google Drive folder name for `export_via="drive"`.
        gcs_bucket: Cloud Storage bucket name for `export_via="gcs"` (the
            service account needs `roles/storage.objectAdmin` on it).
        asset_id: Parent folder asset id for `export_via="asset"` (e.g.
            `"projects/my-project/assets/my-folder"`). Each export's asset
            is created at `<asset_id>/<prefix>`.
        region: Optional `GeoDataFrame` to clip to precisely; when given
            it supersedes the lat/lon bbox for the actual clip (the bbox
            is still used for the `"url"` size estimate).
        http_timeout: Timeout in seconds for the synchronous
            `getDownloadURL` HTTP request (`export_via="url"`). Defaults
            to 300 s.
        auto_split: For `export_via="url"`, when the estimated request
            exceeds Earth Engine's 32768-px per-axis cap, automatically
            split the AOI into tiles each within the cap, download each
            tile separately, and mosaic them back into a single GeoTIFF
            via `pyramids.dataset.merge.merge_rasters`. Defaults to
            `False` — the previous behaviour, which raises `ValueError`
            with an actionable message.
        discover_extent: When the catalog entry's `extent.end_date`
            (and/or `start_date`) is missing, fall back to an EE-side
            `reduceColumns(minMax)` over `system:time_start` to discover
            the collection's actual extent and clamp the request window
            to it. The discovered extent is cached per asset for the
            lifetime of the `GEE` instance. Defaults to `False` — the
            previous behaviour, which uses `now() + 1 day` as the upper
            bound for open-ended catalog entries.
        wait_for_export: For asynchronous sinks (`export_via="drive"` /
            `"gcs"` / `"asset"`), whether `download()` blocks until
            each task reaches a terminal state. Defaults to `True`
            (the historical behaviour — returns the destination
            string). When `False`, each task is started and
            `download()` returns a list of :class:`TaskInfo` objects
            so the caller can track them asynchronously via
            :mod:`earthlens.gee.jobs`. Ignored for `export_via="url"`,
            which is always synchronous.
        cloud_mask: Optional per-image mask `.map`-applied to every image
            in the stack *before* the reducer, so the composite is built
            from cloud-screened pixels — the usual way to get a clean
            optical mosaic. A callable `ee.Image -> ee.Image`; see
            :mod:`earthlens.gee.cloud_masks` (`landsat_sr` /
            `sentinel2_scl`). It runs before the band `select`, so it may
            read quality bands (`QA_PIXEL` / `SCL`) that are not listed in
            `variables`. Meant for image collections; on a static
            `ee_type="image"` dataset it is applied verbatim and a warning
            is logged (see :meth:`_build_collection`). Defaults to `None`
            (no masking).
        filters: Optional iterable of `ee.ImageCollection ->
            ee.ImageCollection` filters applied to the stack after the
            spatial / temporal filters (`filterBounds`, and `filterDate`
            for image collections) and before the `cloud_mask` and
            reducer — e.g. a metadata cloud-cover cap. Each entry takes
            the collection and returns it; wrap the
            :mod:`earthlens.gee.filters` helpers (`by_cloud_cover_lte` /
            `by_property_in` / ...), whose first argument is the
            collection, with `functools.partial` or a lambda —
            `partial(by_cloud_cover_lte, max_pct=60)`. Applied left to
            right, so pass an *ordered* iterable (a `set` would apply in
            arbitrary order); like `cloud_mask`, meant for image
            collections. Defaults to `None` (no extra filters).
        engine: Which layer materialises the pixels for `export_via="url"`.
            `"auto"` (the default) uses the pyramids-eo EEDAI reader when
            the request is a raw read of a materialised asset — no reducer
            over a collection, no `cloud_mask`, no `filters`, and
            `crs="EPSG:4326"` — and the `[eedai]` extra is installed,
            falling back to Earth Engine's
            `getDownloadURL` otherwise. `"ee"` always uses `getDownloadURL`
            (the historical behaviour). `"eedai"` forces the reader and
            raises if the request is not eligible. The EEDAI path reads
            pixels straight from the asset, so Earth Engine's 32768-px
            synchronous cap does not apply and `auto_split` is unnecessary —
            a window too large to materialise is streamed to disk in tiles
            and mosaicked. It cannot run server-side compute, which is why
            composited requests stay on Earth Engine. Ignored for the
            asynchronous `"drive"` / `"gcs"` / `"asset"` sinks, which are
            Earth Engine-only.

            The two engines do not produce byte-identical rasters. Earth
            Engine reads `scale` in a geographic CRS as a uniform
            degree-equivalent, while the EEDAI grid is sized for square
            metres on the ground, so away from the equator the column counts
            differ; and the reader downsamples locally (nearest by default)
            where Earth Engine aggregates server-side. The AOI, CRS and
            values agree — the sampling does not. `"eedai"` still needs the
            `[gee]` extra and Earth Engine credentials: the request is built
            through `ee` before the pixels are fetched.
        cog: Write the EEDAI path's raster as a Cloud Optimized GeoTIFF
            (tiled, with overviews) via `Dataset.cog.to_cog` instead of a
            plain GeoTIFF. Applies only to the EEDAI path — the Earth
            Engine `getDownloadURL` and batch-export sinks are unaffected.
            Defaults to `False`.
        resample: Resampling kernel the EEDAI reader warps the native grid
            with — `"nearest"` (the default), `"average"`, `"bilinear"`, … .
            This path always warps from the asset's native resolution to the
            requested `scale`, so for continuous fields (elevation,
            temperature, reflectance) being read coarser than native,
            `"average"` is closer to Earth Engine's server-side aggregation
            than the point-sampling default; keep `"nearest"` for
            categorical data such as land cover. Ignored on the Earth Engine
            path, which resamples server-side.

    Credentials are not constructor arguments — the constructor describes
    only what to fetch. Supply them at the authentication step:
    :meth:`authenticate` accepts `service_account=` / `service_key=` /
    `project=`, each falling back to the `GEE_SERVICE_ACCOUNT` /
    `GEE_SERVICE_KEY` / `GEE_PROJECT` environment variable when omitted.
    `download()` opens the connection lazily (resolving the same way) if
    `authenticate()` was never called.

    Raises:
        AuthenticationError: If Earth Engine cannot be initialised
            (missing/invalid key, unregistered project, missing IAM role).
        ValueError: At construction for a bad `export_via` (or `"drive"`
            without `drive_folder` / `"gcs"` without `gcs_bucket` /
            `"asset"` without `asset_id`); from the parent on a bad date
            range; from :meth:`_check_input_dates` on an unknown
            `temporal_resolution`; from :meth:`_api` on a missing scale
            or an oversized `"url"` request (unless `auto_split=True`);
            from :meth:`_download_dataset` on an unknown asset id or band.
        TypeError: At construction when `cloud_mask` is not callable, or
            `filters` is a `str` / bytes-like / mapping / non-iterable or
            contains a non-callable entry.
        NotImplementedError: From :meth:`download` when `aggregate=` is
            passed (not yet supported).
        RuntimeError: From :meth:`_api` if a `"drive"` / `"gcs"` export
            task does not complete.

    Examples:
        - Authenticate against a service account, then download SRTM over a small bbox:
            ```python
            >>> from earthlens.gee import GEE  # doctest: +SKIP
            >>> gee = GEE(  # doctest: +SKIP
            ...     start="2000-02-11", end="2000-02-12",
            ...     variables={"USGS/SRTMGL1_003": ["elevation"]},
            ...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3],
            ...     path="data/gee",
            ... )
            >>> paths = gee.authenticate(  # doctest: +SKIP
            ...     service_account="sa@my-project.iam.gserviceaccount.com",
            ...     service_key="/path/to/key.json",
            ... ).download()

            ```
        - Read the same raw asset through the pyramids-eo EEDAI reader and write
          a Cloud Optimized GeoTIFF (no 32768-px cap, no `auto_split`):
            ```python
            >>> from earthlens.gee import GEE  # doctest: +SKIP
            >>> gee = GEE(  # doctest: +SKIP
            ...     start="2000-02-11", end="2000-02-12",
            ...     variables={"USGS/SRTMGL1_003": ["elevation"]},
            ...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3],
            ...     path="data/gee", scale=90,
            ...     engine="eedai", cog=True,
            ... )
            >>> paths = gee.authenticate().download()  # doctest: +SKIP
            >>> paths[0].name  # doctest: +SKIP
            'USGS_SRTMGL1_003_elevation_20000211.tif'

            ```
    """

    OUTPUT_KIND: OutputKind = "raster"

    AGGREGATE_REFUSAL_REASON = (
        "the reducer is not wired for this backend yet (planned — see the GEE "
        "plan task M3)"
    )

    #: Clips to the exact polygon when `aoi=` carries one, not just its bbox.
    SUPPORTS_POLYGON_AOI = True

    @property
    def catalog(self):
        """The bundled GEE :class:`~earthlens.gee.Catalog` (alias of `_catalog`)."""
        return self._catalog

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "raw",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        *,
        scale: float | None = None,
        crs: str = "EPSG:4326",
        reducer: str | None = None,
        export_via: Literal["url", "drive", "gcs", "asset"] = "url",
        drive_folder: str | None = None,
        gcs_bucket: str | None = None,
        asset_id: str | None = None,
        region: GeoDataFrame | None = None,
        http_timeout: float | None = None,
        auto_split: bool = False,
        discover_extent: bool = False,
        wait_for_export: bool = True,
        cloud_mask: CloudMask | None = None,
        filters: Iterable[CollectionFilter] | None = None,
        engine: Literal["auto", "ee", "eedai"] = "auto",
        cog: bool = False,
        resample: str = "nearest",
    ):
        # Validate the cheap (no-I/O) config first so user typos surface
        # before the ~3.3 s cold-cache catalog parse below.
        if export_via not in {"url", "drive", "gcs", "asset"}:
            raise ValueError(
                f"export_via must be 'url', 'drive', 'gcs', or 'asset', "
                f"got {export_via!r}"
            )
        if export_via == "drive" and not drive_folder:
            raise ValueError("export_via='drive' requires drive_folder=")
        if export_via == "gcs" and not gcs_bucket:
            raise ValueError("export_via='gcs' requires gcs_bucket=")
        if export_via == "asset" and not asset_id:
            raise ValueError(
                "export_via='asset' requires asset_id= (the parent folder "
                "asset, e.g. 'projects/my-project/assets/my-folder')"
            )
        if cloud_mask is not None and not callable(cloud_mask):
            raise TypeError(
                "cloud_mask must be a callable ee.Image -> ee.Image (or None), "
                f"got {type(cloud_mask).__name__}"
            )
        collection_filters = _validate_filters(filters)
        if engine not in _ENGINES:
            raise ValueError(
                f"engine must be one of {sorted(_ENGINES)}, got {engine!r}"
            )
        _validate_pure_config(start, end, temporal_resolution, fmt)

        # These must be set before `super().__init__` runs, because the
        # parent constructor immediately calls `self._initialize()` (and
        # `_create_grid` / `_check_input_dates`), which read them.
        self._catalog = Catalog()
        # Credentials are resolved at authenticate()/first-client-access time
        # (explicitly or from the GEE_SERVICE_ACCOUNT / GEE_SERVICE_KEY /
        # GEE_PROJECT environment variables), not at construction.
        self._service_account: str | None = None
        self._service_key: str | None = None
        self._project: str | None = None
        self.project: str | None = None
        self.scale = scale
        self.crs = crs
        self.reducer = reducer
        self.export_via = export_via
        self.drive_folder = drive_folder
        self.gcs_bucket = gcs_bucket
        self.asset_id = asset_id
        self.region = region
        self.http_timeout = (
            float(http_timeout) if http_timeout is not None else _DEFAULT_HTTP_TIMEOUT_S
        )
        self.auto_split = bool(auto_split)
        self.discover_extent = bool(discover_extent)
        self.wait_for_export = bool(wait_for_export)
        #: The per-image `cloud_mask` hook (or `None`), `.map`-applied
        #: before the reducer in :meth:`_build_collection`.
        self.cloud_mask = cloud_mask
        #: The validated `filters` as a tuple (empty when none were given),
        #: applied left to right in :meth:`_build_collection`.
        self.filters: tuple[CollectionFilter, ...] = collection_filters
        #: Which layer materialises the pixels: `"auto"` (the pyramids-eo
        #: EEDAI reader when the request is eligible and installed, else
        #: Earth Engine), `"ee"`, or `"eedai"`.
        self.engine = engine
        #: Write the EEDAI path's output as a Cloud Optimized GeoTIFF.
        self.cog = bool(cog)
        #: Resampling kernel the EEDAI reader warps with (`nearest` by default).
        self.resample = resample
        self._ee_geometry: Any = None  # lazily built in `_ee_region`
        self._eedai_credential: Any = None  # lazily built in `_eedai_credentials`
        self._cog_warned = False  # one-shot guard for the `cog=` notice

        super().__init__(
            start=start,
            end=end,
            variables=variables,
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _resolve_credentials(self) -> tuple[str | None, str | None, str | None]:
        """Resolve credentials from explicit values, then the environment.

        Each credential piece falls back to its environment variable when
        not set explicitly (via :meth:`authenticate`): `GEE_SERVICE_ACCOUNT`,
        `GEE_SERVICE_KEY`, `GEE_PROJECT`.

        Returns:
            tuple: `(service_account, service_key, project)`, each `None`
                when neither an explicit value nor its env var is set.

        Examples:
            - Explicit values (set by :meth:`authenticate`) are returned as-is:
                ```python
                >>> import tempfile
                >>> from earthlens.gee import GEE
                >>> gee = GEE(
                ...     start="2000-02-11", end="2000-02-12",
                ...     variables={"USGS/SRTMGL1_003": ["elevation"]},
                ...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3],
                ...     path=tempfile.mkdtemp(),
                ... )
                >>> gee._service_account = "sa@demo.iam.gserviceaccount.com"
                >>> gee._service_key = "/path/to/key.json"
                >>> gee._project = "demo-project"
                >>> gee._resolve_credentials()
                ('sa@demo.iam.gserviceaccount.com', '/path/to/key.json', 'demo-project')

                ```
        """
        service_account = self._service_account or os.environ.get("GEE_SERVICE_ACCOUNT")
        service_key = self._service_key or os.environ.get("GEE_SERVICE_KEY")
        project = self._project or os.environ.get("GEE_PROJECT")
        return service_account, service_key, project

    def authenticate(
        self,
        service_account: str | None = None,
        service_key: str | None = None,
        project: str | None = None,
    ) -> GEE:
        """Resolve credentials and open the Earth Engine connection.

        The explicit, fail-fast credential step. Pass `service_account=`
        + `service_key=` (and optionally `project=`) to authenticate with
        a service-account key; omit a value to read its `GEE_SERVICE_ACCOUNT`
        / `GEE_SERVICE_KEY` / `GEE_PROJECT` environment variable instead.
        Opening the connection (which `download()` also does lazily if you
        never call this) validates the credentials against Earth Engine.

        Args:
            service_account: Service-account email. When `None`, the
                `GEE_SERVICE_ACCOUNT` environment variable is read.
            service_key: Path to the service-account JSON key file, or the
                JSON content as a string. When `None`, the `GEE_SERVICE_KEY`
                environment variable is read.
            project: Cloud project id to scope Earth Engine calls to. When
                `None`, the `GEE_PROJECT` environment variable is read (or
                the project is taken from the key's `project_id`).

        Returns:
            The backend instance, so it chains
            `EarthLens(...).authenticate(...).download()`.

        Raises:
            AuthenticationError: If no service-account pair and no project
                can be resolved, or Earth Engine rejects the credentials.

        Examples:
            - Authenticate with a service-account key, then download (live;
              skipped here):
                ```python
                >>> from earthlens.gee import GEE  # doctest: +SKIP
                >>> GEE(  # doctest: +SKIP
                ...     start="2000-02-11", end="2000-02-12",
                ...     variables={"USGS/SRTMGL1_003": ["elevation"]},
                ...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3], path="data/gee",
                ... ).authenticate(
                ...     service_account="sa@my-project.iam.gserviceaccount.com",
                ...     service_key="/path/to/key.json",
                ... ).download()

                ```
            - Resolve the same credentials from the environment instead of
              passing them (live; skipped here):
                ```python
                >>> import os  # doctest: +SKIP
                >>> os.environ["GEE_SERVICE_ACCOUNT"] = "sa@my-project.iam.gserviceaccount.com"
                >>> os.environ["GEE_SERVICE_KEY"] = "/path/to/key.json"
                >>> GEE(  # doctest: +SKIP
                ...     start="2000-02-11", end="2000-02-12",
                ...     variables={"USGS/SRTMGL1_003": ["elevation"]},
                ...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3], path="data/gee",
                ... ).authenticate().download()

                ```
        """
        if service_account is not None:
            self._service_account = service_account
        if service_key is not None:
            self._service_key = service_key
        if project is not None:
            self._project = project
        # Re-authenticating may switch identity, so the reader's cached
        # credential must not outlive the values it was built from.
        self._eedai_credential = None
        # LazyClientMixin: first access to `client` runs `_open_client` (auth).
        _ = self.client
        return self

    def _open_client(self) -> Any:
        """Authenticate and initialise the Earth Engine connection (lazily).

        Resolves the credentials (explicit values from :meth:`authenticate`,
        else the `GEE_SERVICE_ACCOUNT` / `GEE_SERVICE_KEY` / `GEE_PROJECT`
        environment variables), then uses a service-account key when a
        `service_account` + `service_key` pair is available (via
        :class:`EarthEngineAuth`); otherwise runs `ee.Authenticate()` and
        `ee.Initialize(project=...)` against the resolved `project`. The
        `ee.Authenticate()` flow is interactive — it opens a browser and
        waits for the user to paste a token, so on a headless box (CI,
        Docker, remote shell) it will hang or fail with whatever the EE
        SDK emits natively; use service-account auth for non-interactive
        use. The resolved project id is stored on :attr:`project`. Called
        by :attr:`~earthlens.base.LazyClientMixin.client` on first use.

        Returns:
            The `ee` module (cached as `self.client`).

        Raises:
            AuthenticationError: If no service-account pair and no project
                can be resolved, the credentials are invalid, the project
                is not registered for Earth Engine, or the service account
                lacks the required IAM role on it.
        """
        service_account, service_key, project = self._resolve_credentials()
        if not (service_account and service_key) and not project:
            raise AuthenticationError(
                "the GEE backend needs either service_account + service_key, "
                "or an explicit project=, supplied to authenticate(...) or via "
                "the GEE_SERVICE_ACCOUNT / GEE_SERVICE_KEY / GEE_PROJECT "
                "environment variables. See "
                "https://developers.google.com/earth-engine/guides/service_account."
            )
        if service_account and service_key:
            self.project = EarthEngineAuth.initialize(
                service_account, service_key, project
            )
            return ee
        try:
            ee.Authenticate()
            ee.Initialize(project=project)
        except ee.EEException as exc:
            message = str(exc)
            if "not registered to use Earth Engine" in message:
                raise AuthenticationError(
                    f"Cloud project {project!r} is not registered to use "
                    "Earth Engine. Register it at "
                    "https://code.earthengine.google.com/register, then retry."
                ) from exc
            raise AuthenticationError(
                f"Earth Engine initialisation failed for project {project!r}: {message}"
            ) from exc
        except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
            raise AuthenticationError(
                f"Earth Engine initialisation failed for project {project!r}: {exc}"
            ) from exc
        self.project = project
        return ee

    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ) -> TemporalExtent:
        """Parse the date range and produce the per-bucket date index.

        Args:
            start: Inclusive start date string.
            end: Inclusive end date string.
            temporal_resolution: `"raw"` (one bucket spanning the whole
                window), `"daily"` (`freq="D"`), `"monthly"` (`"MS"`),
                or `"yearly"` (`"YS"`).
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: `start_date`, `end_date`, `resolution` (the
            string passed in), and `dates` — a :class:`pandas.DatetimeIndex`
            with one entry per time bucket (a single entry for `"raw"`).

        Raises:
            ValueError: If `temporal_resolution` is not one of `"raw"`,
                `"daily"`, `"monthly"`, `"yearly"`, or if `start > end`.
        """
        start_dt = to_datetime(start, fmt)
        end_dt = to_datetime(end, fmt)
        if temporal_resolution == "raw":
            dates = pd.DatetimeIndex([start_dt])
        elif temporal_resolution in _RESOLUTION_FREQ:
            dates = date_windows(
                start_dt, end_dt, _RESOLUTION_FREQ[temporal_resolution]
            )
        else:
            raise ValueError(
                "temporal_resolution must be 'raw', 'daily', 'monthly', or "
                f"'yearly', got {temporal_resolution!r}"
            )
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution=temporal_resolution,
            dates=dates,
        )

    def download(self, progress_bar: bool = True) -> list[Path | str | TaskInfo]:
        """Download every requested band-set of every requested dataset.

        Args:
            progress_bar: Show a per-bucket `tqdm` bar. Defaults to `True`.

        Returns:
            One entry per `(dataset, band-set, time-bucket)`. The
            shape depends on the sink:

            * `export_via="url"` — `pathlib.Path` to the
                written GeoTIFF (always synchronous).
            * `export_via="drive"` / `"gcs"` / `"asset"` with the
                default `wait_for_export=True` — destination string
                (`"drive://<folder>/<prefix>"` / `"gs://<bucket>/<prefix>"` /
                `"ee://<asset_id>/<prefix>"`), populated only once
                the task reaches `COMPLETED`.
            * `export_via="drive"` / `"gcs"` / `"asset"` with
                `wait_for_export=False` — `TaskInfo` captured
                at submission time; follow up via
                `earthlens.gee.jobs` (`get_task_status`,
                `wait_for_task_id`, etc.).

        Raises:
            ValueError: On an unknown asset id, an unknown band, or an
                oversized `"url"` request (see :meth:`_api`).
            RuntimeError: If a `"drive"` / `"gcs"` / `"asset"` export
                task fails. Only raised when `wait_for_export=True`;
                in the non-blocking mode the caller handles failures
                themselves via `wait_for_task_id`.

        Examples:
            - Download one band, one image (needs network + credentials):
                ```python
                >>> gee = GEE(  # doctest: +SKIP
                ...     start="2020-06-01", end="2020-06-30",
                ...     temporal_resolution="monthly",
                ...     variables={"UCSB-CHG/CHIRPS/DAILY": ["precipitation"]},
                ...     lat_lim=[29.0, 30.0], lon_lim=[31.0, 32.0],
                ...     path="data/gee", scale=5566,
                ... )
                >>> gee.authenticate(  # doctest: +SKIP
                ...     service_account="sa@p.iam.gserviceaccount.com",
                ...     service_key="/path/to/key.json",
                ... )
                >>> paths = gee.download()  # doctest: +SKIP
                >>> [p.name for p in paths]  # doctest: +SKIP
                ['UCSB-CHG_CHIRPS_DAILY_precipitation_20200601.tif']

                ```
            - `aggregate=` is not yet supported and is rejected up front:
                ```python
                >>> gee = GEE(  # doctest: +SKIP
                ...     start="2020-06-01", end="2020-06-01",
                ...     variables={"UCSB-CHG/CHIRPS/DAILY": ["precipitation"]},
                ...     lat_lim=[29.0, 30.0], lon_lim=[31.0, 32.0],
                ...     scale=5566,
                ... )
                >>> gee.download(aggregate=object())  # doctest: +SKIP
                Traceback (most recent call last):
                    ...
                NotImplementedError: aggregate= is not yet supported ...

                ```

        See Also:
            earthlens.gee.Catalog: Resolves the `{asset_id: [band, ...]}`
                request against `src/earthlens/gee/catalog/`.
            earthlens.gee.auth.EarthEngineAuth: Performs the one-time
                `ee.Initialize` used by :meth:`_open_client`.
        """
        # Trigger the lazy Earth Engine auth/init before any `ee` call.
        _ = self.client
        self._cog_warned = False  # the cog= notice is once per run, not per object
        outputs: list[Path | str | TaskInfo] = []
        assert isinstance(
            self.vars, dict
        )  # GEE always uses the {asset_id: [band]} form
        for asset_id, bands in self.vars.items():
            outputs.extend(self._download_dataset(asset_id, list(bands), progress_bar))
        return outputs

    def _download_dataset(
        self, asset_id: str, bands: list[str], progress_bar: bool = True
    ) -> list[Path | str | TaskInfo]:
        """Download one dataset's requested bands across the time buckets.

        Validates `asset_id` and every band against the catalog, clamps
        the request window to the dataset's published extent, builds the
        filtered collection, composites it per time bucket, and writes
        each bucket via :meth:`_api`.

        Args:
            asset_id: An Earth Engine asset id present in the catalog.
            bands: Band ids of that dataset to download.
            progress_bar: Show a `tqdm` bar over the time buckets.

        Returns:
            The list of GeoTIFF paths written for this dataset (possibly
            empty if the request window does not overlap the dataset's
            extent).

        Raises:
            ValueError: If `asset_id` or any band is not in the catalog,
                or if a write fails the size guard (see :meth:`_api`).
        """
        var_info = self._catalog.get_dataset(asset_id)
        for band in bands:
            var_info.get_band(band)  # raises ValueError with a suggestion

        start, end = self._clamp_window_to_extent(var_info)
        if start is None:
            logger.warning(
                f"{asset_id}: request window does not overlap the dataset's "
                f"extent ({var_info.extent.start_date}..{var_info.extent.end_date}); "
                "skipping."
            )
            return []

        assert end is not None  # _clamp_window_to_extent returns both bounds or neither
        collection = self._build_collection(var_info, bands, start, end)
        buckets = list(self._composite(collection, var_info, start, end))
        iterator: Iterable = buckets
        if progress_bar:
            iterator = tqdm(buckets, desc=f"{asset_id} [{','.join(bands)}]", unit="img")
        return [self._api(image, var_info, bands, when) for when, image in iterator]

    def _build_collection(
        self, var_info: Dataset, bands: list[str], start: dt.datetime, end: dt.datetime
    ):
        """Build the filtered, cloud-masked, band-selected `ee.ImageCollection`.

        For an `ee_type="image"` dataset the single `ee.Image` is wrapped
        in a one-element collection so the rest of the pipeline is
        uniform. `filterDate` uses a half-open `[start, end]` window
        (Earth Engine convention); the `end` passed here is already
        bumped by one day by :meth:`_clamp_window_to_extent` so the
        user's inclusive end date is covered.

        The pipeline is `filterDate` (image collections only) →
        `filterBounds` → the constructor `filters` (left to right) → the
        per-image `cloud_mask` (`.map`) → `select(bands)`. The
        `cloud_mask` runs *before* `select` on purpose: an optical mask
        reads a quality band (`QA_PIXEL` / `SCL`) that the user's `bands`
        usually omit, so selecting the requested bands first would strip
        it.

        `filters` and `cloud_mask` are meant for image collections. On a
        static `ee_type="image"` dataset they are still applied verbatim
        (and a `logger.warning` is emitted): a metadata filter can drop
        the single wrapped image and empty the collection, while a mask
        reading a band the asset lacks fails when the graph is computed —
        either way it surfaces later as an opaque Earth Engine error at
        download time rather than here.

        Args:
            var_info: The catalog entry.
            bands: Band ids to `.select(...)`.
            start: Inclusive window start (clamped).
            end: Exclusive window end (clamped, already +1 day).

        Returns:
            The `ee.ImageCollection`.
        """
        if var_info.is_image_collection:
            collection = ee.ImageCollection(var_info.id).filterDate(
                start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d")
            )
        else:
            # A static image: no temporal filtering (the asset may not
            # carry a `system:time_start` inside the request window).
            collection = ee.ImageCollection([ee.Image(var_info.id)])
            if self.filters or self.cloud_mask is not None:
                logger.warning(
                    f"filters / cloud_mask were set but {var_info.id!r} is a "
                    "static single-image dataset (ee_type='image'); they are "
                    "applied verbatim — a metadata filter can empty the "
                    "collection, and a mask reading an absent band fails when "
                    "the graph is computed; either way it surfaces as an opaque "
                    "Earth Engine error at download time."
                )
        collection = collection.filterBounds(self._ee_region())
        for image_filter in self.filters:
            collection = image_filter(collection)
        if self.cloud_mask is not None:
            collection = collection.map(self.cloud_mask)
        return collection.select(list(bands))

    def _composite(
        self, collection, var_info: Dataset, start: dt.datetime, end: dt.datetime
    ):
        """Yield one `ee.Image` per time bucket.

        For `temporal_resolution="raw"` (and for static `ee_type="image"`
        datasets) there is a single bucket spanning the whole clamped
        window. Otherwise the window is split into daily / monthly /
        yearly buckets and each is collapsed with the dataset's
        `default_reducer` (or the constructor `reducer` override).

        Args:
            collection: The filtered `ee.ImageCollection` from
                :meth:`_build_collection`.
            var_info: The catalog entry (its `default_reducer`).
            start: Inclusive window start (clamped).
            end: Exclusive window end (clamped, already +1 day).

        Yields:
            `(timestamp, ee.Image)` pairs — `timestamp` is the bucket
            start (a :class:`datetime.datetime`), used in the filename.
        """
        reducer = self.reducer or var_info.default_reducer
        if self.temporal_resolution == "raw" or not var_info.is_image_collection:
            yield start, reduce_collection(collection, reducer)
            return
        freq = _RESOLUTION_FREQ[self.temporal_resolution]
        bucket_starts = date_windows(start, end, freq, inclusive="left")
        for i, bucket_start in enumerate(bucket_starts):
            bucket_end = (
                bucket_starts[i + 1]
                if i + 1 < len(bucket_starts)
                else pd.Timestamp(end)
            )
            window = collection.filterDate(
                bucket_start.strftime("%Y-%m-%d"), bucket_end.strftime("%Y-%m-%d")
            )
            yield bucket_start.to_pydatetime(), reduce_collection(window, reducer)

    def _api(
        self, image, var_info: Dataset, bands: list[str], when: dt.datetime
    ) -> Path | str | TaskInfo:
        """Export one composited `ee.Image` via the configured `export_via`.

        For `export_via="url"`: estimate the request's pixel dimensions
        from the bbox and `scale`; if either axis exceeds Earth Engine's
        32768-px synchronous limit, either auto-split + mosaic via
        pyramids (when `auto_split=True`) or raise a `ValueError`
        pointing the user at a coarser `scale`, a smaller bbox,
        `export_via="drive"`, or `auto_split=True`. Otherwise request a
        GeoTIFF via `getDownloadURL` and stream it to disk as
        `<asset-slug>_<bands>_<YYYYMMDD>.tif`. For `export_via="drive"` /
        `"gcs"` / `"asset"`: queue an
        `ee.batch.Export.image.to{Drive,CloudStorage,Asset}` task, poll
        it to completion (no synchronous size cap, just `maxPixels`),
        and return a destination string — for Drive / GCS the file is
        left in the destination for the caller to pull; for `"asset"`
        a new EE asset is created at `<asset_id>/<prefix>`.

        A raw, no-compute request may instead be served by the pyramids-eo
        EEDAI reader — see :meth:`_use_eedai` for when, and
        :meth:`_export_via_eedai` for what that path does. The composited
        `image` is then unused: the reader materialises the asset's own
        pixels.

        Args:
            image: The `ee.Image` to export (unused on the EEDAI path).
            var_info: The catalog entry (for the asset slug and the
                fallback `spatial_resolution`).
            bands: The band ids in `image` (used in the filename / prefix).
            when: The bucket timestamp (used in the filename / prefix).

        Returns:
            For `"url"`: the :class:`pathlib.Path` of the written GeoTIFF.
            For `"drive"` / `"gcs"` / `"asset"`: a destination string
            (`"drive://<folder>/<prefix>"` / `"gs://<bucket>/<prefix>"` /
            `"ee://<asset_id>/<prefix>"`).

        Raises:
            ValueError: If no output scale can be resolved; for `"url"` with
                `auto_split=False`, when the estimated request exceeds the
                32768-px limit; or, for a forced `engine="eedai"`, when the
                request cannot be served by the reader.
            AuthenticationError: If the EEDAI path cannot build credentials.
            ImportError: If `engine="eedai"` is forced without the `[eedai]`
                extra installed.
            RuntimeError: If Earth Engine returns a zip instead of a
                GeoTIFF (`"url"`), or a `"drive"` / `"gcs"` / `"asset"`
                export task does not complete.
        """
        scale = self.scale or var_info.spatial_resolution
        if scale is None:
            raise ValueError(
                f"no output scale for {var_info.id}: pass scale= (metres) to "
                "GEE(...) — the catalog has no nominal spatial_resolution for it."
            )
        prefix = f"{slug_asset_id(var_info.id)}_{'-'.join(bands)}_{when:%Y%m%d}"
        if self.export_via == "url":
            # An empty request is not one band: upstream opens every band the
            # asset has, so budget for that rather than under-counting.
            use_reader, plan = self._use_eedai(
                var_info, max(len(bands) or len(var_info.bands), 1)
            )
            if use_reader:
                assert plan is not None  # a yes always carries its plan
                return self._export_via_eedai(
                    var_info, bands, float(scale), prefix, plan
                )
        self._warn_cog_ignored(var_info)
        # Only the Earth Engine paths need the `ee.Geometry`; the reader clips
        # to its own bbox / cutline.
        region = self._ee_region()
        if self.export_via == "url":
            return self._export_via_url(image, var_info, float(scale), region, prefix)
        return self._export_via_batch(image, float(scale), region, prefix)

    def _export_via_url(
        self, image, var_info: Dataset, scale: float, region, prefix: str
    ) -> Path:
        """Fetch a GeoTIFF from `image.getDownloadURL`; enforce the 32768-px cap.

        Earth Engine returns a single GeoTIFF when one band is exported and
        a zip archive of per-band GeoTIFFs when several are. Both shapes
        are routed through pyramids: single tifs via :meth:`Dataset.from_bytes`
        (writes the in-memory body to a `/vsimem/` path then materialises it
        on disk), zips via :meth:`Dataset.from_archive` (chained `/vsizip/`,
        merging members into one multi-band tif).

        Oversized AOIs (either axis above :data:`EE_MAX_DIMENSION` px at
        `scale`) take one of two paths: when `auto_split=True` was passed
        to the constructor, the bbox is tiled, each tile is downloaded
        individually, and the tiles are mosaicked into one GeoTIFF via
        :func:`pyramids.dataset.merge.merge_rasters`; otherwise a
        `ValueError` is raised with a coarser-scale / smaller-bbox /
        `export_via="drive"` hint.
        """
        width_px, height_px = self.space.estimate_pixel_dims(scale)
        if max(width_px, height_px) > EE_MAX_DIMENSION:
            if self.auto_split:
                return self._auto_split_and_download(image, var_info, scale, prefix)
            raise ValueError(
                f"{var_info.id}: the requested AOI at scale={scale} m is about "
                f"{width_px}x{height_px} px, over Earth Engine's "
                f"{EE_MAX_DIMENSION}-px per-axis limit for synchronous downloads. "
                "Use a coarser scale, a smaller bbox, export_via='drive', or "
                "auto_split=True."
            )
        return self._download_one_url_tile(image, region, scale, prefix)

    def _eedai_eligible(self, var_info: Dataset) -> bool:
        """Return whether this request is a raw read the EEDAI reader can serve.

        The pyramids-eo reader materialises pixels from a real asset id; it
        cannot execute an Earth Engine computation graph. So it can only
        stand in for `getDownloadURL` when nothing server-side shapes the
        image: a single materialised `ee_type="image"` asset, no per-image
        `cloud_mask`, and no collection `filters`. The asynchronous sinks
        are Earth Engine-only.

        It is also limited to `crs="EPSG:4326"`. The reader interprets its
        `bbox` in the *target* CRS, while this backend's AOI is lat/lon; for
        a projected `crs` those degrees would be read as projected units and
        silently produce a valid-looking raster of the wrong ground area, so
        such requests stay on Earth Engine.

        Args:
            var_info: The catalog entry for the dataset being fetched.

        Returns:
            `True` when the request is a raw, no-compute read.
        """
        return (
            self.export_via == "url"
            and self.cloud_mask is None
            and not self.filters
            and var_info.ee_type == "image"
            and self.crs.upper() == _EEDAI_NATIVE_CRS
        )

    def _eedai_plan(self, var_info: Dataset, band_count: int) -> EedaiPlan:
        """Decide how — or whether — the reader can serve this request.

        A window too large to materialise is no longer a dead end: the reader
        can stream it to disk one tile at a time and mosaic the result, which
        is what retires `auto_split` for this path.

        Tiling is declined — and the request falls back to Earth Engine — in
        five cases:

        * the asset has no catalogued native resolution, so a per-tile read
          cannot be sized;
        * the request is much coarser than the asset (`native_ratio` above
          :data:`_EEDAI_MAX_TILING_RATIO`), where Earth Engine's server-side
          aggregation returns a small raster instead of fetching `ratio**2`
          native pixels per output pixel;
        * the whole read would still fetch more than
          :data:`_EEDAI_MAX_NATIVE_PIXELS` — the tile budget bounds memory,
          this bounds the work;
        * `resample` is not `"nearest"`, which upstream refuses because an
          interpolating kernel would disagree at the tile seams;
        * a polygon cutline is set, which upstream also refuses.

        A sixth case declines late: if the reader's block padding consumes the
        whole per-tile allowance there is no workable tile to cut, so the read
        falls back rather than dividing by a zero-sized tile.

        Args:
            var_info: The catalog entry being fetched.
            band_count: How many bands the read asks for; the reader holds
                them all, so they divide the per-tile budget.

        Returns:
            An :class:`EedaiPlan`. `tile_size` is `None` for a single read, and
            `reason` explains a `False` for the fallback log line or the
            forced-engine error.
        """
        bbox, cutline = self._eedai_window()
        fits, reason = self._eedai_native_fits(var_info, bbox, band_count)
        if fits:
            return EedaiPlan(True, None, 1, "")
        native_scale = var_info.spatial_resolution
        if not native_scale:
            return EedaiPlan(False, None, 0, reason)
        if cutline is not None:
            return EedaiPlan(
                False, None, 0, f"{reason}, and it cannot be tiled behind a cutline"
            )
        if self.resample != _EEDAI_TILING_RESAMPLE:
            return EedaiPlan(
                False,
                None,
                0,
                (
                    f"{reason}, and it cannot be tiled with resample="
                    f"{self.resample!r} — an interpolating resampler would differ "
                    "from the un-tiled read at the tile seams"
                ),
            )
        scale_m = float(self.scale or native_scale)
        native_ratio = max(scale_m / float(native_scale), 1.0)
        if native_ratio > _EEDAI_MAX_TILING_RATIO:
            return EedaiPlan(
                False,
                None,
                0,
                (
                    f"{reason}, and tiling it would be worse than Earth Engine: at "
                    f"scale={scale_m:g} m over a {native_scale:g} m asset the reader "
                    f"fetches about {native_ratio**2:,.0f} native px per output px, "
                    "which Earth Engine aggregates server-side instead"
                ),
            )
        native_rows, native_cols = self._eedai_grid(bbox, float(native_scale))
        native_total = native_rows * native_cols * max(band_count, 1)
        if native_total > _EEDAI_MAX_NATIVE_PIXELS:
            return EedaiPlan(
                False,
                None,
                0,
                (
                    f"{reason}, and tiling it would still fetch about "
                    f"{native_total:,} native px, over the "
                    f"{_EEDAI_MAX_NATIVE_PIXELS:,}-px ceiling on one read's total work"
                ),
            )
        # One tile's native read is `tile_size * scale / native_scale` px per
        # side and is held in memory whole, so shrink the tile until that read
        # satisfies *both* budgets the single-pass gate applies — the per-axis
        # cap and the total-pixel one.
        # Budgets are on the *native* footprint, which is the nominal window
        # plus the reader's block alignment and pad, so the allowance is
        # spent before dividing back into output pixels.
        axis_allowance = EE_MAX_DIMENSION - _EEDAI_WINDOW_PAD
        area_allowance = (
            math.sqrt(_EEDAI_MAX_PIXELS / max(band_count, 1)) - _EEDAI_WINDOW_PAD
        )
        # Not floored: an allowance the padding has already exhausted must
        # reach the guard below, not be rounded up into a one-pixel tile.
        tile_size = int(
            min(
                _EEDAI_TILE_PIXELS,
                axis_allowance / native_ratio,
                area_allowance / native_ratio,
            )
        )
        if tile_size < 1:
            # Defensive: with the shipped constants the ratio bound leaves a
            # workable tile, but this guard is deliberately kept — it was
            # removed once as unreachable, and the next change to the padding
            # made the plan divide by a zero-sized tile.
            return EedaiPlan(
                False,
                None,
                0,
                (
                    f"{reason}, and no tile is small enough: the reader's "
                    f"{_EEDAI_WINDOW_PAD}-px window padding already exceeds the "
                    "per-tile budget here"
                ),
            )
        rows, cols = self._eedai_grid(bbox, scale_m)
        tiles = math.ceil(rows / tile_size) * math.ceil(cols / tile_size)
        if tiles > _EEDAI_MAX_TILES:
            return EedaiPlan(
                False,
                None,
                0,
                (
                    f"{reason}, and tiling it would take {tiles:,} tiles (over the "
                    f"{_EEDAI_MAX_TILES:,}-tile ceiling); every tile is its own "
                    "fetch and they are opened together to mosaic"
                ),
            )
        return EedaiPlan(True, tile_size, tiles, "")

    def _use_eedai(
        self, var_info: Dataset, band_count: int
    ) -> tuple[bool, EedaiPlan | None]:
        """Resolve the configured `engine` against this request's eligibility.

        Args:
            var_info: The catalog entry for the dataset being fetched.
            band_count: How many bands the read asks for; the reader holds
                them all, so they divide the per-tile budget.

        Returns:
            `(use_reader, plan)`. The plan is built here — after the
            short-circuits, so a request that opted out of this engine never
            pays for its sizing nor inherits its failure modes — and handed
            back so the read that follows is the same decision rather than a
            second one. It is `None` whenever `use_reader` is `False`.
            Under `"auto"` a request the plan declines falls back rather than
            failing: the user asked for a download, not for this engine.

        Raises:
            ValueError: If `engine="eedai"` was forced and the request is
                either ineligible — it needs server-side compute (a reduced
                collection, a `cloud_mask` or `filters`) or targets a projected
                `crs` — or eligible but declined by :meth:`_eedai_plan`,
                which the message names: the asset has no native resolution,
                the window is behind a polygon cutline, `resample` is not
                nearest-neighbour, or tiling it would cost more than Earth
                Engine would (too coarse a `scale` over a fine asset, too many
                native pixels, or too many tiles).
        """
        if self.engine == "ee":
            return False, None
        eligible = self._eedai_eligible(var_info)
        if self.engine == "eedai":
            if not eligible:
                raise ValueError(
                    f"engine='eedai' cannot serve {var_info.id}: the EEDAI "
                    "reader materialises pixels from an asset id, so it cannot "
                    "run server-side compute (a reduced collection, cloud_mask "
                    "or filters) and only writes "
                    f"crs={_EEDAI_NATIVE_CRS!r} (got {self.crs!r}). Use "
                    "engine='auto' or engine='ee'."
                )
            plan = self._eedai_plan(var_info, band_count)
            if not plan.can_serve:
                raise ValueError(
                    f"engine='eedai' cannot serve {var_info.id}: {plan.reason}. Use a "
                    "smaller bbox, engine='ee' (with auto_split=True to tile), or "
                    "export_via='drive'."
                )
            return True, plan
        if not (eligible and eedai_available()):
            return False, None
        plan = self._eedai_plan(var_info, band_count)
        if not plan.can_serve:
            logger.info(
                f"Serving {var_info.id} through Earth Engine rather than the EEDAI "
                f"reader: {plan.reason}."
            )
            return False, None
        return True, plan

    def _eedai_window(self) -> tuple[tuple[float, float, float, float], Any]:
        """Return the AOI the reader should read, as `(bbox, cutline)`.

        The reader takes `bbox` as the read window and only falls back to a
        `geometry`'s envelope when no `bbox` is given, so both are returned
        together: the bbox always describes the window the pixel grid is
        sized for, and the cutline (when a `region` was passed) clips the
        result to the exact polygon. Deriving the bbox from the region's own
        bounds is what keeps the window and the grid in agreement — sizing a
        bbox-shaped grid for a region-shaped window would silently change the
        ground resolution by the ratio of the two extents.

        Returns:
            `(bbox, cutline)` — the lat/lon `(min_x, min_y, max_x, max_y)`
            window, and the `region` to clip to or `None`.
        """
        region = self._region_in_native_crs(self.region)
        if region is not None:
            min_x, min_y, max_x, max_y = (float(v) for v in region.total_bounds)
            return (min_x, min_y, max_x, max_y), region
        return (
            self.space.longitude_min,
            self.space.latitude_min,
            self.space.longitude_max,
            self.space.latitude_max,
        ), None

    @staticmethod
    def _region_in_native_crs(region: Any) -> Any:
        """Return `region` in the lat/lon CRS the reader's `bbox` is read in.

        The reader reprojects a CRS-carrying `geometry` to the target CRS but
        takes `bbox` as already being in it. Handing over a projected
        region's bounds unchanged would therefore window in metres-read-as-
        degrees while the cutline landed correctly — two different parts of
        the planet. Reprojecting the region once keeps its bounds and its
        cutline in the same space.

        Args:
            region: The constructor `region`, or `None`.

        Returns:
            The region in EPSG:4326 (`None` passes through). A region with
            no CRS is assumed to be lat/lon already, matching how the Earth
            Engine path treats it.
        """
        if region is None:
            return None
        crs = getattr(region, "crs", None)
        if crs is None:
            return region
        to_epsg = getattr(crs, "to_epsg", None)
        if callable(to_epsg) and to_epsg() == 4326:
            return region
        return region.to_crs(_EEDAI_NATIVE_CRS)

    @staticmethod
    def _eedai_grid(
        bbox: tuple[float, float, float, float], scale: float
    ) -> tuple[int, int]:
        """Size a pixel grid for a lat/lon `bbox` at a metre `scale`.

        The reader sizes its output in the units of the output CRS (degrees
        here), so the metre `scale` has to become an explicit grid. This is
        deliberately not :meth:`SpatialExtent.estimate_pixel_dims`, which
        pyramids documents as a worst-case *upper bound* for cap pre-checks
        — it over-counts both axes (and unevenly, so a square AOI comes out
        non-square). Here the real span is used, with longitude degrees
        shortened by `cos(latitude)` at the AOI's mid-latitude, so the pixels
        are square on the ground at the requested `scale`.

        Args:
            bbox: The lat/lon window `(min_x, min_y, max_x, max_y)`.
            scale: Target ground sample distance in metres.

        Returns:
            `(rows, cols)` — at least one pixel per axis, so a sub-pixel AOI
            still yields a readable raster rather than a zero-sized one, and
            never coarser on the ground than the requested `scale`.

        Raises:
            ValueError: If any bound is not finite, or `scale` is not a
                positive number of metres.

        Examples:
            - A 0.1° box over Cairo at 90 m is taller than it is wide in
              pixels: a degree of longitude is shorter at that latitude, and
              the grid is sized at the box's poleward edge:
                ```python
                >>> from earthlens.gee.backend import GEE
                >>> GEE._eedai_grid((31.2, 29.9, 31.3, 30.0), 90.0)
                (124, 108)

                ```
            - Spanning the equator the two axes match, because the poleward
              edge is 1° and a degree of longitude is barely shortened there:
                ```python
                >>> from earthlens.gee.backend import GEE
                >>> GEE._eedai_grid((0.0, 0.0, 1.0, 1.0), 1000.0)
                (112, 112)

                ```
            - An AOI smaller than one pixel still yields a readable raster:
                ```python
                >>> from earthlens.gee.backend import GEE
                >>> GEE._eedai_grid((31.2, 29.9, 31.2001, 29.9001), 90.0)
                (1, 1)

                ```
        """
        min_x, min_y, max_x, max_y = bbox
        if not all(math.isfinite(bound) for bound in bbox):
            raise ValueError(f"the AOI bounds must be finite, got {bbox}")
        if scale <= 0 or not math.isfinite(scale):
            raise ValueError(f"scale must be a positive number of metres, got {scale}")
        # Take `cos` at the poleward edge rather than the mid-latitude: for a
        # tall AOI the mid-latitude value would under-count columns nearer the
        # pole, sampling coarser than asked. The poleward edge only ever errs
        # finer. Clamped away from the pole itself, where `cos` reaches zero.
        poleward = min(max(abs(min_y), abs(max_y)), 89.9)
        height_m = abs(max_y - min_y) * _METRES_PER_DEGREE
        width_m = (
            abs(max_x - min_x) * _METRES_PER_DEGREE * math.cos(math.radians(poleward))
        )
        # Round up, not to nearest: rounding down would leave the raster a
        # little coarser than the scale that was asked for.
        rows = max(1, math.ceil(height_m / scale))
        cols = max(1, math.ceil(width_m / scale))
        return rows, cols

    def _warn_cog_ignored(self, var_info: Dataset) -> None:
        """Say so, once, when `cog=True` cannot apply to this request.

        `cog=` only reaches the EEDAI writer, so a request that stays on
        Earth Engine silently yields a plain GeoTIFF. Without a notice the
        only symptom is an output that is not a COG.

        Args:
            var_info: The catalog entry being written (named in the notice).
        """
        if not self.cog or self._cog_warned:
            return
        self._cog_warned = True
        logger.warning(
            f"cog=True has no effect for {var_info.id}: it applies to the EEDAI "
            "path, and this request is served by Earth Engine (see engine=). A "
            "plain GeoTIFF is written instead."
        )

    def _eedai_credentials(self) -> Any:
        """Return the pyramids-eo credential for EEDAI reads, built once.

        `EarthEngineCredentials` writes inline key material to a private
        temp file whose removal is left to the garbage collector, so
        rebuilding it per bucket would scatter transient key files across a
        multi-band, multi-date download. It is therefore resolved once per
        instance and reused.

        The Earth Engine `project` is deliberately not forwarded: the reader
        authenticates GDAL's `EEDAI:` driver with the key alone. When no key
        resolves at all the reader falls back to Application Default
        Credentials, which may be a *different* identity from the one the
        Earth Engine half uses, so that case is logged rather than silent.

        Returns:
            The `pyramids_eo.earthengine.EarthEngineCredentials` to read with.

        Raises:
            AuthenticationError: If the credential cannot be built, so the
                failure matches this backend's error contract rather than
                surfacing pyramids-eo's own exception type.
            ImportError: If `pyramids-eo` (the `[eedai]` extra) is missing.
        """
        if self._eedai_credential is not None:
            return self._eedai_credential
        _service_account, service_key, _project = self._resolve_credentials()
        if service_key is None:
            logger.warning(
                "No Earth Engine service key resolved for the EEDAI read; falling "
                "back to Application Default Credentials, which may authenticate "
                "as a different identity than the Earth Engine half of this "
                "request. Pass service_key= (or set GEE_SERVICE_KEY) to pin it."
            )
        try:
            self._eedai_credential = credentials_for(service_key)
        except ImportError:
            raise
        except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
            raise AuthenticationError(
                f"could not build Earth Engine credentials for the EEDAI read: {exc}"
            ) from exc
        return self._eedai_credential

    def _eedai_native_fits(
        self,
        var_info: Dataset,
        bbox: tuple[float, float, float, float],
        band_count: int,
    ) -> tuple[bool, str]:
        """Report whether the reader's native-resolution read is bounded.

        The EEDAI driver's overviews are unreliable, so the reader fetches
        the AOI at the asset's *native* resolution and materialises it in
        memory before downsampling — a wide AOI over a fine-resolution asset
        is a huge read however coarse the requested `scale`.

        An asset with no catalogued `spatial_resolution` counts as not
        fitting, rather than as safe: an unknown native grid is exactly the
        case that cannot be sized up front.

        This answers only "can one pass hold it?". A window that does not fit
        is not necessarily refused — :meth:`_eedai_plan` may still serve it by
        streaming in tiles — so this reports rather than raises.

        Args:
            var_info: The catalog entry (for the asset's native resolution).
            bbox: The lat/lon window the reader would materialise.
            band_count: How many bands the read asks for. The reader holds
                every requested band of the window at once, so the budget is
                spent per band.

        Returns:
            `(fits, reason)` — `reason` is empty when it fits, and otherwise
            explains why in a form suitable for a log line or an error.
        """
        native_scale = var_info.spatial_resolution
        if not native_scale:
            return False, (
                f"{var_info.id} has no catalogued native resolution, so the "
                "reader's native-resolution read cannot be bounded up front"
            )
        # The warp holds whichever grid is larger: a `scale` finer than the
        # asset makes the output bigger than the native window. Fold them
        # together before *either* budget is applied.
        native_rows, native_cols = self._eedai_grid(bbox, float(native_scale))
        out_rows, out_cols = self._eedai_grid(bbox, float(self.scale or native_scale))
        rows = max(native_rows, out_rows)
        cols = max(native_cols, out_cols)
        binding = (
            "native"
            if (rows, cols) == (native_rows, native_cols)
            else f"{self.scale or native_scale} m output"
        )
        if max(rows, cols) > EE_MAX_DIMENSION:
            return False, (
                f"the AOI is about {cols}x{rows} px on {var_info.id}'s {binding} "
                f"grid, over the {EE_MAX_DIMENSION}-px per-axis budget the reader "
                "would hold in memory"
            )
        bands_held = max(band_count, 1)
        total_px = rows * cols * bands_held
        if total_px > _EEDAI_MAX_PIXELS:
            return False, (
                f"the AOI is about {cols * rows:,} px across {bands_held} band(s) "
                f"= {total_px:,} px on {var_info.id}'s {binding} grid, over the "
                f"{_EEDAI_MAX_PIXELS:,}-px budget the reader would hold in memory"
            )
        return True, ""

    def _export_via_eedai(
        self,
        var_info: Dataset,
        bands: list[str],
        scale: float,
        prefix: str,
        plan: EedaiPlan,
    ) -> Path:
        """Materialise one raw asset through the pyramids-eo EEDAI reader.

        Reads the requested bands straight from the asset via GDAL's `EEDAI`
        driver into a pyramids `Dataset` — reprojected to `crs`, clipped to
        the AOI — and writes it to `<prefix>.tif`. There is no
        `getDownloadURL` round-trip, so Earth Engine's 32768-px synchronous
        cap (and `auto_split`) does not apply.

        The reader sizes its output in the units of `crs` (degrees, since
        this path is EPSG:4326-only), whereas `scale` here is Earth Engine's
        metres. :meth:`_eedai_grid` reconciles the two by turning `scale`
        into an explicit `shape` over the same window
        :meth:`_eedai_window` hands the reader, so the grid and the read
        window always describe the same ground area.

        The raster is written as a plain GeoTIFF, or as a Cloud Optimized
        GeoTIFF (tiled, with overviews) when `cog=True` was passed to the
        constructor.

        Args:
            var_info: The catalog entry; its `id` is the Earth Engine asset.
            bands: Band ids to read.
            scale: Output pixel size in metres.
            prefix: Output filename stem (no extension).
            plan: The :class:`EedaiPlan` verdict from :meth:`_eedai_plan`,
                computed once by the caller so the routing decision and the
                read it performs cannot disagree.

        Returns:
            The :class:`pathlib.Path` of the written GeoTIFF.

        Raises:
            ImportError: If `pyramids-eo` (the `[eedai]` extra) is missing.
            AuthenticationError: If the reader's credentials cannot be built.
        """
        reader = import_earthengine_reader()
        credentials = self._eedai_credentials()
        target = self.root_dir / f"{prefix}.tif"
        # Write beside the target and rename on success: `to_file` / `to_cog`
        # write in place, so a mid-write failure would otherwise leave a
        # truncated raster sitting at the final name for a later run to read
        # as a finished product.
        staged = self.root_dir / f"{prefix}.partial.tif"
        # A tiled read has already written `staged`, so a COG conversion needs a
        # second name rather than using its source as its own destination.
        cog_staged = self.root_dir / f"{prefix}.partial-cog.tif"
        bbox, cutline = self._eedai_window()
        if not plan.can_serve:
            # Only reachable if a caller bypasses `_use_eedai`; taking the read
            # anyway would be the unguarded path the plan exists to prevent.
            raise ValueError(
                f"the EEDAI reader cannot serve {var_info.id}: {plan.reason}"
            )
        read_options: dict[str, Any] = {}
        tile_size = plan.tile_size
        if tile_size is not None:
            # Too large for one pass: have the reader stream the mosaic to disk
            # a tile at a time rather than hold the whole window in memory.
            read_options = {"tile_size": tile_size, "path": str(staged)}
            logger.info(
                f"Streaming {var_info.id} through the EEDAI reader as {plan.tiles:,} "
                f"tile(s) of {tile_size} px."
            )
        # The tiled read writes `staged` itself, so it belongs inside the same
        # `try` as the write: a mosaic that fails partway would otherwise leave
        # its partial file behind.
        try:
            dataset = reader.from_earthengine(
                var_info.id,
                bands=list(bands),
                crs=self.crs,
                bbox=bbox,
                geometry=cutline,
                shape=self._eedai_grid(bbox, scale),
                resample=self.resample,
                credentials=credentials,
                **read_options,
            )
            try:
                if self.cog:
                    dataset.cog.to_cog(str(cog_staged))
                elif tile_size is None:
                    dataset.to_file(str(staged))
            finally:
                close_quietly(dataset)
                # Drop the last reference before the rename: closing alone
                # leaves the GDAL object alive in this frame, so a collect
                # inside the retry would have nothing to free.
                dataset = None
            _rename_when_unlocked(cog_staged if self.cog else staged, target)
        finally:
            _discard_quietly(staged)
            _discard_quietly(cog_staged)
        logger.info(f"Wrote {target} (EEDAI{', COG' if self.cog else ''})")
        return target

    def _client(self) -> HttpClient:
        """Return this instance's HTTP client, built once.

        A tiled export issues one download per tile against the same host, so
        the client (and its pooled connection) is held on the instance rather
        than rebuilt per tile. The import stays local, as elsewhere in this
        module, so importing the backend does not pull the HTTP stack.

        Returns:
            HttpClient: The shared client.
        """
        from earthlens.base.http import HttpClient

        if self._http is None:
            self._http = HttpClient(timeout=self.http_timeout)
        return self._http

    def _download_one_url_tile(self, image, region, scale: float, prefix: str) -> Path:
        """Issue one `getDownloadURL` request → tif at `<prefix>.tif`.

        Single-tile worker shared by the small-AOI path and the
        auto-split loop. Stripped of size-checking — callers are
        expected to have already verified that the request fits the
        Earth Engine synchronous limit.
        """
        url = image.getDownloadURL(
            {"scale": scale, "crs": self.crs, "region": region, "format": "GEO_TIFF"}
        )
        target = self.root_dir / f"{prefix}.tif"
        # Route the (single-shot, expiring) getDownloadURL fetch through the
        # shared HttpClient so a transient 429/5xx is retried with back-off
        # instead of failing the tile outright.
        client = self._client()
        # Stream to a temp rather than buffering `response.content`: a tile is
        # capped at 32768 px/axis, so a single band can run to hundreds of
        # megabytes and the old path held it whole just to inspect four bytes.
        # GEE returns either a bare GeoTIFF or a zip of them, so the format is
        # decided from the leading bytes on disk.
        staged = self.root_dir / f"{prefix}.download"
        try:
            client.download(url, staged, progress=False)
            with open(staged, "rb") as handle:
                is_zip = handle.read(4) == _ZIP_MAGIC
            size = staged.stat().st_size
            if is_zip:
                PyramidsDataset.from_archive(
                    staged,
                    kind="zip",
                    member_glob="*.tif",
                    path=str(target),
                )
            else:
                # Release the reader before the `finally` unlink: pyramids keeps
                # the GDAL handle open, which holds a Windows lock on `staged`
                # (the same reason ghsl closes before its rename).
                reader = PyramidsDataset.read_file(str(staged))
                try:
                    reader.to_file(str(target))
                finally:
                    close_quietly(reader)
        finally:
            staged.unlink(missing_ok=True)
        logger.info(f"Wrote {target} ({size} bytes)")
        return target

    def _auto_split_and_download(
        self, image, var_info: Dataset, scale: float, prefix: str
    ) -> Path:
        """Tile an oversized AOI, download each tile, mosaic into one GeoTIFF.

        Only reachable when `auto_split=True` was passed to the
        constructor and the full AOI exceeds :data:`EE_MAX_DIMENSION` px
        per axis. The bbox is split with :func:`split_aoi_for_url`, each
        sub-extent is downloaded via :meth:`_download_one_url_tile`, and
        the per-tile tifs are mosaicked into `<prefix>.tif` with
        :func:`pyramids.dataset.merge.merge_rasters`. Per-tile tifs are
        deleted on success.
        """
        sub_extents = split_aoi_for_url(self.space, scale)
        logger.info(
            f"{var_info.id}: AOI exceeds {EE_MAX_DIMENSION}-px per-axis cap at "
            f"scale={scale} m; auto-splitting into {len(sub_extents)} tile(s)."
        )
        tile_paths: list[Path] = []
        for k, sub in enumerate(sub_extents):
            sub_region = ee.Geometry.Rectangle(
                [sub.west, sub.south, sub.east, sub.north]
            )
            sub_prefix = f"{prefix}_tile_{k:04d}"
            tile_paths.append(
                self._download_one_url_tile(image, sub_region, scale, sub_prefix)
            )
        target = self.root_dir / f"{prefix}.tif"
        merge_rasters([str(p) for p in tile_paths], str(target))
        for p in tile_paths:
            p.unlink(missing_ok=True)
        logger.info(f"Stitched {len(tile_paths)} tile(s) into {target} via pyramids.")
        return target

    def _export_via_batch(
        self, image, scale: float, region, prefix: str
    ) -> str | TaskInfo:
        """Queue an `ee.batch.Export.image.to{Drive,CloudStorage,Asset}` task.

        When `wait_for_export=True` (the default) blocks until the task
        reaches a terminal state via `wait_for_task` and returns the
        destination URL (`drive://...` / `gs://...` / `ee://...`).
        When `wait_for_export=False` returns a :class:`TaskInfo`
        immediately so the caller can track the task asynchronously via
        :mod:`earthlens.gee.jobs`.
        """
        common = {
            "image": image,
            "description": prefix[:100],
            "region": region,
            "scale": scale,
            "crs": self.crs,
            "maxPixels": 1e13,
        }
        if self.export_via == "drive":
            task = ee.batch.Export.image.toDrive(
                folder=self.drive_folder, fileNamePrefix=prefix, **common
            )
            destination = f"drive://{self.drive_folder}/{prefix}"
        elif self.export_via == "gcs":
            task = ee.batch.Export.image.toCloudStorage(
                bucket=self.gcs_bucket, fileNamePrefix=prefix, **common
            )
            destination = f"gs://{self.gcs_bucket}/{prefix}"
        else:
            # The asset sink uses `assetId` instead of `fileNamePrefix` —
            # each export creates one asset at `<self.asset_id>/<prefix>`.
            assert (
                self.asset_id is not None
            )  # constructor requires it for export_via='asset'
            target_asset = f"{self.asset_id.rstrip('/')}/{prefix}"
            task = ee.batch.Export.image.toAsset(assetId=target_asset, **common)
            destination = f"ee://{target_asset}"
        if not self.wait_for_export:
            task.start()
            info = _op_to_taskinfo(task.status())
            logger.info(
                f"Submitted {self.export_via} export {info.id} "
                f"({info.description}); track via earthlens.gee.jobs."
            )
            return info
        wait_for_task(task, progress_bar=True)
        logger.info(
            f"Exported {destination} (pull it from the {self.export_via} destination)"
        )
        return destination

    def _ee_region(self):
        """Return the `ee.Geometry` to clip / filter requests to.

        Uses the constructor `region` `GeoDataFrame` (converted via
        :func:`earthlens.gee.features.create_feature`) when given, else a
        polygon `aoi=` carried on `self.space.geometry` (the unified
        ergonomic channel), and otherwise an `ee.Geometry.Rectangle` built
        from the lat/lon bbox. Computed once and cached.

        Returns:
            The `ee.Geometry`.
        """
        if self._ee_geometry is None:
            aoi_geometry = getattr(self.space, "geometry", None)
            if self.region is not None:
                self._ee_geometry = create_feature(self.region).geometry()
            elif aoi_geometry is not None:
                self._ee_geometry = create_feature(aoi_geometry).geometry()
            else:
                self._ee_geometry = ee.Geometry.Rectangle(
                    [
                        self.space.longitude_min,
                        self.space.latitude_min,
                        self.space.longitude_max,
                        self.space.latitude_max,
                    ]
                )
        return self._ee_geometry

    def _clamp_window_to_extent(
        self, var_info: Dataset
    ) -> tuple[dt.datetime | None, dt.datetime | None]:
        """Clamp the request window to a dataset's published extent.

        Args:
            var_info: The catalog entry (its :class:`Extent`).

        Returns:
            `(start, end_exclusive)` — `start` is the later of the
            request start and the dataset start; `end_exclusive` is the
            earlier of (request end + 1 day) and (dataset end + 1 day, or
            "now" + 1 day for open-ended datasets). Returns
            `(None, None)` if the windows do not overlap.

            When `discover_extent=True` was passed at construction
            and the catalog's `end_date` (or `start_date`) is missing,
            the gap is filled by an EE-side
            `reduceColumns(minMax)` over `system:time_start` via
            :meth:`_discover_ee_extent` (cached per asset for the
            lifetime of the instance).
        """
        req_start = self.time.start_date
        req_end_excl = self.time.end_date + dt.timedelta(days=1)

        ds_start, ds_end_excl = self._effective_extent(var_info)

        start = max(req_start, ds_start)
        end_excl = min(req_end_excl, ds_end_excl)
        if start >= end_excl:
            return None, None
        return start, end_excl

    def _effective_extent(self, var_info: Dataset) -> tuple[dt.datetime, dt.datetime]:
        """Resolve a dataset's effective `(start, end_exclusive)` extent.

        The catalog's `start_date` is always a curated string (the
        `Extent` pydantic field is required); the upper bound comes
        from the curated `end_date` if present, else — when
        `discover_extent=True` — an EE-side `reduceColumns(minMax)`
        query (cached per asset), falling back to `now() + 1 day` if
        the query fails or the catalog has no `end_date` and
        discovery is disabled.

        Args:
            var_info: The catalog entry.

        Returns:
            `(start, end_exclusive)` as naive UTC datetimes.
        """
        ds_start = dt.datetime.strptime(var_info.extent.start_date, "%Y-%m-%d")
        catalog_end_str = var_info.extent.end_date

        if catalog_end_str is not None:
            ds_end_excl = dt.datetime.strptime(
                catalog_end_str, "%Y-%m-%d"
            ) + dt.timedelta(days=1)
            return ds_start, ds_end_excl

        _, ee_end = self._maybe_discover_ee_extent(var_info)
        if ee_end is not None:
            return ds_start, ee_end + dt.timedelta(days=1)

        # `now()` would be local-naive; the rest of the path is naive
        # UTC, so use a naive UTC value.
        ds_end_excl = dt.datetime.now(dt.UTC).replace(tzinfo=None) + dt.timedelta(
            days=1
        )
        return ds_start, ds_end_excl

    def _maybe_discover_ee_extent(
        self, var_info: Dataset
    ) -> tuple[dt.datetime | None, dt.datetime | None]:
        """Cached entry point for :meth:`_discover_ee_extent`.

        Reads / writes :data:`_EXTENT_CACHE` (module-level) so a
        second `GEE(...)` instance querying the same asset doesn't
        re-issue the 2-5 s `reduceColumns(minMax)` round trip.
        """
        if not self.discover_extent:
            return None, None
        cached = _EXTENT_CACHE.get(var_info.id)
        if cached is not None:
            return cached
        discovered = self._discover_ee_extent(var_info)
        _EXTENT_CACHE[var_info.id] = discovered
        return discovered

    def _discover_ee_extent(
        self, var_info: Dataset
    ) -> tuple[dt.datetime | None, dt.datetime | None]:
        """Query a collection's actual `system:time_start` min/max via EE.

        Issues one `reduceColumns(ee.Reducer.minMax(), ["system:time_start"])
        .getInfo()` round-trip per asset (callers cache via
        :meth:`_maybe_discover_ee_extent`). On any EE-side failure
        (network, missing property, image-typed asset) returns
        `(None, None)` and logs a warning — the caller falls back to
        the catalog values or `now()`.

        Args:
            var_info: The catalog entry. Only `var_info.id` is used.

        Returns:
            `(min_dt, max_dt)` as naive UTC datetimes, or `(None,
            None)` if the query failed or the collection has no
            time-stamped images.
        """
        try:
            collection = ee.ImageCollection(var_info.id)
            result = (
                collection.reduceColumns(
                    ee.Reducer.minMax(), ["system:time_start"]
                ).getInfo()
                or {}
            )
        except Exception as exc:  # noqa: BLE001 - downgrade EE errors to a warning
            logger.warning(
                f"discover_extent: reduceColumns(minMax) failed for "
                f"{var_info.id}: {type(exc).__name__}: {exc}; "
                "falling back to catalog / now()."
            )
            return None, None

        min_ms = result.get("min")
        max_ms = result.get("max")
        if min_ms is None or max_ms is None:
            return None, None
        return (
            dt.datetime.fromtimestamp(min_ms / 1000.0, tz=dt.UTC).replace(tzinfo=None),
            dt.datetime.fromtimestamp(max_ms / 1000.0, tz=dt.UTC).replace(tzinfo=None),
        )

catalog property #

The bundled GEE :class:~earthlens.gee.Catalog (alias of _catalog).

authenticate(service_account=None, service_key=None, project=None) #

Resolve credentials and open the Earth Engine connection.

The explicit, fail-fast credential step. Pass service_account= + service_key= (and optionally project=) to authenticate with a service-account key; omit a value to read its GEE_SERVICE_ACCOUNT / GEE_SERVICE_KEY / GEE_PROJECT environment variable instead. Opening the connection (which download() also does lazily if you never call this) validates the credentials against Earth Engine.

Parameters:

Name Type Description Default
service_account str | None

Service-account email. When None, the GEE_SERVICE_ACCOUNT environment variable is read.

 None
service_key str | None

Path to the service-account JSON key file, or the JSON content as a string. When None, the GEE_SERVICE_KEY environment variable is read.

 None
project str | None

Cloud project id to scope Earth Engine calls to. When None, the GEE_PROJECT environment variable is read (or the project is taken from the key's project_id).

 None

Returns:

Type Description
GEE

The backend instance, so it chains
GEE

EarthLens(...).authenticate(...).download().

Raises:

Type Description
AuthenticationError

If no service-account pair and no project can be resolved, or Earth Engine rejects the credentials.

Examples:

  • Authenticate with a service-account key, then download (live; skipped here): 
    >>> from earthlens.gee import GEE  # doctest: +SKIP
>>> GEE(  # doctest: +SKIP
...     start="2000-02-11", end="2000-02-12",
...     variables={"USGS/SRTMGL1_003": ["elevation"]},
...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3], path="data/gee",
... ).authenticate(
...     service_account="sa@my-project.iam.gserviceaccount.com",
...     service_key="/path/to/key.json",
... ).download()
  • Resolve the same credentials from the environment instead of passing them (live; skipped here): 
    >>> import os  # doctest: +SKIP
>>> os.environ["GEE_SERVICE_ACCOUNT"] = "sa@my-project.iam.gserviceaccount.com"
>>> os.environ["GEE_SERVICE_KEY"] = "/path/to/key.json"
>>> GEE(  # doctest: +SKIP
...     start="2000-02-11", end="2000-02-12",
...     variables={"USGS/SRTMGL1_003": ["elevation"]},
...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3], path="data/gee",
... ).authenticate().download()
Source code in libs/providers/imagery/src/earthlens/gee/backend.py 
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
def authenticate(
    self,
    service_account: str | None = None,
    service_key: str | None = None,
    project: str | None = None,
) -> GEE:
    """Resolve credentials and open the Earth Engine connection.

    The explicit, fail-fast credential step. Pass `service_account=`
    + `service_key=` (and optionally `project=`) to authenticate with
    a service-account key; omit a value to read its `GEE_SERVICE_ACCOUNT`
    / `GEE_SERVICE_KEY` / `GEE_PROJECT` environment variable instead.
    Opening the connection (which `download()` also does lazily if you
    never call this) validates the credentials against Earth Engine.

    Args:
        service_account: Service-account email. When `None`, the
            `GEE_SERVICE_ACCOUNT` environment variable is read.
        service_key: Path to the service-account JSON key file, or the
            JSON content as a string. When `None`, the `GEE_SERVICE_KEY`
            environment variable is read.
        project: Cloud project id to scope Earth Engine calls to. When
            `None`, the `GEE_PROJECT` environment variable is read (or
            the project is taken from the key's `project_id`).

    Returns:
        The backend instance, so it chains
        `EarthLens(...).authenticate(...).download()`.

    Raises:
        AuthenticationError: If no service-account pair and no project
            can be resolved, or Earth Engine rejects the credentials.

    Examples:
        - Authenticate with a service-account key, then download (live;
          skipped here):
            ```python
            >>> from earthlens.gee import GEE  # doctest: +SKIP
            >>> GEE(  # doctest: +SKIP
            ...     start="2000-02-11", end="2000-02-12",
            ...     variables={"USGS/SRTMGL1_003": ["elevation"]},
            ...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3], path="data/gee",
            ... ).authenticate(
            ...     service_account="sa@my-project.iam.gserviceaccount.com",
            ...     service_key="/path/to/key.json",
            ... ).download()

            ```
        - Resolve the same credentials from the environment instead of
          passing them (live; skipped here):
            ```python
            >>> import os  # doctest: +SKIP
            >>> os.environ["GEE_SERVICE_ACCOUNT"] = "sa@my-project.iam.gserviceaccount.com"
            >>> os.environ["GEE_SERVICE_KEY"] = "/path/to/key.json"
            >>> GEE(  # doctest: +SKIP
            ...     start="2000-02-11", end="2000-02-12",
            ...     variables={"USGS/SRTMGL1_003": ["elevation"]},
            ...     lat_lim=[29.9, 30.0], lon_lim=[31.2, 31.3], path="data/gee",
            ... ).authenticate().download()

            ```
    """
    if service_account is not None:
        self._service_account = service_account
    if service_key is not None:
        self._service_key = service_key
    if project is not None:
        self._project = project
    # Re-authenticating may switch identity, so the reader's cached
    # credential must not outlive the values it was built from.
    self._eedai_credential = None
    # LazyClientMixin: first access to `client` runs `_open_client` (auth).
    _ = self.client
    return self

download(progress_bar=True) #

Download every requested band-set of every requested dataset.

Parameters:

Name Type Description Default
progress_bar bool

Show a per-bucket tqdm bar. Defaults to True.

 True

Returns:

Type Description
list[Path | str | TaskInfo]

One entry per (dataset, band-set, time-bucket). The
list[Path | str | TaskInfo]

shape depends on the sink:
list[Path | str | TaskInfo]
  • export_via="url" — pathlib.Path to the written GeoTIFF (always synchronous).
list[Path | str | TaskInfo]
  • export_via="drive" / "gcs" / "asset" with the default wait_for_export=True — destination string ("drive://<folder>/<prefix>" / "gs://<bucket>/<prefix>" / "ee://<asset_id>/<prefix>"), populated only once the task reaches COMPLETED.
list[Path | str | TaskInfo]
  • export_via="drive" / "gcs" / "asset" with wait_for_export=False — TaskInfo captured at submission time; follow up via earthlens.gee.jobs (get_task_status, wait_for_task_id, etc.).

Raises:

Type Description
ValueError

On an unknown asset id, an unknown band, or an oversized "url" request (see :meth:_api).
RuntimeError

If a "drive" / "gcs" / "asset" export task fails. Only raised when wait_for_export=True; in the non-blocking mode the caller handles failures themselves via wait_for_task_id.

Examples:

  • Download one band, one image (needs network + credentials): 
    >>> gee = GEE(  # doctest: +SKIP
...     start="2020-06-01", end="2020-06-30",
...     temporal_resolution="monthly",
...     variables={"UCSB-CHG/CHIRPS/DAILY": ["precipitation"]},
...     lat_lim=[29.0, 30.0], lon_lim=[31.0, 32.0],
...     path="data/gee", scale=5566,
... )
>>> gee.authenticate(  # doctest: +SKIP
...     service_account="sa@p.iam.gserviceaccount.com",
...     service_key="/path/to/key.json",
... )
>>> paths = gee.download()  # doctest: +SKIP
>>> [p.name for p in paths]  # doctest: +SKIP
['UCSB-CHG_CHIRPS_DAILY_precipitation_20200601.tif']
  • aggregate= is not yet supported and is rejected up front: 
    >>> gee = GEE(  # doctest: +SKIP
...     start="2020-06-01", end="2020-06-01",
...     variables={"UCSB-CHG/CHIRPS/DAILY": ["precipitation"]},
...     lat_lim=[29.0, 30.0], lon_lim=[31.0, 32.0],
...     scale=5566,
... )
>>> gee.download(aggregate=object())  # doctest: +SKIP
Traceback (most recent call last):
    ...
NotImplementedError: aggregate= is not yet supported ...
See Also

earthlens.gee.Catalog: Resolves the {asset_id: [band, ...]} request against src/earthlens/gee/catalog/. earthlens.gee.auth.EarthEngineAuth: Performs the one-time ee.Initialize used by :meth:_open_client.

Source code in libs/providers/imagery/src/earthlens/gee/backend.py 
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
def download(self, progress_bar: bool = True) -> list[Path | str | TaskInfo]:
    """Download every requested band-set of every requested dataset.

    Args:
        progress_bar: Show a per-bucket `tqdm` bar. Defaults to `True`.

    Returns:
        One entry per `(dataset, band-set, time-bucket)`. The
        shape depends on the sink:

        * `export_via="url"` — `pathlib.Path` to the
            written GeoTIFF (always synchronous).
        * `export_via="drive"` / `"gcs"` / `"asset"` with the
            default `wait_for_export=True` — destination string
            (`"drive://<folder>/<prefix>"` / `"gs://<bucket>/<prefix>"` /
            `"ee://<asset_id>/<prefix>"`), populated only once
            the task reaches `COMPLETED`.
        * `export_via="drive"` / `"gcs"` / `"asset"` with
            `wait_for_export=False` — `TaskInfo` captured
            at submission time; follow up via
            `earthlens.gee.jobs` (`get_task_status`,
            `wait_for_task_id`, etc.).

    Raises:
        ValueError: On an unknown asset id, an unknown band, or an
            oversized `"url"` request (see :meth:`_api`).
        RuntimeError: If a `"drive"` / `"gcs"` / `"asset"` export
            task fails. Only raised when `wait_for_export=True`;
            in the non-blocking mode the caller handles failures
            themselves via `wait_for_task_id`.

    Examples:
        - Download one band, one image (needs network + credentials):
            ```python
            >>> gee = GEE(  # doctest: +SKIP
            ...     start="2020-06-01", end="2020-06-30",
            ...     temporal_resolution="monthly",
            ...     variables={"UCSB-CHG/CHIRPS/DAILY": ["precipitation"]},
            ...     lat_lim=[29.0, 30.0], lon_lim=[31.0, 32.0],
            ...     path="data/gee", scale=5566,
            ... )
            >>> gee.authenticate(  # doctest: +SKIP
            ...     service_account="sa@p.iam.gserviceaccount.com",
            ...     service_key="/path/to/key.json",
            ... )
            >>> paths = gee.download()  # doctest: +SKIP
            >>> [p.name for p in paths]  # doctest: +SKIP
            ['UCSB-CHG_CHIRPS_DAILY_precipitation_20200601.tif']

            ```
        - `aggregate=` is not yet supported and is rejected up front:
            ```python
            >>> gee = GEE(  # doctest: +SKIP
            ...     start="2020-06-01", end="2020-06-01",
            ...     variables={"UCSB-CHG/CHIRPS/DAILY": ["precipitation"]},
            ...     lat_lim=[29.0, 30.0], lon_lim=[31.0, 32.0],
            ...     scale=5566,
            ... )
            >>> gee.download(aggregate=object())  # doctest: +SKIP
            Traceback (most recent call last):
                ...
            NotImplementedError: aggregate= is not yet supported ...

            ```

    See Also:
        earthlens.gee.Catalog: Resolves the `{asset_id: [band, ...]}`
            request against `src/earthlens/gee/catalog/`.
        earthlens.gee.auth.EarthEngineAuth: Performs the one-time
            `ee.Initialize` used by :meth:`_open_client`.
    """
    # Trigger the lazy Earth Engine auth/init before any `ee` call.
    _ = self.client
    self._cog_warned = False  # the cog= notice is once per run, not per object
    outputs: list[Path | str | TaskInfo] = []
    assert isinstance(
        self.vars, dict
    )  # GEE always uses the {asset_id: [band]} form
    for asset_id, bands in self.vars.items():
        outputs.extend(self._download_dataset(asset_id, list(bands), progress_bar))
    return outputs

clear_extent_cache() #

Forget every cached EE-discovered temporal extent.

Mirrors :func:earthlens.gee.catalog.clear_catalog_cache for the discover_extent=True cache. Primarily useful in tests that need a fresh cache between runs (or when the EE-side data was updated and the in-process cache has gone stale).

Source code in libs/providers/imagery/src/earthlens/gee/backend.py 
184
185
186
187
188
189
190
191
192
def clear_extent_cache() -> None:
    """Forget every cached EE-discovered temporal extent.

    Mirrors :func:`earthlens.gee.catalog.clear_catalog_cache` for the
    `discover_extent=True` cache. Primarily useful in tests that need
    a fresh cache between runs (or when the EE-side data was updated
    and the in-process cache has gone stale).
    """
    _EXTENT_CACHE.clear()

earthlens.gee.catalog #

Dataset/band catalog loader for the Google Earth Engine backend.

Hosts :class:Catalog, the pydantic-backed reader for the bundled GEE catalog — the analogue of earthlens.ecmwf.Catalog / cds_data_catalog.yaml. The catalog ships as a directory of per-category YAML files at src/earthlens/gee/catalog/ (optical-multispectral.yaml, climate-reanalysis.yaml, land-cover-change.yaml, hydrology-water.yaml, community.yaml for projects/... user-contributed assets, …), plus a single _index.yaml carrying the merged available_datasets: list. Per-file sections each map to a typed field on :class:Catalog once merged:

  • available_datasets (informational list of Earth Engine asset ids) → :attr:Catalog.available_datasets
  • datasets (curated map of collections, each with band + aggregation metadata) → :attr:Catalog.datasets, with each value a :class:Dataset and each band a :class:Band.

Datasets are addressed by their Earth Engine asset id (e.g. "USGS/SRTMGL1_003", "COPERNICUS/S2_SR_HARMONIZED"); bands by (asset_id, band_id) via :meth:Catalog.get_band (aliased as :meth:Catalog.get_variable for parity with the ECMWF catalog).

The path to the bundled catalog directory lives at :data:CATALOG_PATH; tests can monkey-patch that module attribute to redirect the loader at a temporary directory or single YAML file.

Examples:

  • Construct the catalog and look up a dataset / band:

    >>> from earthlens.gee.catalog import Catalog
>>> cat = Catalog()
>>> cat.get_dataset("USGS/SRTMGL1_003").spatial_resolution
30.0
>>> cat.get_band("USGS/SRTMGL1_003", "elevation").units
'm'

Band #

Bases: BaseModel

Per-band metadata for one band of an Earth Engine dataset.

A frozen value object; the band id is injected from the YAML mapping key at load time, so the YAML body does not repeat it.

Attributes:

Name Type Description
id str

The Earth Engine band id (e.g. "SR_B4", "precipitation").
description str | None

Human description of the band, or None when only the band id is known. The bundled catalog YAMLs were swept once (M4 / commit 9cf1085) to drop 1334 redundant "Band <id>" stub descriptions that carried no information beyond the id; the loader itself stores whatever the YAML says — read the id if you just need a label.
units str | None

Physical unit string, or None (common for reflectance and indices).
scale float | None

Multiply the raw DN by this to get physical units, or None if no scaling applies.
offset float | None

Add this after scaling, or None.
wavelength float | None

Centre wavelength in micrometres for optical bands, or None.
min float | None

Typical / valid minimum DN, or None.
max float | None

Typical / valid maximum DN, or None.
estimated_range bool

True if min/max are sample-based estimates rather than hard bounds.

Examples:

  • Build a reflectance band and read its scaling: 
    >>> b = Band(id="SR_B4", description="Red surface reflectance", scale=2.75e-05, offset=-0.2)
>>> b.id
'SR_B4'
>>> b.scale
2.75e-05
>>> b.units is None
True
  • A band with only an id is fine — description is optional: 
    >>> Band(id="b1").description is None
True
  • An unknown field is rejected by extra="forbid": 
    >>> Band(id="x", description="d", not_a_band_field="x")  # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
    ...
pydantic_core._pydantic_core.ValidationError: 1 validation error for Band
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
class Band(BaseModel):
    """Per-band metadata for one band of an Earth Engine dataset.

    A frozen value object; the band id is injected from the YAML
    mapping key at load time, so the YAML body does not repeat it.

    Attributes:
        id: The Earth Engine band id (e.g. `"SR_B4"`, `"precipitation"`).
        description: Human description of the band, or `None` when only
            the band id is known. The bundled catalog YAMLs were swept
            once (M4 / commit `9cf1085`) to drop 1334 redundant
            `"Band <id>"` stub descriptions that carried no information
            beyond the id; the loader itself stores whatever the YAML
            says — read the id if you just need a label.
        units: Physical unit string, or `None` (common for reflectance
            and indices).
        scale: Multiply the raw DN by this to get physical units, or
            `None` if no scaling applies.
        offset: Add this after scaling, or `None`.
        wavelength: Centre wavelength in micrometres for optical bands,
            or `None`.
        min: Typical / valid minimum DN, or `None`.
        max: Typical / valid maximum DN, or `None`.
        estimated_range: `True` if `min`/`max` are sample-based
            estimates rather than hard bounds.

    Examples:
        - Build a reflectance band and read its scaling:
            ```python
            >>> b = Band(id="SR_B4", description="Red surface reflectance", scale=2.75e-05, offset=-0.2)
            >>> b.id
            'SR_B4'
            >>> b.scale
            2.75e-05
            >>> b.units is None
            True

            ```
        - A band with only an id is fine — `description` is optional:
            ```python
            >>> Band(id="b1").description is None
            True

            ```
        - An unknown field is rejected by `extra="forbid"`:
            ```python
            >>> Band(id="x", description="d", not_a_band_field="x")  # doctest: +IGNORE_EXCEPTION_DETAIL
            Traceback (most recent call last):
                ...
            pydantic_core._pydantic_core.ValidationError: 1 validation error for Band

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    id: str
    description: str | None = None
    units: str | None = None
    scale: float | None = None
    offset: float | None = None
    wavelength: float | None = None
    min: float | None = None
    max: float | None = None
    estimated_range: bool = False

Cadence #

Bases: BaseModel

Native temporal step of an Earth Engine collection.

A frozen value object derived from the STAC gee:interval field (which is sometimes inaccurate — the catalog YAML hand-corrects known cases).

Attributes:

Name Type Description
interval int

Number of unit periods between successive images (e.g. 16 for a 16-day composite).
unit Literal['minute', 'hour', 'day', 'pentad', 'dekad', 'month', 'year']

The period unit.

Examples:

  • Build a 16-day cadence and read its parts: 
    >>> c = Cadence(interval=16, unit="day")
>>> c.interval
16
>>> c.unit
'day'
  • A non-positive interval is rejected: 
    >>> Cadence(interval=0, unit="day")  # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
    ...
pydantic_core._pydantic_core.ValidationError: 1 validation error for Cadence
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
class Cadence(BaseModel):
    """Native temporal step of an Earth Engine collection.

    A frozen value object derived from the STAC `gee:interval` field
    (which is sometimes inaccurate — the catalog YAML hand-corrects
    known cases).

    Attributes:
        interval: Number of `unit` periods between successive images
            (e.g. `16` for a 16-day composite).
        unit: The period unit.

    Examples:
        - Build a 16-day cadence and read its parts:
            ```python
            >>> c = Cadence(interval=16, unit="day")
            >>> c.interval
            16
            >>> c.unit
            'day'

            ```
        - A non-positive interval is rejected:
            ```python
            >>> Cadence(interval=0, unit="day")  # doctest: +IGNORE_EXCEPTION_DETAIL
            Traceback (most recent call last):
                ...
            pydantic_core._pydantic_core.ValidationError: 1 validation error for Cadence

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    interval: int = Field(gt=0)
    unit: Literal["minute", "hour", "day", "pentad", "dekad", "month", "year"]

Catalog #

Bases: AbstractCatalog

YAML-backed catalog of Earth Engine datasets for the GEE backend.

Reads every *.yaml file under :data:CATALOG_PATH (the per-category catalog/ directory shipped with the package) on construction plus the canonical provider registry at :data:PROVIDERS_PATH, merging them into one logical catalog and validating every entry into typed :class:Dataset / :class:Band / :class:Provider models. A duplicate dataset/band key in the YAML (within a file or across files), an unknown band field, a curated dataset not listed in available_datasets, or a provider: slug that is missing from providers.yaml is a load-time error.

Attributes:

Name Type Description
available_datasets list[str]

Informational list of every Earth Engine asset id the package knows about.
datasets dict[str, Dataset]

Curated asset id → :class:Dataset.
providers dict[str, Provider]

Canonical provider slug → :class:Provider.

Examples:

  • Construct the catalog and look at what it holds: 
    >>> cat = Catalog()
>>> "USGS/SRTMGL1_003" in cat.datasets
True
>>> "USGS/SRTMGL1_003" in cat.available_datasets
True
>>> cat.get_dataset("UCSB-CHG/CHIRPS/DAILY").default_reducer
'mean'
  • Reach a band's metadata in one call: 
    >>> Catalog().get_band("MODIS/061/MOD11A1", "LST_Day_1km").scale
0.02
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
class Catalog(AbstractCatalog):
    """YAML-backed catalog of Earth Engine datasets for the GEE backend.

    Reads every `*.yaml` file under :data:`CATALOG_PATH` (the
    per-category `catalog/` directory shipped with the package) on
    construction plus the canonical provider registry at
    :data:`PROVIDERS_PATH`, merging them into one logical catalog and
    validating every entry into typed :class:`Dataset` / :class:`Band`
    / :class:`Provider` models. A duplicate dataset/band key in the
    YAML (within a file or across files), an unknown band field, a
    curated dataset not listed in `available_datasets`, or a
    `provider:` slug that is missing from `providers.yaml` is a
    load-time error.

    Attributes:
        available_datasets: Informational list of every Earth Engine
            asset id the package knows about.
        datasets: Curated asset id → :class:`Dataset`.
        providers: Canonical provider slug → :class:`Provider`.

    Examples:
        - Construct the catalog and look at what it holds:
            ```python
            >>> cat = Catalog()
            >>> "USGS/SRTMGL1_003" in cat.datasets
            True
            >>> "USGS/SRTMGL1_003" in cat.available_datasets
            True
            >>> cat.get_dataset("UCSB-CHG/CHIRPS/DAILY").default_reducer
            'mean'

            ```
        - Reach a band's metadata in one call:
            ```python
            >>> Catalog().get_band("MODIS/061/MOD11A1", "LST_Day_1km").scale
            0.02

            ```
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    _catalog_kind: str = "GEE catalog"

    available_datasets: list[str] = Field(default_factory=list)
    datasets: dict[str, Dataset] = Field(default_factory=dict)
    providers: dict[str, Provider] = Field(default_factory=dict)

    def model_post_init(self, __context: Any) -> None:
        """Auto-load the bundled catalog when the user didn't supply one.

        `Catalog()` with no args is sugar for `Catalog.load()` — it
        reads the bundled `catalog/*.yaml` and `providers.yaml`. If
        the caller passed `datasets=...` (or any non-empty field),
        the disk read is skipped: that path is for tests and ad-hoc
        in-memory catalogs (see :meth:`load` for the heavy-lifting
        classmethod).

        Raises:
            ValueError: When auto-loading, propagates the same errors
                as :meth:`load` (missing YAML, duplicate key, unknown
                band field, unregistered provider slug, …).
        """
        if self.datasets:
            super().model_post_init(__context)
            return
        loaded = Catalog.load()
        self.available_datasets = loaded.available_datasets
        self.datasets = loaded.datasets
        self.providers = loaded.providers
        super().model_post_init(__context)

    @classmethod
    def load(
        cls,
        catalog_path: Path | None = None,
        providers_path: Path | None = None,
    ) -> Catalog:
        """Read the catalog directory + providers registry from disk.

        Factored out of `model_post_init` so callers can:

        * point at a non-default catalog tree without monkey-patching
          `CATALOG_PATH` (`Catalog.load(my_dir)`);
        * hand-build a `Catalog` in tests via `Catalog(datasets=...)`
          without triggering disk I/O;
        * keep the parse cached on `(path, mtime_ns)` for both paths.

        Args:
            catalog_path: Per-category catalog directory (or a single
                `*.yaml` file for tests). Defaults to module-level
                :data:`CATALOG_PATH` at call time (so test
                monkey-patches take effect).
            providers_path: Path to `providers.yaml`. Defaults to
                module-level :data:`PROVIDERS_PATH` at call time.

        Returns:
            A fully-populated :class:`Catalog`.

        Raises:
            ValueError: If a YAML is missing, declares the same key
                twice, contains an unknown band field, lists a curated
                dataset absent from `available_datasets`, or references
                an unregistered provider slug.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        providers_path = (
            providers_path if providers_path is not None else PROVIDERS_PATH
        )
        available_datasets, datasets = _load_catalog_data(catalog_path)
        providers = _load_providers(providers_path)
        unknown = sorted(
            {
                d.provider
                for d in datasets.values()
                if d.provider and d.provider not in providers
            }
        )
        if unknown:
            raise ValueError(
                f"the following provider slugs are referenced by "
                f"`catalog/*.yaml` but missing from {providers_path}: {unknown}. "
                "Add them to providers.yaml or fix the typo."
            )
        return cls(
            available_datasets=list(available_datasets),
            datasets=dict(datasets),
            providers=dict(providers),
        )

    def health(self) -> dict[str, list[str]]:
        """Report structural hygiene issues across the loaded catalog.

        Returns a mapping from a check name to the list of asset ids
        (or provider slugs) that fail it. An empty list means the
        check is currently passing; an empty dict means the catalog
        is clean. Surfaced by `earthlens datasets validate gee` and
        CI hygiene checks.

        Checks reported:

        * `long_title` — datasets whose `title` exceeds 180 chars.
        * `html_in_title` — titles containing an `<html-tag>` pattern.
        * `raster_no_bands` — `is_raster` datasets with zero curated
          bands (often access-restricted or hydration-failed assets).
        * `unregistered_provider` — `Dataset.provider` slugs absent
          from `providers.yaml`. Should always be `[]` since the
          loader fails fast on this; included for defence in depth.
        * `unused_provider` — providers in `providers.yaml` that no
          dataset references. Lets you prune dead registry entries.

        Returns:
            `{check_name: [offending_ids, ...], ...}` with at least
            one key per check (empty list if nothing offends).
        """
        long_title: list[str] = []
        html_in_title: list[str] = []
        raster_no_bands: list[str] = []
        unregistered_provider: list[str] = []
        used_providers: set[str] = set()
        html_re = re.compile(r"<[A-Za-z][^>]*>")
        for asset_id, d in self.datasets.items():
            if d.title and len(d.title) > 180:
                long_title.append(asset_id)
            if d.title and html_re.search(d.title):
                html_in_title.append(asset_id)
            if d.is_raster and not d.bands:
                raster_no_bands.append(asset_id)
            if d.provider:
                used_providers.add(d.provider)
                if d.provider not in self.providers:
                    unregistered_provider.append(asset_id)
        unused_provider = sorted(set(self.providers) - used_providers)
        return {
            "long_title": sorted(long_title),
            "html_in_title": sorted(html_in_title),
            "raster_no_bands": sorted(raster_no_bands),
            "unregistered_provider": sorted(unregistered_provider),
            "unused_provider": unused_provider,
        }

    # `get_provider(slug)` (with did-you-mean hint) lifted to
    # :class:`earthlens.base.AbstractCatalog`.

    def get_catalog(self) -> dict[str, Dataset]:
        """Return the curated dataset map (asset id → :class:`Dataset`).

        Returns:
            The :attr:`datasets` mapping.

        Examples:
            - The map is keyed by Earth Engine asset id:
                ```python
                >>> cat = Catalog()
                >>> "ESA/WorldCover/v200" in cat.get_catalog()
                True
                >>> cat.get_catalog()["ESA/WorldCover/v200"].title
                'ESA WorldCover 10m v200 (2021)'

                ```
        """
        return self.datasets

    def get_band(self, dataset_id: str, band_id: str) -> Band:
        """Return the :class:`Band` for `(dataset_id, band_id)`.

        Args:
            dataset_id: The Earth Engine asset id.
            band_id: The band id within that dataset.

        Returns:
            The matching :class:`Band`.

        Raises:
            ValueError: If the dataset or the band is unknown.

        Examples:
            - Read a precipitation band's unit:
                ```python
                >>> Catalog().get_band("UCSB-CHG/CHIRPS/DAILY", "precipitation").units
                'mm/d'

                ```
            - Read a Sentinel-2 band's centre wavelength:
                ```python
                >>> Catalog().get_band("COPERNICUS/S2_SR_HARMONIZED", "B4").wavelength
                0.6645

                ```

        See Also:
            get_variable: Identical; provided for naming parity with
                `earthlens.ecmwf.Catalog.get_variable`.
        """
        return cast("Band", self.get_dataset(dataset_id).get_band(band_id))

    def get_variable(self, dataset_id: str, band_id: str) -> Band:
        """Alias of :meth:`get_band` (name parity with the ECMWF catalog).

        Args:
            dataset_id: The Earth Engine asset id.
            band_id: The band id within that dataset.

        Returns:
            The matching :class:`Band`.

        Examples:
            - Same result as :meth:`get_band`:
                ```python
                >>> Catalog().get_variable("USGS/SRTMGL1_003", "elevation").units
                'm'

                ```
        """
        return self.get_band(dataset_id, band_id)

    # -- job / task tracking shortcuts -----------------------------------
    # Thin delegations to `earthlens.gee.jobs` so callers can stay on the
    # catalog object instead of importing the jobs module separately —
    # parity with `earthlens.ecmwf.Catalog.list_recent_jobs`.

    def list_recent_tasks(self, **kwargs: Any) -> list[TaskInfo]:
        """List recent Earth Engine batch tasks (delegates to `gee.jobs`).

        Args:
            **kwargs: Forwarded verbatim to
                :func:`earthlens.gee.jobs.list_recent_tasks` —
                `state` / `max_age_min` / `task_type` /
                `description_prefix` / `project` / `limit`.

        Returns:
            A list of :class:`earthlens.gee.jobs.TaskInfo`, newest first.
        """
        from earthlens.gee.jobs import list_recent_tasks

        return list_recent_tasks(**kwargs)

    def get_task_status(self, task_id: str, **kwargs: Any) -> TaskInfo:
        """Fetch one task's status by id (delegates to `gee.jobs`).

        Args:
            task_id: Bare task id or full operation name.
            **kwargs: Forwarded to
                :func:`earthlens.gee.jobs.get_task_status` —
                currently just `project`.

        Returns:
            A :class:`earthlens.gee.jobs.TaskInfo`.
        """
        from earthlens.gee.jobs import get_task_status

        return get_task_status(task_id, **kwargs)

    def audit_recent_tasks(
        self,
        max_age_min: int = 7 * 24 * 60,
        **kwargs: Any,
    ) -> dict[str, list[TaskInfo]]:
        """Group recent batch tasks by state — the task-side `health()`.

        Walks :func:`earthlens.gee.jobs.list_recent_tasks` with the
        given `max_age_min` (default: 7 days) and any extra filters
        from `**kwargs`, then groups the results by `TaskInfo.state`.
        Useful for "are any of yesterday's exports stuck or failed?"
        eyeballing.

        Args:
            max_age_min: Window length in minutes. Defaults to 7 days
                — long enough to cover a typical weekly batch.
            **kwargs: Forwarded to
                :func:`earthlens.gee.jobs.list_recent_tasks` —
                `task_type` / `description_prefix` / `project` / `limit`.
                Do not pass `state` here (the helper groups by state
                itself); a `state` kwarg would silently narrow the
                report.

        Returns:
            `{state_name: [TaskInfo, ...], ...}` for every state that
            appears in the window. Empty dict if no tasks match.

        Examples:
            - Eyeball this week's exports for the SRTM asset:
                ```python
                >>> report = Catalog().audit_recent_tasks(  # doctest: +SKIP
                ...     description_prefix="USGS_SRTMGL1_003",
                ... )
                >>> for failed in report.get("FAILED", []):  # doctest: +SKIP
                ...     print(failed.id, failed.error_message)

                ```
        """
        from earthlens.gee.jobs import list_recent_tasks

        if "state" in kwargs:
            raise ValueError(
                "audit_recent_tasks groups by state — don't pass `state=`; "
                "filter on the returned dict instead."
            )
        tasks = list_recent_tasks(max_age_min=max_age_min, **kwargs)
        report: dict[str, list[TaskInfo]] = {}
        for t in tasks:
            report.setdefault(t.state, []).append(t)
        return report

audit_recent_tasks(max_age_min=7 * 24 * 60, **kwargs) #

Group recent batch tasks by state — the task-side health().

Walks :func:earthlens.gee.jobs.list_recent_tasks with the given max_age_min (default: 7 days) and any extra filters from **kwargs, then groups the results by TaskInfo.state. Useful for "are any of yesterday's exports stuck or failed?" eyeballing.

Parameters:

Name Type Description Default
max_age_min int

Window length in minutes. Defaults to 7 days — long enough to cover a typical weekly batch.

 7 * 24 * 60
**kwargs Any

Forwarded to :func:earthlens.gee.jobs.list_recent_tasks — task_type / description_prefix / project / limit. Do not pass state here (the helper groups by state itself); a state kwarg would silently narrow the report.

 {}

Returns:

Type Description
dict[str, list[TaskInfo]]

{state_name: [TaskInfo, ...], ...} for every state that
dict[str, list[TaskInfo]]

appears in the window. Empty dict if no tasks match.

Examples:

  • Eyeball this week's exports for the SRTM asset: 
    >>> report = Catalog().audit_recent_tasks(  # doctest: +SKIP
...     description_prefix="USGS_SRTMGL1_003",
... )
>>> for failed in report.get("FAILED", []):  # doctest: +SKIP
...     print(failed.id, failed.error_message)
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
def audit_recent_tasks(
    self,
    max_age_min: int = 7 * 24 * 60,
    **kwargs: Any,
) -> dict[str, list[TaskInfo]]:
    """Group recent batch tasks by state — the task-side `health()`.

    Walks :func:`earthlens.gee.jobs.list_recent_tasks` with the
    given `max_age_min` (default: 7 days) and any extra filters
    from `**kwargs`, then groups the results by `TaskInfo.state`.
    Useful for "are any of yesterday's exports stuck or failed?"
    eyeballing.

    Args:
        max_age_min: Window length in minutes. Defaults to 7 days
            — long enough to cover a typical weekly batch.
        **kwargs: Forwarded to
            :func:`earthlens.gee.jobs.list_recent_tasks` —
            `task_type` / `description_prefix` / `project` / `limit`.
            Do not pass `state` here (the helper groups by state
            itself); a `state` kwarg would silently narrow the
            report.

    Returns:
        `{state_name: [TaskInfo, ...], ...}` for every state that
        appears in the window. Empty dict if no tasks match.

    Examples:
        - Eyeball this week's exports for the SRTM asset:
            ```python
            >>> report = Catalog().audit_recent_tasks(  # doctest: +SKIP
            ...     description_prefix="USGS_SRTMGL1_003",
            ... )
            >>> for failed in report.get("FAILED", []):  # doctest: +SKIP
            ...     print(failed.id, failed.error_message)

            ```
    """
    from earthlens.gee.jobs import list_recent_tasks

    if "state" in kwargs:
        raise ValueError(
            "audit_recent_tasks groups by state — don't pass `state=`; "
            "filter on the returned dict instead."
        )
    tasks = list_recent_tasks(max_age_min=max_age_min, **kwargs)
    report: dict[str, list[TaskInfo]] = {}
    for t in tasks:
        report.setdefault(t.state, []).append(t)
    return report

get_band(dataset_id, band_id) #

Return the :class:Band for (dataset_id, band_id).

Parameters:

Name Type Description Default
dataset_id str

The Earth Engine asset id.

 required
band_id str

The band id within that dataset.

 required

Returns:

Type Description
Band

The matching :class:Band.

Raises:

Type Description
ValueError

If the dataset or the band is unknown.

Examples:

  • Read a precipitation band's unit: 
    >>> Catalog().get_band("UCSB-CHG/CHIRPS/DAILY", "precipitation").units
'mm/d'
  • Read a Sentinel-2 band's centre wavelength: 
    >>> Catalog().get_band("COPERNICUS/S2_SR_HARMONIZED", "B4").wavelength
0.6645
See Also

get_variable: Identical; provided for naming parity with earthlens.ecmwf.Catalog.get_variable.

Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
def get_band(self, dataset_id: str, band_id: str) -> Band:
    """Return the :class:`Band` for `(dataset_id, band_id)`.

    Args:
        dataset_id: The Earth Engine asset id.
        band_id: The band id within that dataset.

    Returns:
        The matching :class:`Band`.

    Raises:
        ValueError: If the dataset or the band is unknown.

    Examples:
        - Read a precipitation band's unit:
            ```python
            >>> Catalog().get_band("UCSB-CHG/CHIRPS/DAILY", "precipitation").units
            'mm/d'

            ```
        - Read a Sentinel-2 band's centre wavelength:
            ```python
            >>> Catalog().get_band("COPERNICUS/S2_SR_HARMONIZED", "B4").wavelength
            0.6645

            ```

    See Also:
        get_variable: Identical; provided for naming parity with
            `earthlens.ecmwf.Catalog.get_variable`.
    """
    return cast("Band", self.get_dataset(dataset_id).get_band(band_id))

get_catalog() #

Return the curated dataset map (asset id → :class:Dataset).

Returns:

Name Type Description
The dict[str, Dataset]

attr:datasets mapping.

Examples:

  • The map is keyed by Earth Engine asset id: 
    >>> cat = Catalog()
>>> "ESA/WorldCover/v200" in cat.get_catalog()
True
>>> cat.get_catalog()["ESA/WorldCover/v200"].title
'ESA WorldCover 10m v200 (2021)'
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
def get_catalog(self) -> dict[str, Dataset]:
    """Return the curated dataset map (asset id → :class:`Dataset`).

    Returns:
        The :attr:`datasets` mapping.

    Examples:
        - The map is keyed by Earth Engine asset id:
            ```python
            >>> cat = Catalog()
            >>> "ESA/WorldCover/v200" in cat.get_catalog()
            True
            >>> cat.get_catalog()["ESA/WorldCover/v200"].title
            'ESA WorldCover 10m v200 (2021)'

            ```
    """
    return self.datasets

get_task_status(task_id, **kwargs) #

Fetch one task's status by id (delegates to gee.jobs).

Parameters:

Name Type Description Default
task_id str

Bare task id or full operation name.

 required
**kwargs Any

Forwarded to :func:earthlens.gee.jobs.get_task_status — currently just project.

 {}

Returns:

Name Type Description
A TaskInfo

class:earthlens.gee.jobs.TaskInfo.
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
def get_task_status(self, task_id: str, **kwargs: Any) -> TaskInfo:
    """Fetch one task's status by id (delegates to `gee.jobs`).

    Args:
        task_id: Bare task id or full operation name.
        **kwargs: Forwarded to
            :func:`earthlens.gee.jobs.get_task_status` —
            currently just `project`.

    Returns:
        A :class:`earthlens.gee.jobs.TaskInfo`.
    """
    from earthlens.gee.jobs import get_task_status

    return get_task_status(task_id, **kwargs)

get_variable(dataset_id, band_id) #

Alias of :meth:get_band (name parity with the ECMWF catalog).

Parameters:

Name Type Description Default
dataset_id str

The Earth Engine asset id.

 required
band_id str

The band id within that dataset.

 required

Returns:

Type Description
Band

The matching :class:Band.

Examples:

  • Same result as :meth:get_band: 
    >>> Catalog().get_variable("USGS/SRTMGL1_003", "elevation").units
'm'
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
def get_variable(self, dataset_id: str, band_id: str) -> Band:
    """Alias of :meth:`get_band` (name parity with the ECMWF catalog).

    Args:
        dataset_id: The Earth Engine asset id.
        band_id: The band id within that dataset.

    Returns:
        The matching :class:`Band`.

    Examples:
        - Same result as :meth:`get_band`:
            ```python
            >>> Catalog().get_variable("USGS/SRTMGL1_003", "elevation").units
            'm'

            ```
    """
    return self.get_band(dataset_id, band_id)

health() #

Report structural hygiene issues across the loaded catalog.

Returns a mapping from a check name to the list of asset ids (or provider slugs) that fail it. An empty list means the check is currently passing; an empty dict means the catalog is clean. Surfaced by earthlens datasets validate gee and CI hygiene checks.

Checks reported:

  • long_title — datasets whose title exceeds 180 chars.
  • html_in_title — titles containing an <html-tag> pattern.
  • raster_no_bands — is_raster datasets with zero curated bands (often access-restricted or hydration-failed assets).
  • unregistered_provider — Dataset.provider slugs absent from providers.yaml. Should always be [] since the loader fails fast on this; included for defence in depth.
  • unused_provider — providers in providers.yaml that no dataset references. Lets you prune dead registry entries.

Returns:

Type Description
dict[str, list[str]]

{check_name: [offending_ids, ...], ...} with at least
dict[str, list[str]]

one key per check (empty list if nothing offends).
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
def health(self) -> dict[str, list[str]]:
    """Report structural hygiene issues across the loaded catalog.

    Returns a mapping from a check name to the list of asset ids
    (or provider slugs) that fail it. An empty list means the
    check is currently passing; an empty dict means the catalog
    is clean. Surfaced by `earthlens datasets validate gee` and
    CI hygiene checks.

    Checks reported:

    * `long_title` — datasets whose `title` exceeds 180 chars.
    * `html_in_title` — titles containing an `<html-tag>` pattern.
    * `raster_no_bands` — `is_raster` datasets with zero curated
      bands (often access-restricted or hydration-failed assets).
    * `unregistered_provider` — `Dataset.provider` slugs absent
      from `providers.yaml`. Should always be `[]` since the
      loader fails fast on this; included for defence in depth.
    * `unused_provider` — providers in `providers.yaml` that no
      dataset references. Lets you prune dead registry entries.

    Returns:
        `{check_name: [offending_ids, ...], ...}` with at least
        one key per check (empty list if nothing offends).
    """
    long_title: list[str] = []
    html_in_title: list[str] = []
    raster_no_bands: list[str] = []
    unregistered_provider: list[str] = []
    used_providers: set[str] = set()
    html_re = re.compile(r"<[A-Za-z][^>]*>")
    for asset_id, d in self.datasets.items():
        if d.title and len(d.title) > 180:
            long_title.append(asset_id)
        if d.title and html_re.search(d.title):
            html_in_title.append(asset_id)
        if d.is_raster and not d.bands:
            raster_no_bands.append(asset_id)
        if d.provider:
            used_providers.add(d.provider)
            if d.provider not in self.providers:
                unregistered_provider.append(asset_id)
    unused_provider = sorted(set(self.providers) - used_providers)
    return {
        "long_title": sorted(long_title),
        "html_in_title": sorted(html_in_title),
        "raster_no_bands": sorted(raster_no_bands),
        "unregistered_provider": sorted(unregistered_provider),
        "unused_provider": unused_provider,
    }

list_recent_tasks(**kwargs) #

List recent Earth Engine batch tasks (delegates to gee.jobs).

Parameters:

Name Type Description Default
**kwargs Any

Forwarded verbatim to :func:earthlens.gee.jobs.list_recent_tasks — state / max_age_min / task_type / description_prefix / project / limit.

 {}

Returns:

Type Description
list[TaskInfo]

A list of :class:earthlens.gee.jobs.TaskInfo, newest first.
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
def list_recent_tasks(self, **kwargs: Any) -> list[TaskInfo]:
    """List recent Earth Engine batch tasks (delegates to `gee.jobs`).

    Args:
        **kwargs: Forwarded verbatim to
            :func:`earthlens.gee.jobs.list_recent_tasks` —
            `state` / `max_age_min` / `task_type` /
            `description_prefix` / `project` / `limit`.

    Returns:
        A list of :class:`earthlens.gee.jobs.TaskInfo`, newest first.
    """
    from earthlens.gee.jobs import list_recent_tasks

    return list_recent_tasks(**kwargs)

load(catalog_path=None, providers_path=None) classmethod #

Read the catalog directory + providers registry from disk.

Factored out of model_post_init so callers can:

  • point at a non-default catalog tree without monkey-patching CATALOG_PATH (Catalog.load(my_dir));
  • hand-build a Catalog in tests via Catalog(datasets=...) without triggering disk I/O;
  • keep the parse cached on (path, mtime_ns) for both paths.

Parameters:

Name Type Description Default
catalog_path Path | None

Per-category catalog directory (or a single *.yaml file for tests). Defaults to module-level :data:CATALOG_PATH at call time (so test monkey-patches take effect).

 None
providers_path Path | None

Path to providers.yaml. Defaults to module-level :data:PROVIDERS_PATH at call time.

 None

Returns:

Type Description
Catalog

A fully-populated :class:Catalog.

Raises:

Type Description
ValueError

If a YAML is missing, declares the same key twice, contains an unknown band field, lists a curated dataset absent from available_datasets, or references an unregistered provider slug.
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
@classmethod
def load(
    cls,
    catalog_path: Path | None = None,
    providers_path: Path | None = None,
) -> Catalog:
    """Read the catalog directory + providers registry from disk.

    Factored out of `model_post_init` so callers can:

    * point at a non-default catalog tree without monkey-patching
      `CATALOG_PATH` (`Catalog.load(my_dir)`);
    * hand-build a `Catalog` in tests via `Catalog(datasets=...)`
      without triggering disk I/O;
    * keep the parse cached on `(path, mtime_ns)` for both paths.

    Args:
        catalog_path: Per-category catalog directory (or a single
            `*.yaml` file for tests). Defaults to module-level
            :data:`CATALOG_PATH` at call time (so test
            monkey-patches take effect).
        providers_path: Path to `providers.yaml`. Defaults to
            module-level :data:`PROVIDERS_PATH` at call time.

    Returns:
        A fully-populated :class:`Catalog`.

    Raises:
        ValueError: If a YAML is missing, declares the same key
            twice, contains an unknown band field, lists a curated
            dataset absent from `available_datasets`, or references
            an unregistered provider slug.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    providers_path = (
        providers_path if providers_path is not None else PROVIDERS_PATH
    )
    available_datasets, datasets = _load_catalog_data(catalog_path)
    providers = _load_providers(providers_path)
    unknown = sorted(
        {
            d.provider
            for d in datasets.values()
            if d.provider and d.provider not in providers
        }
    )
    if unknown:
        raise ValueError(
            f"the following provider slugs are referenced by "
            f"`catalog/*.yaml` but missing from {providers_path}: {unknown}. "
            "Add them to providers.yaml or fix the typo."
        )
    return cls(
        available_datasets=list(available_datasets),
        datasets=dict(datasets),
        providers=dict(providers),
    )

model_post_init(__context) #

Auto-load the bundled catalog when the user didn't supply one.

Catalog() with no args is sugar for Catalog.load() — it reads the bundled catalog/*.yaml and providers.yaml. If the caller passed datasets=... (or any non-empty field), the disk read is skipped: that path is for tests and ad-hoc in-memory catalogs (see :meth:load for the heavy-lifting classmethod).

Raises:

Type Description
ValueError

When auto-loading, propagates the same errors as :meth:load (missing YAML, duplicate key, unknown band field, unregistered provider slug, …).
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
def model_post_init(self, __context: Any) -> None:
    """Auto-load the bundled catalog when the user didn't supply one.

    `Catalog()` with no args is sugar for `Catalog.load()` — it
    reads the bundled `catalog/*.yaml` and `providers.yaml`. If
    the caller passed `datasets=...` (or any non-empty field),
    the disk read is skipped: that path is for tests and ad-hoc
    in-memory catalogs (see :meth:`load` for the heavy-lifting
    classmethod).

    Raises:
        ValueError: When auto-loading, propagates the same errors
            as :meth:`load` (missing YAML, duplicate key, unknown
            band field, unregistered provider slug, …).
    """
    if self.datasets:
        super().model_post_init(__context)
        return
    loaded = Catalog.load()
    self.available_datasets = loaded.available_datasets
    self.datasets = loaded.datasets
    self.providers = loaded.providers
    super().model_post_init(__context)

Dataset #

Bases: BaseModel

One Earth Engine collection/image with curated metadata.

A frozen value object; the asset id is injected from the YAML mapping key at load time.

Attributes:

Name Type Description
id str

The Earth Engine asset id (e.g. "COPERNICUS/S2_SR_HARMONIZED").
title str

Human title.
provider str | None

Canonical provider slug (e.g. "nasa-lp-daac", "copernicus"), or None. The catalog validates that every non-None slug is registered in providers.yaml; resolve to a display name via Catalog.get_provider(slug).
ee_type Literal['image', 'image_collection', 'table', 'table_collection', 'bigquery_table']

"image" (a single static raster), "image_collection" (a time series), "table" (a FeatureCollection — out of scope for the raster backend), "table_collection" (a collection of FeatureCollections, e.g. GEDI footprint shots), or "bigquery_table" (a BigQuery-backed table — also out of scope for the raster backend; included for catalog completeness).
cadence Cadence | None

Native temporal step, or None for static images.
spatial_resolution float | None

Nominal pixel size in metres, or None.
extent Extent

Spatial/temporal coverage.
default_reducer Literal['mean', 'median', 'mosaic', 'min', 'max', 'mode', 'sum']

Earth Engine reducer name used to collapse a temporal composite. One of "mean" (continuous fields / rates), "median" (cloud-screened optical scenes), "mosaic" (tiled / annual static maps), "min", "max", "mode", or "sum". Constrained by Literal to catch typos at YAML-load time.
license str | None

SPDX identifier ("CC-BY-4.0", "CC-BY-SA-4.0", "CC-BY-NC-SA-4.0", "CC0-1.0", "ODbL-1.0", …) or one of the conventional values "public-domain", "proprietary" (publisher-specific terms-of-service), or "unknown". None for stanzas that pre-date the licence-normalisation pass.
terms_note str | None

Free-text note that doesn't fit the SPDX id — attribution requirements, custom commercial clauses, links to publisher terms-of-use pages, etc. None when the license field alone conveys everything.
source Literal['ee_native', 'republished', 'community']

Where the asset originated, used to disambiguate the three publication paths Earth Engine exposes. One of:

  • "ee_native" — first-party Earth Engine catalog entry published by the data provider directly (the default; the vast majority of assets).
  • "republished" — a copy of an external dataset that Google or a partner re-hosts in the EE catalog under an organisation-style asset id.
  • "community" — a user-uploaded asset whose path starts with projects/.... Often documented less rigorously than the first two.

Replaces the older ambiguous user_uploaded: bool flag.
bands dict[str, Band]

Band id → :class:Band.
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
class Dataset(BaseModel):
    """One Earth Engine collection/image with curated metadata.

    A frozen value object; the asset id is injected from the YAML
    mapping key at load time.

    Attributes:
        id: The Earth Engine asset id (e.g. `"COPERNICUS/S2_SR_HARMONIZED"`).
        title: Human title.
        provider: Canonical provider slug (e.g. `"nasa-lp-daac"`,
            `"copernicus"`), or `None`. The catalog validates that
            every non-`None` slug is registered in `providers.yaml`;
            resolve to a display name via `Catalog.get_provider(slug)`.
        ee_type: `"image"` (a single static raster), `"image_collection"`
            (a time series), `"table"` (a `FeatureCollection` — out of
            scope for the raster backend), `"table_collection"` (a
            collection of FeatureCollections, e.g. GEDI footprint shots),
            or `"bigquery_table"` (a BigQuery-backed table — also out of
            scope for the raster backend; included for catalog completeness).
        cadence: Native temporal step, or `None` for static images.
        spatial_resolution: Nominal pixel size in metres, or `None`.
        extent: Spatial/temporal coverage.
        default_reducer: Earth Engine reducer name used to collapse a
            temporal composite. One of `"mean"` (continuous fields /
            rates), `"median"` (cloud-screened optical scenes),
            `"mosaic"` (tiled / annual static maps), `"min"`, `"max"`,
            `"mode"`, or `"sum"`. Constrained by `Literal` to catch
            typos at YAML-load time.
        license: SPDX identifier (`"CC-BY-4.0"`, `"CC-BY-SA-4.0"`,
            `"CC-BY-NC-SA-4.0"`, `"CC0-1.0"`, `"ODbL-1.0"`, …) or one of
            the conventional values `"public-domain"`, `"proprietary"`
            (publisher-specific terms-of-service), or `"unknown"`. `None`
            for stanzas that pre-date the licence-normalisation pass.
        terms_note: Free-text note that doesn't fit the SPDX id —
            attribution requirements, custom commercial clauses, links
            to publisher terms-of-use pages, etc. `None` when the
            `license` field alone conveys everything.
        source: Where the asset originated, used to disambiguate the
            three publication paths Earth Engine exposes. One of:

            * `"ee_native"` — first-party Earth Engine catalog entry
              published by the data provider directly (the default;
              the vast majority of assets).
            * `"republished"` — a copy of an external dataset that
              Google or a partner re-hosts in the EE catalog under
              an organisation-style asset id.
            * `"community"` — a user-uploaded asset whose path starts
              with `projects/...`. Often documented less rigorously
              than the first two.

            Replaces the older ambiguous `user_uploaded: bool` flag.
        bands: Band id → :class:`Band`.
    """

    model_config = ConfigDict(frozen=True)

    id: str
    title: str
    provider: str | None = None
    ee_type: Literal[
        "image", "image_collection", "table", "table_collection", "bigquery_table"
    ] = "image_collection"
    cadence: Cadence | None = None
    spatial_resolution: float | None = None
    extent: Extent
    default_reducer: Literal[
        "mean", "median", "mosaic", "min", "max", "mode", "sum"
    ] = "median"
    license: str | None = None
    terms_note: str | None = None
    source: Literal["ee_native", "republished", "community"] = "ee_native"
    bands: dict[str, Band] = Field(default_factory=dict)

    @property
    def is_raster(self) -> bool:
        """Whether the asset is a raster (image or image_collection).

        Returns:
            `True` if :attr:`ee_type` is `"image"` or `"image_collection"`,
            else `False`. Use this to gate "can the backend download
            this?" decisions — :attr:`is_image_collection` excludes
            static `Image` assets (e.g. SRTM) and is therefore misleading
            for that question.

        Examples:
            - Both `image` and `image_collection` are rasters:
                ```python
                >>> from earthlens.gee.catalog import Catalog
                >>> cat = Catalog()
                >>> cat.get_dataset("USGS/SRTMGL1_003").is_raster
                True
                >>> cat.get_dataset("LANDSAT/LC09/C02/T1_L2").is_raster
                True

                ```
        """
        return self.ee_type in {"image", "image_collection"}

    @property
    def is_tabular(self) -> bool:
        """Whether the asset is tabular (FeatureCollection / table / BigQuery).

        Returns:
            `True` if :attr:`ee_type` is `"table"`, `"table_collection"`,
            or `"bigquery_table"`, else `False`. Tabular assets are out
            of scope for the raster backend.
        """
        return self.ee_type in {"table", "table_collection", "bigquery_table"}

    @property
    def is_image_collection(self) -> bool:
        """Whether the asset is an `ImageCollection` (vs. a single `Image`).

        Returns:
            `True` if :attr:`ee_type` is `"image_collection"`, else `False`.

        .. deprecated::
            Prefer :attr:`is_raster` for the "can the backend download
            this?" question — it correctly includes static `image`
            assets like SRTM. This narrower property is kept for
            back-compat with older callers.

        Examples:
            - SRTM is a single static image; Landsat 9 is a collection:
                ```python
                >>> from earthlens.gee.catalog import Catalog
                >>> cat = Catalog()
                >>> cat.get_dataset("USGS/SRTMGL1_003").is_image_collection
                False
                >>> cat.get_dataset("LANDSAT/LC09/C02/T1_L2").is_image_collection
                True

                ```
        """
        return self.ee_type == "image_collection"

    def get_band(self, band_id: str) -> Band:
        """Return the :class:`Band` for `band_id`.

        Args:
            band_id: The Earth Engine band id.

        Returns:
            The matching :class:`Band`.

        Raises:
            ValueError: If `band_id` is not a band of this dataset; the
                message suggests the closest known band id.

        Examples:
            - Look up a Landsat band and read its centre wavelength:
                ```python
                >>> from earthlens.gee.catalog import Catalog
                >>> ds = Catalog().get_dataset("LANDSAT/LC09/C02/T1_L2")
                >>> ds.get_band("SR_B4").description
                'Band 4 (red) surface reflectance'
                >>> ds.get_band("SR_B4").wavelength
                0.655

                ```
            - A misspelt band raises with a suggestion:
                ```python
                >>> from earthlens.gee.catalog import Catalog
                >>> Catalog().get_dataset("USGS/SRTMGL1_003").get_band("elevashun")  # doctest: +ELLIPSIS
                Traceback (most recent call last):
                    ...
                ValueError: 'elevashun' is not a band of 'USGS/SRTMGL1_003'. ... Did you mean 'elevation'?

                ```
        """
        try:
            return self.bands[band_id]
        except KeyError:
            close = difflib.get_close_matches(band_id, self.bands, n=1)
            hint = f" Did you mean {close[0]!r}?" if close else ""
            raise ValueError(
                f"{band_id!r} is not a band of {self.id!r}. "
                f"Known bands: {sorted(self.bands)}.{hint}"
            ) from None

is_image_collection property #

Whether the asset is an ImageCollection (vs. a single Image).

Returns:

Type Description
bool

True if :attr:ee_type is "image_collection", else False.

.. deprecated:: Prefer :attr:is_raster for the "can the backend download this?" question — it correctly includes static image assets like SRTM. This narrower property is kept for back-compat with older callers.

Examples:

  • SRTM is a single static image; Landsat 9 is a collection: 
    >>> from earthlens.gee.catalog import Catalog
>>> cat = Catalog()
>>> cat.get_dataset("USGS/SRTMGL1_003").is_image_collection
False
>>> cat.get_dataset("LANDSAT/LC09/C02/T1_L2").is_image_collection
True

is_raster property #

Whether the asset is a raster (image or image_collection).

Returns:

Type Description
bool

True if :attr:ee_type is "image" or "image_collection",
bool

else False. Use this to gate "can the backend download
bool

this?" decisions — :attr:is_image_collection excludes
bool

static Image assets (e.g. SRTM) and is therefore misleading
bool

for that question.

Examples:

  • Both image and image_collection are rasters: 
    >>> from earthlens.gee.catalog import Catalog
>>> cat = Catalog()
>>> cat.get_dataset("USGS/SRTMGL1_003").is_raster
True
>>> cat.get_dataset("LANDSAT/LC09/C02/T1_L2").is_raster
True

is_tabular property #

Whether the asset is tabular (FeatureCollection / table / BigQuery).

Returns:

Type Description
bool

True if :attr:ee_type is "table", "table_collection",
bool

or "bigquery_table", else False. Tabular assets are out
bool

of scope for the raster backend.

get_band(band_id) #

Return the :class:Band for band_id.

Parameters:

Name Type Description Default
band_id str

The Earth Engine band id.

 required

Returns:

Type Description
Band

The matching :class:Band.

Raises:

Type Description
ValueError

If band_id is not a band of this dataset; the message suggests the closest known band id.

Examples:

  • Look up a Landsat band and read its centre wavelength: 
    >>> from earthlens.gee.catalog import Catalog
>>> ds = Catalog().get_dataset("LANDSAT/LC09/C02/T1_L2")
>>> ds.get_band("SR_B4").description
'Band 4 (red) surface reflectance'
>>> ds.get_band("SR_B4").wavelength
0.655
  • A misspelt band raises with a suggestion: 
    >>> from earthlens.gee.catalog import Catalog
>>> Catalog().get_dataset("USGS/SRTMGL1_003").get_band("elevashun")  # doctest: +ELLIPSIS
Traceback (most recent call last):
    ...
ValueError: 'elevashun' is not a band of 'USGS/SRTMGL1_003'. ... Did you mean 'elevation'?
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
def get_band(self, band_id: str) -> Band:
    """Return the :class:`Band` for `band_id`.

    Args:
        band_id: The Earth Engine band id.

    Returns:
        The matching :class:`Band`.

    Raises:
        ValueError: If `band_id` is not a band of this dataset; the
            message suggests the closest known band id.

    Examples:
        - Look up a Landsat band and read its centre wavelength:
            ```python
            >>> from earthlens.gee.catalog import Catalog
            >>> ds = Catalog().get_dataset("LANDSAT/LC09/C02/T1_L2")
            >>> ds.get_band("SR_B4").description
            'Band 4 (red) surface reflectance'
            >>> ds.get_band("SR_B4").wavelength
            0.655

            ```
        - A misspelt band raises with a suggestion:
            ```python
            >>> from earthlens.gee.catalog import Catalog
            >>> Catalog().get_dataset("USGS/SRTMGL1_003").get_band("elevashun")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'elevashun' is not a band of 'USGS/SRTMGL1_003'. ... Did you mean 'elevation'?

            ```
    """
    try:
        return self.bands[band_id]
    except KeyError:
        close = difflib.get_close_matches(band_id, self.bands, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        raise ValueError(
            f"{band_id!r} is not a band of {self.id!r}. "
            f"Known bands: {sorted(self.bands)}.{hint}"
        ) from None

Extent #

Bases: BaseModel

Spatial/temporal coverage of an Earth Engine dataset.

Attributes:

Name Type Description
start_date str

First available date, YYYY-MM-DD.
end_date str | None

Last available date (YYYY-MM-DD), or None for a continuously updated collection.
bbox tuple[float, float, float, float] | None

Spatial bounding box as [west, south, east, north] in EPSG:4326, or None for global coverage.

Examples:

  • A bounded, completed dataset (e.g. SRTM): 
    >>> e = Extent(start_date="2000-02-11", end_date="2000-02-22")
>>> e.start_date
'2000-02-11'
>>> e.end_date
'2000-02-22'
>>> e.bbox is None
True
  • A continuously updated, regionally bounded dataset (e.g. CHIRPS): 
    >>> e = Extent(start_date="1981-01-01", bbox=(-180.0, -50.0, 180.0, 50.0))
>>> e.end_date is None
True
>>> e.bbox
(-180.0, -50.0, 180.0, 50.0)
Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
class Extent(BaseModel):
    """Spatial/temporal coverage of an Earth Engine dataset.

    Attributes:
        start_date: First available date, `YYYY-MM-DD`.
        end_date: Last available date (`YYYY-MM-DD`), or `None` for a
            continuously updated collection.
        bbox: Spatial bounding box as `[west, south, east, north]` in
            EPSG:4326, or `None` for global coverage.

    Examples:
        - A bounded, completed dataset (e.g. SRTM):
            ```python
            >>> e = Extent(start_date="2000-02-11", end_date="2000-02-22")
            >>> e.start_date
            '2000-02-11'
            >>> e.end_date
            '2000-02-22'
            >>> e.bbox is None
            True

            ```
        - A continuously updated, regionally bounded dataset (e.g. CHIRPS):
            ```python
            >>> e = Extent(start_date="1981-01-01", bbox=(-180.0, -50.0, 180.0, 50.0))
            >>> e.end_date is None
            True
            >>> e.bbox
            (-180.0, -50.0, 180.0, 50.0)

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    start_date: str
    end_date: str | None = None
    bbox: tuple[float, float, float, float] | None = None

clear_catalog_cache() #

Empty the module-level catalog + providers caches.

Useful in tests that rewrite the catalog on disk and want to force a re-parse. Production callers do not need this — the cache keys include st_mtime_ns, so any real file mutation invalidates the entry on its own.

Source code in libs/providers/imagery/src/earthlens/gee/catalog.py 
233
234
235
236
237
238
239
240
241
242
def clear_catalog_cache() -> None:
    """Empty the module-level catalog + providers caches.

    Useful in tests that rewrite the catalog on disk and want to force a
    re-parse. Production callers do not need this — the cache keys
    include `st_mtime_ns`, so any real file mutation invalidates the
    entry on its own.
    """
    _CATALOG_CACHE.clear()
    _clear_providers_cache_base()

earthlens.gee.auth #

Service-account authentication for the Google Earth Engine backend.

Hosts :class:EarthEngineAuth, a thin wrapper over the earthengine-api (ee) authentication entry points. The Earth Engine backend authenticates with a Google Cloud service account plus a JSON key file (no interactive browser login on the machine that runs the download); :class:EarthEngineAuth.initialize performs the one-time ee.Initialize against a registered Cloud project.

The Cloud project the calls are scoped/billed to is mandatory on current earthengine-api releases: it is taken from the explicit project argument when given, else from the key file's project_id field. A project that has never been registered for Earth Engine, or that the service account lacks permission on, surfaces as an :class:AuthenticationError with a pointer at the registration / permissions docs rather than a raw ee exception.

See

AuthenticationError #

Bases: AuthenticationError

Raised when the Earth Engine connection cannot be established.

Wraps the underlying ee / Google credential errors with an actionable message — most commonly a missing or malformed service key, an unregistered Cloud project, or a service account that lacks an Earth Engine IAM role on the target project.

A subclass of the cross-backend :class:earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause; backward compatible with existing except earthlens.gee.AuthenticationError consumers.

Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class AuthenticationError(_BaseAuthenticationError):
    """Raised when the Earth Engine connection cannot be established.

    Wraps the underlying `ee` / Google credential errors with an
    actionable message — most commonly a missing or malformed service
    key, an unregistered Cloud project, or a service account that lacks
    an Earth Engine IAM role on the target project.

    A subclass of the cross-backend
    :class:`earthlens.base.AuthenticationError` so callers can catch
    every backend's auth failure with one `except` clause; backward
    compatible with existing `except earthlens.gee.AuthenticationError`
    consumers.
    """

EarthEngineAuth #

Bases: AbstractAuth[EarthEngineCredentials]

Authenticate and initialise a connection to Google Earth Engine.

Construct this with a service-account email and key (file path or raw JSON); construction performs the one-time ee.Initialize. The Cloud project is read from the project argument or, failing that, from the key file's project_id.

Conforms to the cross-backend :class:earthlens.base.AbstractAuth contract (C2): construction still authenticates eagerly for backward compatibility, but the underlying work lives in :meth:configure and is idempotent — the second call after :meth:is_authenticated returns True short-circuits.

Parameters:

Name Type Description Default
service_account str

The service-account email, e.g. my-sa@my-project.iam.gserviceaccount.com.

 required
service_key str

Path to the service-account JSON key file, or the JSON content as a string.

 required
project str | None

Cloud project id to scope the Earth Engine calls to. If omitted, the key file's project_id is used.

 None

Raises:

Type Description
AuthenticationError

If the credentials are missing/invalid, no project can be determined, or the project is not registered for Earth Engine / not accessible to the service account.

Examples:

  • Authenticate with a key file:

    >>> auth = EarthEngineAuth(  # doctest: +SKIP
...     "my-sa@my-project.iam.gserviceaccount.com",
...     "/path/to/key.json",
... )
Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
class EarthEngineAuth(AbstractAuth[EarthEngineCredentials]):
    """Authenticate and initialise a connection to Google Earth Engine.

    Construct this with a service-account email and key (file path or
    raw JSON); construction performs the one-time `ee.Initialize`. The
    Cloud project is read from the `project` argument or, failing that,
    from the key file's `project_id`.

    Conforms to the cross-backend
    :class:`earthlens.base.AbstractAuth` contract (C2): construction
    still authenticates eagerly for backward compatibility, but the
    underlying work lives in :meth:`configure` and is idempotent —
    the second call after :meth:`is_authenticated` returns `True`
    short-circuits.

    Args:
        service_account: The service-account email, e.g.
            `my-sa@my-project.iam.gserviceaccount.com`.
        service_key: Path to the service-account JSON key file, or the
            JSON content as a string.
        project: Cloud project id to scope the Earth Engine calls to.
            If omitted, the key file's `project_id` is used.

    Raises:
        AuthenticationError: If the credentials are missing/invalid, no
            project can be determined, or the project is not registered
            for Earth Engine / not accessible to the service account.

    Examples:
        - Authenticate with a key file:

            ```python
            >>> auth = EarthEngineAuth(  # doctest: +SKIP
            ...     "my-sa@my-project.iam.gserviceaccount.com",
            ...     "/path/to/key.json",
            ... )
            ```
    """

    def __init__(
        self,
        service_account: str,
        service_key: str,
        project: str | None = None,
    ):
        """Authenticate and call `ee.Initialize`; see the class docstring.

        Args:
            service_account: The service-account email.
            service_key: Path to the service-account JSON key file, or
                the JSON content as a string.
            project: Cloud project id; if omitted, read from the key
                file's `project_id`.

        Raises:
            AuthenticationError: As described on :class:`EarthEngineAuth`.
        """
        creds = EarthEngineCredentials(
            service_account=service_account,
            service_key=service_key,
            project=project,
        )
        super().__init__(creds)
        # Backward-compat surface: existing callers reach for
        # `auth.service_account` and `auth.project` as plain attrs.
        self.service_account = service_account
        self.project: str | None = None
        self.configure()

    def configure(self) -> None:
        """Authenticate against Earth Engine; idempotent.

        Calls `initialize` on first invocation and caches the
        resolved Cloud project id on `self.project`. Subsequent
        calls short-circuit when `is_authenticated` returns `True`,
        so it is safe to call repeatedly from long-lived workers.

        Raises:
            AuthenticationError: As described on `EarthEngineAuth`
                — missing/invalid key, unresolved project,
                unregistered Earth Engine project, or insufficient
                IAM permissions on the service account.

        Examples:
            - Calling `configure` twice does the network work once
              (the second call short-circuits via
              `is_authenticated`):

                ```python
                >>> auth = EarthEngineAuth(  # doctest: +SKIP
                ...     "my-sa@my-project.iam.gserviceaccount.com",
                ...     "/path/to/key.json",
                ... )
                >>> auth.is_authenticated()  # doctest: +SKIP
                True
                >>> auth.configure()  # no-op  # doctest: +SKIP

                ```
        """
        if self.is_authenticated():
            return
        self.project = self.initialize(
            self._creds.service_account,
            self._creds.service_key,
            self._creds.project,
        )

    def is_authenticated(self) -> bool:
        """`True` once `ee.Initialize` has succeeded for this instance.

        Cheap predicate — does not call into the `ee` library or
        the network. Returns `True` exactly when `self.project` is
        set to a non-empty string (the success signal from
        `initialize`).

        Returns:
            bool: `True` after a successful `configure()` /
                construction, `False` otherwise.

        Examples:
            - A fresh, configured instance is authenticated:
                ```python
                >>> auth = EarthEngineAuth(  # doctest: +SKIP
                ...     "my-sa@my-project.iam.gserviceaccount.com",
                ...     "/path/to/key.json",
                ... )
                >>> auth.is_authenticated()  # doctest: +SKIP
                True
                >>> auth.project  # doctest: +SKIP
                'my-project'

                ```
        """
        return bool(self.project)

    @staticmethod
    def initialize(
        service_account: str,
        service_key: str,
        project: str | None = None,
    ) -> str:
        """Authenticate the service account and call `ee.Initialize`.

        Args:
            service_account: The service-account email.
            service_key: Path to the service-account JSON key file, or
                the JSON content as a string.
            project: Cloud project id to scope the calls to. If omitted,
                the key file's `project_id` is used.

        Returns:
            The Cloud project id the connection was initialised with.

        Raises:
            AuthenticationError: If the key cannot be loaded, no project
                can be resolved, the project is not registered for Earth
                Engine, or the service account lacks permission on it.

        Examples:
            - Initialise from a key file (requires network + a registered project):
                ```python
                >>> EarthEngineAuth.initialize(  # doctest: +SKIP
                ...     "my-sa@my-project.iam.gserviceaccount.com",
                ...     "/path/to/key.json",
                ... )
                'my-project'

                ```
            - A key with no `project_id` and no explicit `project` fails fast:
                ```python
                >>> import json
                >>> bad_key = json.dumps({"type": "service_account"})
                >>> EarthEngineAuth.initialize("sa@x.iam", bad_key)  # doctest: +IGNORE_EXCEPTION_DETAIL
                Traceback (most recent call last):
                    ...
                earthlens.gee.auth.AuthenticationError: no Earth Engine Cloud project

                ```
        """
        key_dict = _load_key_dict(service_key)
        resolved_project = project or (key_dict or {}).get("project_id")
        if not resolved_project:
            raise AuthenticationError(
                "no Earth Engine Cloud project: pass project=, or use a "
                "service-account key file that includes a 'project_id' "
                f"field. See {_SERVICE_ACCOUNT_DOCS}."
            )

        try:
            credentials = ee.ServiceAccountCredentials(service_account, service_key)
        except ValueError:
            try:
                credentials = ee.ServiceAccountCredentials(
                    service_account, key_data=service_key
                )
            except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
                raise AuthenticationError(
                    "could not build service-account credentials from the "
                    f"supplied key (account={service_account!r}). Check that "
                    f"the key file/JSON is valid. See {_SERVICE_ACCOUNT_DOCS}."
                ) from exc

        try:
            ee.Initialize(credentials=credentials, project=resolved_project)
        except ee.EEException as exc:
            message = str(exc)
            if "not registered to use Earth Engine" in message:
                raise AuthenticationError(
                    f"Cloud project {resolved_project!r} is not registered "
                    f"to use Earth Engine. Register it at {_REGISTER_URL} "
                    "(pick the noncommercial track if eligible), then retry."
                ) from exc
            if (
                "does not have required permission" in message
                or "serviceUsageConsumer" in message
                or "PERMISSION_DENIED" in message
            ):
                raise AuthenticationError(
                    f"service account {service_account!r} cannot use project "
                    f"{resolved_project!r}: grant it the "
                    "'roles/serviceusage.serviceUsageConsumer' and "
                    "'roles/earthengine.viewer' IAM roles on that project."
                ) from exc
            raise AuthenticationError(
                f"Earth Engine initialisation failed for project "
                f"{resolved_project!r}: {message}"
            ) from exc
        except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
            raise AuthenticationError(
                f"Earth Engine initialisation failed for project "
                f"{resolved_project!r}: {exc}"
            ) from exc

        return resolved_project

    @staticmethod
    def encode_service_account(service_key_path: str) -> bytes:
        """Base64-encode a service-account JSON key file.

        Useful for shipping a key through an environment variable or CI
        secret without newlines.

        Args:
            service_key_path: Path to the service-account JSON key file.

        Returns:
            The base64-encoded JSON content as a byte string.

        Examples:
            - Encode a tiny key file and inspect the result:
                ```python
                >>> import json, os, tempfile
                >>> p = os.path.join(tempfile.mkdtemp(), "key.json")
                >>> _ = open(p, "w").write(json.dumps({"type": "service_account", "project_id": "demo"}))
                >>> blob = EarthEngineAuth.encode_service_account(p)
                >>> EarthEngineAuth.decode_service_account(blob)
                {'type': 'service_account', 'project_id': 'demo'}

                ```

        See Also:
            decode_service_account: The inverse operation.
        """
        content = json.loads(Path(service_key_path).read_text())
        return base64.b64encode(json.dumps(content).encode())

    @staticmethod
    def decode_service_account(service_key_bytes: bytes) -> dict[str, Any]:
        """Decode a base64-encoded service-account key back to a mapping.

        Inverse of :meth:`encode_service_account`.

        Args:
            service_key_bytes: The base64-encoded JSON content.

        Returns:
            The decoded service-account key as a dictionary.

        Examples:
            - Round-trip a key dict through encode then decode:
                ```python
                >>> import base64, json
                >>> blob = base64.b64encode(json.dumps({"client_email": "sa@p.iam", "project_id": "p"}).encode())
                >>> decoded = EarthEngineAuth.decode_service_account(blob)
                >>> decoded["client_email"]
                'sa@p.iam'
                >>> decoded["project_id"]
                'p'

                ```

        See Also:
            encode_service_account: The inverse operation.
        """
        return cast(
            "dict[str, Any]", json.loads(base64.b64decode(service_key_bytes).decode())
        )

__init__(service_account, service_key, project=None) #

Authenticate and call ee.Initialize; see the class docstring.

Parameters:

Name Type Description Default
service_account str

The service-account email.

 required
service_key str

Path to the service-account JSON key file, or the JSON content as a string.

 required
project str | None

Cloud project id; if omitted, read from the key file's project_id.

 None

Raises:

Type Description
AuthenticationError

As described on :class:EarthEngineAuth.
Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def __init__(
    self,
    service_account: str,
    service_key: str,
    project: str | None = None,
):
    """Authenticate and call `ee.Initialize`; see the class docstring.

    Args:
        service_account: The service-account email.
        service_key: Path to the service-account JSON key file, or
            the JSON content as a string.
        project: Cloud project id; if omitted, read from the key
            file's `project_id`.

    Raises:
        AuthenticationError: As described on :class:`EarthEngineAuth`.
    """
    creds = EarthEngineCredentials(
        service_account=service_account,
        service_key=service_key,
        project=project,
    )
    super().__init__(creds)
    # Backward-compat surface: existing callers reach for
    # `auth.service_account` and `auth.project` as plain attrs.
    self.service_account = service_account
    self.project: str | None = None
    self.configure()

configure() #

Authenticate against Earth Engine; idempotent.

Calls initialize on first invocation and caches the resolved Cloud project id on self.project. Subsequent calls short-circuit when is_authenticated returns True, so it is safe to call repeatedly from long-lived workers.

Raises:

Type Description
AuthenticationError

As described on EarthEngineAuth — missing/invalid key, unresolved project, unregistered Earth Engine project, or insufficient IAM permissions on the service account.

Examples:

  • Calling configure twice does the network work once (the second call short-circuits via is_authenticated):

    >>> auth = EarthEngineAuth(  # doctest: +SKIP
...     "my-sa@my-project.iam.gserviceaccount.com",
...     "/path/to/key.json",
... )
>>> auth.is_authenticated()  # doctest: +SKIP
True
>>> auth.configure()  # no-op  # doctest: +SKIP
Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def configure(self) -> None:
    """Authenticate against Earth Engine; idempotent.

    Calls `initialize` on first invocation and caches the
    resolved Cloud project id on `self.project`. Subsequent
    calls short-circuit when `is_authenticated` returns `True`,
    so it is safe to call repeatedly from long-lived workers.

    Raises:
        AuthenticationError: As described on `EarthEngineAuth`
            — missing/invalid key, unresolved project,
            unregistered Earth Engine project, or insufficient
            IAM permissions on the service account.

    Examples:
        - Calling `configure` twice does the network work once
          (the second call short-circuits via
          `is_authenticated`):

            ```python
            >>> auth = EarthEngineAuth(  # doctest: +SKIP
            ...     "my-sa@my-project.iam.gserviceaccount.com",
            ...     "/path/to/key.json",
            ... )
            >>> auth.is_authenticated()  # doctest: +SKIP
            True
            >>> auth.configure()  # no-op  # doctest: +SKIP

            ```
    """
    if self.is_authenticated():
        return
    self.project = self.initialize(
        self._creds.service_account,
        self._creds.service_key,
        self._creds.project,
    )

decode_service_account(service_key_bytes) staticmethod #

Decode a base64-encoded service-account key back to a mapping.

Inverse of :meth:encode_service_account.

Parameters:

Name Type Description Default
service_key_bytes bytes

The base64-encoded JSON content.

 required

Returns:

Type Description
dict[str, Any]

The decoded service-account key as a dictionary.

Examples:

  • Round-trip a key dict through encode then decode: 
    >>> import base64, json
>>> blob = base64.b64encode(json.dumps({"client_email": "sa@p.iam", "project_id": "p"}).encode())
>>> decoded = EarthEngineAuth.decode_service_account(blob)
>>> decoded["client_email"]
'sa@p.iam'
>>> decoded["project_id"]
'p'
See Also

encode_service_account: The inverse operation.

Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
@staticmethod
def decode_service_account(service_key_bytes: bytes) -> dict[str, Any]:
    """Decode a base64-encoded service-account key back to a mapping.

    Inverse of :meth:`encode_service_account`.

    Args:
        service_key_bytes: The base64-encoded JSON content.

    Returns:
        The decoded service-account key as a dictionary.

    Examples:
        - Round-trip a key dict through encode then decode:
            ```python
            >>> import base64, json
            >>> blob = base64.b64encode(json.dumps({"client_email": "sa@p.iam", "project_id": "p"}).encode())
            >>> decoded = EarthEngineAuth.decode_service_account(blob)
            >>> decoded["client_email"]
            'sa@p.iam'
            >>> decoded["project_id"]
            'p'

            ```

    See Also:
        encode_service_account: The inverse operation.
    """
    return cast(
        "dict[str, Any]", json.loads(base64.b64decode(service_key_bytes).decode())
    )

encode_service_account(service_key_path) staticmethod #

Base64-encode a service-account JSON key file.

Useful for shipping a key through an environment variable or CI secret without newlines.

Parameters:

Name Type Description Default
service_key_path str

Path to the service-account JSON key file.

 required

Returns:

Type Description
bytes

The base64-encoded JSON content as a byte string.

Examples:

  • Encode a tiny key file and inspect the result: 
    >>> import json, os, tempfile
>>> p = os.path.join(tempfile.mkdtemp(), "key.json")
>>> _ = open(p, "w").write(json.dumps({"type": "service_account", "project_id": "demo"}))
>>> blob = EarthEngineAuth.encode_service_account(p)
>>> EarthEngineAuth.decode_service_account(blob)
{'type': 'service_account', 'project_id': 'demo'}
See Also

decode_service_account: The inverse operation.

Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
@staticmethod
def encode_service_account(service_key_path: str) -> bytes:
    """Base64-encode a service-account JSON key file.

    Useful for shipping a key through an environment variable or CI
    secret without newlines.

    Args:
        service_key_path: Path to the service-account JSON key file.

    Returns:
        The base64-encoded JSON content as a byte string.

    Examples:
        - Encode a tiny key file and inspect the result:
            ```python
            >>> import json, os, tempfile
            >>> p = os.path.join(tempfile.mkdtemp(), "key.json")
            >>> _ = open(p, "w").write(json.dumps({"type": "service_account", "project_id": "demo"}))
            >>> blob = EarthEngineAuth.encode_service_account(p)
            >>> EarthEngineAuth.decode_service_account(blob)
            {'type': 'service_account', 'project_id': 'demo'}

            ```

    See Also:
        decode_service_account: The inverse operation.
    """
    content = json.loads(Path(service_key_path).read_text())
    return base64.b64encode(json.dumps(content).encode())

initialize(service_account, service_key, project=None) staticmethod #

Authenticate the service account and call ee.Initialize.

Parameters:

Name Type Description Default
service_account str

The service-account email.

 required
service_key str

Path to the service-account JSON key file, or the JSON content as a string.

 required
project str | None

Cloud project id to scope the calls to. If omitted, the key file's project_id is used.

 None

Returns:

Type Description
str

The Cloud project id the connection was initialised with.

Raises:

Type Description
AuthenticationError

If the key cannot be loaded, no project can be resolved, the project is not registered for Earth Engine, or the service account lacks permission on it.

Examples:

  • Initialise from a key file (requires network + a registered project): 
    >>> EarthEngineAuth.initialize(  # doctest: +SKIP
...     "my-sa@my-project.iam.gserviceaccount.com",
...     "/path/to/key.json",
... )
'my-project'
  • A key with no project_id and no explicit project fails fast: 
    >>> import json
>>> bad_key = json.dumps({"type": "service_account"})
>>> EarthEngineAuth.initialize("sa@x.iam", bad_key)  # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
    ...
earthlens.gee.auth.AuthenticationError: no Earth Engine Cloud project
Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
@staticmethod
def initialize(
    service_account: str,
    service_key: str,
    project: str | None = None,
) -> str:
    """Authenticate the service account and call `ee.Initialize`.

    Args:
        service_account: The service-account email.
        service_key: Path to the service-account JSON key file, or
            the JSON content as a string.
        project: Cloud project id to scope the calls to. If omitted,
            the key file's `project_id` is used.

    Returns:
        The Cloud project id the connection was initialised with.

    Raises:
        AuthenticationError: If the key cannot be loaded, no project
            can be resolved, the project is not registered for Earth
            Engine, or the service account lacks permission on it.

    Examples:
        - Initialise from a key file (requires network + a registered project):
            ```python
            >>> EarthEngineAuth.initialize(  # doctest: +SKIP
            ...     "my-sa@my-project.iam.gserviceaccount.com",
            ...     "/path/to/key.json",
            ... )
            'my-project'

            ```
        - A key with no `project_id` and no explicit `project` fails fast:
            ```python
            >>> import json
            >>> bad_key = json.dumps({"type": "service_account"})
            >>> EarthEngineAuth.initialize("sa@x.iam", bad_key)  # doctest: +IGNORE_EXCEPTION_DETAIL
            Traceback (most recent call last):
                ...
            earthlens.gee.auth.AuthenticationError: no Earth Engine Cloud project

            ```
    """
    key_dict = _load_key_dict(service_key)
    resolved_project = project or (key_dict or {}).get("project_id")
    if not resolved_project:
        raise AuthenticationError(
            "no Earth Engine Cloud project: pass project=, or use a "
            "service-account key file that includes a 'project_id' "
            f"field. See {_SERVICE_ACCOUNT_DOCS}."
        )

    try:
        credentials = ee.ServiceAccountCredentials(service_account, service_key)
    except ValueError:
        try:
            credentials = ee.ServiceAccountCredentials(
                service_account, key_data=service_key
            )
        except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
            raise AuthenticationError(
                "could not build service-account credentials from the "
                f"supplied key (account={service_account!r}). Check that "
                f"the key file/JSON is valid. See {_SERVICE_ACCOUNT_DOCS}."
            ) from exc

    try:
        ee.Initialize(credentials=credentials, project=resolved_project)
    except ee.EEException as exc:
        message = str(exc)
        if "not registered to use Earth Engine" in message:
            raise AuthenticationError(
                f"Cloud project {resolved_project!r} is not registered "
                f"to use Earth Engine. Register it at {_REGISTER_URL} "
                "(pick the noncommercial track if eligible), then retry."
            ) from exc
        if (
            "does not have required permission" in message
            or "serviceUsageConsumer" in message
            or "PERMISSION_DENIED" in message
        ):
            raise AuthenticationError(
                f"service account {service_account!r} cannot use project "
                f"{resolved_project!r}: grant it the "
                "'roles/serviceusage.serviceUsageConsumer' and "
                "'roles/earthengine.viewer' IAM roles on that project."
            ) from exc
        raise AuthenticationError(
            f"Earth Engine initialisation failed for project "
            f"{resolved_project!r}: {message}"
        ) from exc
    except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
        raise AuthenticationError(
            f"Earth Engine initialisation failed for project "
            f"{resolved_project!r}: {exc}"
        ) from exc

    return resolved_project

is_authenticated() #

True once ee.Initialize has succeeded for this instance.

Cheap predicate — does not call into the ee library or the network. Returns True exactly when self.project is set to a non-empty string (the success signal from initialize).

Returns:

Name Type Description
bool bool

True after a successful configure() / construction, False otherwise.

Examples:

  • A fresh, configured instance is authenticated: 
    >>> auth = EarthEngineAuth(  # doctest: +SKIP
...     "my-sa@my-project.iam.gserviceaccount.com",
...     "/path/to/key.json",
... )
>>> auth.is_authenticated()  # doctest: +SKIP
True
>>> auth.project  # doctest: +SKIP
'my-project'
Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def is_authenticated(self) -> bool:
    """`True` once `ee.Initialize` has succeeded for this instance.

    Cheap predicate — does not call into the `ee` library or
    the network. Returns `True` exactly when `self.project` is
    set to a non-empty string (the success signal from
    `initialize`).

    Returns:
        bool: `True` after a successful `configure()` /
            construction, `False` otherwise.

    Examples:
        - A fresh, configured instance is authenticated:
            ```python
            >>> auth = EarthEngineAuth(  # doctest: +SKIP
            ...     "my-sa@my-project.iam.gserviceaccount.com",
            ...     "/path/to/key.json",
            ... )
            >>> auth.is_authenticated()  # doctest: +SKIP
            True
            >>> auth.project  # doctest: +SKIP
            'my-project'

            ```
    """
    return bool(self.project)

EarthEngineCredentials #

Bases: BaseModel

Frozen value object holding the Earth Engine service-account creds.

Used internally by EarthEngineAuth to satisfy the earthlens.base.AbstractAuth generic-type bound. The public EarthEngineAuth constructor still accepts the three positional kwargs (service_account, service_key, project) for backward compatibility — the credentials object is built internally and stored on self._creds.

Attributes:

Name Type Description
service_account str

Service-account email, e.g. my-sa@my-project.iam.gserviceaccount.com.
service_key str

Path to the JSON key file, or the JSON content as a string. EarthEngineAuth.initialize distinguishes the two by leading character.
project str | None

Cloud project id; if None, falls back to the key file's project_id field at configure() time.

Examples:

  • Build a credentials object from a file path: 
    >>> from earthlens.gee.auth import EarthEngineCredentials
>>> creds = EarthEngineCredentials(
...     service_account="sa@my-project.iam.gserviceaccount.com",
...     service_key="/path/to/key.json",
...     project="my-project",
... )
>>> creds.service_account
'sa@my-project.iam.gserviceaccount.com'
>>> creds.project
'my-project'
  • project is optional — None defers resolution to configure(): 
    >>> from earthlens.gee.auth import EarthEngineCredentials
>>> creds = EarthEngineCredentials(
...     service_account="sa@p.iam",
...     service_key='{"type": "service_account"}',
... )
>>> creds.project is None
True
Source code in libs/providers/imagery/src/earthlens/gee/auth.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class EarthEngineCredentials(BaseModel):
    """Frozen value object holding the Earth Engine service-account creds.

    Used internally by `EarthEngineAuth` to satisfy the
    `earthlens.base.AbstractAuth` generic-type bound. The public
    `EarthEngineAuth` constructor still accepts the three positional
    kwargs (`service_account`, `service_key`, `project`) for
    backward compatibility — the credentials object is built
    internally and stored on `self._creds`.

    Attributes:
        service_account: Service-account email, e.g.
            `my-sa@my-project.iam.gserviceaccount.com`.
        service_key: Path to the JSON key file, or the JSON content
            as a string. `EarthEngineAuth.initialize` distinguishes
            the two by leading character.
        project: Cloud project id; if `None`, falls back to the key
            file's `project_id` field at `configure()` time.

    Examples:
        - Build a credentials object from a file path:
            ```python
            >>> from earthlens.gee.auth import EarthEngineCredentials
            >>> creds = EarthEngineCredentials(
            ...     service_account="sa@my-project.iam.gserviceaccount.com",
            ...     service_key="/path/to/key.json",
            ...     project="my-project",
            ... )
            >>> creds.service_account
            'sa@my-project.iam.gserviceaccount.com'
            >>> creds.project
            'my-project'

            ```
        - `project` is optional — `None` defers resolution to `configure()`:
            ```python
            >>> from earthlens.gee.auth import EarthEngineCredentials
            >>> creds = EarthEngineCredentials(
            ...     service_account="sa@p.iam",
            ...     service_key='{"type": "service_account"}',
            ... )
            >>> creds.project is None
            True

            ```
    """

    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

    service_account: str
    service_key: str
    project: str | None = None

earthlens.gee.features #

Shapely / GeoDataFrame → Earth Engine geometry converters.

The GEE backend lets a caller pass a GeoDataFrame as the clip region; this module turns Shapely geometries and GeoDataFrames into the ee.Geometry / ee.FeatureCollection objects Earth Engine expects. Only Polygon and Point geometries are supported (LineString is not yet implemented); a GeoDataFrame of MultiPolygons is exploded to one feature per polygon part, and the non-geometry columns become each feature's property dictionary.

create_feature(gdf, columns=None) #

Build an ee.FeatureCollection from a GeoDataFrame.

Each row becomes an ee.Feature whose geometry is the converted Shapely geometry (via :func:create_geometry) and whose properties are that row's non-geometry columns (optionally narrowed to columns). A row holding a MultiPolygon is exploded into one feature per constituent polygon.

Parameters:

Name Type Description Default
gdf GeoDataFrame

A GeoDataFrame whose geometry column holds Polygon / Point / MultiPolygon geometries.

 required
columns list[str] | None

If given, only these (non-geometry) columns become feature properties; otherwise all non-geometry columns are used. If the frame has no non-geometry columns (or columns is empty/None with a geometry-only frame), features are created without properties.

 None

Returns:

Type Description
FeatureCollection

An ee.FeatureCollection with one feature per (exploded) row.

Raises:

Type Description
ValueError

If any row's geometry cannot be converted via create_geometry (e.g. a LineString).
KeyError

If gdf has no geometry column, or if any of the requested columns is missing from gdf.
Note

Other exceptions raised by pandas / geopandas / earthengine-api propagate verbatim (with their original type and traceback).

Examples:

  • Build a collection from two polygons with a name property (needs the ee SDK initialised): 
    >>> import geopandas as gpd
>>> from shapely.geometry import Polygon
>>> from earthlens.gee.features import create_feature
>>> gdf = gpd.GeoDataFrame(
...     {"name": ["a", "b"],
...      "geometry": [Polygon([(0, 0), (1, 0), (1, 1)]),
...                   Polygon([(2, 2), (3, 2), (3, 3)])]},
...     crs="EPSG:4326",
... )
>>> fc = create_feature(gdf)  # doctest: +SKIP
  • Restricting which columns become properties: 
    >>> import geopandas as gpd
>>> from shapely.geometry import Polygon
>>> from earthlens.gee.features import create_feature
>>> gdf = gpd.GeoDataFrame(
...     {"name": ["a"], "value": [1],
...      "geometry": [Polygon([(0, 0), (1, 0), (1, 1)])]},
...     crs="EPSG:4326",
... )
>>> fc = create_feature(gdf, columns=["name"])  # doctest: +SKIP
See Also

create_geometry: Converts a single Shapely geometry; called per row.

Source code in libs/providers/imagery/src/earthlens/gee/features.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def create_feature(
    gdf: GeoDataFrame, columns: list[str] | None = None
) -> FeatureCollection:
    """Build an `ee.FeatureCollection` from a `GeoDataFrame`.

    Each row becomes an `ee.Feature` whose geometry is the converted
    Shapely geometry (via :func:`create_geometry`) and whose properties
    are that row's non-geometry columns (optionally narrowed to
    `columns`). A row holding a `MultiPolygon` is exploded into one
    feature per constituent polygon.

    Args:
        gdf: A `GeoDataFrame` whose `geometry` column holds `Polygon` /
            `Point` / `MultiPolygon` geometries.
        columns: If given, only these (non-geometry) columns become
            feature properties; otherwise all non-geometry columns are
            used. If the frame has no non-geometry columns (or `columns`
            is empty/`None` with a geometry-only frame), features are
            created without properties.

    Returns:
        An `ee.FeatureCollection` with one feature per (exploded) row.

    Raises:
        ValueError: If any row's geometry cannot be converted via
            `create_geometry` (e.g. a `LineString`).
        KeyError: If `gdf` has no `geometry` column, or if any of the
            requested `columns` is missing from `gdf`.

    Note:
        Other exceptions raised by `pandas` / `geopandas` /
        `earthengine-api` propagate verbatim (with their original type
        and traceback).

    Examples:
        - Build a collection from two polygons with a `name` property
          (needs the `ee` SDK initialised):
            ```python
            >>> import geopandas as gpd
            >>> from shapely.geometry import Polygon
            >>> from earthlens.gee.features import create_feature
            >>> gdf = gpd.GeoDataFrame(
            ...     {"name": ["a", "b"],
            ...      "geometry": [Polygon([(0, 0), (1, 0), (1, 1)]),
            ...                   Polygon([(2, 2), (3, 2), (3, 3)])]},
            ...     crs="EPSG:4326",
            ... )
            >>> fc = create_feature(gdf)  # doctest: +SKIP

            ```
        - Restricting which columns become properties:
            ```python
            >>> import geopandas as gpd
            >>> from shapely.geometry import Polygon
            >>> from earthlens.gee.features import create_feature
            >>> gdf = gpd.GeoDataFrame(
            ...     {"name": ["a"], "value": [1],
            ...      "geometry": [Polygon([(0, 0), (1, 0), (1, 1)])]},
            ...     crs="EPSG:4326",
            ... )
            >>> fc = create_feature(gdf, columns=["name"])  # doctest: +SKIP

            ```

    See Also:
        create_geometry: Converts a single Shapely geometry; called per row.
    """
    geotype = [i.geom_type for i in gdf["geometry"]]
    # if any is "MultiPolygon" explode the dataframe to single polygons
    # (`index_parts=True` makes the resulted index multi-index if a multi-polygon
    #  resulted in many different polygons)
    if "MultiPolygon" in geotype:
        gdf = gdf.explode(index_parts=True)

    # Convert per-row; on the first failure raise a `ValueError` naming
    # the offending row index so the user can spot it in a large frame —
    # never hand `None` or an opaque error down to `ee.Geometry`.
    ee_geom_list: list[Geometry] = []
    for i, geom in enumerate(gdf.geometry):
        try:
            ee_geom_list.append(create_geometry(geom))
        except NotImplementedError as exc:
            raise ValueError(
                f"create_feature cannot convert row {i} ({geom.geom_type}): {exc}"
            ) from exc
    records_df = pd.DataFrame(gdf.drop("geometry", axis=1))
    if columns:
        records_df = records_df[columns]
    records = records_df.to_dict("records")
    if not records:
        ee_feature_list = [ee.Feature(geom) for geom in ee_geom_list]
    else:
        ee_feature_list = [
            ee.Feature(geom, record) for geom, record in zip(ee_geom_list, records)
        ]
    return cast("FeatureCollection", ee.FeatureCollection(ee_feature_list))

create_geometry(shapely_geometry, epsg=4326) #

Convert a Shapely Polygon or Point to an ee.Geometry.

The geometry's GeoJSON coordinates are passed straight to ee.Geometry.Polygon / ee.Geometry.Point along with an "epsg:<epsg>" projection string.

Parameters:

Name Type Description Default
shapely_geometry Polygon | Point | LineString

A Shapely Polygon or Point.

 required
epsg int

EPSG code of the geometry's coordinates. Defaults to 4326 (WGS84 lon/lat).

 4326

Returns:

Type Description
Geometry

The corresponding ee.Geometry (Polygon or Point).

Raises:

Type Description
NotImplementedError

If shapely_geometry is any geometry type other than Polygon or Point (e.g. LineString, MultiPoint, GeometryCollection). Use :func:create_feature for MultiPolygon inputs — it explodes them into per-polygon rows before calling this.

Examples:

  • Convert a unit-square polygon (needs the ee SDK initialised): 
    >>> from shapely.geometry import Polygon
>>> from earthlens.gee.features import create_geometry
>>> square = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
>>> geom = create_geometry(square)  # doctest: +SKIP
  • The GeoJSON coordinates that get handed to Earth Engine: 
    >>> from shapely.geometry import Polygon
>>> Polygon([(0, 0), (1, 0), (1, 1), (0, 1)]).__geo_interface__["coordinates"]
(((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0), (0.0, 0.0)),)
  • A LineString is rejected: 
    >>> from shapely.geometry import LineString
>>> from earthlens.gee.features import create_geometry
>>> create_geometry(LineString([(0, 0), (1, 1)]))
Traceback (most recent call last):
    ...
NotImplementedError: LineString geometries are not yet supported by the GEE backend.
See Also

create_feature: Builds an ee.FeatureCollection from a GeoDataFrame, calling this for each row's geometry.

Source code in libs/providers/imagery/src/earthlens/gee/features.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def create_geometry(
    shapely_geometry: Polygon | Point | LineString,
    epsg: int = 4326,
) -> Geometry:
    """Convert a Shapely `Polygon` or `Point` to an `ee.Geometry`.

    The geometry's GeoJSON coordinates are passed straight to
    `ee.Geometry.Polygon` / `ee.Geometry.Point` along with an
    `"epsg:<epsg>"` projection string.

    Args:
        shapely_geometry: A Shapely `Polygon` or `Point`.
        epsg: EPSG code of the geometry's coordinates. Defaults to
            `4326` (WGS84 lon/lat).

    Returns:
        The corresponding `ee.Geometry` (`Polygon` or `Point`).

    Raises:
        NotImplementedError: If `shapely_geometry` is any geometry type
            other than `Polygon` or `Point` (e.g. `LineString`,
            `MultiPoint`, `GeometryCollection`). Use
            :func:`create_feature` for `MultiPolygon` inputs — it
            explodes them into per-polygon rows before calling this.

    Examples:
        - Convert a unit-square polygon (needs the `ee` SDK initialised):
            ```python
            >>> from shapely.geometry import Polygon
            >>> from earthlens.gee.features import create_geometry
            >>> square = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
            >>> geom = create_geometry(square)  # doctest: +SKIP

            ```
        - The GeoJSON coordinates that get handed to Earth Engine:
            ```python
            >>> from shapely.geometry import Polygon
            >>> Polygon([(0, 0), (1, 0), (1, 1), (0, 1)]).__geo_interface__["coordinates"]
            (((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0), (0.0, 0.0)),)

            ```
        - A `LineString` is rejected:
            ```python
            >>> from shapely.geometry import LineString
            >>> from earthlens.gee.features import create_geometry
            >>> create_geometry(LineString([(0, 0), (1, 1)]))
            Traceback (most recent call last):
                ...
            NotImplementedError: LineString geometries are not yet supported by the GEE backend.

            ```

    See Also:
        create_feature: Builds an `ee.FeatureCollection` from a
            `GeoDataFrame`, calling this for each row's geometry.
    """
    coords = shapely_geometry.__geo_interface__["coordinates"]
    geom_type = shapely_geometry.geom_type
    if geom_type == "Polygon":
        return cast("Geometry", ee.Geometry.Polygon(coords, f"epsg:{epsg}"))
    if geom_type == "Point":
        return cast("Geometry", ee.Geometry.Point(coords, f"epsg:{epsg}"))
    if geom_type == "LineString":
        raise NotImplementedError(
            "LineString geometries are not yet supported by the GEE backend."
        )
    raise NotImplementedError(
        f"{geom_type} geometries are not supported by the GEE backend; "
        "only Polygon and Point are accepted (MultiPolygon is auto-exploded "
        "by create_feature)."
    )