forked from afanasyev.aa/stratoflights
71 lines
2.9 KiB
Python
71 lines
2.9 KiB
Python
import requests
|
|
from urllib.parse import urlencode
|
|
from datetime import datetime
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
from collections import OrderedDict
|
|
|
|
# Request value meaning "let the predictor use whatever run it has loaded".
|
|
# Lives here because this module is the one that decides what reaches the
|
|
# predictor; the serializer imports it as its default.
|
|
LATEST_DATASET_KEYWORD = "latest"
|
|
|
|
|
|
class TawhiriClient:
|
|
BASE_URL = "http://127.0.0.1:8080/api/v1/prediction"
|
|
TIMEOUT = 15
|
|
|
|
@staticmethod
|
|
def _convert_value(value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
return value.isoformat().replace("+00:00", "Z")
|
|
return value
|
|
|
|
@classmethod
|
|
def get_prediction(cls, params: dict) -> dict:
|
|
url = cls.build_url(params)
|
|
print("🔍 URL:", url)
|
|
response = requests.get(url, timeout=cls.TIMEOUT)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
@classmethod
|
|
def build_url(cls, params: dict) -> str:
|
|
query = OrderedDict()
|
|
|
|
query["profile"] = params.get("profile")
|
|
query["launch_datetime"] = cls._convert_value(params.get("launch_datetime"))
|
|
query["launch_latitude"] = params.get("launch_latitude")
|
|
query["launch_longitude"] = params.get("launch_longitude")
|
|
query["launch_altitude"] = params.get("launch_altitude", 0)
|
|
query["ascent_rate"] = params.get("ascent_rate")
|
|
query["burst_altitude"] = params.get("burst_altitude")
|
|
query["descent_rate"] = params.get("descent_rate")
|
|
query["interpolate"] = str(params.get("interpolate", False)).lower()
|
|
# Forwarded only when the operator actually named a run.
|
|
#
|
|
# The predictor now honours this parameter and refuses a run it does not
|
|
# hold, so anything that is not a real epoch must not be sent. Two values
|
|
# reach here that are not: "" from the UI, and LATEST_DATASET_KEYWORD, the
|
|
# serializer's default, which means "server chooses" — exactly what
|
|
# omitting the parameter does. The `filtered` comprehension below drops
|
|
# None but neither of these.
|
|
#
|
|
# A malformed epoch is deliberately still forwarded, so the predictor
|
|
# rejects it instead of the value being quietly ignored.
|
|
dataset = params.get("dataset")
|
|
if dataset and dataset != LATEST_DATASET_KEYWORD:
|
|
query["dataset"] = cls._convert_value(dataset)
|
|
query["format"] = params.get("format", "json")
|
|
query["pred_type"] = "single" # <-- в конце
|
|
|
|
filtered = {k: v for k, v in query.items() if v is not None}
|
|
return f"{cls.BASE_URL}?{urlencode(filtered)}"
|
|
|
|
|
|
@classmethod
|
|
def get_prediction(cls, params: dict) -> dict:
|
|
url = cls.build_url(params)
|
|
response = requests.get(url, timeout=cls.TIMEOUT)
|
|
response.raise_for_status()
|
|
return response.json()
|