diff --git a/akd_ext/agents/__init__.py b/akd_ext/agents/__init__.py index bf18e9d..45f3a7d 100644 --- a/akd_ext/agents/__init__.py +++ b/akd_ext/agents/__init__.py @@ -28,6 +28,13 @@ GapAgentOutputSchema, ) +from akd_ext.agents.ieso.worldview import ( + IESOWorldviewAgent, + IESOWorldviewAgentConfig, + IESOWorldviewAgentInputSchema, + IESOWorldviewAgentOutputSchema, +) + from akd_ext.agents.image_analyzer import ( ImageAnalyzerAgent, ImageAnalyzerConfig, @@ -92,6 +99,10 @@ "GapAgentConfig", "GapAgentInputSchema", "GapAgentOutputSchema", + "IESOWorldviewAgent", + "IESOWorldviewAgentConfig", + "IESOWorldviewAgentInputSchema", + "IESOWorldviewAgentOutputSchema", # Image analyzer "ImageAnalyzerAgent", "ImageAnalyzerConfig", diff --git a/akd_ext/agents/ieso.worldview.py b/akd_ext/agents/ieso.worldview.py new file mode 100644 index 0000000..81772f4 --- /dev/null +++ b/akd_ext/agents/ieso.worldview.py @@ -0,0 +1,337 @@ +"""IESO CARE Agent for NASA Worldview visualization. + +This module implements the IESO Agent for guided, +reproducible discovery of NASA Worldview visualizations. + +Public API: + IESOWorldviewAgent, IESOWorldviewAgentInputSchema, IESOWorldviewAgentOutputSchema, IESOWorldviewAgentConfig +""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import Field + +from akd._base import ( + InputSchema, + OutputSchema, + TextOutput, +) +from akd_ext.agents._base import ( + PydanticAIBaseAgent, + PydanticAIBaseAgentConfig, +) + + +# ----------------------------------------------------------------------------- +# System Prompts +# ----------------------------------------------------------------------------- + +IESO_WORLDVIEW_AGENT_SYSTEM_PROMPT = """ + ## **ROLE** + + You are a **NASA Worldview Scientific Data Assistant Agent**. + + You act as a **non-authoritative, transparency-first guide** that helps users: + + * Discover NASA datasets + * Understand dataset meaning, limitations, and proxies + * Configure and generate Worldview visualizations + * Perform **Worldview-native exploratory analysis only** + + You **do not interpret, conclude, recommend scientifically, or make decisions for the user**. + + ## **OBJECTIVE** + + Enable users to: + + 1. Translate their intent into scientifically relevant datasets + 2. Explore datasets via **NASA Worldview deep links** + 3. Understand dataset caveats, uncertainty, and limitations + 4. **Support visualization-driven and all the analysis workflows aligned with Worldview capabilities** + 5. Maintain **full human control over dataset selection and interpretation** + + ## **CONTEXT & INPUTS** + + ### **Available Systems & Tools** + + * **NASA Worldview (Primary Interface)** + * Generate deep links using URL parameters + * Support layers, time, comparison modes, charting + * **CMR API (Metadata Authority)** + * `search_collections`, `get_collection_metadata` + * UMM-based authoritative metadata + * **Earthdata Search Links** + * Dataset landing pages (no downloads or execution) + * **EONET** + * Event context (wildfires, storms, etc.) + * **Science Discovery Engine (Fallback)** + * Used only if dataset not found in primary sources + * **Worldview Layer Vector DB** + * Semantic mapping (non-authoritative) + * **Document Fetch Tool (ATBD/User Guide)** + * Triggered after dataset identification + + --- + + ### **User Inputs** + + * Natural language query (scientific or colloquial) + * Optional constraints: + * Time range + * Location + * Variable + * User expertise level: + * Beginner / Intermediate / Advanced (must be requested if unknown) + + ## **CONSTRAINTS & STYLE RULES** + + ### **Hard Constraints** + + * No scientific interpretation or conclusions + * No data inference or fabrication + * No predictive analysis + * No dataset ranking as final decision + * No autonomous dataset selection (user confirmation required) + * Only **pre-defined metrics and Worldview-supported analysis** + + ### **Guardrail Enforcement** + + * If violation detected → **REFUSE with explanation** + * If ambiguity → **ASK clarification** + * If partial data → **EXPLICITLY FLAG** + * Always include: + * Dataset provenance + * Uncertainty statement + * Non-authoritative disclaimer + + ### **Language Policy** + + * Avoid: + * “This shows…” + * “This means…” + * “This indicates…” + * Use: + * “This dataset represents…” + * “This visualization displays…” + * “Possible interpretation requires user judgment” + + ### **Output Style** + + Hybrid format: + + 1. **Structured schema (deterministic)** + 2. **User-adapted narrative** + 3. **Optional metadata expansion (on request)** + + ## **PROCESS** + + ### **Step 1: Intent Interpretation** + + * Extract: + * Goal + * Variables + * Constraints + * Normalize into scientific terms + * Ask clarification if ambiguity is high + + ### **Step 2: Expertise Detection** + + * Ask user to classify (Beginner / Intermediate / Advanced) + * Adapt: + * Vocabulary + * Detail level + * Explanation depth + + ### **Step 3: Feasibility Validation (HARD GATE)** + + * Check: + * Dataset availability + * Physical plausibility + * System capability + * If invalid: + * STOP + * Provide alternatives + + ### **Step 4: Dataset Retrieval** + + * Query: + * Worldview layers + * CMR metadata (Should be parallel) + * Use NASA SDE only if needed + * Do not override authoritative metadata + + ### **Step 5: Candidate Structuring** + + * Group datasets + * Explain: + * What dataset represents + * What it does NOT represent + * Proxy relationships + * Highlight a **recommended option (non-binding)** + + ### **Step 6: Mandatory User Confirmation** + + * Present options + * Ask: + * “Which dataset would you like to use?” + * DO NOT proceed without confirmation + + ### **Step 7: Visualization Construction** + + * Generate **Worldview deep link** + * Configure: + * Layers + * Time + * Viewport + * Comparison (if relevant) + * Ensure only valid parameters used + + ### **Step 8: Analysis Support (Limited)** + + * Provide: + * Time series (if supported) + * Regional statistics (if supported) + * Do NOT interpret results + + ### **Step 9: Provenance & Uncertainty** + + * Include: + * Dataset name + * Source + * Timestamp + * Resolution + * Add: + * Dataset uncertainty OR fallback statement + + ### **Step 10: Misuse Detection** + + Detect and block: + + * Causal inference + * Trend interpretation + * Invalid comparisons + * Proxy misuse + + ### **Step 11: Optional Expansion** + + Offer: + + * “Show dataset details” + * “Show metadata” + * “Open Earthdata page” + + ## **OUTPUT FORMAT** + + ### **1\. STRUCTURED RESPONSE** + + INTENT: + DATASET\_OPTIONS: + SELECTED\_DATASET: (ONLY after user confirmation) + WORLDVIEW\_URL: + + Options \[provide with more options\] + PARAMETERS\_USED: + PROVENANCE: + UNCERTAINTY: + LIMITATIONS: + MISSING\_FIELDS: + + ### **2\. USER NARRATIVE** + + * Beginner → simplified explanation + * Intermediate → moderate detail + * Advanced → technical description + + ### **3\. OPTIONAL ACTIONS** + + * View metadata + * Open dataset page + * Fetch documentation + + ### **4\. REQUIRED DISCLAIMER** + + “This information is derived from publicly available datasets and visualization tools on NASA Worldview . It is intended for exploratory and informational purposes only and does not constitute scientific analysis, interpretation, or validated conclusions.” +""" + +# ----------------------------------------------------------------------------- +# Configuration +# ----------------------------------------------------------------------------- + + +class IESOWorldviewAgentConfig(PydanticAIBaseAgentConfig): + """Configuration for IESO CARE Agent.""" + + description: str = Field( + default=( + """Earth science Worldview-visualization agent. Helps users translate + Earth science queries into NASA Worldview permalinks by clarifying intent, + surfacing candidate datasets, awaiting user confirmation, and producing a + reproducible visualization URL with provenance and uncertainty annotations. + Outputs are delivered via a structured schema and interactive chat with the + user for clarification, dataset selection, approval gates, and disclaimers.""" + ) + ) + system_prompt: str = Field(default=IESO_WORLDVIEW_AGENT_SYSTEM_PROMPT) + model_name: str = Field(default="openai:gpt-5.2") + reasoning_effort: Literal["low", "medium", "high"] | None = Field(default="medium") + + +# ----------------------------------------------------------------------------- +# Input/Output Schemas +# ----------------------------------------------------------------------------- + + +class IESOWorldviewAgentInputSchema(InputSchema): + """Input schema for the IESO Worldview-discovery agent.""" + + query: str = Field(..., description="Earth science query to interact with worldview visualization") + + +class IESOWorldviewAgentOutputSchema(OutputSchema): + """Structured Worldview-discovery response. Use this on the final turn, + after the user has confirmed a dataset; populate `result` with the full + sectioned response and `url` with the Worldview permalink. + Use TextOutput for clarification questions or when no dataset has been + confirmed yet.""" + + __response_field__ = "result" + result: str = Field( + ..., + description=( + "Full sectioned response: INTENT, DATASET_OPTIONS, " + "SELECTED_DATASET, WORLDVIEW_URL, PARAMETERS_USED, PROVENANCE, " + "UNCERTAINTY, LIMITATIONS, MISSING_FIELDS, USER NARRATIVE " + "OPTIONAL ACTIONS, and the REQUIRED DISCLAIMER" + "Format is defined by the system prompt." + ), + ) + url: str = Field( + ..., + description="Worldview permalink that resolves the science query.", + ) + + +# ----------------------------------------------------------------------------- +# IESO Agent +# ----------------------------------------------------------------------------- + + +class IESOWorldviewAgent(PydanticAIBaseAgent[IESOWorldviewAgentInputSchema, IESOWorldviewAgentOutputSchema]): + """Earth science Worldview-visualization agent. + + Resolves an Earth science query into a NASA Worldview permalink. + """ + + input_schema = IESOWorldviewAgentInputSchema + output_schema = IESOWorldviewAgentOutputSchema | TextOutput + config_schema = IESOWorldviewAgentConfig + + def check_output(self, output) -> str | None: + if isinstance(output, IESOWorldviewAgentOutputSchema): + if not output.result.strip(): + return "Result is empty. Provide the structured Worldview-discovery response." + if not output.url.strip(): + return "URL is empty. Provide a valid url" + return super().check_output(output) diff --git a/akd_ext/tools/__init__.py b/akd_ext/tools/__init__.py index fb1328e..8472043 100644 --- a/akd_ext/tools/__init__.py +++ b/akd_ext/tools/__init__.py @@ -1,41 +1,52 @@ """Tools module for akd_ext.""" -from .dummy import DummyInputSchema, DummyOutputSchema, DummyTool -from .sde_search import ( - SDEDocument, - SDESearchTool, - SDESearchToolConfig, - SDESearchToolInputSchema, - SDESearchToolOutputSchema, -) -from .code_search.code_signals import ( - CodeSignalsSearchInputSchema, - CodeSignalsSearchOutputSchema, - CodeSignalsSearchTool, - CodeSignalsSearchToolConfig, -) -from .code_search.repository_search import ( - RepositorySearchTool, - RepositorySearchToolInputSchema, - RepositorySearchToolOutputSchema, - RepositorySearchToolConfig, +# from .dummy import DummyInputSchema, DummyOutputSchema, DummyTool +# from .sde_search import ( +# SDEDocument, +# SDESearchTool, +# SDESearchToolConfig, +# SDESearchToolInputSchema, +# SDESearchToolOutputSchema, +# ) +# from .code_search.code_signals import ( +# CodeSignalsSearchInputSchema, +# CodeSignalsSearchOutputSchema, +# CodeSignalsSearchTool, +# CodeSignalsSearchToolConfig, +# ) +# from .code_search.repository_search import ( +# RepositorySearchTool, +# RepositorySearchToolInputSchema, +# RepositorySearchToolOutputSchema, +# RepositorySearchToolConfig, +# ) + +from .worldview import ( + LayerSpec, + WorldviewPermalinkInputSchema, + WorldviewPermalinkOutputSchema, + WorldviewPermalinkTool, ) __all__ = [ - "DummyTool", - "DummyInputSchema", - "DummyOutputSchema", - "SDESearchTool", - "SDESearchToolInputSchema", - "SDESearchToolOutputSchema", - "SDESearchToolConfig", - "SDEDocument", - "CodeSignalsSearchInputSchema", - "CodeSignalsSearchOutputSchema", - "CodeSignalsSearchTool", - "CodeSignalsSearchToolConfig", - "RepositorySearchTool", - "RepositorySearchToolInputSchema", - "RepositorySearchToolOutputSchema", - "RepositorySearchToolConfig", + # "DummyTool", + # "DummyInputSchema", + # "DummyOutputSchema", + # "SDESearchTool", + # "SDESearchToolInputSchema", + # "SDESearchToolOutputSchema", + # "SDESearchToolConfig", + # "SDEDocument", + # "CodeSignalsSearchInputSchema", + # "CodeSignalsSearchOutputSchema", + # "CodeSignalsSearchTool", + # "CodeSignalsSearchToolConfig", + # "RepositorySearchTool", + # "RepositorySearchToolInputSchema", + # "RepositorySearchToolOutputSchema", + # "RepositorySearchToolConfig", + "LayerSpec", + "WorldviewPermalinkInputSchema", + "WorldviewPermalinkOutputSchema", + "WorldviewPermalinkTool", ] diff --git a/akd_ext/tools/worldview/__init__.py b/akd_ext/tools/worldview/__init__.py new file mode 100644 index 0000000..483b5f9 --- /dev/null +++ b/akd_ext/tools/worldview/__init__.py @@ -0,0 +1,13 @@ +from akd_ext.tools.worldview.permalink import ( + LayerSpec, + WorldviewPermalinkInputSchema, + WorldviewPermalinkOutputSchema, + WorldviewPermalinkTool, +) + +__all__ = [ + "LayerSpec", + "WorldviewPermalinkInputSchema", + "WorldviewPermalinkOutputSchema", + "WorldviewPermalinkTool", +] diff --git a/akd_ext/tools/worldview/permalink.py b/akd_ext/tools/worldview/permalink.py new file mode 100644 index 0000000..8e73090 --- /dev/null +++ b/akd_ext/tools/worldview/permalink.py @@ -0,0 +1,566 @@ +"""AKD Tool for the NASA Worldview permalink builder. + +URL-assembly helpers and the public `build_url` live as static/classmethods on the tool class. +Field descriptions on the input schema are written as agent-facing guidance. +""" + +import os +from datetime import date, datetime, timedelta, timezone +from typing import Literal, Self +from urllib.parse import urlencode + +from akd._base import InputSchema, OutputSchema +from akd.tools import BaseTool, BaseToolConfig +from dateutil import parser as date_parser +from pydantic import BaseModel, Field, ValidationInfo, field_validator, model_validator + +from akd_ext.mcp import mcp_tool + +# ----------------------------------------------------------------------------- +# Module global variables +# ----------------------------------------------------------------------------- + +DEFAULT_BASE_URL = "https://worldview.earthdata.nasa.gov/" + +BASE_LAYERS: tuple[str, ...] = ( + "MODIS_Terra_CorrectedReflectance_TrueColor", + "MODIS_Aqua_CorrectedReflectance_TrueColor", + "VIIRS_SNPP_CorrectedReflectance_TrueColor", + "VIIRS_NOAA20_CorrectedReflectance_TrueColor", + "VIIRS_NOAA21_CorrectedReflectance_TrueColor", +) +BASE_LAYERS_SET: frozenset[str] = frozenset(BASE_LAYERS) +DEFAULT_BASE_LAYER: str = BASE_LAYERS[0] +DEFAULT_REFERENCE_OVERLAYS: tuple[str, ...] = ("Coastlines_15m", "Reference_Features_15m") + +_FORBIDDEN_LAYER_CHARS: frozenset[str] = frozenset(",()") + + +# ----------------------------------------------------------------------------- +# Input/Output Schema with Validators +# ----------------------------------------------------------------------------- + + +def _reject_grammar_chars(value: str, field_name: str) -> str: + """Reject ',', '(', ')' — reserved by Worldview's layer-list grammar. + + The URL format `l=LayerID(mod,mod),LayerID2` uses these characters as + structural delimiters; embedding them inside an ID, style, or palette + silently corrupts the URL when Worldview parses it back. + """ + bad = _FORBIDDEN_LAYER_CHARS.intersection(value) + if bad: + raise ValueError( + f"{field_name}={value!r} contains forbidden character(s) {sorted(bad)}; " + f"',', '(', and ')' are reserved by Worldview's layer-list grammar" + ) + return value + + +def _coerce_to_datetime(t: str | date | datetime) -> datetime: + """Best-effort coerce a time value to a TZ-aware UTC datetime for comparison. + + Used in model validator during time comparision to check semantics. + """ + if isinstance(t, str): + try: + t = date_parser.parse(t) + except (ValueError, OverflowError) as e: + raise ValueError(f"could not parse time {t!r}: {e}") from e + # datetime is a subclass of date — check it first + if isinstance(t, datetime): + return t if t.tzinfo else t.replace(tzinfo=timezone.utc) + if isinstance(t, date): + return datetime.combine(t, datetime.min.time(), tzinfo=timezone.utc) + raise TypeError(f"unsupported time type: {type(t).__name__}") + + +class LayerSpec(BaseModel): + """A single Worldview layer, optionally with rendering modifiers. + + Field defaults match Worldview's own defaults; omitted fields are not + emitted into the URL. + """ + + id: str = Field( + ..., + description=( + "GIBS layer identifier (e.g. 'MODIS_Terra_CorrectedReflectance_TrueColor', " + "'VIIRS_SNPP_AOD'). Stable strings published by NASA's GIBS service." + ), + ) + hidden: bool = Field( + default=False, + description=( + "If True, the layer is included in the layer stack but rendered invisibly. " + "Useful for pre-loading toggleable layers without rebuilding the link." + ), + ) + opacity: float | None = Field( + default=None, + ge=0.0, + le=1.0, + description="Layer opacity, 0.0 (fully transparent) to 1.0 (fully opaque). None for full opacity.", + ) + palettes: list[str] | None = Field( + default=None, + description=( + "Custom palette IDs to apply, in order. Only meaningful for raster layers that support palette swapping." + ), + ) + style: str | None = Field( + default=None, + description="Vector style ID. Only meaningful for vector layers.", + ) + min: float | None = Field( + default=None, + description="Lower bound of the palette/data range. Set together with `max` to clamp the visible range.", + ) + max: float | None = Field( + default=None, + description="Upper bound of the palette/data range.", + ) + squash: bool = Field( + default=False, + description=( + "If True, the palette is squashed to the designated min/max values " + "rather than spanning the layer's full data range." + ), + ) + + @field_validator("id", "style") + @classmethod + def _check_string_field(cls, v: str | None, info: ValidationInfo) -> str | None: + if v is not None: + _reject_grammar_chars(v, info.field_name) + return v + + @field_validator("palettes") + @classmethod + def _check_palettes(cls, v: list[str] | None) -> list[str] | None: + if v is not None: + for item in v: + _reject_grammar_chars(item, "palettes") + return v + + +class WorldviewPermalinkInputSchema(InputSchema): + """Input schema for the Worldview Permalink Tool.""" + + layers: list[LayerSpec] = Field( + ..., + description=( + "One or more LayerSpec instances, in render order (top of stack last). " + "Each LayerSpec carries a GIBS layer ID plus optional per-layer modifiers " + "(hidden, opacity, palettes, style, min/max, squash). REQUIRED — at least " + "one layer must be supplied." + ), + ) + projection: Literal["geographic", "arctic", "antarctic"] = Field( + default="geographic", + description=( + "Map projection. 'geographic' for global Mercator (the default), 'arctic' " + "for north polar stereographic, or 'antarctic' for south polar stereographic." + ), + ) + time: str | date | datetime | None = Field( + default=None, + description=( + "Map time. Accepts a date (daily resolution), a datetime (subdaily, " + "normalised to UTC), or a string in any reasonable date/datetime format — " + "ISO 8601, 'Sep 15, 2025', '2025/09/15', TZ-aware forms, etc. (parsed via " + "dateutil). If None, defaults to yesterday (UTC) — Worldview's own 'today' " + "default can show partially-rendered scenes because daily MODIS/VIIRS data " + "is still being ingested into GIBS; yesterday guarantees a fully-rendered " + "scene. Note: ambiguous slash-separated strings like '01/02/2025' are " + "interpreted as month/day/year by default; pass an unambiguous form " + "('2025-01-02') if the order matters." + ), + ) + bbox: list[float] | None = Field( + default=None, + min_length=4, + max_length=4, + description=( + "Map viewport extent as [west, south, east, north]. Degrees for the " + "geographic projection; projected meters for arctic/antarctic. " + "If None, Worldview opens at its default global extent." + ), + ) + rotation: float | None = Field( + default=None, + ge=-180.0, + le=180.0, + description=( + "Map rotation in degrees, range -180 to 180. Honored only by arctic/" + "antarctic projections; ignored by geographic." + ), + ) + + compare_active: bool | None = Field( + default=None, + description=( + "Activates Worldview's compare mode. Tri-state: None (default) — compare " + "OFF, no compare params emitted, all other compare_* args silently ignored. " + "True — compare ON with the A state shown as active. False — compare ON " + "with the B state shown as active. When set (True or False), compare_layers " + "is REQUIRED." + ), + ) + compare_layers: list[LayerSpec] | None = Field( + default=None, + description=( + "LayerSpec instances for the B state, same shape as `layers`. REQUIRED when compare_active is not None." + ), + ) + compare_time: str | date | datetime | None = Field( + default=None, + description=( + "Time for the B state, same accepted forms as `time`. Optional even when " + "compare is on; if omitted, the B state uses the same time as the A state." + ), + ) + compare_mode: Literal["swipe", "spy", "opacity"] = Field( + default="swipe", + description=( + "Comparison style. 'swipe' (vertical swiper between A and B), 'spy' " + "(lens-style hover view of B over A), or 'opacity' (A overlaid on B with " + "adjustable opacity). Only consulted when compare is active." + ), + ) + compare_value: int = Field( + default=50, + ge=0, + le=100, + description=( + "Position of the swiper or value of the opacity overlay, integer 0–100. " + "Only consulted when compare is active." + ), + ) + + chart_active: bool = Field( + default=False, + description=( + "Activates Worldview's charting mode (time-series of regional statistics " + "over a drawn area). False (default) — charting OFF, no chart params " + "emitted, all other chart_* args silently ignored. True — charting ON. " + "When True, chart_layer is REQUIRED." + ), + ) + chart_layer: str | None = Field( + default=None, + description=("GIBS layer ID to chart. Charting supports one layer at a time. REQUIRED when chart_active=True."), + ) + chart_area: list[float] | None = Field( + default=None, + min_length=4, + max_length=4, + description=( + "Area-of-interest for the chart, as [x1, y1, x2, y2] in the same coordinate " + "system as `bbox`. Statistics are computed over this region." + ), + ) + chart_time_start: str | date | datetime | None = Field( + default=None, + description="Start of the chart's time range, same accepted forms as `time`.", + ) + chart_time_end: str | date | datetime | None = Field( + default=None, + description="End of the chart's time range.", + ) + chart_autoload: bool = Field( + default=False, + description=( + "If True, the chart computes and renders the moment the link is opened. " + "Default False (user must click 'Generate Chart' in the Worldview UI)." + ), + ) + + @field_validator("chart_layer") + @classmethod + def _check_chart_layer(cls, v: str | None) -> str | None: + if v is not None: + _reject_grammar_chars(v, "chart_layer") + return v + + @model_validator(mode="after") + def _enforce_feature_gates(self) -> Self: + if self.compare_active is not None and self.compare_layers is None: + raise ValueError( + "compare_active is set, so compare_layers is required " + "(cannot enable compare mode without a B-state layer list)" + ) + if self.chart_active and self.chart_layer is None: + raise ValueError( + "chart_active is True, so chart_layer is required (cannot enable charting without a layer to chart)" + ) + return self + + @model_validator(mode="after") + def _validate_semantics(self) -> Self: + if self.bbox is not None: + west, south, east, north = self.bbox + if south >= north: + raise ValueError(f"bbox south ({south}) must be < north ({north})") + if west == east: + raise ValueError(f"bbox west ({west}) must differ from east ({east}); zero-width bbox is invalid") + # west > east is allowed (antimeridian crossing in geographic projection) + if self.projection == "geographic": + if not (-180 <= west <= 180 and -180 <= east <= 180): + raise ValueError(f"bbox lon out of [-180, 180] for geographic projection: {self.bbox}") + if not (-90 <= south <= 90 and -90 <= north <= 90): + raise ValueError(f"bbox lat out of [-90, 90] for geographic projection: {self.bbox}") + + if self.chart_active and self.chart_area is not None: + x1, y1, x2, y2 = self.chart_area + if y1 >= y2: + raise ValueError(f"chart_area y1 ({y1}) must be < y2 ({y2})") + if x1 == x2: + raise ValueError(f"chart_area x1 ({x1}) must differ from x2 ({x2}); zero-width area is invalid") + if self.projection == "geographic": + if not (-180 <= x1 <= 180 and -180 <= x2 <= 180): + raise ValueError(f"chart_area lon out of [-180, 180] for geographic projection: {self.chart_area}") + if not (-90 <= y1 <= 90 and -90 <= y2 <= 90): + raise ValueError(f"chart_area lat out of [-90, 90] for geographic projection: {self.chart_area}") + + if self.chart_time_start is not None and self.chart_time_end is not None: + start = _coerce_to_datetime(self.chart_time_start) + end = _coerce_to_datetime(self.chart_time_end) + if start > end: + raise ValueError( + f"chart_time_start ({self.chart_time_start}) must be <= chart_time_end ({self.chart_time_end})" + ) + + return self + + +class WorldviewPermalinkOutputSchema(OutputSchema): + """Output schema for the Worldview Permalink Tool.""" + + url: str = Field( + ..., + description="A complete NASA Worldview permalink URL that opens the map at the requested state.", + ) + + +# ----------------------------------------------------------------------------- +# Tool Configuration +# ----------------------------------------------------------------------------- + + +class WorldviewPermalinkToolConfig(BaseToolConfig): + """Configuration for the WorldviewPermalinkTool Tool.""" + + base_url: str = Field( + default=os.getenv("WORLDVIEW_BASE_URL", DEFAULT_BASE_URL), + description="Base URL for the NASA WORLDVIEW", + ) + + +# ----------------------------------------------------------------------------- +# Permalink generation (mcp) Tool +# ----------------------------------------------------------------------------- + + +@mcp_tool +class WorldviewPermalinkTool(BaseTool[WorldviewPermalinkInputSchema, WorldviewPermalinkOutputSchema]): + """ + Build a NASA Worldview permalink URL. + + Generates a deep link to NASA Worldview (https://worldview.earthdata.nasa.gov) + that opens the interactive map at a specific layer configuration, time, + viewport, and (optionally) comparison or charting state. No I/O — pure URL + string assembly. + + Use this tool after a dataset has been confirmed with the user, to produce + the visualization link the user will open. + p.s. The IESO Worldview agent calls this in its "Visualization Construction" and + "Analysis Support" steps. + + Required: + - layers: at least one LayerSpec (GIBS layer ID + optional rendering modifiers). + Layer IDs must be GIBS layer IDs (e.g. `MODIS_Terra_CorrectedReflectance_TrueColor`, + `VIIRS_SNPP_AOD`), NOT CMR collection concept_ids (`C-`). The + latter will produce a valid-looking URL that renders blank. + + Optional viewport / time: + - projection, time, bbox, rotation — omit any to inherit Worldview's defaults + + Optional feature blocks (each gated by an _active flag; the rest of the + block is silently ignored when the gate is off): + - Comparison: set compare_active=True (A side) or False (B side) and + provide compare_layers; optionally compare_time / compare_mode / + compare_value. + - Charting: set chart_active=True and provide chart_layer; optionally + chart_area / chart_time_start / chart_time_end / chart_autoload. + """ + + input_schema = WorldviewPermalinkInputSchema + output_schema = WorldviewPermalinkOutputSchema + config_schema = WorldviewPermalinkToolConfig + + async def _arun(self, params: WorldviewPermalinkInputSchema) -> WorldviewPermalinkOutputSchema: + return WorldviewPermalinkOutputSchema(url=self.build_url(params, self.config.base_url)) + + @classmethod + def build_url(cls, params: WorldviewPermalinkInputSchema, base_url: str = DEFAULT_BASE_URL) -> str: + """Pure URL-string assembly from a validated input schema. No I/O. + + The schema's `_enforce_feature_gates` validator guarantees the + compare/chart not-None invariants before this method runs, so the body + does not re-check them. + """ + out: dict[str, str] = {} + + layers = cls._apply_layer_preprocessing(params.layers) + out["l"] = ",".join(cls._format_layer(s) for s in layers) + + if params.compare_active is not None: + assert params.compare_layers is not None + compare_layers = cls._apply_layer_preprocessing(params.compare_layers) + out["l1"] = ",".join(cls._format_layer(s) for s in compare_layers) + + time_value = params.time if params.time is not None else (datetime.now(timezone.utc) - timedelta(days=1)).date() + if (formatted := cls._format_time(time_value)) is not None: + out["t"] = formatted + + if params.compare_active is not None and (formatted := cls._format_time(params.compare_time)) is not None: + out["t1"] = formatted + + if params.bbox is not None: + out["v"] = ",".join(cls._fmt_num(x) for x in params.bbox) + + out["p"] = params.projection + + if params.rotation is not None: + out["r"] = cls._fmt_num(params.rotation) + + if params.compare_active is not None: + out["ca"] = "true" if params.compare_active else "false" + out["cm"] = params.compare_mode + out["cv"] = str(params.compare_value) + + if params.chart_active: + assert params.chart_layer is not None + out["cha"] = "true" + out["chl"] = params.chart_layer + if params.chart_area is not None: + out["chc"] = ",".join(cls._fmt_num(x) for x in params.chart_area) + if (formatted := cls._format_time(params.chart_time_start)) is not None: + out["cht"] = formatted + if (formatted := cls._format_time(params.chart_time_end)) is not None: + out["cht2"] = formatted + out["chch"] = "true" if params.chart_autoload else "false" + + out["em"] = "true" + return f"{base_url}?{urlencode(out, safe=',()=:')}" + + # ----------------------------------------------------------------------------- + # Utility static methods + # ----------------------------------------------------------------------------- + + @staticmethod + def _fmt_num(n: float | int) -> str: + if isinstance(n, float) and n.is_integer(): + return str(int(n)) + return str(n) + + @classmethod + def _format_layer(cls, spec: LayerSpec) -> str: + tokens: list[str] = [] + if spec.hidden: + tokens.append("hidden") + if spec.opacity is not None: + tokens.append(f"opacity={cls._fmt_num(spec.opacity)}") + if spec.palettes: + tokens.append(f"palettes={','.join(spec.palettes)}") + if spec.style is not None: + tokens.append(f"style={spec.style}") + if spec.min is not None: + tokens.append(f"min={cls._fmt_num(spec.min)}") + if spec.max is not None: + tokens.append(f"max={cls._fmt_num(spec.max)}") + if spec.squash: + tokens.append("squash") + if not tokens: + return spec.id + return f"{spec.id}({','.join(tokens)})" + + @staticmethod + def _apply_layer_preprocessing(layers: list[LayerSpec]) -> list[LayerSpec]: + """Pre-process a layer list before URL emission. Applied unconditionally. + + Three steps: + 1. Prepend the default base layer if none of the supplied layers' ids are in + BASE_LAYERS_SET. Load-bearing — Worldview shows a black background when + l= contains only overlays. + 2. Append default reference overlays (Coastlines_15m, Reference_Features_15m) + that aren't already present. Provides land/water clarity + political borders. + 3. Canonical reorder: baselayers first, overlays after; user-supplied order + preserved within each partition. + + Returns a new list; the input is not mutated. + """ + result = list(layers) + + if not any(layer.id in BASE_LAYERS_SET for layer in result): + result = [LayerSpec(id=DEFAULT_BASE_LAYER), *result] + + existing_ids = {layer.id for layer in result} + for ref_id in DEFAULT_REFERENCE_OVERLAYS: + if ref_id not in existing_ids: + result.append(LayerSpec(id=ref_id)) + + baselayers = [layer for layer in result if layer.id in BASE_LAYERS_SET] + overlays = [layer for layer in result if layer.id not in BASE_LAYERS_SET] + return [*baselayers, *overlays] + + @staticmethod + def _format_time(t: str | date | datetime | None) -> str | None: + if t is None: + return None + if isinstance(t, str): + try: + t = date_parser.parse(t) + except (ValueError, OverflowError) as e: + raise ValueError(f"Could not parse time {t!r}: {e}") from e + if isinstance(t, datetime): + if t.tzinfo is not None: + t = t.astimezone(timezone.utc) + if t.time() == datetime.min.time(): + return t.date().isoformat() + return t.strftime("%Y-%m-%dT%H:%M:%SZ") + if isinstance(t, date): + return t.isoformat() + raise TypeError(f"Unsupported time type: {type(t).__name__}") + + +if __name__ == "__main__": + core = WorldviewPermalinkTool.build_url( + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="MODIS_Terra_CorrectedReflectance_TrueColor"), LayerSpec(id="MODIS_Aqua_Aerosol")], + time="2025-09-15", + bbox=(-125, 32, -114, 42), + ) + ) + print("Core:", core) + + rich = WorldviewPermalinkTool.build_url( + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="MODIS_Aqua_Aerosol", opacity=0.8)], + time="September 15, 2025", + bbox=(-125, 32, -114, 42), + compare_active=True, + compare_layers=[LayerSpec(id="MODIS_Aqua_Aerosol")], + compare_time="2025-09-14", + compare_mode="swipe", + compare_value=60, + chart_active=True, + chart_layer="MODIS_Aqua_Aerosol", + chart_area=(-125, 32, -114, 42), + chart_time_start="2025-09-01", + chart_time_end="2025-09-30", + chart_autoload=True, + ) + ) + print("Rich:", rich) diff --git a/pyproject.toml b/pyproject.toml index 5acf35e..37e4b1f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,12 +17,13 @@ classifiers = [ ] requires-python = ">=3.12" dependencies = [ - "akd @ git+https://github.com/NASA-IMPACT/akd-core.git@develop", + "akd @ git+https://${GITHUB_TOKEN}@github.com/NASA-IMPACT/akd-core.git@develop", "fastmcp>=2.0.0,<3.2.4", "griffe>=1.0.0,<2", "openai-agents>=0.6.7", "pydantic-ai>=1.81.0", "PyGithub>=2.1.1", + "python-dateutil>=2.8", ] [project.urls] diff --git a/tests/tools/worldview/__init__.py b/tests/tools/worldview/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/tools/worldview/test_permalink.py b/tests/tools/worldview/test_permalink.py new file mode 100644 index 0000000..abca0e4 --- /dev/null +++ b/tests/tools/worldview/test_permalink.py @@ -0,0 +1,271 @@ +"""Unit tests for the WorldviewPermalinkTool wrapper.""" + +import pytest +from pydantic import ValidationError + +from akd_ext.tools.worldview import ( + LayerSpec, + WorldviewPermalinkInputSchema, + WorldviewPermalinkOutputSchema, + WorldviewPermalinkTool, +) + + +class TestWorldviewPermalinkTool: + """Tool-level behaviour: input schema validation and _arun delegation.""" + + async def test_arun_returns_output_schema_with_url(self): + tool = WorldviewPermalinkTool() + result = await tool.arun( + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="MODIS_Terra_CorrectedReflectance_TrueColor")], + time="2025-09-15", + bbox=(-125, 32, -114, 42), + ) + ) + assert isinstance(result, WorldviewPermalinkOutputSchema) + assert result.url.startswith("https://worldview.earthdata.nasa.gov/?") + assert "l=MODIS_Terra_CorrectedReflectance_TrueColor" in result.url + assert "t=2025-09-15" in result.url + assert "v=-125,32,-114,42" in result.url + + async def test_arun_compare_block(self): + tool = WorldviewPermalinkTool() + result = await tool.arun( + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L_A")], + compare_active=True, + compare_layers=[LayerSpec(id="L_B")], + compare_time="2025-09-14", + compare_mode="spy", + compare_value=70, + ) + ) + assert "ca=true" in result.url + assert "cm=spy" in result.url + assert "cv=70" in result.url + # B-state list is pre-processed: base prepended, refs appended. + assert ",L_B," in result.url + assert "t1=2025-09-14" in result.url + + async def test_arun_chart_block(self): + tool = WorldviewPermalinkTool() + result = await tool.arun( + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + chart_active=True, + chart_layer="L_CHART", + chart_area=(-125, 32, -114, 42), + chart_time_start="2025-09-01", + chart_time_end="2025-09-30", + chart_autoload=True, + ) + ) + assert "cha=true" in result.url + assert "chl=L_CHART" in result.url + assert "chc=-125,32,-114,42" in result.url + assert "cht=2025-09-01" in result.url + assert "cht2=2025-09-30" in result.url + assert "chch=true" in result.url + + +class TestSchemaValidation: + """Cross-field gate constraints enforced by model_validator.""" + + def test_compare_active_without_compare_layers_rejected(self): + with pytest.raises(ValidationError, match="compare_layers is required"): + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + compare_active=True, + compare_layers=None, + ) + + def test_chart_active_without_chart_layer_rejected(self): + with pytest.raises(ValidationError, match="chart_layer is required"): + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + chart_active=True, + chart_layer=None, + ) + + def test_compare_value_out_of_range_rejected(self): + with pytest.raises(ValidationError): + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + compare_active=True, + compare_layers=[LayerSpec(id="LB")], + compare_value=150, + ) + + def test_minimal_input_validates(self): + # Just layers — no compare, no chart, no extras + schema = WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")]) + assert schema.compare_active is None + assert schema.chart_active is False + assert schema.projection == "geographic" + + +class TestGrammarCharRejection: + """Reject ',', '(', ')' anywhere they would corrupt the layer-list grammar.""" + + @pytest.mark.parametrize("bad_id", ["MODIS,Aqua", "A(B)", "A(", "A)", "X,Y,Z"]) + def test_layer_id_with_grammar_char_rejected(self, bad_id): + with pytest.raises(ValidationError, match="forbidden character"): + LayerSpec(id=bad_id) + + @pytest.mark.parametrize("bad_style", ["dashed,thick", "fancy(thing)", "a)b"]) + def test_layer_style_with_grammar_char_rejected(self, bad_style): + with pytest.raises(ValidationError, match="forbidden character"): + LayerSpec(id="L", style=bad_style) + + @pytest.mark.parametrize("bad_palette", ["red,blue", "p(1)", "x)"]) + def test_palette_item_with_grammar_char_rejected(self, bad_palette): + with pytest.raises(ValidationError, match="forbidden character"): + LayerSpec(id="L", palettes=[bad_palette]) + + def test_chart_layer_with_grammar_char_rejected(self): + with pytest.raises(ValidationError, match="forbidden character"): + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + chart_active=True, + chart_layer="X(Y)", + ) + + def test_safe_strings_accepted(self): + # Realistic GIBS-shaped IDs must not trip the validator. + spec = LayerSpec( + id="MODIS_Terra_CorrectedReflectance_TrueColor", + style="dashed", + palettes=["red_blue", "viridis"], + ) + assert spec.id == "MODIS_Terra_CorrectedReflectance_TrueColor" + assert spec.style == "dashed" + assert spec.palettes == ["red_blue", "viridis"] + + +class TestFieldBounds: + """Field-level numeric bounds added alongside the validators.""" + + @pytest.mark.parametrize("bad_opacity", [-0.1, 1.1, 2.0]) + def test_opacity_out_of_range_rejected(self, bad_opacity): + with pytest.raises(ValidationError): + LayerSpec(id="L", opacity=bad_opacity) + + @pytest.mark.parametrize("good_opacity", [0.0, 0.5, 1.0]) + def test_opacity_in_range_accepted(self, good_opacity): + spec = LayerSpec(id="L", opacity=good_opacity) + assert spec.opacity == good_opacity + + @pytest.mark.parametrize("bad_rotation", [-200, 200, 360]) + def test_rotation_out_of_range_rejected(self, bad_rotation): + with pytest.raises(ValidationError): + WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")], rotation=bad_rotation) + + @pytest.mark.parametrize("good_rotation", [-180, 0, 180]) + def test_rotation_in_range_accepted(self, good_rotation): + schema = WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")], rotation=good_rotation) + assert schema.rotation == good_rotation + + +class TestBboxValidation: + """bbox ordering, geographic-projection bounds, antimeridian carve-out, polar-projection skip.""" + + def test_zero_width_bbox_rejected(self): + with pytest.raises(ValidationError, match="zero-width bbox"): + WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")], bbox=(10.0, 0.0, 10.0, 5.0)) + + def test_inverted_latitude_rejected(self): + with pytest.raises(ValidationError, match="south .* must be < north"): + WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")], bbox=(-10.0, 20.0, 10.0, 0.0)) + + def test_antimeridian_crossing_accepted(self): + # Pacific-spanning bbox: west (170) > east (-170) is valid in geographic. + schema = WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")], bbox=(170.0, -10.0, -170.0, 10.0)) + assert schema.bbox == [170.0, -10.0, -170.0, 10.0] + + def test_lon_out_of_range_rejected_for_geographic(self): + with pytest.raises(ValidationError, match="bbox lon out of"): + WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")], bbox=(-200.0, 0.0, 0.0, 10.0)) + + def test_lat_out_of_range_rejected_for_geographic(self): + with pytest.raises(ValidationError, match="bbox lat out of"): + WorldviewPermalinkInputSchema(layers=[LayerSpec(id="L")], bbox=(-10.0, -100.0, 10.0, 0.0)) + + @pytest.mark.parametrize("projection", ["arctic", "antarctic"]) + def test_polar_projection_skips_lonlat_bounds(self, projection): + # Same out-of-lonlat-range bbox is accepted for polar projections (those use projected meters). + schema = WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + projection=projection, + bbox=(-4_000_000.0, -4_000_000.0, 4_000_000.0, 4_000_000.0), + ) + assert schema.projection == projection + + @pytest.mark.parametrize("projection", ["arctic", "antarctic"]) + def test_polar_projection_still_enforces_ordering(self, projection): + with pytest.raises(ValidationError, match="south .* must be < north"): + WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + projection=projection, + bbox=(-1000.0, 1000.0, 1000.0, -1000.0), + ) + + +class TestChartAreaValidation: + """chart_area follows the same rules as bbox; only consulted when chart_active.""" + + def _base(self, **overrides): + params = dict( + layers=[LayerSpec(id="L")], + chart_active=True, + chart_layer="L", + ) + params.update(overrides) + return params + + def test_zero_width_chart_area_rejected(self): + with pytest.raises(ValidationError, match="zero-width area"): + WorldviewPermalinkInputSchema(**self._base(chart_area=(10.0, 0.0, 10.0, 5.0))) + + def test_inverted_y_rejected(self): + with pytest.raises(ValidationError, match="y1 .* must be < y2"): + WorldviewPermalinkInputSchema(**self._base(chart_area=(-10.0, 20.0, 10.0, 0.0))) + + def test_lon_out_of_range_rejected_for_geographic(self): + with pytest.raises(ValidationError, match="chart_area lon out of"): + WorldviewPermalinkInputSchema(**self._base(chart_area=(-200.0, 0.0, 0.0, 10.0))) + + def test_chart_area_ignored_when_chart_inactive(self): + # When chart_active is False, chart_area validation is skipped — only emission is gated. + schema = WorldviewPermalinkInputSchema( + layers=[LayerSpec(id="L")], + chart_active=False, + chart_area=(10.0, 0.0, 10.0, 5.0), # would be rejected if chart_active=True + ) + assert schema.chart_active is False + + +class TestChartTimeOrdering: + """chart_time_start <= chart_time_end when both are supplied.""" + + def _base(self, **overrides): + params = dict(layers=[LayerSpec(id="L")], chart_active=True, chart_layer="L") + params.update(overrides) + return params + + def test_inverted_chart_time_rejected(self): + with pytest.raises(ValidationError, match="chart_time_start .* must be <="): + WorldviewPermalinkInputSchema(**self._base(chart_time_start="2025-09-30", chart_time_end="2025-09-01")) + + def test_equal_chart_time_accepted(self): + schema = WorldviewPermalinkInputSchema(**self._base(chart_time_start="2025-09-15", chart_time_end="2025-09-15")) + assert schema.chart_time_start == "2025-09-15" + + def test_ordered_chart_time_accepted(self): + schema = WorldviewPermalinkInputSchema(**self._base(chart_time_start="2025-09-01", chart_time_end="2025-09-30")) + assert schema.chart_time_end == "2025-09-30" + + def test_one_sided_chart_time_skipped(self): + # Only one of (start, end) supplied — order check is skipped. + schema = WorldviewPermalinkInputSchema(**self._base(chart_time_start="2025-09-15")) + assert schema.chart_time_end is None diff --git a/tests/tools/worldview/test_permalink_assembly.py b/tests/tools/worldview/test_permalink_assembly.py new file mode 100644 index 0000000..4fa8a5e --- /dev/null +++ b/tests/tools/worldview/test_permalink_assembly.py @@ -0,0 +1,368 @@ +"""Unit tests for `WorldviewPermalinkTool.build_url` URL-string assembly. + +Schema-level validation (gate constraints, range constraints) is covered in +`test_permalink.py`. Tests here exercise the pure URL-assembly path: pass a +validated `WorldviewPermalinkInputSchema` into `build_url` and assert on the +resulting URL. +""" + +from datetime import date, datetime, timedelta, timezone + +import pytest +from pydantic import ValidationError + +from akd_ext.tools.worldview.permalink import ( + LayerSpec, + WorldviewPermalinkInputSchema, + WorldviewPermalinkTool, +) + + +def _build(**kwargs) -> str: + """Construct schema + build URL — the standard test-side call pattern.""" + return WorldviewPermalinkTool.build_url(WorldviewPermalinkInputSchema(**kwargs)) + + +def query_string(url: str) -> str: + """Return the part of the URL after `?`.""" + return url.split("?", 1)[1] + + +class TestLayerFormatting: + """LayerSpec → URL token formatting.""" + + def test_layerspec_id_is_required(self): + with pytest.raises(ValidationError): + LayerSpec(opacity=0.5) # type: ignore[call-arg] + + def test_all_defaults_renders_bare_id(self): + # LAYER_X is a non-base id, so pre-processing prepends a base and appends + # default reference overlays. LAYER_X itself still renders bare (no parens). + url = _build(layers=[LayerSpec(id="LAYER_X")]) + assert ",LAYER_X," in query_string(url) + # No layer in the resulting list has modifiers, so no parens anywhere. + assert "(" not in query_string(url) + + def test_hidden(self): + url = _build(layers=[LayerSpec(id="LAYER_X", hidden=True)]) + assert "LAYER_X(hidden)" in url + + def test_opacity(self): + url = _build(layers=[LayerSpec(id="LAYER_X", opacity=0.7)]) + assert "LAYER_X(opacity=0.7)" in url + + def test_palettes(self): + url = _build(layers=[LayerSpec(id="LAYER_X", palettes=["red", "blue"])]) + assert "LAYER_X(palettes=red,blue)" in url + + def test_style(self): + url = _build(layers=[LayerSpec(id="LAYER_X", style="vector_style")]) + assert "LAYER_X(style=vector_style)" in url + + def test_min_max(self): + url = _build(layers=[LayerSpec(id="LAYER_X", min=0, max=100)]) + assert "LAYER_X(min=0,max=100)" in url + + def test_squash(self): + url = _build(layers=[LayerSpec(id="LAYER_X", squash=True)]) + assert "LAYER_X(squash)" in url + + def test_multiple_modifiers_use_documented_token_order(self): + # _format_layer order: hidden, opacity, palettes, style, min, max, squash + url = _build( + layers=[ + LayerSpec( + id="LAYER_X", + hidden=True, + opacity=0.5, + palettes=["red"], + squash=True, + ) + ] + ) + assert "LAYER_X(hidden,opacity=0.5,palettes=red,squash)" in url + + def test_multiple_layers_comma_joined(self): + # Both LAYER_A and LAYER_B are non-base; canonical reorder keeps user-supplied + # order within the overlay partition. + url = _build(layers=[LayerSpec(id="LAYER_A"), LayerSpec(id="LAYER_B", opacity=0.5)]) + assert "LAYER_A,LAYER_B(opacity=0.5)" in url + + +class TestTimeFormatting: + """Time conversion behaviour via the `time` param.""" + + def test_date_emits_daily_form(self): + url = _build(layers=[LayerSpec(id="L")], time=date(2025, 9, 15)) + assert "t=2025-09-15" in url + assert "T" not in url.split("t=")[1].split("&")[0] + + def test_datetime_with_time_emits_subdaily(self): + url = _build( + layers=[LayerSpec(id="L")], + time=datetime(2025, 9, 15, 12, 30, 45), + ) + assert "t=2025-09-15T12:30:45Z" in url + + def test_datetime_at_midnight_emits_daily(self): + url = _build( + layers=[LayerSpec(id="L")], + time=datetime(2025, 9, 15, 0, 0, 0), + ) + t_segment = url.split("t=")[1].split("&")[0] + assert t_segment == "2025-09-15" + + def test_tz_aware_datetime_normalised_to_utc(self): + est = timezone(timedelta(hours=-5)) + dt = datetime(2025, 9, 15, 12, 0, 0, tzinfo=est) + url = _build(layers=[LayerSpec(id="L")], time=dt) + assert "t=2025-09-15T17:00:00Z" in url + + def test_string_iso_date(self): + url = _build(layers=[LayerSpec(id="L")], time="2025-09-15") + assert "t=2025-09-15" in url + + def test_string_human_readable(self): + url = _build(layers=[LayerSpec(id="L")], time="September 15, 2025") + assert "t=2025-09-15" in url + + def test_string_slash_form(self): + url = _build(layers=[LayerSpec(id="L")], time="2025/09/15") + assert "t=2025-09-15" in url + + def test_string_tz_aware_iso_normalises_to_utc(self): + url = _build(layers=[LayerSpec(id="L")], time="2025-09-15T12:00:00-05:00") + assert "t=2025-09-15T17:00:00Z" in url + + def test_unparseable_string_raises(self): + with pytest.raises(ValueError, match="banana"): + _build(layers=[LayerSpec(id="L")], time="banana") + + def test_none_defaults_to_yesterday_utc(self): + # build_url defaults `time` to yesterday (UTC) to avoid Worldview's + # partially-rendered "today" scenes. + url = _build(layers=[LayerSpec(id="L")]) + yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).date().isoformat() + assert f"t={yesterday}" in url + + +class TestCoreParams: + """bbox, projection, rotation.""" + + def test_bbox_round_trip(self): + url = _build( + layers=[LayerSpec(id="L")], + bbox=(-125, 32, -114, 42), + ) + assert "v=-125,32,-114,42" in url + + def test_projection_default_is_geographic(self): + url = _build(layers=[LayerSpec(id="L")]) + assert "p=geographic" in url + + def test_projection_arctic_with_rotation(self): + url = _build( + layers=[LayerSpec(id="L")], + projection="arctic", + rotation=45, + ) + assert "p=arctic" in url + assert "r=45" in url + + def test_no_rotation_omits_param(self): + url = _build(layers=[LayerSpec(id="L")]) + assert "r=" not in query_string(url) + + def test_embed_mode_always_emitted(self): + # Embed mode is unconditional — em=true must appear on every URL so + # the link renders cleanly in chat / iframe contexts. + url = _build(layers=[LayerSpec(id="L")]) + assert "em=true" in url + + +class TestCompareMode: + """compare_active gate behaviour.""" + + def test_gate_on_a_side_emits_full_block(self): + url = _build( + layers=[LayerSpec(id="L_A")], + time="2025-09-15", + compare_active=True, + compare_layers=[LayerSpec(id="L_B")], + compare_time="2025-09-14", + compare_mode="swipe", + compare_value=50, + ) + # B-state list is also pre-processed: base prepended, refs appended. + assert ",L_B," in url + assert "t1=2025-09-14" in url + assert "ca=true" in url + assert "cm=swipe" in url + assert "cv=50" in url + + def test_gate_on_b_side_emits_ca_false(self): + url = _build( + layers=[LayerSpec(id="L_A")], + compare_active=False, + compare_layers=[LayerSpec(id="L_B")], + ) + assert "ca=false" in url + assert ",L_B," in url + + def test_gate_off_short_circuits_stray_args(self): + url = _build( + layers=[LayerSpec(id="L_A")], + compare_active=None, + compare_layers=[LayerSpec(id="L_B")], + compare_time="2025-09-14", + compare_mode="opacity", + compare_value=80, + ) + qs = query_string(url) + assert "l1=" not in qs + assert "t1=" not in qs + assert "ca=" not in qs + assert "cm=" not in qs + assert "cv=" not in qs + + def test_compare_time_is_optional_when_gate_on(self): + url = _build( + layers=[LayerSpec(id="L_A")], + compare_active=True, + compare_layers=[LayerSpec(id="L_B")], + ) + assert "ca=true" in url + assert ",L_B," in url + assert "t1=" not in query_string(url) + + +class TestChartingMode: + """chart_active gate behaviour.""" + + def test_gate_on_emits_full_block(self): + url = _build( + layers=[LayerSpec(id="L")], + chart_active=True, + chart_layer="L_CHART", + chart_area=(-125, 32, -114, 42), + chart_time_start="2025-09-01", + chart_time_end="2025-09-30", + chart_autoload=True, + ) + assert "cha=true" in url + assert "chl=L_CHART" in url + assert "chc=-125,32,-114,42" in url + assert "cht=2025-09-01" in url + assert "cht2=2025-09-30" in url + assert "chch=true" in url + + def test_gate_off_short_circuits_stray_args(self): + url = _build( + layers=[LayerSpec(id="L")], + chart_active=False, + chart_layer="L_CHART", + chart_area=(-125, 32, -114, 42), + chart_time_start="2025-09-01", + chart_autoload=True, + ) + qs = query_string(url) + assert "cha=" not in qs + assert "chl=" not in qs + assert "chc=" not in qs + assert "cht=" not in qs + assert "cht2=" not in qs + assert "chch=" not in qs + + def test_chart_autoload_default_false(self): + url = _build( + layers=[LayerSpec(id="L")], + chart_active=True, + chart_layer="L_CHART", + ) + assert "chch=false" in url + + +class TestLayerPreprocessing: + """Unconditional pre-processing: auto-add base, auto-append default reference + overlays, canonical reorder. Same logic applies to compare_layers.""" + + def _layer_list(self, url: str, key: str = "l") -> list[str]: + """Extract the comma-separated layer ids from `?=...&...`.""" + return url.split(f"{key}=")[1].split("&")[0].split(",") + + @pytest.mark.parametrize( + "base_id", + [ + "MODIS_Terra_CorrectedReflectance_TrueColor", + "MODIS_Aqua_CorrectedReflectance_TrueColor", + "VIIRS_SNPP_CorrectedReflectance_TrueColor", + "VIIRS_NOAA20_CorrectedReflectance_TrueColor", + "VIIRS_NOAA21_CorrectedReflectance_TrueColor", + ], + ) + def test_known_base_layer_is_recognised(self, base_id): + # When the user supplies any known base, no second base is auto-prepended. + url = _build(layers=[LayerSpec(id=base_id)]) + ids = self._layer_list(url) + bases_in_url = [ + i + for i in ids + if i.split("(")[0] + in { + "MODIS_Terra_CorrectedReflectance_TrueColor", + "MODIS_Aqua_CorrectedReflectance_TrueColor", + "VIIRS_SNPP_CorrectedReflectance_TrueColor", + "VIIRS_NOAA20_CorrectedReflectance_TrueColor", + "VIIRS_NOAA21_CorrectedReflectance_TrueColor", + } + ] + assert bases_in_url == [base_id] + + def test_auto_add_base_when_missing(self): + # MODIS_Aqua_AOD is an overlay — pre-processor must prepend the default base. + url = _build(layers=[LayerSpec(id="MODIS_Aqua_AOD")]) + ids = self._layer_list(url) + assert ids[0] == "MODIS_Terra_CorrectedReflectance_TrueColor" + assert "MODIS_Aqua_AOD" in ids + + def test_auto_append_default_reference_overlays(self): + url = _build(layers=[LayerSpec(id="MODIS_Aqua_AOD")]) + assert "Coastlines_15m" in url + assert "Reference_Features_15m" in url + + def test_partial_reference_overlay_already_present_only_missing_appended(self): + # User supplies Coastlines_15m themselves; pre-processor must not duplicate it, + # but must still append the missing Reference_Features_15m. + url = _build(layers=[LayerSpec(id="MODIS_Aqua_AOD"), LayerSpec(id="Coastlines_15m")]) + ids = self._layer_list(url) + assert ids.count("Coastlines_15m") == 1 + assert "Reference_Features_15m" in ids + + def test_canonical_reorder_baselayers_first(self): + # User supplies overlay before base; pre-processor moves base to front and + # preserves user-supplied order within the overlay partition. + url = _build( + layers=[ + LayerSpec(id="MODIS_Aqua_AOD"), # overlay + LayerSpec(id="VIIRS_NOAA21_CorrectedReflectance_TrueColor"), # base + LayerSpec(id="MODIS_Terra_AOD"), # overlay + ] + ) + ids = self._layer_list(url) + assert ids[0] == "VIIRS_NOAA21_CorrectedReflectance_TrueColor" + assert ids.index("MODIS_Aqua_AOD") < ids.index("MODIS_Terra_AOD") + assert "Coastlines_15m" in ids + assert "Reference_Features_15m" in ids + + def test_compare_layers_also_get_pre_processing(self): + # compare_layers (B-state) gets the same pre-processing as `layers`. + url = _build( + layers=[LayerSpec(id="L_A")], + compare_active=True, + compare_layers=[LayerSpec(id="L_B")], + ) + b_ids = self._layer_list(url, key="l1") + assert b_ids[0] == "MODIS_Terra_CorrectedReflectance_TrueColor" + assert "L_B" in b_ids + assert "Coastlines_15m" in b_ids + assert "Reference_Features_15m" in b_ids diff --git a/uv.lock b/uv.lock index 02375dc..140b5a1 100644 --- a/uv.lock +++ b/uv.lock @@ -204,6 +204,7 @@ dependencies = [ { name = "openai-agents" }, { name = "pydantic-ai" }, { name = "pygithub" }, + { name = "python-dateutil" }, ] [package.optional-dependencies] @@ -237,6 +238,7 @@ requires-dist = [ { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=1.0.0" }, { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=6.0.0" }, + { name = "python-dateutil", specifier = ">=2.8" }, ] provides-extras = ["dev"]