Initial import

src/geo_mask/pipeline.py

@@ -0,0 +1,73 @@
+from pathlib import Path
+import importlib.util
+import subprocess
+import sys
+import time
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+DOWNLOADER = "src.geo_mask.downloader"
+GENERATOR = "src.geo_mask.raster_tile_generator"
+
+STEP_DEPENDENCIES = {
+    "Geo Mask Downloader": ("requests",),
+    "Geo Mask Raster Generator": ("mercantile", "shapely", "numpy"),
+}
+
+
+def check_dependencies():
+    missing = []
+    for step_name, modules in STEP_DEPENDENCIES.items():
+        missing_modules = [module for module in modules if importlib.util.find_spec(module) is None]
+        if missing_modules:
+            missing.append(f"{step_name}: {', '.join(missing_modules)}")
+    return missing
+
+
+def run_step(name, module_name):
+    print("\n==========================")
+    print("Running:", name)
+    print("==========================\n")
+
+    start = time.time()
+    command = [sys.executable, "-u", "-m", module_name]
+    process = subprocess.Popen(
+        command,
+        cwd=str(SCRIPT_DIR.parent.parent),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,
+    )
+
+    assert process.stdout is not None
+    for line in process.stdout:
+        print(line, end="")
+
+    return_code = process.wait()
+    if return_code != 0:
+        raise RuntimeError(f"{name} failed with exit code {return_code}")
+
+    print("\nFinished:", name)
+    print("Time:", round(time.time() - start, 2), "seconds")
+
+
+def main():
+    print("Geo Mask Pipeline Starting...")
+    print("Date:", time.strftime("%Y-%m-%d %H:%M:%S"))
+
+    missing_dependencies = check_dependencies()
+    if missing_dependencies:
+        print("\n" + "!" * 50)
+        print("Pipeline Failed: missing runtime dependencies")
+        for item in missing_dependencies:
+            print("-", item)
+        print("!" * 50)
+        raise SystemExit(1)
+
+    run_step("Geo Mask Downloader", DOWNLOADER)
+    run_step("Geo Mask Raster Generator", GENERATOR)
+    print("\nGeo Mask Pipeline Completed Successfully!")
+
+
+if __name__ == "__main__":
+    main()