"""Synchronize official population and land-use time series for one approved scope. This is an explicit operator command, never an application-startup task. It coordinates the existing Statbel and Departement Omgeving import paths and keeps all persistence behind the canonical dataset upload API. """ from __future__ import annotations import argparse import json import os import subprocess import sys from pathlib import Path from typing import Any from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope DEFAULT_API_URL = "http://127.0.0.1:8000" DEFAULT_SCOPE_KEY = "kempen-transport-region" DEFAULT_SCOPE_OUTPUT_ROOT = Path("/app/storage/operator-data/geographic-scopes") DEFAULT_TIMESERIES_OUTPUT_ROOT = Path("/app/storage/operator-data/regional-timeseries") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Synchronize official GeoIntel time series for an approved scope.") parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY) parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL)) parser.add_argument("--population-years", default="2021,2022,2023,2024,2025") parser.add_argument("--landuse-years", default="2013,2016,2019,2022,2025") parser.add_argument("--historical-years", default="1778,1873,1969") parser.add_argument("--historical-themes", default="buildings,water,roads") parser.add_argument( "--scope-output-root", type=Path, default=Path(os.environ.get("GEOINTEL_SCOPE_OUTPUT_ROOT", DEFAULT_SCOPE_OUTPUT_ROOT)), ) parser.add_argument( "--output-root", type=Path, default=Path(os.environ.get("GEOINTEL_REGIONAL_TIMESERIES_OUTPUT_ROOT", DEFAULT_TIMESERIES_OUTPUT_ROOT)), ) parser.add_argument("--max-landuse-features", type=int, default=500000) parser.add_argument("--max-historical-features", type=int, default=500000) parser.add_argument("--request-timeout", type=int, default=300) parser.add_argument("--import-timeout", type=int, default=3600) parser.add_argument("--skip-population", action="store_true") parser.add_argument("--skip-landuse", action="store_true") parser.add_argument("--skip-historical", action="store_true") parser.add_argument("--fetch-only", action="store_true") parser.add_argument("--force", action="store_true") return parser.parse_args() def resolve_boundary(scope: GeographicScope, scope_output_root: Path) -> tuple[Path, Path, Path]: scope_dir = scope_output_root / scope.key manifest_path = scope_dir / f"{scope.key.replace('-', '_')}_scope_manifest.json" if not manifest_path.is_file(): raise RuntimeError( f"Official scope manifest is missing at {manifest_path}; run provision_geographic_scope.py --scope {scope.key} first" ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if ( manifest.get("status") != "complete" or manifest.get("scope_key") != scope.key or int(manifest.get("member_count") or 0) != len(scope.members) ): raise RuntimeError(f"Official scope manifest at {manifest_path} is incomplete or inconsistent") boundary_path = scope_dir / str(manifest.get("boundary_filename") or "") members_path = scope_dir / str(manifest.get("municipalities_filename") or "") if not boundary_path.is_file(): raise RuntimeError(f"Official scope boundary referenced by {manifest_path} is missing") if not members_path.is_file(): raise RuntimeError(f"Official scope member boundaries referenced by {manifest_path} are missing") return boundary_path, members_path, manifest_path def build_operator_commands( args: argparse.Namespace, scope: GeographicScope, boundary_path: Path, members_path: Path | None = None, ) -> list[tuple[str, list[str]]]: scripts_dir = Path(__file__).resolve().parent scope_output = args.output_root / scope.key common_flags = ["--fetch-only"] if args.fetch_only else [] if args.force: common_flags.append("--force") commands: list[tuple[str, list[str]]] = [] if not args.skip_population: commands.append( ( "population", [ sys.executable, str(scripts_dir / "provision_mol_population_history.py"), "--scope", scope.key, "--base-url", args.base_url, "--project-name", scope.project_name, "--area-name", scope.area_name, "--years", args.population_years, "--boundary-path", str(boundary_path), "--output-dir", str(scope_output / "population"), "--request-timeout", str(args.request_timeout), "--import-timeout", str(args.import_timeout), *common_flags, ], ) ) if not args.skip_landuse: partition_flags = ["--partition-boundaries-path", str(members_path)] if members_path else [] commands.append( ( "forest", [ sys.executable, str(scripts_dir / "provision_official_landuse_timeseries.py"), "--base-url", args.base_url, "--project-name", scope.project_name, "--area-name", scope.area_name, "--municipality-name", scope.display_name, "--nis-code", ",".join(scope.nis_codes), "--scope-key", scope.key, "--years", args.landuse_years, "--themes", "forest,water,built,transport", "--boundary-path", str(boundary_path), *partition_flags, "--output-dir", str(scope_output / "landuse"), "--request-timeout", str(args.request_timeout), "--import-timeout", str(args.import_timeout), "--max-features", str(args.max_landuse_features), *common_flags, ], ) ) if not args.skip_historical: commands.append( ( "historical_landuse", [ sys.executable, str(scripts_dir / "provision_regional_historical_landuse.py"), "--scope", scope.key, "--base-url", args.base_url, "--years", args.historical_years, "--themes", args.historical_themes, "--scope-output-root", str(args.scope_output_root), "--output-root", str(args.output_root / "historical"), "--request-timeout", str(args.request_timeout), "--import-timeout", str(args.import_timeout), "--max-total-features", str(args.max_historical_features), *common_flags, ], ) ) if not commands: raise ValueError("At least one regional time-series synchronization must remain enabled") return commands def run_operator(label: str, command: list[str]) -> dict[str, Any]: completed = subprocess.run(command, check=False, capture_output=True, text=True, encoding="utf-8") if completed.returncode != 0: detail = completed.stderr.strip() or completed.stdout.strip() or "operator returned no diagnostic output" raise RuntimeError(f"{label} synchronization failed with exit {completed.returncode}: {detail[-2000:]}") try: payload = json.loads(completed.stdout) except json.JSONDecodeError as exc: raise RuntimeError(f"{label} synchronization returned invalid JSON: {completed.stdout[-1000:]}") from exc if payload.get("status") != "ok": raise RuntimeError(f"{label} synchronization did not report success") return payload def main() -> int: args = parse_args() scope = GEOGRAPHIC_SCOPES[args.scope] try: if args.max_landuse_features <= 0 or args.max_historical_features <= 0: raise ValueError("land-use feature safety limits must be greater than zero") boundary_path, members_path, manifest_path = resolve_boundary(scope, args.scope_output_root) commands = build_operator_commands(args, scope, boundary_path, members_path) results = {label: run_operator(label, command) for label, command in commands} except (OSError, RuntimeError, ValueError, KeyError) as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) return 1 print( json.dumps( { "status": "ok", "mode": "fetch_only" if args.fetch_only else "synchronized", "scope": scope.key, "display_name": scope.display_name, "member_count": len(scope.members), "boundary_path": str(boundary_path), "municipality_boundaries_path": str(members_path), "scope_manifest_path": str(manifest_path), "results": results, }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": sys.exit(main())