diff --git a/example/app.py b/example/app.py
index b8e11e4..a8e7424 100755
--- a/example/app.py
+++ b/example/app.py
@@ -89,6 +89,12 @@ def api_home(endpoint):
WMS GetLegend Example (PNG) (v1.3.0)
+ WMTS: Open Geospatial Consortium (OGC) Web Map Tile Service (WMTS) (v1.0.0)
+
+
EDR: Open Geospatial Consortium (OGC) Environmental Data Retrieval (EDR) (v1.0.1)
- EDR Landing Page (HTML) (v1.0.1)
diff --git a/ogc/core.py b/ogc/core.py
index 776161a..d214d2f 100755
--- a/ogc/core.py
+++ b/ogc/core.py
@@ -7,6 +7,7 @@
import gc
import logging
import traitlets as tl
+from typing import Dict, Any
from . import settings
from . import wcs_request_1_0_0
@@ -14,8 +15,9 @@
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)
diff --git a/ogc/edr/test/conftest.py b/ogc/edr/test/conftest.py
index d566273..6da1229 100644
--- a/ogc/edr/test/conftest.py
+++ b/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()
diff --git a/ogc/ogc_common.py b/ogc/ogc_common.py
index 4bf3dd4..db091ad 100755
--- a/ogc/ogc_common.py
+++ b/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):
""".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)
diff --git a/ogc/servers.py b/ogc/servers.py
index 2c1093f..64dcc77 100755
--- a/ogc/servers.py
+++ b/ogc/servers.py
@@ -76,6 +76,19 @@ def home(endpoint):
else ""
)
+ wmts_list_item = (
+ f"""
+ - WMTS: Open Geospatial Consortium (OGC) Web Map Tile Service (WMTS) (v1.0.0)
+
+
+ """
+ if settings.WMTS_ENABLED
+ else ""
+ )
+
edr_list_item = (
f"""
- EDR: Open Geospatial Consortium (OGC) Environmental Data Retrieval (EDR) (v1.0.1)
@@ -96,6 +109,7 @@ def home(endpoint):
{wcs_list_item}
{wms_list_item}
+ {wmts_list_item}
{edr_list_item}
"""
@@ -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)
diff --git a/ogc/settings.py b/ogc/settings.py
index 5387048..ee516a6 100755
--- a/ogc/settings.py
+++ b/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()
diff --git a/ogc/test/test_servers.py b/ogc/test/test_servers.py
index 9d96229..a58e6fb 100644
--- a/ogc/test/test_servers.py
+++ b/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
diff --git a/ogc/wmts/__init__.py b/ogc/wmts/__init__.py
new file mode 100644
index 0000000..3aeb15c
--- /dev/null
+++ b/ogc/wmts/__init__.py
@@ -0,0 +1,3 @@
+from .wmts_routes import WmtsRoutes
+
+__all__ = ["WmtsRoutes"]
diff --git a/ogc/wmts/test/conftest.py b/ogc/wmts/test/conftest.py
new file mode 100644
index 0000000..a331778
--- /dev/null
+++ b/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)
diff --git a/ogc/wmts/test/test_wmts_request_1_0_0.py b/ogc/wmts/test/test_wmts_request_1_0_0.py
new file mode 100644
index 0000000..0db2160
--- /dev/null
+++ b/ogc/wmts/test/test_wmts_request_1_0_0.py
@@ -0,0 +1,217 @@
+import pytest
+from typing import Dict, Any
+from ogc import settings
+from ogc.wmts.wmts_request_1_0_0 import GetCapabilities, GetTile, WebMercatorQuad, WorldCRS84Quad, TileMatrixSet
+
+
+def make_valid_tile_arguments() -> Dict[str, Any]:
+ """Create a valid arguments dictionary for a get tile request.
+
+ Returns
+ -------
+ Dict[str, Any]
+ Valid dictionary for a get tile request.
+ """
+ return {
+ "request": "GetTile",
+ "service": "wmts",
+ "version": "1.0.0",
+ "layer": "layer",
+ "tilematrixset": "WebMercatorQuad",
+ "tilerow": "0",
+ "tilecol": "0",
+ "tilematrix": "0",
+ "format": "image/png",
+ }
+
+
+def make_valid_capabilities_arguments() -> Dict[str, Any]:
+ """Create a valid arguments dictionary for a get capabilities request.
+
+ Returns
+ -------
+ Dict[str, Any]
+ Valid dictionary for a get capabilities request.
+ """
+ return {
+ "request": "GetCapabilities",
+ "service": "wmts",
+ "version": "1.0.0",
+ }
+
+
+def test_get_capabilities_validate_success():
+ """Validate a correct GetCapabilities request."""
+ capabilities = GetCapabilities()
+ capabilities.load_from_kv(make_valid_capabilities_arguments())
+ capabilities.validate()
+
+
+def test_getcapabilities_validate_invalid_service():
+ """Ensure validation fails when service is not WMTS."""
+ capabilities = GetCapabilities()
+ capabilities.load_from_kv(make_valid_capabilities_arguments())
+ capabilities.service = "WMS"
+
+ with pytest.raises(AssertionError):
+ capabilities.validate()
+
+
+def test_getcapabilities_load_from_kv_invalid_request():
+ """Ensure invalid request type raises assertion."""
+ capabilities = GetCapabilities()
+ args = make_valid_capabilities_arguments()
+ args["request"] = "invalid"
+
+ with pytest.raises(AssertionError):
+ capabilities.load_from_kv(args)
+
+
+def test_get_tile_validate_success():
+ """Validate a correct GetTile request."""
+ tile = GetTile()
+ tile.load_from_kv(make_valid_tile_arguments())
+ tile.validate()
+
+
+def test_get_tile_validate_missing_layer():
+ """Ensure validation fails when layer is missing."""
+ tile = GetTile()
+ args = make_valid_tile_arguments()
+ args["layer"] = None
+ tile.load_from_kv(args)
+
+ with pytest.raises(AssertionError):
+ tile.validate()
+
+
+def test_get_tile_validate_invalid_tile_row():
+ """Ensure tile row outside valid range raises error."""
+ tile = GetTile()
+ args = make_valid_tile_arguments()
+ args["tilerow"] = "-1"
+ tile.load_from_kv(args)
+
+ with pytest.raises(AssertionError):
+ tile.validate()
+
+
+@pytest.mark.parametrize(
+ "input_number,expected",
+ [
+ (1.234567891234, "1.234567891"),
+ (10, "10"),
+ ],
+)
+def test_format_number(input_number: float | int, expected: str):
+ """Validate number formatting logic.
+
+ Parameters
+ ----------
+ input_number : float | int
+ Input number to format.
+ expected : str
+ Expected string output.
+ """
+ result = TileMatrixSet().format_number(input_number)
+ assert result == expected
+
+
+def test_convert_to_map_args_structure():
+ """Ensure GetTile converts correctly to map request args."""
+ tile = GetTile()
+ tile.load_from_kv(make_valid_tile_arguments())
+ tile.params = {"time": "0"}
+ tile.tile_matrix_set = WebMercatorQuad()
+
+ result = tile.convert_to_map_args()
+
+ assert "bbox" in result
+ assert result["service"] == "WMTS"
+ assert result["crs"] == tile.tile_matrix_set.crs
+ assert result["height"] == settings.WMTS_TILE_SIZE
+ assert result["width"] == settings.WMTS_TILE_SIZE
+ assert result["format"] == "image/png"
+ assert result["PARAMS"] == '{"time": "0"}'
+
+
+def test_bbox_web_mercator_returns_valid_string():
+ """Validate Web Mercator bounding box output format."""
+ tile = GetTile()
+ tile.load_from_kv(make_valid_tile_arguments())
+ tile.tile_matrix_set = WebMercatorQuad(row=0, column=0, matrix_level=0)
+ bbox = tile.tile_matrix_set.calculate_eval_bounding_box()
+
+ assert isinstance(bbox, str)
+ assert len(bbox.split(",")) == 4
+
+
+def test_bbox_world_crs84_returns_valid_string():
+ """Validate CRS84 bounding box output format."""
+ tile = GetTile()
+ tile.load_from_kv(make_valid_tile_arguments())
+ tile.tile_matrix_set = WorldCRS84Quad(row=0, column=0, matrix_level=0)
+ bbox = tile.tile_matrix_set.calculate_eval_bounding_box()
+
+ assert isinstance(bbox, str)
+ assert len(bbox.split(",")) == 4
+
+
+@pytest.mark.parametrize(
+ "tile_matrix_set, matrix_level, expected_scale_denominator",
+ [
+ (WebMercatorQuad(), 0, 559082264),
+ (WebMercatorQuad(), 1, 279541132),
+ (WorldCRS84Quad(), 0, 279541132),
+ (WorldCRS84Quad(), 1, 139770566),
+ ],
+)
+def test_calculate_scale_denominator(
+ tile_matrix_set: TileMatrixSet, matrix_level: int, expected_scale_denominator: float
+):
+ """Ensure scale denominator scales correctly by zoom level.
+
+ Parameters
+ ----------
+ tile_matrix_set : TileMatrixSet
+ The tile matrix set.
+ matrix_level: int
+ The zoom level for the calculation.
+ expected_scale_denominator : float
+ Expected scale denominator.
+ """
+ result = tile_matrix_set.calculate_scale_denominator(matrix_level)
+ assert float(result) == pytest.approx(expected_scale_denominator, rel=1e-6)
+
+
+@pytest.mark.parametrize(
+ "tile_matrix_set, matrix_level, expected_width, expected_height",
+ [
+ (WebMercatorQuad(), 0, "1", "1"),
+ (WebMercatorQuad(), 1, "2", "2"),
+ (WebMercatorQuad(), 2, "4", "4"),
+ (WorldCRS84Quad(), 0, "2", "1"),
+ (WorldCRS84Quad(), 1, "4", "2"),
+ (WorldCRS84Quad(), 2, "8", "4"),
+ ],
+)
+def test_calculate_matrix_sizes(
+ tile_matrix_set: TileMatrixSet, matrix_level: int, expected_width: str, expected_height: str
+):
+ """Calculate matrix sizes based on zoom level.
+
+ Parameters
+ ----------
+ tile_matrix_set : TileMatrixSet
+ The tile matrix set.
+ matrix_level: int
+ The zoom level for the calculation.
+ expected_width : str
+ The expected matrix width.
+ expected_height : str
+ The expected matrix height.
+ """
+ width, height = tile_matrix_set.calculate_matrix_sizes(matrix_level)
+
+ assert width == expected_width
+ assert height == expected_height
diff --git a/ogc/wmts/test/test_wmts_response_1_0_0.py b/ogc/wmts/test/test_wmts_response_1_0_0.py
new file mode 100644
index 0000000..ed35059
--- /dev/null
+++ b/ogc/wmts/test/test_wmts_response_1_0_0.py
@@ -0,0 +1,101 @@
+import re
+import pytest
+from ogc.wmts.wmts_response_1_0_0 import Capabilities, SERVICE_VERSION
+
+
+def contains_text(xml_str: str, expected: str) -> bool:
+ """Check if a string is contained in XML data.
+ Uses the entire XML string in single search without extra whitespace.
+
+ Parameters
+ ----------
+ xml_str : str
+ The XML data as a string.
+ expected : str
+ The string to check in the XML data.
+
+ Returns
+ -------
+ bool
+ True if contained, False otherwise.
+ """
+ return expected in re.sub(r"\s+", " ", xml_str).strip()
+
+
+def make_capabilities() -> Capabilities:
+ """Create a valid Capabilities object.
+
+ Returns
+ -------
+ Capabilities
+ Initialized capabilities instance.
+ """
+ capabilities = Capabilities(service_title="Test Service", service_abstract="Abstract", service_keywords=["Keyword"])
+ capabilities.coverages = []
+ return capabilities
+
+
+def test_to_xml_contains_required_sections():
+ """Ensure main XML output contains required top-level sections."""
+ capabilities = make_capabilities()
+ xml = capabilities.to_xml()
+ assert contains_text(xml, "")
+ assert contains_text(xml, f"{capabilities.service_type}")
+ assert contains_text(xml, f"{SERVICE_VERSION}")
+
+
+def test_operations_metadata_contains_supported_requests():
+ """Ensure supported requests are properly defined in operations metadata."""
+ capabilities = make_capabilities()
+ xml = capabilities._operations_metadata()
+
+ assert contains_text(xml, '')
+ assert contains_text(xml, '')
+ assert not contains_text(xml, '')
+
+
+def test_bounding_box_tile_matrix_set():
+ """Validate tile matrix set bounding box generation."""
+ capabilities = make_capabilities()
+ xml = capabilities._tile_matrix_sets()
+
+ assert contains_text(xml, "")
+ assert contains_text(xml, "")
+ assert contains_text(xml, "")
+
+
+@pytest.mark.parametrize(
+ "value,expected",
+ [
+ ("2026-05-01 11:28:45", True),
+ ("2026-05-01T11:28:45", True),
+ ("2026-05-01", True),
+ ("invalid", False),
+ ("10", False),
+ ],
+)
+def test_is_iso_datetime(value: str, expected: bool):
+ """Validate datetime detection.
+
+ Parameters
+ ----------
+ value : str
+ Input string.
+ expected : bool
+ Expected result.
+ """
+ assert Capabilities._is_iso_datetime(value) == expected
diff --git a/ogc/wmts/test/test_wmts_routes.py b/ogc/wmts/test/test_wmts_routes.py
new file mode 100644
index 0000000..c71153c
--- /dev/null
+++ b/ogc/wmts/test/test_wmts_routes.py
@@ -0,0 +1,183 @@
+import pytest
+import struct
+from typing import List, Dict, Any
+from ogc import core
+from ogc import podpac as pogc
+from ogc.wmts.wmts_routes import WmtsRoutes
+from ogc.wms_response_1_3_0 import Coverage
+
+
+def make_valid_tile_arguments(coverages: List[Coverage]) -> Dict[str, Any]:
+ """Create a valid arguments dictionary for a get tile request.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages to get a layer from.
+
+ Returns
+ -------
+ Dict[str, Any]
+ Valid dictionary for a get tile request.
+ """
+ return {
+ "request": "GetTile",
+ "service": "WMTS",
+ "version": "1.0.0",
+ "layer": coverages[0].layer.identifier,
+ "format": "image/png",
+ "tilerow": 0,
+ "tilecol": 0,
+ "tilematrixset": "WebMercatorQuad",
+ "tilematrix": "0",
+ "time": str(coverages[0].layer.get_coordinates()["time"].coordinates[0]),
+ }
+
+
+def test_handle_wmts_kv_get_capabilities_from_ogc_core(layers: List[pogc.Layer]):
+ """Test the handle_wmts_kv method of the OGC class with a valid WMTS GetCapabilities request.
+
+ Parameters
+ ----------
+ layers : List[pogc.Layer]
+ Layers provided by a test fixture.
+ """
+ ogc = core.OGC(layers=layers)
+ args = {
+ "request": "GetCapabilities",
+ "service": "WMTS",
+ "version": "1.0.0",
+ "base_url": "/",
+ }
+ response = ogc.handle_wmts_kv(args)
+ assert isinstance(response, str)
+ assert "Capabilities" in response and "wmts/1.0" in response
+
+
+def test_handle_kv_get_capabilities(coverages: List[Coverage]):
+ """Test the handle_wmts_kv method of the WMTS routes class with a valid GetCapabilities request.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages provided by a test fixture.
+ """
+ wmts_routes = WmtsRoutes(coverages=coverages)
+ args = {
+ "request": "GetCapabilities",
+ "service": "WMTS",
+ "version": "1.0.0",
+ "base_url": None,
+ }
+ response = wmts_routes.handle_kv(args)
+ assert isinstance(response, str)
+ assert "Capabilities" in response and "wmts/1.0" in response
+
+
+def test_handle_kv_get_capabilities_invalid_service(coverages: List[Coverage]):
+ """Test the handle_kv method of the WMTS routes class with an invalid GetCapabilities request.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages provided by a test fixture.
+ """
+ wmts_routes = WmtsRoutes(coverages=coverages)
+ args = {
+ "request": "GetCapabilities",
+ "service": "InvalidService",
+ "version": "1.0.0",
+ "base_url": None,
+ }
+
+ with pytest.raises(core.WMTSException):
+ wmts_routes.handle_kv(args)
+
+
+def test_handle_kv_get_feature_info_unsupported(coverages: List[Coverage]):
+ """Test the handle_kv method of the WMTS routes class with an unsupported GetFeatureInfo request.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages provided by a test fixture.
+ """
+ wmts_routes = WmtsRoutes(coverages=coverages)
+ args = {
+ "request": "GetFeatureInfo",
+ "service": "WMTS",
+ "version": "1.0.0",
+ "coverage": coverages[0].layer.identifier,
+ }
+
+ with pytest.raises(core.WMTSException):
+ wmts_routes.handle_kv(args)
+
+
+def test_handle_kv_get_tile(coverages: List[Coverage]):
+ """Test the handle_kv method of the WMTS routes class with a GetTile request.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages provided by a test fixture.
+ """
+ wmts_routes = WmtsRoutes(coverages=coverages)
+ args = make_valid_tile_arguments(coverages)
+ response = wmts_routes.handle_kv(args)
+ assert isinstance(response["fn"], str)
+ assert response["fn"].endswith(".png")
+ assert response["fp"] is not None
+
+ # Check height and width without additional packages
+ response["fp"].seek(16)
+ chunk = response["fp"].read(8)
+ assert struct.unpack(">II", chunk) == (256, 256)
+
+
+def test_handle_kv_get_tile_invalid_version(coverages: List[Coverage]):
+ """Test the handle_kv method of the WMTS routes class with a GetTile request with an invalid version.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages provided by a test fixture.
+ """
+ wmts_routes = WmtsRoutes(coverages=coverages)
+ args = make_valid_tile_arguments(coverages)
+ args["version"] = "invalid"
+
+ with pytest.raises(core.WMTSException):
+ wmts_routes.handle_kv(args)
+
+
+def test_handle_kv_get_tile_invalid_tile_indices(coverages: List[Coverage]):
+ """Test the handle_kv method of the WMTS routes class with a GetTile request with an invalid tile row.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages provided by a test fixture.
+ """
+ wmts_routes = WmtsRoutes(coverages=coverages)
+ args = make_valid_tile_arguments(coverages)
+ args["tilerow"] = "-1"
+
+ with pytest.raises(core.WMTSException):
+ wmts_routes.handle_kv(args)
+
+
+def test_handle_kv_with_invalid_request(coverages: List[Coverage]):
+ """Test the handle_kv method of the WMTS routes class with a GetTile request with an invalid request.
+
+ Parameters
+ ----------
+ coverages : List[Coverage]
+ Coverages provided by a test fixture.
+ """
+ wmts_routes = WmtsRoutes(coverages=coverages)
+ args = make_valid_tile_arguments(coverages)
+ args["request"] = "invalid"
+
+ with pytest.raises(core.WMTSException):
+ wmts_routes.handle_kv(args)
diff --git a/ogc/wmts/wmts_request_1_0_0.py b/ogc/wmts/wmts_request_1_0_0.py
new file mode 100755
index 0000000..436b6e0
--- /dev/null
+++ b/ogc/wmts/wmts_request_1_0_0.py
@@ -0,0 +1,453 @@
+import logging
+import json
+import traitlets as tl
+from typing import Dict, Any, Tuple
+
+from ogc import ogc_common
+from ogc import settings
+from podpac.core.utils import VALID_DIMENSION_NAMES
+from ogc.wcs_request_1_0_0 import Identifier
+
+logger = logging.getLogger(__file__)
+escape_format = ogc_common.EscapeFormatter().format
+WMTS_VALIDATION_ERROR = "WMTS Request validation error: service should be WMTS"
+
+
+class TileMatrixSet(ogc_common.XMLNode):
+ """Tile matrix set defining how OGC tile data is available at different scales."""
+
+ name = tl.Unicode()
+ crs = tl.Unicode()
+ wkss = tl.Unicode()
+ bounds = tl.Dict(key_trait=tl.Unicode(), value_trait=tl.Float())
+ matrix_levels = tl.List(tl.Int(), default_value=list(range(0, 21)))
+ depth = tl.Int(default_value=2)
+ indent = " "
+
+ row = tl.Int()
+ column = tl.Int()
+ matrix_level = tl.Int()
+
+ def validate(self):
+ """Validate the tile matrix set."""
+ assert len(self.name) > 0, "WMTS Request validation error: tile matrix set name not found"
+ assert len(self.crs) > 0, "WMTS Request validation error: tile matrix set crs not found"
+ assert len(self.wkss) > 0, "WMTS Request validation error: tile matrix set wkss not found"
+ assert len(self.bounds) > 0, "WMTS Request validation error: tile matrix set bounds not found"
+ assert len(self.matrix_levels) > 0, "WMTS Request validation error: tile matrix levels not found"
+ assert self.matrix_level >= 0 and self.matrix_level <= max(
+ self.matrix_levels
+ ), "WMTS Request validation error: tile matrix level not in range"
+
+ def to_xml(self):
+ xml = ""
+ xml += self.indent * self.depth + "\n"
+ xml += self.indent * (self.depth + 1) + escape_format("{}\n", self.name)
+ xml += self.indent * (self.depth + 1) + escape_format("{}\n", self.crs)
+ xml += self._bounding_box_tile_matrix_set() if self.crs is not None else ""
+ xml += self.indent * (self.depth + 1) + escape_format("{}\n", self.wkss)
+ top_left_corner = self._top_left_corner() if self.crs is not None else ""
+
+ for matrix_level in self.matrix_levels:
+ scale_denominator = self.calculate_scale_denominator(matrix_level)
+ matrix_width, matrix_height = self.calculate_matrix_sizes(matrix_level) if self.crs is not None else ""
+ xml += self.indent * (self.depth + 1) + """\n"""
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{}\n""", matrix_level
+ )
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{}\n""", scale_denominator
+ )
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{}\n""", top_left_corner
+ )
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{}\n""", settings.WMTS_TILE_SIZE
+ )
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{}\n""", settings.WMTS_TILE_SIZE
+ )
+ xml += self.indent * (self.depth + 2) + escape_format("""{}\n""", matrix_width)
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{}\n""", matrix_height
+ )
+ xml += self.indent * (self.depth + 1) + """\n"""
+
+ xml += self.indent * self.depth + """\n"""
+ return xml
+
+ def calculate_eval_bounding_box(self) -> str:
+ """Determine the bounding box for evaluation."""
+ raise NotImplementedError
+
+ def _top_left_corner(self) -> str | None:
+ """Get the top left corner of the bounded area defined by the tile matrix set.
+
+ Returns
+ -------
+ str | None
+ The top left corner as a string or None if the CRS is not valid for WMTS.
+ """
+ if self.bounds is not None:
+ x_min = self.format_number(self.bounds["minx"])
+ y_max = self.format_number(self.bounds["maxy"])
+ return "{} {}".format(x_min, y_max)
+
+ return None
+
+ def _bounding_box_tile_matrix_set(self) -> str:
+ """Metadata for the bounding box for a specific tile matrix set based on the supported CRS.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ xml = ""
+ if self.bounds is not None:
+ xml += self.indent * (self.depth + 1) + """\n"""
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{x} {y}\n""",
+ x=self.format_number(self.bounds["minx"]),
+ y=self.format_number(self.bounds["miny"]),
+ )
+ xml += self.indent * (self.depth + 2) + escape_format(
+ """{x} {y}\n""",
+ x=self.format_number(self.bounds["maxx"]),
+ y=self.format_number(self.bounds["maxy"]),
+ )
+ xml += self.indent * (self.depth + 1) + """\n"""
+
+ return xml
+
+ @staticmethod
+ def calculate_scale_denominator(matrix_level: int) -> str:
+ """Calculate the scale denominator for the zoom level."""
+ raise NotImplementedError
+
+ @staticmethod
+ def calculate_matrix_sizes(matrix_level) -> Tuple[str, str]:
+ """Calculate the scale denominator for the matrix level."""
+ raise NotImplementedError
+
+ @staticmethod
+ def format_number(input_number: float | int, float_decimals: int = 9) -> str:
+ """Format a number into a string with specified decimal count.
+
+ Parameters
+ ----------
+ input_number : float | int
+ The number to convert to a string.
+ float_decimals : int, optional
+ The decimals to display in the string, by default 9.
+
+ Returns
+ -------
+ str
+ The number formatted as a string.
+ """
+ if isinstance(input_number, float):
+ return "{number:.{decimals}f}".format(number=input_number, decimals=float_decimals)
+ else:
+ return "{}".format(input_number)
+
+
+class WebMercatorQuad(TileMatrixSet):
+ """Tile matrix set for Web Mercator Quad format."""
+
+ name = tl.Unicode(default_value="WebMercatorQuad")
+ crs = tl.Unicode(default_value="urn:ogc:def:crs:EPSG::3857")
+ wkss = tl.Unicode(default_value="urn:ogc:def:wkss:OGC:1.0:GoogleMapsCompatible")
+ bounds = tl.Dict(
+ key_trait=tl.Unicode(),
+ value_trait=tl.Float(),
+ default_value={
+ "minx": -20037508.342789244,
+ "miny": -20037508.342789244,
+ "maxx": 20037508.342789244,
+ "maxy": 20037508.342789244,
+ },
+ )
+
+ def validate(self):
+ """Validate the tile matrix set."""
+ super().validate()
+ assert self.row >= 0 and self.row < 2**self.matrix_level, "WMTS Request validation error: invalid tile row"
+ assert (
+ self.column >= 0 and self.column < 2**self.matrix_level
+ ), "WMTS Request validation error: invalid tile column"
+
+ def calculate_eval_bounding_box(self) -> str:
+ """Determine the bounding box for evaluation.
+
+ A zero pixel row and column corresponds to the top left corner.
+
+ Returns
+ -------
+ str
+ The bounding box for the provided tile indices.
+ """
+ resolution = settings.WMTS_INITIAL_RESOLUTION / (2**self.matrix_level)
+ origin_x = self.bounds["minx"]
+ origin_y = self.bounds["maxy"]
+ tile_size_m = settings.WMTS_TILE_SIZE * resolution
+
+ min_x = self.column * tile_size_m + origin_x
+ max_x = (self.column + 1) * tile_size_m + origin_x
+ min_y = origin_y - (self.row * tile_size_m)
+ max_y = origin_y - ((self.row + 1) * tile_size_m)
+
+ # Hack: Reverse x, y order to get around issues with Coordinates.from_url assuming lat, lon order
+ min_x, min_y = min_y, min_x
+ max_x, max_y = max_y, max_x
+
+ return ",".join(str(x) for x in [min_x, min_y, max_x, max_y])
+
+ @staticmethod
+ def calculate_scale_denominator(matrix_level: int) -> str:
+ """Calculate the scale denominator for the matrix level.
+
+ Parameters
+ ----------
+ matrix_level : int
+ The matrix level to calculate scale denominator for.
+
+ Returns
+ -------
+ str
+ The scaled denominator as a string or empty string on failure.
+ """
+ resolution = settings.WMTS_INITIAL_RESOLUTION / (2 ** (matrix_level))
+ scale_denominator = resolution / settings.WMTS_PIXEL_SIZE_METERS
+ return TileMatrixSet.format_number(scale_denominator)
+
+ @staticmethod
+ def calculate_matrix_sizes(matrix_level: int) -> Tuple[str, str]:
+ """Calculate the matrix width and height from matrix level.
+
+ Parameters
+ ----------
+ matrix_level : int
+ The matrix level to calculate width and height for.
+
+ Returns
+ -------
+ Tuple[str, str]
+ The matrix width and height as strings or empty strings on failure.
+ """
+ matrix_width = 2**matrix_level
+ matrix_height = 2**matrix_level
+ return TileMatrixSet.format_number(matrix_width), TileMatrixSet.format_number(matrix_height)
+
+
+class WorldCRS84Quad(TileMatrixSet):
+ """Tile matrix set for World CRS84 Quad format."""
+
+ name = tl.Unicode(default_value="WorldCRS84Quad")
+ crs = tl.Unicode(default_value="urn:ogc:def:crs:OGC:2:84")
+ wkss = tl.Unicode(default_value="urn:ogc:def:wkss:OGC:1.0:GlobalCRS84Pixel")
+ bounds = tl.Dict(
+ key_trait=tl.Unicode(),
+ value_trait=tl.Float(),
+ default_value={"minx": -180, "miny": -90, "maxx": 180, "maxy": 90},
+ )
+
+ def validate(self):
+ """Validate the tile matrix set."""
+ super().validate()
+ assert self.row >= 0 and self.row < 2**self.matrix_level, "WMTS Request validation error: invalid tile row"
+ assert self.column >= 0 and self.column < 2 ** (
+ self.matrix_level + 1
+ ), "WMTS Request validation error: invalid tile column"
+
+ def calculate_eval_bounding_box(self) -> str:
+ """Determine the bounding box for evaluation.
+
+ A zero pixel row and column corresponds to the top left corner.
+
+ Returns
+ -------
+ str
+ The bounding box for the provided tile indices.
+ """
+ tiles_per_width = 2 ** (self.matrix_level + 1)
+ tiles_per_height = 2**self.matrix_level
+
+ tile_width = abs(self.bounds["maxx"] - self.bounds["minx"]) / tiles_per_width
+ tile_height = abs(self.bounds["maxy"] - self.bounds["miny"]) / tiles_per_height
+
+ min_x = self.column * tile_width + self.bounds["minx"]
+ max_x = (self.column + 1) * tile_width + self.bounds["minx"]
+ min_y = self.bounds["maxy"] - (self.row * tile_height)
+ max_y = self.bounds["maxy"] - ((self.row + 1) * tile_height)
+
+ # Hack: Reverse x, y order to get around issues with Coordinates.from_url assuming lat, lon order
+ min_x, min_y = min_y, min_x
+ max_x, max_y = max_y, max_x
+
+ return ",".join(str(x) for x in [min_x, min_y, max_x, max_y])
+
+ @staticmethod
+ def calculate_scale_denominator(matrix_level: int) -> str:
+ """Calculate the scale denominator for the matrix level.
+
+ Parameters
+ ----------
+ matrix_level : int
+ The matrix level to calculate scale denominator for.
+
+ Returns
+ -------
+ str
+ The scaled denominator as a string or empty string on failure.
+ """
+ resolution = settings.WMTS_INITIAL_RESOLUTION / (2 ** (matrix_level + 1))
+ scale_denominator = resolution / settings.WMTS_PIXEL_SIZE_METERS
+ return TileMatrixSet.format_number(scale_denominator)
+
+ @staticmethod
+ def calculate_matrix_sizes(matrix_level: int) -> Tuple[str, str]:
+ """Calculate the matrix width and height from matrix level.
+
+ Parameters
+ ----------
+ matrix_level : int
+ The matrix level to calculate width and height for.
+
+ Returns
+ -------
+ Tuple[str, str]
+ The matrix width and height as strings or empty strings on failure.
+ """
+ matrix_width = 2 ** (matrix_level + 1)
+ matrix_height = 2**matrix_level
+ return TileMatrixSet.format_number(matrix_width), TileMatrixSet.format_number(matrix_height)
+
+
+class GetCapabilities(ogc_common.XMLNode):
+ """
+ Request to a WMTS server to perform the GetCapabilities operation.
+ This operation allows a client to retrieve a Capabilities XML document providing
+ metadata for the specific WMTS server. In this XML encoding, no "request" parameter
+ is included, since the element name specifies the specific operation.
+ """
+
+ service = tl.Unicode(default_value=None, allow_none=True)
+
+ accept_versions = tl.List(trait=tl.Unicode(default_value=None, allow_none=True))
+ accept_formats = tl.List(trait=tl.Instance(klass=ogc_common.OutputFormat))
+
+ def validate(self):
+ """Validate the get capabilities request."""
+ assert self.service == "WMTS", WMTS_VALIDATION_ERROR
+
+ for obj in self.accept_formats:
+ obj.validate()
+
+ def _load_from_kv(self, args: Dict[str, Any]):
+ """Initialize the properties of the class from a key-value argument dictionary.
+
+ Parameters
+ ----------
+ args : Dict[str, Any]
+ Request arguments for the get capabilities request.
+ """
+ assert args["request"] == "GetCapabilities", args["request"]
+ self.service = args["service"].upper()
+
+
+class GetTile(ogc_common.XMLNode):
+ """
+ Request to a WMTS to perform the GetTile operation.
+
+ This operation allows a client to retrieve a
+ subset of one coverage. In this XML encoding, no "request" parameter is
+ included, since the element name specifies the specific operation.
+ """
+
+ service = tl.Unicode(default_value=None, allow_none=True)
+ version = tl.Unicode(default_value=None, allow_none=True)
+
+ layer = tl.Instance(klass=Identifier)
+ tile_matrix_set = tl.Instance(klass=TileMatrixSet)
+ output_format = tl.Enum(
+ default_value=None,
+ values=["image/png", "image/png; mode=8bit", "image/png;mode=8-bit"],
+ )
+ params = tl.Dict(tl.Unicode(), tl.Unicode(), default_value={})
+
+ def validate(self):
+ """Validate the get tile request."""
+ assert self.service == "WMTS", WMTS_VALIDATION_ERROR
+ assert self.output_format is not None, "WMTS Request validation error: no output format specified"
+ self.layer.validate()
+ self.tile_matrix_set.validate()
+
+ def _load_from_kv(self, args: Dict[str, Any]):
+ """Initialize traitlet properties from a key-value argument dictionary.
+
+ Parameters
+ ----------
+ args : Dict[str, Any]
+ Request arguments for the get tile request.
+ """
+ assert args["request"].lower() == "gettile", args["request"]
+ self.service = args["service"].upper()
+ self.version = args["version"]
+ self.layer = Identifier(value=args["layer"])
+
+ row = args.get("tilerow")
+ column = args.get("tilecol")
+ matrix_level = args.get("tilematrix")
+ row = int(row) if row is not None else None
+ column = int(column) if column is not None else None
+ matrix_level = int(matrix_level) if matrix_level is not None else None
+ tile_matrix_set_name = args.get("tilematrixset", "")
+
+ if tile_matrix_set_name.lower() == "webmercatorquad":
+ self.tile_matrix_set = WebMercatorQuad(row=row, column=column, matrix_level=matrix_level)
+ elif tile_matrix_set_name.lower() == "worldcrs84quad":
+ self.tile_matrix_set = WorldCRS84Quad(row=row, column=column, matrix_level=matrix_level)
+
+ if "format" in args:
+ self.output_format = str(args["format"])
+
+ # Check for dimensions and add to params
+ self.params = {}
+ for key, value in args.items():
+ key_dimension = next((dim for dim in VALID_DIMENSION_NAMES if dim.lower() == key.lower()), None)
+ if key_dimension:
+ self.params[key_dimension] = value
+
+ def _load_xml_doc(self, xml_doc: str):
+ """Initialize traitlet properties from a XML document.
+ Parameters
+ ----------
+ xml_doc : str
+ The XML document to unpack.
+
+ Raises
+ ------
+ NotImplementedError
+ Not implemented.
+ """
+ raise NotImplementedError()
+
+ def convert_to_map_args(self) -> Dict[str, Any]:
+ """Converts the arguments from a get tile request into a get map request.
+
+ Returns
+ -------
+ Dict[str, Any]
+ The arguments which are valid for a get map request.
+ """
+ return {
+ "PARAMS": json.dumps(self.params), # this needs to be capitalized for podpac
+ "crs": self.tile_matrix_set.crs,
+ "height": settings.WMTS_TILE_SIZE,
+ "width": settings.WMTS_TILE_SIZE,
+ "bbox": self.tile_matrix_set.calculate_eval_bounding_box(),
+ "format": self.output_format,
+ "version": self.version,
+ "service": self.service,
+ }
diff --git a/ogc/wmts/wmts_response_1_0_0.py b/ogc/wmts/wmts_response_1_0_0.py
new file mode 100755
index 0000000..3a591ad
--- /dev/null
+++ b/ogc/wmts/wmts_response_1_0_0.py
@@ -0,0 +1,337 @@
+import logging
+import traitlets as tl
+from datetime import datetime
+from xml.sax.saxutils import escape
+
+from ogc import ogc_common
+from ogc import settings
+from ogc.wcs_response_1_0_0 import Coverage
+from .wmts_request_1_0_0 import WebMercatorQuad, WorldCRS84Quad, TileMatrixSet
+
+logger = logging.getLogger(__name__)
+escape_format = ogc_common.EscapeFormatter().format
+
+SERVICE_VERSION = "1.0.0"
+
+
+class Capabilities(ogc_common.XMLNode):
+ """
+ XML encoded WMTS GetCapabilities operation response.
+ The Capabilities document provides clients with service
+ metadata about a specific service instance, including metadata
+ about the coverages served. If the server does not
+ implement the updateSequence parameter, the server shall always
+ return the Capabilities document, without the updateSequence
+ parameter. When the server implements the updateSequence
+ parameter and the GetCapabilities operation request included
+ the updateSequence parameter with the current value, the server
+ shall return this element with only the "version" and
+ "updateSequence" attributes. Otherwise, all optional sections
+ shall be included or not depending on the actual value of the
+ Contents parameter in the GetCapabilities operation request.
+ """
+
+ # Description Part
+ service_title = tl.Unicode(default_value=None, allow_none=True)
+ service_abstract = tl.Unicode(default_value=None, allow_none=True)
+ service_keywords = tl.List(trait=tl.Unicode(default_value=None, allow_none=True))
+ service_group_title = tl.Unicode(default_value=None, allow_none=True)
+
+ # Service Identification Part
+ service_type = tl.Unicode(default_value="WMTS", allow_none=True)
+ service_type_version = tl.List([SERVICE_VERSION])
+
+ coverages = tl.List(trait=tl.Instance(klass=Coverage))
+ base_url = tl.Unicode(default_value=None, allow_none=True)
+
+ version = tl.Unicode(default_value=SERVICE_VERSION)
+ provider_name = tl.Unicode(default_value="Creare LLC")
+ provider_site = tl.Unicode(default_value="https://github.com/creare-com")
+ indent = " "
+
+ def to_xml(self) -> str:
+ """Generate the XML response for service metadata.
+
+ Returns
+ -------
+ str
+ XML response as a string.
+ """
+ return """\
+
+
+{service_identification}
+{service_provider}
+{operations_metadata}
+{contents}
+
+
+ """.format(
+ version=escape(self.version),
+ base_url=escape(self.base_url if self.base_url else "?"),
+ service_identification=self._service_identification(),
+ service_provider=self._service_provider(),
+ operations_metadata=self._operations_metadata(),
+ contents=self._contents(),
+ )
+
+ def _service_identification(self) -> str:
+ """Metadata about the specific service.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ keywords = ""
+ if self.service_keywords:
+ for service_keyword in self.service_keywords:
+ keywords += escape_format("""{}\n""", service_keyword)
+
+ return escape_format(
+ """\
+
+ {self.service_title}
+ {self.service_abstract}
+ {keywords}
+ {self.service_type}
+ {self.version}
+
+ {constraints}
+ """,
+ self=self,
+ keywords=keywords,
+ constraints=settings.CONSTRAINTS,
+ )
+
+ def _service_provider(self) -> str:
+ """Metadata about the organization providing the service.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ return escape_format(
+ """\
+
+ {self.provider_name}
+
+ """,
+ self=self,
+ )
+
+ def _operations_metadata(self) -> str:
+ """Metadata about the operations specified by the service.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ return escape_format(
+ """\
+
+
+
+
+
+
+
+ KVP
+
+
+
+
+
+
+
+
+
+
+
+
+ KVP
+
+
+
+
+
+
+ """,
+ self=self,
+ )
+
+ def _contents(self) -> str:
+ """Metadata about the data provided by the service.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ return """\
+ \n{layers}{tile_matrix_sets}""".format(
+ layers=self._layers(), tile_matrix_sets=self._tile_matrix_sets()
+ )
+
+ def _coverage_layer(self, coverage: Coverage, depth: int = 2) -> str:
+ """Metadata about an individual coverage layer.
+
+ Parameters
+ ----------
+ coverage : Coverage
+ The coverage used to define layer metadata.
+ depth : int
+ The indentation depth for the XML data, by default 2.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ xml = ""
+ xml = self.indent * depth + """\n"""
+ if coverage.identifier:
+ xml += self.indent * (depth + 1) + escape_format(
+ "{identifier}\n", identifier=coverage.identifier
+ )
+ if coverage.title:
+ xml += self.indent * (depth + 1) + escape_format("{}\n", coverage.title)
+ else:
+ logger.info("Invalid layer. Missing title.")
+ if coverage.abstract:
+ xml += self.indent * (depth + 1) + escape_format("{}\n", coverage.abstract)
+
+ xml += self._bounding_box_layer(coverage)
+
+ coordinates = coverage.layer.get_coordinates()
+ extra_dims = [dim for dim in coordinates.udims if dim not in ["lat", "lon"]]
+ for extra_dim in extra_dims:
+ if coordinates[extra_dim].coordinates.size > 0:
+ units = "number"
+ if extra_dim == "alt":
+ units = coordinates.alt_units if coordinates.alt_units is not None else units
+ elif self._is_iso_datetime(str(coordinates[extra_dim].coordinates[-1])):
+ units = "ISO8601"
+
+ xml += self.indent * (depth + 1) + """\n"""
+ xml += self.indent * (depth + 2) + escape_format("""{}\n""", extra_dim)
+ xml += self.indent * (depth + 2) + escape_format("""{}\n""", units)
+ xml += self.indent * (depth + 2) + escape_format(
+ """{}\n""", coordinates[extra_dim].coordinates[-1]
+ )
+ for value in coordinates[extra_dim].coordinates:
+ xml += self.indent * (depth + 2) + escape_format("""{}\n""", value)
+ xml += self.indent * (depth + 1) + """\n"""
+
+ # Write the style section and format
+ xml += self.indent * (depth + 1) + """\n"""
+ xml += self.indent * (depth + 1) + """image/png\n"""
+
+ # Tile matrix set links
+ matrix_sets = [WebMercatorQuad(), WorldCRS84Quad()]
+ for matrix_set in matrix_sets:
+ xml += self.indent * (depth + 1) + """\n"""
+ xml += self.indent * (depth + 2) + escape_format("""{}\n""", matrix_set.name)
+ xml += self.indent * (depth + 1) + """\n"""
+
+ xml += self.indent * depth + """\n"""
+
+ return xml
+
+ def _layers(self) -> str:
+ """Metadata for all layers in the service.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ xml = ""
+ for coverage in self.coverages:
+ xml += self._coverage_layer(coverage)
+
+ return xml
+
+ def _tile_matrix_sets(self, depth: int = 2) -> str:
+ """Metadata about the supported tile matrix sets.
+
+ Parameters
+ ----------
+ depth : int
+ The indentation depth for the XML data, by default 2.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ matrix_sets = [WebMercatorQuad(), WorldCRS84Quad()]
+ xml = ""
+ for matrix_set in matrix_sets:
+ xml += matrix_set.to_xml()
+
+ return xml
+
+ def _bounding_box_layer(self, coverage: Coverage, depth: int = 3) -> str:
+ """Metadata for the WGS84 bounding box for a specific coverage layer.
+
+ Parameters
+ ----------
+ coverage : Coverage
+ The coverage used to define layer metadata.
+ depth : int
+ The indentation depth for the XML data, by default 3.
+
+ Returns
+ -------
+ str
+ XML data as a string.
+ """
+ xml = ""
+ xml += self.indent * (depth) + """\n"""
+ xml += self.indent * (depth + 1) + escape_format(
+ """{x} {y}\n""",
+ x=TileMatrixSet.format_number(coverage.wgs84_bounding_box_lower_corner_lat_lon[1]),
+ y=TileMatrixSet.format_number(coverage.wgs84_bounding_box_lower_corner_lat_lon[0]),
+ )
+ xml += self.indent * (depth + 1) + escape_format(
+ """{x} {y}\n""",
+ x=TileMatrixSet.format_number(coverage.wgs84_bounding_box_upper_corner_lat_lon[1]),
+ y=TileMatrixSet.format_number(coverage.wgs84_bounding_box_upper_corner_lat_lon[0]),
+ )
+ xml += self.indent * (depth) + """\n"""
+
+ return xml
+
+ @staticmethod
+ def _is_iso_datetime(datetime_string: str) -> bool:
+ """Determine whether a string is a valid datetime.
+
+ Parameters
+ ----------
+ datetime_string : str
+ The datetime string to validate.
+
+ Returns
+ -------
+ bool
+ True if the provided string is a valid datetime, False otherwise.
+ """
+ try:
+ datetime.fromisoformat(datetime_string)
+ return True
+ except ValueError:
+ return False
diff --git a/ogc/wmts/wmts_routes.py b/ogc/wmts/wmts_routes.py
new file mode 100644
index 0000000..15ca355
--- /dev/null
+++ b/ogc/wmts/wmts_routes.py
@@ -0,0 +1,194 @@
+import traitlets as tl
+import logging
+import gc
+from typing import Dict, Any
+
+from ogc.ogc_common import WMTSException
+from ogc.wcs_response_1_0_0 import Coverage
+
+from . import wmts_response_1_0_0
+from . import wmts_request_1_0_0
+
+LOAD_FAILURE = "Failed to load and validate: "
+INVALID_ARGUMENTS = "Invalid arguments"
+
+logger = logging.getLogger(__name__)
+
+
+class WmtsRoutes(tl.HasTraits):
+ """Class responsible for WMTS requests."""
+
+ coverages = tl.List(trait=tl.Instance(klass=Coverage))
+ base_url = tl.Unicode(default_value=None, allow_none=True)
+ service_title = tl.Unicode(default_value=None, allow_none=True)
+ service_abstract = tl.Unicode(default_value=None, allow_none=True)
+ service_group_title = tl.Unicode(default_value=None, allow_none=True)
+
+ def handle_kv(self, args: Dict[str, Any]) -> Dict[str, Any] | str:
+ """Handle WMTS key value requests.
+
+ 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 unsupported get feature info request.
+ WMTSException
+ Exception for invalid parameter value in the request arguments.
+ WMTSException
+ Exception for unsupported request argument.
+ """
+ if args["request"] == "GetCapabilities":
+ return self.get_capabilities(args)
+
+ if args["request"] == "GetFeatureInfo":
+ raise WMTSException(
+ exception_code="OperationNotSupported",
+ locator="REQUEST",
+ exception_text="Unsupported request",
+ )
+
+ if args["request"].lower() == "gettile":
+ return self.get_tile(args)
+
+ raise WMTSException(exception_text="KV Request not handled properly: " + str(args))
+
+ def get_coverage_from_id(self, identifier: str) -> Coverage:
+ """Find the coverage for a given identifier.
+
+ Parameters
+ ----------
+ identifier : str
+ The coverage identifier.
+
+ Returns
+ -------
+ Coverage
+ The coverage matching the provided identifier.
+
+ Raises
+ ------
+ WMTSException
+ Exception for an invalid coverage identifier.
+ """
+ for coverage in self.coverages:
+ if coverage.identifier == identifier:
+ return coverage
+ raise WMTSException(
+ exception_code="InvalidParameterValue",
+ locator="COVERAGE",
+ exception_text="Invalid coverage {}".format(identifier),
+ )
+
+ def get_capabilities(self, args: Dict[str, Any]) -> str:
+ """Get capabilities for the WMTS server.
+
+ Parameters
+ ----------
+ args : Dict[str, Any]
+ The filtered request arguments.
+
+ Returns
+ -------
+ str
+ The xml response as a string.
+
+ Raises
+ ------
+ WMTSException
+ Exception for invalid specified version.
+ WMTSException
+ Exception for invalid arguments found during validation.
+ """
+ if args["base_url"]:
+ self.base_url = args["base_url"]
+
+ # Version is optional, use 1.0.0 as default
+ if "version" not in args or args["version"] == "1.0.0":
+ get_capabilities = wmts_request_1_0_0.GetCapabilities()
+ capabilities = wmts_response_1_0_0.Capabilities(
+ coverages=self.coverages,
+ base_url=self.base_url,
+ service_title=self.service_title,
+ service_abstract=self.service_abstract,
+ service_group_title=self.service_group_title,
+ )
+ else:
+ raise WMTSException(
+ exception_code="InvalidParameterValue",
+ locator="VERSION",
+ exception_text="Unsupported version: %s" % (args["version"]),
+ )
+
+ try:
+ get_capabilities.load_from_kv(args)
+ get_capabilities.validate()
+ except AssertionError:
+ logger.error(LOAD_FAILURE, exc_info=True)
+ raise WMTSException(exception_text=INVALID_ARGUMENTS)
+
+ return capabilities.to_xml()
+
+ def get_tile(self, args: Dict[str, Any]) -> Dict[str, Any]:
+ """Retrieve a tile using the requested arguments.
+
+ Parameters
+ ----------
+ args : Dict[str, Any]
+ The filtered request arguments.
+
+ Returns
+ -------
+ Dict[str, Any]
+ A dictionary containing the tile response.
+
+ Raises
+ ------
+ WMTSException
+ Exception for invalid specified version or missing version.
+ WMTSException
+ Exception for invalid arguments found during validation.
+ WMTSException
+ Exception for errors found during the layer evaluation.
+ """
+ if "version" in args and args["version"] == "1.0.0":
+ get_tile = wmts_request_1_0_0.GetTile()
+ else:
+ raise WMTSException(
+ exception_code="InvalidParameterValue",
+ locator="VERSION",
+ exception_text="Unsupported version: %s" % (args["version"] if "version" in args else "None"),
+ )
+
+ try:
+ get_tile.load_from_kv(args)
+ get_tile.validate()
+ except AssertionError:
+ logger.error(LOAD_FAILURE, exc_info=True)
+ raise WMTSException(exception_text=INVALID_ARGUMENTS)
+
+ coverage = self.get_coverage_from_id(get_tile.layer.value)
+ map_args = get_tile.convert_to_map_args()
+
+ try:
+ fp = coverage.layer.get_map(map_args)
+ except Exception:
+ logger.error("Failed to get_tile from layer: ", exc_info=True)
+ raise WMTSException(exception_text=INVALID_ARGUMENTS)
+
+ fn = coverage.identifier.split(".")[-1] + ".png"
+
+ # Collect garbage
+ gc.collect()
+
+ response = {"fp": fp, "fn": fn}
+
+ return response