Implement WMTS endpoints

example/app.py

@@ -89,6 +89,12 @@ def api_home(endpoint):
             <li><a href="?SERVICE=WMS&VERSION=1.3.0&REQUEST=GetLegendGraphic&LAYER={test_layer}&STYLE=default&FORMAT=image/png">WMS GetLegend Example (PNG)</a> <i>(v1.3.0)</i></li>
         </ul>
         </li>
+        <li> WMTS: Open Geospatial Consortium (OGC) Web Map Tile Service (WMTS) <i>(v1.0.0)</i>
+        <ul>
+            <li><a href="?SERVICE=WMTS&REQUEST=GetCapabilities&VERSION=1.0.0">WMTS GetCapabilities (XML)</a> <i>(v1.0.0)</i></li>
+            <li><a href="?SERVICE=WMTS&REQUEST=GetTile&VERSION=1.0.0&LAYER={test_layer}&STYLES=&FORMAT=image%2Fpng&TILEMATRIXSET=WebMercatorQuad&TILEMATRIX=0&TILEROW=0&TILECOL=0">WMTS GetTile Example (PNG)</a> <i>(v1.0.0)</i></li>
+        </ul>
+        </li>
         <li> EDR: Open Geospatial Consortium (OGC) Environmental Data Retrieval (EDR) <i>(v1.0.1)</i>
         <ul>
             <li><a href="{endpoint}/edr?f=html">EDR Landing Page (HTML)</a> <i>(v1.0.1)</i></li>

ogc/core.py

@@ -7,15 +7,17 @@
 import gc
 import logging
 import traitlets as tl
+from typing import Dict, Any
 
 from . import settings
 from . import wcs_request_1_0_0
 from . import wms_request_1_3_0
 from . import wcs_response_1_0_0
 from . import wms_response_1_3_0
 from .edr import EdrRoutes
+from .wmts import WmtsRoutes
 
-from ogc.ogc_common import WCSException
+from ogc.ogc_common import WCSException, WMTSException
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,17 @@ def __init__(self, layers=None, **kwargs):
             if settings.EDR_ENABLED
             else None
         )
+        self.wmts_routes = (
+            WmtsRoutes(
+                coverages=coverages,
+                base_url=self.base_url,
+                service_title=self.service_title,
+                service_abstract=self.service_abstract,
+                service_group_title=self.service_group_title,
+            )
+            if settings.WMTS_ENABLED
+            else None
+        )
 
     def get_coverage_from_id(self, identifier):
         for coverage in self.wcs_capabilities.coverages:
@@ -280,3 +293,30 @@ def handle_wms_kv(self, args):
             return self.get_map(args, wms_request)
 
         raise WCSException(exception_text="KV Request not handled properly: " + str(args))
+
+    def handle_wmts_kv(self, args: Dict[str, Any]) -> Dict[str, Any] | str:
+        """Handle WMTS key value server requests if support is enabled.
+
+        Parameters
+        ----------
+        args : Dict[str, Any]
+            The filtered request arguments.
+
+        Returns
+        -------
+        Dict[str, Any] | str
+            A dictionary containing the tile response or a string of service metadata.
+
+        Raises
+        ------
+        WMTSException
+            Exception for errors related to WMTS operations.
+        """
+        if self.wmts_routes is None:
+            raise WMTSException(
+                exception_code="OperationNotSupported",
+                locator="REQUEST",
+                exception_text="Unsupported request",
+            )
+
+        return self.wmts_routes.handle_kv(args)

ogc/edr/test/conftest.py

@@ -1,8 +1,10 @@
-import os
 import pytest
 import numpy as np
 import datetime
+import importlib
 import podpac
+from unittest.mock import patch
+from ogc import settings
 from ogc import podpac as pogc
 from typing import Dict, List, Any
 from ogc.settings import EDR_TIME_INSTANCE_DIMENSION
@@ -49,10 +51,15 @@
 )
 
 
-@pytest.fixture(scope="session", autouse=True)
+@pytest.fixture(scope="module", autouse=True)
 def set_env_vars():
-    """Setup the environmental variables for the session to support EDR."""
-    os.environ["OGC_SUPPORTED_FORMATS"] = "edr"
+    """Setup the environmental variables for the module to support EDR."""
+    with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": "edr"}):
+        importlib.reload(settings)
+        yield
+
+    # Fix imports after patching for test
+    importlib.reload(settings)
 
 
 @pytest.fixture()

ogc/ogc_common.py

@@ -1,9 +1,12 @@
 import logging
+import string
 
 import lxml
 import lxml.etree
 import numpy as np
 import traitlets as tl
+from xml.sax.saxutils import escape
+from typing import Any
 
 logger = logging.getLogger(__file__)
 
