Run RasterFlow as a Job

#!/usr/bin/env python3
"""
RasterFlow FTW Job Script - Extract field boundaries using Fields of the World model

This script runs the Fields of the World (FTW) model on Haskell County, Kansas
and vectorizes the results using Wherobots RasterFlow.
"""

import os
import wkls
import geopandas as gpd
from datetime import datetime
from rasterflow_remote import RasterflowClient
from rasterflow_remote.data_models import ModelRecipes, VectorizeMethodEnum

def main():
    print("Starting RasterFlow FTW Job...")

    # Initialize RasterFlow client
    rf_client = RasterflowClient(cache=False)

    # Generate AOI for Haskell County, Kansas
    print("Generating AOI for Haskell County, Kansas...")
    gdf = gpd.read_file(wkls['us']['ks']['Haskell County'].geojson())
    aoi_path = os.getenv("USER_S3_PATH") + "haskell_job.parquet"
    gdf.to_parquet(aoi_path)
    print(f"AOI saved to: {aoi_path}")

    # Run FTW model
    print("Running Fields of the World model...")
    model_outputs = rf_client.build_and_predict_mosaic_recipe(
        aoi=aoi_path,
        start=datetime(2023, 1, 1),
        end=datetime(2024, 1, 1),
        crs_epsg=3857,
        model_recipe=ModelRecipes.FTW,
    )
    print(f"Model outputs saved to: {model_outputs}")

    print("RasterFlow FTW Job completed successfully!")
    print(f"Results:")
    print(f"  Model outputs: {model_outputs}")

    return {
         "model_outputs": model_outputs,
    }

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Submit a RasterFlow workflow as a Job Run using the Wherobots Runs REST API.

This script submits a job that runs the Fields of the World (FTW) model
on Haskell County, Kansas.

This is a subset of the code included in the Solution Notebook for the Fields of the World model.
See https://cloud.wherobots.com/model-hub/fields-of-the-world for more details.
"""

import os
import requests
from datetime import datetime


def submit_rasterflow_job(
    api_key: str,  # Loaded from WHEROBOTS_API_KEY env var in main()
    script_s3_path: str,
    region: str = "aws-us-west-2",
    runtime: str = "micro",
    job_name: str = "rasterflow-ftw-job",
    timeout_seconds: int = 3600
) -> dict:
    """
    Submit a RasterFlow workflow using the Wherobots Runs REST API.

    Args:
        api_key: Wherobots API key
        script_s3_path: S3 path to the Python script file
        region: Compute region (default: aws-us-west-2)
        runtime: Wherobots runtime size (default: micro). RasterFlow manages its own compute, so use micro unless also running WherobotsDB workloads.
        job_name: Name for the job run
        timeout_seconds: Job timeout in seconds (default: 3600)

    Returns:
        dict: API response containing job run details
    """

    # API endpoint
    url = f"https://api.cloud.wherobots.com/runs?region={region}"

    # Headers
    headers = {
        "accept": "application/json",
        "X-API-Key": api_key,
        "Content-Type": "application/json"
    }

    # Job payload
    payload = {
        "runtime": runtime,
        "name": job_name,
        "runPython": {
            "uri": script_s3_path
        },
        "timeoutSeconds": timeout_seconds,
        "environment": {
            "dependencies": [
            ]
        }
    }

    # Make the request
    print(f"Submitting job to: {url}")
    print(f"Job name: {job_name}")
    print(f"Runtime: {runtime}")
    print(f"Script path: {script_s3_path}")

    response = requests.post(url, headers=headers, json=payload)

    # Handle response
    if response.status_code == 200 or response.status_code == 201:
        result = response.json()
        print(f"Job submitted successfully!")
        print(f"Job ID: {result.get('id')}")
        print(f"Status: {result.get('status')}")
        return result
    else:
        print(f"Job submission failed!")
        print(f"Status Code: {response.status_code}")
        print(f"Response: {response.text}")
        response.raise_for_status()


def main():
    """Main function to run the job submission script."""

    # Configuration - Update these values
    API_KEY = os.getenv("WHEROBOTS_API_KEY")
    USER_S3_PATH = os.getenv("USER_S3_PATH")

    # Validate required environment variables
    if API_KEY is None or API_KEY == "":
        raise ValueError(
            "WHEROBOTS_API_KEY environment variable is required. "
            "Get your API key from https://cloud.wherobots.com/settings/api-keys"
        )

    if USER_S3_PATH is None or USER_S3_PATH == "":
        raise ValueError(
            "USER_S3_PATH environment variable is required. "
            "This should be your Wherobots managed storage path."
        )

    SCRIPT_S3_PATH = USER_S3_PATH + "rasterflow_ftw_job.py"

    print("=== RasterFlow FTW Job Submission ===")
    print("This script will submit a job to run the Fields of the World model")
    print("on Haskell County, Kansas using Wherobots RasterFlow.\n")

    print(f"Make sure you have uploaded rasterflow_ftw_job.py to: {SCRIPT_S3_PATH}")
    print("The job script should already be available in this directory.\n")

    # Submit the job
    result = submit_rasterflow_job(
        api_key=API_KEY,
        script_s3_path=SCRIPT_S3_PATH,
        job_name=f"rasterflow-ftw-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
    )

    print(f"\nJob submission completed!")
    print(f"Monitor your job at: https://cloud.wherobots.com/job-runs")


if __name__ == "__main__":
    exit(main())