TiPG STAC Items

tipg-pgstac.ipynb

from tipg.collections import get_collection_index
from buildpg import asyncpg
from tipg.database import connection_factory
import json

def box2geojson(box):
    return {
        "type": "Polygon",
        "geometry": {
            "coordinates": [[
                [box[0],box[1]],
                [box[2],box[1]],
                [box[2],box[3]],
                [box[0],box[3]],
                [box[0],box[1]]
            ]]
        }
    }

con_init=connection_factory()
pool = await asyncpg.create_pool_b(init=con_init)
idx = await get_collection_index(pool)
for c in idx['collections'].values():
    extent = c.extent
    spatial = extent.spatial
    bbox = extent.spatial.bbox[0]
    geom = box2geojson(bbox)
    temporal = extent.temporal
    if temporal is None:
        start = '-infinity'
        end = 'infinity'
    else:
        start = temporal.interval[0][0]
        end = temporal.interval[0][1]
    cols = [{
        "name": col.name,
        "description": col.description,
        "type": col.json_type
        } for col in c.properties]
    print(start,end)
    gcol = c.get_geometry_column()
    if gcol:
        gcolname = gcol.name
    else:
        gcolname = None
    dcol = c.get_datetime_column()
    if dcol:
        dcolname = dcol.name
    else:
        dcolname = None

    feature = {
        "type": "Feature",
        "stac_version": "1.0.0",
        "id": c.id,
        "collection": "tipg",
        "geometry": geom,
        "bbox": bbox,
        "links": [],
        "assets": {
            "features" : {
                "title": "OGC Feature Service",
                "description": "OGC Feature Service Endpoint.",
                "href": f"/items/{c.id}",
                "roles": ["data"]
            },
            "vectortiles" : {
                "title": "OGC Vector Tiles",
                "description": "OGC Vector Tiles Endpoint.",
                "href": f"/items/{c.id}",
                "roles": ["data"]
            }
        },
        "stac_extensions": [
            "https://stac-extensions.github.io/table/v1.2.0/schema.json"
        ],
        "properties":{
            "description": c.description,
            "table:columns": cols,
            "table:primary_geometry": gcolname,
            "table:primary_datetime": dcolname
        }
    }
    print(json.dumps(feature, indent=2))