"""Pygeoapi instance of NLDI get cross-section at point."""
from __future__ import annotations
import logging
import time
from typing import Any, Dict, List, Tuple
from geopandas.geodataframe import GeoDataFrame
from pygeoapi.process.base import BaseProcessor
from nldi_xstool.nldi_xstool import getxsatpoint
LOGGER = logging.getLogger(__name__)
PROCESS_METADATA = {
"version": "0.1.0",
"id": "nldi-xsatpoint",
"title": "NLDI xsatpoint process",
"description": "NLDI xsatpoint process",
"jobControlOptions": ["sync-execute"],
"keywords": ["NLDI xsatpoint"],
"links": [
{
"type": "text/html",
"rel": "canonical",
"title": "information",
"href": "https://example.org/process",
"hreflang": "en-US",
}
],
"inputs": {
"lat": {
"title": "lat",
"schema": {"type": "float"},
"input": {
"literalDataDomain": {
"dataType": "float",
"valueDefinition": {"anyValue": True},
}
},
"minOccurs": 1,
"maxOccurs": 1,
},
"lon": {
"title": "lon",
"schema": {"type": "float"},
"input": {
"literalDataDomain": {
"dataType": "float",
"valueDefinition": {"anyValue": True},
}
},
"minOccurs": 1,
"maxOccurs": 1,
},
"width": {
"title": "width",
"schema": {"type": "float"},
"input": {
"literalDataDomain": {
"dataType": "float",
"valueDefinition": {"anyValue": True},
}
},
"minOccurs": 1,
"maxOccurs": 1,
},
"numpts": {
"title": "numpts",
"schema": {"type": "int"},
"input": {
"literalDataDomain": {
"dataType": "int",
"valueDefinition": {"anyValue": True},
}
},
"minOccurs": 1,
"maxOccurs": 1,
},
},
"outputs": {
"nldi-xsatpoint-response": {
"title": "output nldi-xsatpoint",
"schema": {"type": "object", "contentMediaType": "application/json"},
"output": {"formats": [{"mimeType": "application/json"}]},
}
},
"example": {
"inputs": [
{"id": "lat", "value": "39.064867", "type": "text/plain"},
{"id": "lon", "value": "-96.168776", "type": "text/plain"},
{"id": "width", "value": "1000.0", "type": "text/plain"},
{"id": "numpts", "value": "101", "type": "text/plain"},
]
},
}
[docs]
class NLDIxsatpointProcessor(BaseProcessor): # type: ignore
"""NLDI Get Cross-sectin at Point."""
def __init__(self, provider_def: Dict[str, Any]) -> None:
"""Initialize object.
:param provider_def: provider definition
:returns: pygeoapi.process.nldi_delineate.NLDIDelineateProcessor
"""
BaseProcessor.__init__(self, provider_def, PROCESS_METADATA)
[docs]
def execute(self, data: List[Dict[str, Any]]) -> Tuple[str, Dict[str, Any]]:
"""Execute processor."""
mimetype = "application/json"
# reformat data into dict with just needed values
newdata = {d["id"]: d["value"] for d in data}
lat = float(newdata["lat"])
lon = float(newdata["lon"])
numpts = int(newdata["numpts"])
width = float(newdata["width"])
timebefore = time.perf_counter()
results = GeoDataFrame(getxsatpoint((lon, lat), numpts, width))
timeafter = time.perf_counter()
totaltime = timeafter - timebefore
print("Total Time:", totaltime)
return mimetype, results.__geo_interface__
def __repr__(self) -> str:
"""Get representation."""
return "<NLDIxsatpointProcessor> {}".format("get xsatpoint")