kmplus-api

kmplus.py

# -*- coding: utf-8 -*-
import os
import logging
import requests
import pandas as pd
from time import sleep

KEY = os.environ.get("kmplus", "")
KMP = "https://kmplus.ukravtodor.gov.ua/api/v1/kmp"
ROUTE = "https://kmplus.ukravtodor.gov.ua/api/v1/route/id"

def remote_call(endpoint):
    """Виконує запит до АРІ."""
    try:
        data = requests.get(
            url=endpoint,
            headers={"Accept-Language": "uk", "Authorization": KEY},
            data={},
            timeout=5,
            #verify=False # if SSLCertVerificationError occurs
        )
        if data.status_code == 200:
            return data.json()
        logging.info(f"{'/'.join(endpoint.split('/')[-2:])} resulting in {data.status_code} status code")
        return
    except KeyboardInterrupt:
        raise 
    except:
        logging.exception("Timed out")
        return



def process(data):
    """Опрацьовує вміст відповіді."""
    yield {
        "id": data["detail"]["id"],
        "name": data["detail"]["name"],
        "sc": data["detail"]["sc"],
        "ut": data["detail"]["ut"],
        "owner": data["detail"]["owner"],
        "kmp": data["detail"]["kmp"],
        "latitude": data["detail"]["coord"][0],
        "longitude": data["detail"]["coord"][1]
    }
    
    
def fetch(sc, min_km=0, max_km=224, m_freq=2):
    """Завантажує дані з АРІ v1/kmp/
    
    
    Parameters
    ----------
    sc: str
        Індекс дороги
    min_km: int
        Початок відрізку дороги
    max_km: int
        Кінець відрізку дороги
    m_freq: int
        Кількість відрізків в межах одного кілометру потрібно завантажити. 
        Наприклад, якщо m_freq = 2, то 1000/2 = 500, або [0, 500];
                   якщо m_freq = 5, то 1000/5 = 200, або [0, 200, 400, 600, 800]
    """
    for km in range(min_km, max_km+1):
        for m in range(0, 1000, 1000 // m_freq):
            kmp = f"{km}+{str(m).zfill(3)}"
            if m == 0:
                logging.info(f"{sc} - {kmp}")
            response = remote_call(f"{KMP}/{sc}/{kmp}")
            if response:
                yield from process(response)
            

if __name__ == "__main__":

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s]: %(message)s",
        handlers=[
            logging.FileHandler("log.txt"),
            logging.StreamHandler()
        ]
    )

    for road in (
        "T-04-06",
        "T-04-28",
        "T-05-09",
        "T-05-12",
        "T-05-13",
        "T-05-14",
        "T-05-15",
        "T-05-16",
        "T-05-18",
        "T-05-21",
        "T-05-23",
        "T-05-24",
        "T-05-25",
        "T-13-02",
        "T-13-06",
        "T-13-09"
    ):
        meta = remote_call(f"{ROUTE}/{road}")
        if not meta:
            continue
        start = int(meta["data"]["kmp_begin"].split(".")[0])
        end = int(meta["data"]["kmp_end"].split(".")[0])
        
        logging.info(f"{road=}, {start=}, {end=}")

        data = fetch(sc=road, min_km=start, max_km=end, m_freq=100)
        result = pd.DataFrame(data)
        result.drop_duplicates("kmp").to_csv(f"{road}.csv", index=False)