added validation, tawhiri request creation
This commit is contained in:
parent
456551cd4e
commit
2aef4d4756
12 changed files with 184 additions and 18 deletions
BIN
api/services/__pycache__/tawhiri.cpython-313.pyc
Normal file
BIN
api/services/__pycache__/tawhiri.cpython-313.pyc
Normal file
Binary file not shown.
52
api/services/tawhiri.py
Normal file
52
api/services/tawhiri.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
import requests
|
||||
from urllib.parse import urlencode
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from zoneinfo import ZoneInfo
|
||||
from collections import OrderedDict
|
||||
|
||||
class TawhiriClient:
|
||||
BASE_URL = "https://fly.stratonautica.ru/api/v2/"
|
||||
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()
|
||||
query["dataset"] = cls._convert_value(params.get("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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue