"""
Wherobots Geometry Operations
Functions for dissolve, simplify, and buffer operations on Wherobots Cloud.
"""
from datetime import datetime
from typing import Optional
from .wherobots_config import (
DEFAULT_API_KEY,
DEFAULT_BUFFER_POINTS_SCRIPT,
DEFAULT_DISSOLVE_SIMPLIFY_SCRIPT,
DEFAULT_REGION,
DEFAULT_RUNTIME,
DEFAULT_SCRIPT_BASE_URI,
)
from .wherobots_status import submit_job
[docs]
def dissolve_simplify_wherobots(
input_path: str,
output_path: str,
dissolve_by: Optional[str] = None,
simplify_tolerance: Optional[float] = None,
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 = "dissolve-simplify",
) -> dict:
"""
Submit a dissolve and/or simplify job to Wherobots Cloud.
Args:
input_path: Input GeoParquet file path
output_path: Output GeoParquet file path
dissolve_by: Column name to dissolve by (optional)
simplify_tolerance: Simplify tolerance (optional, Douglas-Peucker algorithm)
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 = dissolve_simplify_wherobots(
... input_path="s3://bucket/input.parquet",
... output_path="s3://bucket/output.parquet",
... dissolve_by="region_name",
... simplify_tolerance=0.001,
... )
"""
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_DISSOLVE_SIMPLIFY_SCRIPT}"
script_args = ["--input-path", input_path, "--output-path", output_path]
if dissolve_by:
script_args.extend(["--dissolve-by", dissolve_by])
if simplify_tolerance is not None:
script_args.extend(["--simplify-tolerance", str(simplify_tolerance)])
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
[docs]
def buffer_points_wherobots(
input_path: str,
output_path: str,
buffer_distance: float,
distance_unit: str = "meter",
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 = "buffer-points",
) -> dict:
"""
Submit a buffer generation job to Wherobots Cloud.
Args:
input_path: Input GeoParquet file path (must contain point geometries)
output_path: Output GeoParquet file path
buffer_distance: Buffer distance in the specified unit
distance_unit: Unit for buffer distance (meter, kilometer, foot, mile)
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 = buffer_points_wherobots(
... input_path="s3://bucket/points.parquet",
... output_path="s3://bucket/buffered.parquet",
... buffer_distance=1000,
... distance_unit="meter",
... )
"""
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_BUFFER_POINTS_SCRIPT}"
script_args = [
"--input-path",
input_path,
"--output-path",
output_path,
"--buffer-distance",
str(buffer_distance),
"--distance-unit",
distance_unit,
]
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