Generate the OGC Moving Features read routes over the MEOS dispatcher

The temporal-geometry sequence, its derived velocity and distance queries, and a
stored temporal property are served by a FastAPI router
(mobilityapi/routers/movfeat.py) that dispatches the matching MEOS catalog
function (temporal_as_mfjson, tpoint_speed, tpoint_cumulative_length) through the
Dispatcher and WireCodec and shapes the MF-JSON result as the OGC envelope.
acceleration returns 501 for the piecewise-constant motion model. A FeatureStore
port abstracts the database read so the routes carry stub tests without a live
MEOS; create_app mounts the router and takes the store.

Author: estebanzimanyi

mobilityapi/app.py

@@ -18,14 +18,15 @@
 from fastapi import FastAPI
 
 from .dispatcher import Dispatcher
-from .routers import catalog, functions
+from .routers import catalog, functions, movfeat
 from .wire import WireCodec
 
 
 def create_app(
     dispatcher: Dispatcher,
     codec: WireCodec,
     *,
+    feature_store: object | None = None,
     title: str = "MobilityAPI",
     version: str = "0.1.0",
 ) -> FastAPI:
@@ -52,8 +53,10 @@ def create_app(
     )
     app.state.dispatcher = dispatcher
     app.state.codec = codec
+    app.state.feature_store = feature_store
 
     app.include_router(catalog.router, prefix="/catalog", tags=["catalog"])
     app.include_router(functions.router, prefix="/functions", tags=["functions"])
+    app.include_router(movfeat.router, tags=["movingfeatures"])
 
     return app

mobilityapi/routers/movfeat.py

@@ -0,0 +1,128 @@
+"""OGC API – Moving Features read routes generated over the MEOS dispatcher.
+
+The "pure MEOS round-trip" resources — the temporal-geometry sequence and its
+derived velocity / distance queries, and a stored temporal property — are
+served by dispatching the matching MEOS function from the vendored catalog
+through the ``Dispatcher`` + ``WireCodec``, then shaping the MF-JSON result as
+the OGC envelope. The collection / feature lifecycle, persistence and the
+GeoJSON envelope have no MEOS equivalent and stay hand-written.
+
+OGC resource → MEOS catalog function (the alignment map):
+
+    tgsequence (export)        → temporal_as_mfjson
+    tgsequence/{tg}/velocity   → tpoint_speed
+    tgsequence/{tg}/distance   → tpoint_cumulative_length
+    tproperties/{name}         → temporal_as_mfjson
+
+``acceleration`` returns 501: with linearly interpolated position the speed is
+piecewise-constant, so its derivative is zero within each segment and undefined
+at the vertices; the value is not approximated.
+
+A ``FeatureStore`` port abstracts the database read so the routes are testable
+with stubs; production wires it to PyMEOS + psycopg2. Writes (sub-trajectory
+append, temporal-property creation) remain on the hand-written path until the
+catalog's input-function shapes are confirmed against a live MEOS.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, runtime_checkable
+
+from fastapi import APIRouter, HTTPException, Request
+
+#: OGC derived measure → the MEOS catalog function that computes it.
+DERIVED_FN = {"velocity": "tpoint_speed", "distance": "tpoint_cumulative_length"}
+#: MEOS serialiser used to export a temporal value as MF-JSON.
+EXPORT_FN = "temporal_as_mfjson"
+
+
+@runtime_checkable
+class FeatureStore(Protocol):
+    """Database port. Values are MF-JSON (the wire shape the codec decodes).
+
+    Production wires this to PyMEOS + psycopg2; tests pass a stub.
+    """
+
+    def get_trajectory(self, cid: str, fid: str) -> Any | None: ...
+    def get_property(self, cid: str, fid: str, name: str) -> Any | None: ...
+
+
+router = APIRouter()
+
+
+def _ctx(request: Request):
+    state = request.app.state
+    store = getattr(state, "feature_store", None)
+    if store is None:
+        raise HTTPException(501, "no FeatureStore is wired into the app")
+    return state.dispatcher, state.codec, store
+
+
+def _dispatch_unary(dispatcher, codec, fn: str, value_mfjson: Any) -> Any:
+    """Decode an MF-JSON temporal value, dispatch a single-temporal-argument
+    MEOS function, and re-encode the result as MF-JSON."""
+    if not dispatcher.has(fn):
+        raise HTTPException(501, f"MEOS function `{fn}` is not in the vendored catalog")
+    sig = dispatcher.signature(fn)
+    if not sig.params:
+        raise HTTPException(500, f"MEOS function `{fn}` has no parameters to dispatch")
+    pname = sig.params[0]["name"]
+    arg = codec.decode("mfjson", value_mfjson)
+    result = dispatcher.dispatch(fn, {pname: arg})
+    return codec.encode("mfjson", result)
+
+
+def _temporal_property(name: str, type_token: str, mfjson: Any, self_href: str) -> dict:
+    """Shape a MEOS MF-JSON temporal value as an OGC ``temporalProperty``."""
+    seq = mfjson if isinstance(mfjson, list) else [mfjson]
+    return {
+        "name": name,
+        "type": type_token,
+        "valueSequence": seq,
+        "links": [{"rel": "self", "href": self_href}],
+    }
+
+
+@router.get("/collections/{cid}/items/{fid}/tgsequence", summary="Temporal geometry (MF-JSON)")
+def get_tgsequence(cid: str, fid: str, request: Request) -> Any:
+    dispatcher, codec, store = _ctx(request)
+    trip = store.get_trajectory(cid, fid)
+    if trip is None:
+        raise HTTPException(404, "feature not found")
+    return _dispatch_unary(dispatcher, codec, EXPORT_FN, trip)
+
+
+@router.get(
+    "/collections/{cid}/items/{fid}/tgsequence/{tg}/{measure}",
+    summary="Derived temporal-geometry query: velocity | distance (acceleration → 501)",
+)
+def get_derived(cid: str, fid: str, tg: str, measure: str, request: Request) -> dict:
+    if measure == "acceleration":
+        raise HTTPException(
+            501,
+            "acceleration is not derivable: linearly interpolated position gives a "
+            "piecewise-constant speed, whose derivative is zero within each segment "
+            "and undefined at the vertices",
+        )
+    fn = DERIVED_FN.get(measure)
+    if fn is None:
+        raise HTTPException(404, f"unknown temporal-geometry query: {measure}")
+    dispatcher, codec, store = _ctx(request)
+    trip = store.get_trajectory(cid, fid)
+    if trip is None:
+        raise HTTPException(404, "feature not found")
+    out = _dispatch_unary(dispatcher, codec, fn, trip)
+    return _temporal_property(measure, "TReal", out, request.url.path)
+
+
+@router.get(
+    "/collections/{cid}/items/{fid}/tproperties/{name}",
+    summary="A stored temporal property as an OGC temporalProperty",
+)
+def get_tproperty(cid: str, fid: str, name: str, request: Request) -> dict:
+    dispatcher, codec, store = _ctx(request)
+    value = store.get_property(cid, fid, name)
+    if value is None:
+        raise HTTPException(404, f"unknown temporal property: {name}")
+    out = _dispatch_unary(dispatcher, codec, EXPORT_FN, value)
+    return _temporal_property(name, "TReal", out, request.url.path)

tests/test_movfeat.py

@@ -0,0 +1,106 @@
+"""HTTP-surface tests for the generated OGC API – Moving Features routes.
+
+The routes are exercised via Starlette's ``TestClient`` against a tiny in-test
+catalog, a stub resolver (so the "MEOS function" is a Python lambda), a stub
+``WireCodec``, and a stub ``FeatureStore``. PyMEOS and a database are NOT
+required — the test asserts the dispatch + OGC-shaping wiring, not a live MEOS.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from mobilityapi import create_app, stub_codec
+from mobilityapi.dispatcher import Dispatcher
+
+# Catalog: the three MEOS functions the OGC read routes dispatch to, each a
+# single-temporal-argument function named `t`.
+_CATALOG = {
+    "functions": [
+        {"name": "temporal_as_mfjson", "category": "io", "params": [{"name": "t", "type": "Temporal *"}], "return_type": "text"},
+        {"name": "tpoint_speed", "category": "analysis", "params": [{"name": "t", "type": "Temporal *"}], "return_type": "Temporal *"},
+        {"name": "tpoint_cumulative_length", "category": "analysis", "params": [{"name": "t", "type": "Temporal *"}], "return_type": "Temporal *"},
+    ]
+}
+
+# Stub resolver: each "MEOS function" tags its single argument so the response
+# is traceable back to the function that ran.
+_RESOLVERS = {
+    "temporal_as_mfjson": lambda **kw: {"mfjson_of": kw["t"]},
+    "tpoint_speed": lambda **kw: {"speed_of": kw["t"]},
+    "tpoint_cumulative_length": lambda **kw: {"distance_of": kw["t"]},
+}
+
+
+class _StubStore:
+    """In-memory FeatureStore: vessel 1 exists with a trip + a `fuel` property."""
+
+    def get_trajectory(self, cid, fid):
+        return {"type": "MovingPoint", "id": fid} if fid == "1" else None
+
+    def get_property(self, cid, fid, name):
+        return {"type": "MovingFloat", "prop": name} if (fid == "1" and name == "fuel") else None
+
+
+@pytest.fixture
+def client(tmp_path: Path) -> TestClient:
+    cat = tmp_path / "meos-idl.json"
+    cat.write_text(json.dumps(_CATALOG))
+    dispatcher = Dispatcher(catalog_path=cat, resolver=lambda n: _RESOLVERS[n])
+    codec = stub_codec(
+        decoders={"mfjson": lambda v: {"decoded": v}},
+        encoders={"mfjson": lambda o: o},
+    )
+    return TestClient(create_app(dispatcher, codec, feature_store=_StubStore()))
+
+
+def test_tgsequence_export_dispatches_as_mfjson(client):
+    r = client.get("/collections/ships/items/1/tgsequence")
+    assert r.status_code == 200
+    assert "mfjson_of" in r.json()
+
+
+def test_tgsequence_missing_feature_404(client):
+    assert client.get("/collections/ships/items/999/tgsequence").status_code == 404
+
+
+@pytest.mark.parametrize("measure,tag", [("velocity", "speed_of"), ("distance", "distance_of")])
+def test_derived_measure_is_a_temporal_property(client, measure, tag):
+    r = client.get(f"/collections/ships/items/1/tgsequence/0/{measure}")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["name"] == measure
+    assert body["type"] == "TReal"
+    assert tag in json.dumps(body["valueSequence"])
+
+
+def test_acceleration_is_501(client):
+    r = client.get("/collections/ships/items/1/tgsequence/0/acceleration")
+    assert r.status_code == 501
+    assert "not derivable" in r.json()["detail"]
+
+
+def test_unknown_measure_404(client):
+    assert client.get("/collections/ships/items/1/tgsequence/0/heading").status_code == 404
+
+
+def test_stored_temporal_property(client):
+    r = client.get("/collections/ships/items/1/tproperties/fuel")
+    assert r.status_code == 200
+    assert r.json()["name"] == "fuel"
+
+
+def test_unknown_temporal_property_404(client):
+    assert client.get("/collections/ships/items/1/tproperties/nope").status_code == 404
+
+
+def test_routes_501_without_a_feature_store(tmp_path: Path):
+    cat = tmp_path / "meos-idl.json"
+    cat.write_text(json.dumps(_CATALOG))
+    dispatcher = Dispatcher(catalog_path=cat, resolver=lambda n: _RESOLVERS[n])
+    app = create_app(dispatcher, stub_codec(), feature_store=None)
+    assert TestClient(app).get("/collections/ships/items/1/tgsequence").status_code == 501