@@ -15,6 +18,28 @@
 )
 
 
+class EscapeFormatter(string.Formatter):
+    """Formatter that escapes all values before formatting."""
+
+    def format_field(self, value: Any, format_spec: str) -> str:
+        """Format a single field with escaping applied.
+
+        Parameters
+        ----------
+        value : Any
+            The value to be escaped and formatted.
+        format_spec : str
+            The format specification.
+
+        Returns
+        -------
+        str
+            The escaped and formatted string representation of the value.
+        """
+        escaped_value = escape(str(value))
+        return format(escaped_value, format_spec)
+
+
 class XMLNode(tl.HasTraits):
     """Base class for all Traits objects that correspond to XML schemas."""
 
@@ -126,3 +151,23 @@ def to_xml(self):
 </ExceptionReport>
 """.format(self=self, exception_text=exception_text)
         return xml
+
+
+class WMTSException(WCSException):
+    """
+    WMTS Exception based on the WCS Exception.
+    This ensures it can still be caught as a WCS Exception which is the baseline handled server exception.
+    """
+
+    def __init__(
+        self,
+        exception_text="Internal application error.",
+        exception_code="NoApplicableCode",
+        locator="",
+    ):
+        """
+        exception_code: 'NoApplicableCode', 'VersionNegotiationFailed',
+        'InvalidUpdateSequence', 'MissingParameterValue', 'InvalidParameterValue',
+        'OperationNotSupported', 'TileOutOfRange'
+        """
+        super().__init__(exception_text, exception_code, locator)

ogc/servers.py

@@ -76,6 +76,19 @@ def home(endpoint):
         else ""
     )
 
+    wmts_list_item = (
+        f"""
+        <li> WMTS: Open Geospatial Consortium (OGC) Web Map Tile Service (WMTS) <i>(v1.0.0)</i>
+        <ul>
+            <li><a href="?SERVICE=WMTS&REQUEST=GetCapabilities&VERSION=1.0.0">WMTS GetCapabilities (XML)</a> <i>(v1.0.0)</i></li>
+            <li><a href="?SERVICE=WMTS&REQUEST=GetTile&VERSION=1.0.0&LAYER={test_layer}&STYLES=&FORMAT=image%2Fpng&TILEMATRIXSET=WebMercatorQuad&TILEMATRIX=0&TILEROW=0&TILECOL=0">WMTS GetTile Example (PNG)</a> <i>(v1.0.0)</i></li>
+        </ul>
+        </li>
+    """
+        if settings.WMTS_ENABLED
+        else ""
+    )
+
     edr_list_item = (
         f"""
         <li> EDR: Open Geospatial Consortium (OGC) Environmental Data Retrieval (EDR) <i>(v1.0.1)</i>
@@ -96,6 +109,7 @@ def home(endpoint):
     <ul>
         {wcs_list_item}
         {wms_list_item}
+        {wmts_list_item}
         {edr_list_item}
     </ul>
     """
@@ -278,6 +292,8 @@ def ogc_render(self, ogc_idx):
                 ogc_response = ogc.handle_wcs_kv(args)
             elif args["service"].lower() == "wms" and settings.WMS_ENABLED:
                 ogc_response = ogc.handle_wms_kv(args)
+            elif args["service"].lower() == "wmts" and settings.WMTS_ENABLED:
+                ogc_response = ogc.handle_wmts_kv(args)
             if ogc_response is not None:
                 if isinstance(ogc_response, six.string_types):
                     return respond_xml(ogc_response, status=200)

ogc/settings.py

@@ -39,6 +39,12 @@
     crs_84h_uri_format: {"minx": -180.0, "miny": -90.0, "maxx": 180.0, "maxy": 90.0},
 }
 
+# WMTS tiling parameters
+EARTH_CIRCUMFERENCE_METERS = 40075016.686
+WMTS_TILE_SIZE = 256  # pixels
+WMTS_PIXEL_SIZE_METERS = 0.00028  # meters/pixel screen equivalent
+WMTS_INITIAL_RESOLUTION = EARTH_CIRCUMFERENCE_METERS / WMTS_TILE_SIZE  # meters/pixel for the crs bounds (global)
+
 # EDR query output formats
 GEOTIFF = "GeoTIFF"
 JSON = "JSON"
@@ -98,4 +104,5 @@
 OGC_SUPPORTED_FORMATS = os.environ.get("OGC_SUPPORTED_FORMATS", "wms,wcs")
 WMS_ENABLED = "wms" in OGC_SUPPORTED_FORMATS.lower()
 WCS_ENABLED = "wcs" in OGC_SUPPORTED_FORMATS.lower()
+WMTS_ENABLED = "wmts" in OGC_SUPPORTED_FORMATS.lower()
 EDR_ENABLED = "edr" in OGC_SUPPORTED_FORMATS.lower()

ogc/test/test_servers.py

@@ -3,7 +3,7 @@
 from ogc import podpac as pogc
 from ogc import settings
 from unittest.mock import patch
-from typing import Callable
+from typing import Callable, Generator
 
 import importlib
 import podpac
@@ -12,13 +12,13 @@
 
 
 @pytest.fixture
-def supported_formats() -> Callable[[str], None]:
+def supported_formats() -> Generator[Callable[[str], None], None, None]:
     """Fixture used to patch OGC supported formats.
 
     Returns
     -------
