Era5 pipeline

.gitignore

@@ -172,7 +172,7 @@ experiments/tensorboard
 OLD_*/
 logs/
 scripts/modis/logs/
-scripts/run.sh
 scripts/run_era5.sh
 
 era5_predictions.csv
+era5_pipeline/plot/tmp/

scripts/modis/update_igbp_from_c1.py → era5_pipeline/assign_igbp_from_modis.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Update coord_data.igbp from a MODIS MCD12C1 GeoTIFF."""
+"""Assign coord_data.igbp from a MODIS MCD12C1 GeoTIFF."""
 
 from __future__ import annotations
 
@@ -18,13 +18,13 @@
 from tqdm.auto import tqdm
 
 
-REPO_ROOT = Path(__file__).resolve().parents[2]
+REPO_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO_ROOT))
 
 from ecoperceiver.constants import IGBP_ACRONYMS_MODIS
 
 DEFAULT_DB_PATH = Path("/home/l/luislara/links/projects/aip-pal/luislara/ep/data/era5.db")
-DEFAULT_C1_PATH = REPO_ROOT / "experiments" / "data" / "raw_modis" / "201801011200C1.tiff"
+DEFAULT_MODIS_PATH = REPO_ROOT / "experiments" / "data" / "raw_modis" / "201801011200C1.tiff"
 DEFAULT_TABLE = "coord_data"
 DEFAULT_BATCH_SIZE = 10_000
 SQLITE_TIMEOUT_SECONDS = 60.0
@@ -45,15 +45,15 @@ def width(self) -> int:
 
 
 @dataclass(frozen=True)
-class PlannedUpdate:
+class PlannedAssignment:
     rowid: int
     new_igbp: str
 
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         description=(
-            "Sample MODIS MCD12C1 Majority_Land_Cover_Type_1 values and update "
+            "Sample MODIS MCD12C1 Majority_Land_Cover_Type_1 values and assign "
             "coord_data.igbp labels in an ERA5 SQLite database. The default is "
             "a dry run; pass --write to modify the database."
         )
@@ -62,34 +62,34 @@ def parse_args() -> argparse.Namespace:
         "--db-path",
         type=Path,
         default=DEFAULT_DB_PATH,
-        help=f"SQLite database to update. Default: {DEFAULT_DB_PATH}",
+        help=f"SQLite database to modify. Default: {DEFAULT_DB_PATH}",
     )
     parser.add_argument(
-        "--c1-path",
+        "--modis-path",
         type=Path,
-        default=DEFAULT_C1_PATH,
-        help=f"MODIS MCD12C1 GeoTIFF. Default: {DEFAULT_C1_PATH}",
+        default=DEFAULT_MODIS_PATH,
+        help=f"MODIS MCD12C1 GeoTIFF. Default: {DEFAULT_MODIS_PATH}",
     )
     parser.add_argument(
         "--table",
         default=DEFAULT_TABLE,
-        help=f"Coordinate table to update. Default: {DEFAULT_TABLE}",
+        help=f"Coordinate table to assign. Default: {DEFAULT_TABLE}",
     )
     parser.add_argument(
         "--only-null",
         action="store_true",
-        help="Only update rows where coord_data.igbp is NULL.",
+        help="Only assign rows where coord_data.igbp is NULL.",
     )
     parser.add_argument(
         "--write",
         action="store_true",
-        help="Apply the planned updates. Without this flag, only a dry run is printed.",
+        help="Apply the planned assignments. Without this flag, only a dry run is printed.",
     )
     parser.add_argument(
         "--batch-size",
         type=int,
         default=DEFAULT_BATCH_SIZE,
-        help=f"SQLite update batch size. Default: {DEFAULT_BATCH_SIZE}",
+        help=f"SQLite assignment batch size. Default: {DEFAULT_BATCH_SIZE}",
     )
     return parser.parse_args()
 
@@ -229,15 +229,15 @@ def iter_coord_row_batches(
         yield rows
 
 
-def plan_updates(
+def plan_assignments(
     conn: sqlite3.Connection,
     table: str,
     grid: RasterGrid,
     *,
     only_null: bool,
     batch_size: int,
-) -> tuple[list[PlannedUpdate], Counter[str], Counter[str], Counter[tuple[str, str]], int, int]:
-    updates: list[PlannedUpdate] = []
+) -> tuple[list[PlannedAssignment], Counter[str], Counter[str], Counter[tuple[str, str]], int, int]:
+    assignments: list[PlannedAssignment] = []
     old_counts: Counter[str] = Counter()
     new_counts: Counter[str] = Counter()
     transitions: Counter[tuple[str, str]] = Counter()
@@ -246,7 +246,7 @@ def plan_updates(
 
     with tqdm(
         total=row_count,
-        desc="update_igbp plan",
+        desc="assign_igbp plan",
         unit="coord",
         dynamic_ncols=True,
     ) as progress:
@@ -287,16 +287,16 @@ def plan_updates(
                     transitions[(old_label, new_igbp)] += 1
 
                     if old_igbp != new_igbp:
-                        updates.append(
-                            PlannedUpdate(
+                        assignments.append(
+                            PlannedAssignment(
                                 rowid=int(rowid),
                                 new_igbp=new_igbp,
                             )
                         )
 
             progress.update(len(batch))
 
-    return updates, old_counts, new_counts, transitions, skipped_missing_coords, row_count
+    return assignments, old_counts, new_counts, transitions, skipped_missing_coords, row_count
 
 
 def print_counts(title: str, counts: Counter[str]) -> None:
@@ -308,18 +308,18 @@ def print_counts(title: str, counts: Counter[str]) -> None:
 def print_summary(
     *,
     db_path: Path,
-    c1_path: Path,
+    modis_path: Path,
     table: str,
     only_null: bool,
     row_count: int,
-    updates: list[PlannedUpdate],
+    assignments: list[PlannedAssignment],
     old_counts: Counter[str],
     new_counts: Counter[str],
     transitions: Counter[tuple[str, str]],
     skipped_missing_coords: int,
 ) -> None:
     print(f"DB: {db_path}")
-    print(f"C1: {c1_path}")
+    print(f"MODIS: {modis_path}")
     print(f"Table: {table}")
     print("Sampling: nearest coord_data lat/lon point")
     print(f"Rows scanned: {row_count:,}")
@@ -328,7 +328,7 @@ def print_summary(
     else:
         print("Scope: all coord_data rows")
     print(f"Rows with missing lat/lon skipped: {skipped_missing_coords:,}")
-    print(f"Rows that would change: {len(updates):,}")
+    print(f"Rows that would be assigned/changed: {len(assignments):,}")
     print()
     print_counts("Current labels in scan scope:", old_counts)
     print()
@@ -340,50 +340,50 @@ def print_summary(
             print(f"  {old_label} -> {new_label}: {count:,}")
 
 
-def apply_updates(
+def apply_assignments(
     conn: sqlite3.Connection,
     table: str,
-    updates: list[PlannedUpdate],
+    assignments: list[PlannedAssignment],
     *,
     batch_size: int,
 ) -> None:
     if batch_size <= 0:
         raise SystemExit("--batch-size must be positive.")
 
     table_sql = quote_identifier(table)
-    update_sql = f"UPDATE {table_sql} SET igbp = ? WHERE rowid = ?"
+    assignment_sql = f"UPDATE {table_sql} SET igbp = ? WHERE rowid = ?"
     started_at = time.monotonic()
 
     with conn:
         with tqdm(
-            total=len(updates),
-            desc="update_igbp write",
+            total=len(assignments),
+            desc="assign_igbp write",
             unit="row",
             dynamic_ncols=True,
         ) as progress:
-            for offset in range(0, len(updates), batch_size):
-                batch = updates[offset : offset + batch_size]
+            for offset in range(0, len(assignments), batch_size):
+                batch = assignments[offset : offset + batch_size]
                 conn.executemany(
-                    update_sql,
-                    [(update.new_igbp, update.rowid) for update in batch],
+                    assignment_sql,
+                    [(assignment.new_igbp, assignment.rowid) for assignment in batch],
                 )
                 progress.update(len(batch))
 
-    print(f"Applied updates in {format_duration(time.monotonic() - started_at)}")
+    print(f"Applied assignments in {format_duration(time.monotonic() - started_at)}")
 
 
 def main() -> int:
     args = parse_args()
     db_path = resolve_path(args.db_path)
-    c1_path = resolve_path(args.c1_path)
+    modis_path = resolve_path(args.modis_path)
 
     if not db_path.exists():
         raise SystemExit(f"Database does not exist: {db_path}")
 
-    grid = load_c1_raster(c1_path)
+    grid = load_c1_raster(modis_path)
     with connect_database(db_path, readonly=not args.write) as conn:
         ensure_coord_table(conn, args.table)
-        updates, old_counts, new_counts, transitions, skipped_missing_coords, row_count = plan_updates(
+        assignments, old_counts, new_counts, transitions, skipped_missing_coords, row_count = plan_assignments(
             conn,
             args.table,
             grid,
@@ -392,11 +392,11 @@ def main() -> int:
         )
         print_summary(
             db_path=db_path,
-            c1_path=c1_path,
+            modis_path=modis_path,
             table=args.table,
             only_null=args.only_null,
             row_count=row_count,
-            updates=updates,
+            assignments=assignments,
             old_counts=old_counts,
             new_counts=new_counts,
             transitions=transitions,
@@ -405,14 +405,14 @@ def main() -> int:
 
         if not args.write:
             print()
-            print("Dry run only; rerun with --write to update the database.")
+            print("Dry run only; rerun with --write to assign IGBP in the database.")
             return 0
 
-        if not updates:
-            print("No updates needed.")
+        if not assignments:
+            print("No assignments needed.")
             return 0
 
-        apply_updates(conn, args.table, updates, batch_size=args.batch_size)
+        apply_assignments(conn, args.table, assignments, batch_size=args.batch_size)
 
     return 0