From c59176872bbb78d23f0a1a74ce4c86584b334bec Mon Sep 17 00:00:00 2001 From: Stefan Amberger <1277330+snamber@users.noreply.github.com> Date: Fri, 11 Sep 2026 07:06:45 +0000 Subject: [PATCH 1/2] Update Lightning to 2.6.6 Amp-Thread-ID: https://ampcode.com/threads/T-01a08f47-7060-72ce-88f9-287d7ed8c442 Co-authored-by: Amp --- uv.lock | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/uv.lock b/uv.lock index 534926c..d34caf4 100644 --- a/uv.lock +++ b/uv.lock @@ -1527,7 +1527,7 @@ wheels = [ [[package]] name = "lightning" -version = "2.6.5" +version = "2.6.6" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "fsspec", extra = ["http"] }, @@ -1541,9 +1541,9 @@ dependencies = [ { name = "tqdm" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/c9/1d/83be8536bec71a0173e762a9a1fd92a24a5ad0d0f74c59550c3c4e6c103b/lightning-2.6.5.tar.gz", hash = "sha256:16a30310ed69afde3748491feb5d13508908effd70390d2bfc203dc0812a4b4a", size = 659201, upload-time = "2026-05-27T14:33:41.806Z" } +sdist = { url = "https://files.pythonhosted.org/packages/39/b8/fbf1469d917c3ed3b22e00ab3a852fb9a5f85b9dc7f7060eedc3449098e7/lightning-2.6.6.tar.gz", hash = "sha256:1c6329e241f8a85f29fedead89c1ee138b36caa08de60c260c6a7eaee81c8295", size = 659902, upload-time = "2026-09-10T09:41:03.009Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/e7/c5/fca7144236b6fa3279d0fb3172b32576c5ad8b84a63b9432ad6592d24847/lightning-2.6.5-py3-none-any.whl", hash = "sha256:3702fb7ef4ab51a8c3d4a140f4674514fe72973a9673dfa05e07a078e2767389", size = 848611, upload-time = "2026-05-27T14:33:39.714Z" }, + { url = "https://files.pythonhosted.org/packages/12/43/7da7fe42f254ce57777d1a86eece3f86d1b3329fc568d48e2bc16c8b7ebc/lightning-2.6.6-py3-none-any.whl", hash = "sha256:2bcbd6ee840071cd076c7dc9953d7ac1040df92f1c5fa76583aae644f037ab85", size = 849265, upload-time = "2026-09-10T09:41:00.882Z" }, ] [[package]] @@ -2813,7 +2813,7 @@ wheels = [ [[package]] name = "pytorch-lightning" -version = "2.6.5" +version = "2.6.6" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "fsspec", extra = ["http"] }, @@ -2826,9 +2826,9 @@ dependencies = [ { name = "tqdm" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/52/2c/8e73a3929b4c4bd600cafd38a97aaf7242a8cf518fb9f33d27c274ec898f/pytorch_lightning-2.6.5.tar.gz", hash = "sha256:1c32cefa76a1a9c4c5250338272d961d1e48b180e68396849efe128538ddb28e", size = 661673, upload-time = "2026-05-27T14:33:41.961Z" } +sdist = { url = "https://files.pythonhosted.org/packages/97/e4/5025cb42311b4f8cc6d1b12cd81399ca433159f8e3831ab229c5a6c5177e/pytorch_lightning-2.6.6.tar.gz", hash = "sha256:52c52b20522dd3d3feb3bb3aae2dbe3f291e8e836d4111a74cfd72e9df018834", size = 662507, upload-time = "2026-09-10T09:41:16.684Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/8b/4d/5740c27110b83634d8491c3b5facf0111b3e554c3164f4fb953be9bddaf6/pytorch_lightning-2.6.5-py3-none-any.whl", hash = "sha256:62d9c8549b2278fedc3364f0a5607a56c6063d18635008f8cf3fae8d802b0d76", size = 852407, upload-time = "2026-05-27T14:33:39.856Z" }, + { url = "https://files.pythonhosted.org/packages/4f/cf/ba8e4cc1b982ff6dc6f9fdef0e4074605438b82640f6270888a10e7e42c3/pytorch_lightning-2.6.6-py3-none-any.whl", hash = "sha256:71f95c7b22f25c4f91cc659e1734bcdc6c69c8b66543f2ed2d78dbe29ce2f167", size = 853045, upload-time = "2026-09-10T09:41:14.923Z" }, ] [[package]] From 8bd42145c8a69943f13b47b938494290c57befc7 Mon Sep 17 00:00:00 2001 From: Stefan Amberger <1277330+snamber@users.noreply.github.com> Date: Fri, 11 Sep 2026 09:30:01 +0000 Subject: [PATCH 2/2] Fix workflow static checks Amp-Thread-ID: https://ampcode.com/threads/T-01a08f47-7060-72ce-88f9-287d7ed8c442 Co-authored-by: Amp --- README.md | 6 ++++++ datacenters/workflow.py | 43 +++++++++++++++++++++++++++-------------- 2 files changed, 34 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index c25a9b3..c117b5c 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,12 @@ Install dependencies: uv sync ``` +For local development without Google Cloud credentials, use the local workflow cache: + +```bash +export WORKFLOW_CACHE_BUCKET="" +``` + Publish and deploy the workflow release: ```bash diff --git a/datacenters/workflow.py b/datacenters/workflow.py index 266d2f0..8ebf615 100644 --- a/datacenters/workflow.py +++ b/datacenters/workflow.py @@ -7,10 +7,10 @@ import random import tempfile from dataclasses import asdict, dataclass -from datetime import date, datetime, timedelta +from datetime import UTC, date, datetime, timedelta from functools import lru_cache from pathlib import Path -from typing import Any +from typing import Any, cast import niquests import numpy as np @@ -206,7 +206,10 @@ def _first_column(columns: list[str], candidates: list[str]) -> str: def _download_sites_csv(csv_url: str) -> pd.DataFrame: response = niquests.get(csv_url, timeout=60) response.raise_for_status() - return pd.read_csv(io.BytesIO(response.content)) + content = response.content + if content is None: + raise ValueError(f"CSV response from {csv_url!r} had no content") + return pd.read_csv(io.BytesIO(content)) def _merge_sites( # noqa: C901 @@ -306,11 +309,15 @@ def _dataset_candidates( # noqa: PLR0913 ) -> list[dict[str, Any]]: start, end = _date_window(target_date, window_days) area = _site_crop_polygon(latitude, longitude, crop_size_m) - data = DatasetClient().dataset("open_data.copernicus.sentinel2_msi").query( - collections=SENTINEL2_COLLECTIONS, - temporal_extent=(start, end), - spatial_extent=area, - show_progress=False, + data = ( + DatasetClient() + .dataset("open_data.copernicus.sentinel2_msi") + .query( + collections=SENTINEL2_COLLECTIONS, + temporal_extent=(start, end), + spatial_extent=area, + show_progress=False, + ) ) if data.sizes.get("time", 0) == 0: return [] @@ -459,7 +466,7 @@ def _save_npz(arrays: dict[str, np.ndarray], metadata: dict[str, Any]) -> bytes: buffer = io.BytesIO() np.savez( buffer, - **{band_name: arrays[band_name] for band_name in ALL_BAND_NAMES}, + **{band_name: arrays[band_name] for band_name in ALL_BAND_NAMES}, # ty: ignore[invalid-argument-type] SCL=arrays["SCL"], metadata=json.dumps(metadata), ) @@ -708,13 +715,14 @@ def _clay_model() -> Any: checkpoint_path = _ensure_clay_checkpoint() original_create_model = timm.create_model + timm_module = cast(Any, timm) def create_model_without_pretrained_weights(*args: Any, **kwargs: Any) -> Any: kwargs["pretrained"] = False return original_create_model(*args, **kwargs) try: - timm.create_model = create_model_without_pretrained_weights + timm_module.create_model = create_model_without_pretrained_weights model = ClayMAEModule.load_from_checkpoint( checkpoint_path, map_location="cpu", @@ -739,7 +747,7 @@ def create_model_without_pretrained_weights(*args: Any, **kwargs: Any) -> Any: shuffle=False, ) finally: - timm.create_model = original_create_model + timm_module.create_model = original_create_model torch.set_num_threads(max(1, min(4, os.cpu_count() or 1))) return model.to(torch.device("cpu")).eval() @@ -752,7 +760,7 @@ def _normalize_latlon(latitude: float, longitude: float) -> tuple[tuple[float, f def _normalize_timestamp(value: str | None) -> tuple[tuple[float, float], tuple[float, float]]: - timestamp = datetime.fromisoformat(value) if value else datetime.utcnow() + timestamp = datetime.fromisoformat(value) if value else datetime.now(UTC) week = timestamp.isocalendar().week * 2 * np.pi / 52 hour = timestamp.hour * 2 * np.pi / 24 return (math.sin(week), math.cos(week)), (math.sin(hour), math.cos(hour)) @@ -845,7 +853,9 @@ def _clay_change_metrics( after_metadata: dict[str, Any], ) -> dict[str, float]: before, after, common_shape = _align_common_shape(before, after) - before_patches = _clay_patch_embeddings(before, site.latitude, site.longitude, before_metadata.get("acquisition_time")) + before_patches = _clay_patch_embeddings( + before, site.latitude, site.longitude, before_metadata.get("acquisition_time") + ) after_patches = _clay_patch_embeddings(after, site.latitude, site.longitude, after_metadata.get("acquisition_time")) patch_distances = _cosine_distances(before_patches, after_patches) grid_size = int(math.sqrt(patch_distances.size)) @@ -1076,6 +1086,7 @@ def identifier() -> tuple[str, str]: return "tilebox.com/datacenters/RankDataCenterBuildout", "v1.14" def execute(self, context: ExecutionContext): # noqa: ANN201 + context = cast(Any, context) context.current_task.display = "RankDataCenterBuildout" status_filter = self.status_filter if self.status_filter is not None else DEFAULT_STATUS_FILTER sites = _merge_sites(self.csv_url, self.max_sites, self.random_seed, status_filter) @@ -1138,6 +1149,7 @@ def identifier() -> tuple[str, str]: return "tilebox.com/datacenters/SelectAndCacheScene", "v1.14" def execute(self, context: ExecutionContext): # noqa: ANN201, PLR0915 + context = cast(Any, context) site = _sites_by_id(context.job_cache["sites.json"])[self.site_id] context.current_task.display = f"Select {self.label} {site.site_id}" metadata_key = f"scenes/{site.site_id}/{self.label}/metadata.json" @@ -1342,6 +1354,7 @@ def identifier() -> tuple[str, str]: return "tilebox.com/datacenters/ComputeSiteChange", "v1.14" def execute(self, context: ExecutionContext): # noqa: ANN201 + context = cast(Any, context) site = _sites_by_id(context.job_cache["sites.json"])[self.site_id] context.current_task.display = f"Compute {site.site_id}" before_metadata = _json_loads(context.job_cache[f"scenes/{site.site_id}/before/metadata.json"]) @@ -1392,12 +1405,12 @@ def execute(self, context: ExecutionContext): # noqa: ANN201 class WriteRankingOutput(Task): - @staticmethod def identifier() -> tuple[str, str]: return "tilebox.com/datacenters/WriteRankingOutput", "v1.14" def execute(self, context: ExecutionContext): # noqa: ANN201 + context = cast(Any, context) site_ids = list(_sites_by_id(context.job_cache["sites.json"])) context.current_task.display = f"WriteRankingOutput(n={len(site_ids)})" results = [_json_loads(context.job_cache[f"results/{site_id}.json"]) for site_id in site_ids] @@ -1405,7 +1418,7 @@ def execute(self, context: ExecutionContext): # noqa: ANN201 for rank, item in enumerate(results, start=1): item["rank"] = rank output = { - "generated_at": datetime.utcnow().isoformat(timespec="seconds") + "Z", + "generated_at": datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"), "ranking": results, } context.job_cache["outputs/ranking.json"] = _json_dumps(output)