-    Callable[[str], None]
-        A function which patches the OGC supported formats based on input string.
+    Generator[Callable[[str], None], None, None]
+        A generator which yields a function which patches the OGC supported formats based on input string.
     """
 
     def _supported_formats(formats: str):
@@ -32,7 +32,10 @@ def _supported_formats(formats: str):
         with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": formats}):
             importlib.reload(settings)
 
-    return _supported_formats
+    yield _supported_formats
+
+    # Fix imports after patching for test
+    importlib.reload(settings)
 
 
 @pytest.fixture
@@ -136,6 +139,9 @@ def test_server_with_default_supported_services(client):
     response = client.get("/ogc?service=WCS&request=GetCapabilities")
     assert response.status_code == 200
 
+    response = client.get("/ogc?service=WMTS&request=GetCapabilities")
+    assert response.status_code == 400
+
     response = client.get("/ogc/edr")
     assert response.status_code == 404
 
@@ -152,6 +158,9 @@ def test_server_without_wcs_supported_service(supported_formats, client):
     response = client.get("/ogc?service=WCS&request=GetCapabilities")
     assert response.status_code == 400
 
+    response = client.get("/ogc?service=WMTS&request=GetCapabilities")
+    assert response.status_code == 400
+
     response = client.get("/ogc/edr")
     assert response.status_code == 404
 
@@ -168,5 +177,8 @@ def test_server_without_wms_supported_service(supported_formats, client):
     response = client.get("/ogc?service=WMS&request=GetCapabilities")
     assert response.status_code == 400
 
+    response = client.get("/ogc?service=WMTS&request=GetCapabilities")
+    assert response.status_code == 400
+
     response = client.get("/ogc/edr")
     assert response.status_code == 404

ogc/wmts/__init__.py

@@ -0,0 +1,3 @@
+from .wmts_routes import WmtsRoutes
+
+__all__ = ["WmtsRoutes"]

ogc/wmts/test/conftest.py

@@ -0,0 +1,78 @@
+import pytest
+import podpac
+import numpy as np
+import importlib
+from typing import List
+from ogc import settings
+from unittest.mock import patch
+from ogc import podpac as pogc
+from ogc.wms_response_1_3_0 import Coverage
+
+# Create some podpac nodes
+lat = np.linspace(90, -90, 11)
+lon = np.linspace(-180, 180, 21)
+time = np.array(["2025-10-24T12:00:00"], dtype="datetime64")
+data = np.random.default_rng(1).random((11, 21, 1))
+coords = podpac.Coordinates([lat, lon, time], dims=["lat", "lon", "time"])
+node1 = podpac.data.Array(source=data, coordinates=coords).interpolate()
+node2 = podpac.data.Array(source=data, coordinates=coords).interpolate()
+
+# Use podpac nodes to create some test OGC layers
+layer1 = pogc.Layer(
+    node=node1,
+    identifier="layer1",
+    title="Layer 1",
+    abstract="Layer1 Data",
+    group="Layers",
+)
+
+layer2 = pogc.Layer(
+    node=node2,
+    identifier="layer2",
+    title="Layer 2",
+    abstract="Layer2 Data",
+    group="Layers",
+)
+
+
+@pytest.fixture()
+def layers() -> List[pogc.Layer]:
+    """List of test layers.
+
+    Returns
+    -------
+    List[pogc.Layer]
+        The test layers.
+    """
+    return [layer1, layer2]
+
+
+@pytest.fixture()
+def coverages() -> List[Coverage]:
+    """List of coverages based on layers.
+
+    Returns
+    -------
+    List[Coverage]
+        The test coverages.
+    """
+    return [
+        Coverage(
+            layer=layer,
+            title=layer.title,
+            abstract=layer.abstract,
+            identifier=layer.id_str,
+        )
+        for layer in [layer1, layer2]
+    ]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def set_env_vars():
+    """Setup the environmental variables for the module to support WMTS."""
+    with patch.dict("os.environ", {"OGC_SUPPORTED_FORMATS": "wmts"}):
+        importlib.reload(settings)
+        yield
+
+    # Fix imports after patching for test
+    importlib.reload(settings)