Source code for satorbis_kit.vector_operation.wherobots_geojson

"""
Wherobots GeoJSON Operations

Functions for GeoJSON to GeoParquet conversion on Wherobots Cloud.
"""

from datetime import datetime
from typing import List, Optional

from .wherobots_config import (
    DEFAULT_API_KEY,
    DEFAULT_GEOJSON_TO_GEOPARQUET_SCRIPT,
    DEFAULT_REGION,
    DEFAULT_RUNTIME,
    DEFAULT_SCRIPT_BASE_URI,
)
from .wherobots_status import submit_job




def geojson_to_geoparquet_wherobots(
    input_paths: List[str],
    output_path: str,
    api_key: Optional[str] = None,
    region: Optional[str] = None,
    script_base_uri: Optional[str] = None,
    runtime: Optional[str] = None,
    timeout_seconds: Optional[int] = None,
    job_name_prefix: str = "geojson-to-geoparquet",
) -> dict:
    """
    Submit a GeoJSON to GeoParquet conversion job to Wherobots Cloud.

    Args:
        input_paths: List of input GeoJSON file paths (can include wildcards)
        output_path: Output GeoParquet file path
        api_key: Wherobots API key (optional, uses hardcoded default if None)
        region: Wherobots region (optional, uses hardcoded default if None)
        script_base_uri: Base URI for scripts (optional, uses hardcoded default if None)
        runtime: Runtime size (optional, defaults to "tiny")
        timeout_seconds: Job timeout (optional, defaults to 3600)
        job_name_prefix: Job name prefix (optional)

    Returns:
        Dictionary with job submission result

    Example:
        >>> result = geojson_to_geoparquet_wherobots(
        ...     input_paths=["s3://bucket/input.geojson"],
        ...     output_path="s3://bucket/output.parquet",
        ... )
    """

    api_key = api_key or DEFAULT_API_KEY
    region = region or DEFAULT_REGION
    script_base_uri = script_base_uri or DEFAULT_SCRIPT_BASE_URI
    runtime = runtime or DEFAULT_RUNTIME
    timeout_seconds = timeout_seconds or 3600

    script_uri = f"{script_base_uri.rstrip('/')}/{DEFAULT_GEOJSON_TO_GEOPARQUET_SCRIPT}"

    script_args = ["--input-paths"] + input_paths + ["--output-path", output_path]

    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    job_name = f"{job_name_prefix}-{timestamp}"

    result = submit_job(
        api_key=api_key,
        region=region,
        script_uri=script_uri,
        script_args=script_args,
        runtime=runtime,
        name=job_name,
        timeout_seconds=timeout_seconds,
    )

    return result