Skip to content

async_radiosonde_client

Async client for KMA Upper-Air Observations (Radiosonde) API.

This module provides access to upper-air meteorological observations from radiosondes, including temperature, humidity, wind data at various altitude levels.

AsyncRadiosondeClient

Async client for accessing KMA Upper-Air (Radiosonde) observation data.

Provides access to atmospheric profiles including: - Temperature and dew point at altitude - Wind direction and speed at various levels - Pressure levels and heights - Atmospheric stability indices

Source code in python/src/kma_mcp/upper_air/async_radiosonde_client.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class AsyncRadiosondeClient:
    """Async client for accessing KMA Upper-Air (Radiosonde) observation data.

    Provides access to atmospheric profiles including:
    - Temperature and dew point at altitude
    - Wind direction and speed at various levels
    - Pressure levels and heights
    - Atmospheric stability indices
    """

    BASE_URL = 'https://apihub.kma.go.kr/api/typ01/url'

    def __init__(self, auth_key: str, timeout: float = 30.0):
        """Initialize the Radiosonde client.

        Args:
            auth_key: KMA API authentication key
            timeout: Request timeout in seconds (default: 30.0)
        """
        self.auth_key = auth_key
        self._client = httpx.AsyncClient(timeout=timeout)

    async def __aenter__(self) -> 'AsyncRadiosondeClient':
        """Async context manager entry."""
        return self

    async def __aexit__(self, *args: object) -> None:
        """Async context manager exit."""
        await self.close()

    async def close(self) -> None:
        """Close the HTTP client."""
        await self._client.aclose()

    async def _make_request(self, endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
        """Make an API request.

        Args:
            endpoint: API endpoint name
            params: Query parameters

        Returns:
            JSON response as dictionary
        """
        params['authKey'] = self.auth_key
        url = f'{self.BASE_URL}/{endpoint}'
        response = await self._client.get(url, params=params)
        response.raise_for_status()
        return response.json()

    async def get_upper_air_data(
        self, tm: str | datetime, stn: int | str = 0, pa: float | None = None
    ) -> dict[str, Any]:
        """Get upper-air (TEMP) radiosonde data.

        Args:
            tm: UTC observation time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations, default: 0)
            pa: Pressure level in hPa (optional, e.g., 850, 500, 250)

        Returns:
            Upper-air observation data including temperature, humidity, wind

        Example:
            >>> async with AsyncRadiosondeClient('your_api_key')
            >>> ...     data = await client.get_upper_air_data('202501010000', stn=47122, pa=850)
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params: dict[str, Any] = {'tm': tm, 'stn': str(stn), 'help': '0'}
        if pa is not None:
            params['pa'] = str(pa)

        return await self._make_request('upp_temp.php', params)

    async def get_stability_indices(
        self, tm1: str | datetime, tm2: str | datetime, stn: int | str = 0
    ) -> dict[str, Any]:
        """Get atmospheric stability analysis data.

        Provides derived indices for convective forecasting including
        CAPE, K-index, lifted index, and cloud layer information.

        Args:
            tm1: Start UTC time in 'YYYYMMDDHHmm' format or datetime object
            tm2: End UTC time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations, default: 0)

        Returns:
            Atmospheric stability indices

        Example:
            >>> async with AsyncRadiosondeClient('your_api_key')
            >>> ...     data = await client.get_stability_indices('202501010000', '202501020000')
        """
        if isinstance(tm1, datetime):
            tm1 = tm1.strftime('%Y%m%d%H%M')
        if isinstance(tm2, datetime):
            tm2 = tm2.strftime('%Y%m%d%H%M')

        params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
        return await self._make_request('upp_idx.php', params)

    async def get_maximum_altitude_data(
        self, tm1: str | datetime, tm2: str | datetime, stn: int | str = 0
    ) -> dict[str, Any]:
        """Get maximum altitude reached during radiosonde ascent.

        Args:
            tm1: Start date in 'YYYYMMDD' format or datetime object
            tm2: End date in 'YYYYMMDD' format or datetime object
            stn: Station number (0 for all stations, default: 0)

        Returns:
            Maximum altitude data including flight time, height, horizontal distance

        Example:
            >>> async with AsyncRadiosondeClient('your_api_key')
            >>> ...     data = await client.get_maximum_altitude_data('20250101', '20250131')
        """
        if isinstance(tm1, datetime):
            tm1 = tm1.strftime('%Y%m%d')
        if isinstance(tm2, datetime):
            tm2 = tm2.strftime('%Y%m%d')

        params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
        return await self._make_request('upp_raw_max.php', params)

close() async

Close the HTTP client.

Source code in python/src/kma_mcp/upper_air/async_radiosonde_client.py 
43
44
45
async def close(self) -> None:
    """Close the HTTP client."""
    await self._client.aclose()

get_maximum_altitude_data(tm1, tm2, stn=0) async

Get maximum altitude reached during radiosonde ascent.

Parameters:

  • tm1 (str | datetime) –

    Start date in 'YYYYMMDD' format or datetime object

  • tm2 (str | datetime) –

    End date in 'YYYYMMDD' format or datetime object

  • stn (int | str, default: 0 ) –

    Station number (0 for all stations, default: 0)

Returns:

  • dict[str, Any] –

    Maximum altitude data including flight time, height, horizontal distance

Example

async with AsyncRadiosondeClient('your_api_key') ... data = await client.get_maximum_altitude_data('20250101', '20250131')

Source code in python/src/kma_mcp/upper_air/async_radiosonde_client.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def get_maximum_altitude_data(
    self, tm1: str | datetime, tm2: str | datetime, stn: int | str = 0
) -> dict[str, Any]:
    """Get maximum altitude reached during radiosonde ascent.

    Args:
        tm1: Start date in 'YYYYMMDD' format or datetime object
        tm2: End date in 'YYYYMMDD' format or datetime object
        stn: Station number (0 for all stations, default: 0)

    Returns:
        Maximum altitude data including flight time, height, horizontal distance

    Example:
        >>> async with AsyncRadiosondeClient('your_api_key')
        >>> ...     data = await client.get_maximum_altitude_data('20250101', '20250131')
    """
    if isinstance(tm1, datetime):
        tm1 = tm1.strftime('%Y%m%d')
    if isinstance(tm2, datetime):
        tm2 = tm2.strftime('%Y%m%d')

    params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
    return await self._make_request('upp_raw_max.php', params)

get_stability_indices(tm1, tm2, stn=0) async

Get atmospheric stability analysis data.

Provides derived indices for convective forecasting including CAPE, K-index, lifted index, and cloud layer information.

Parameters:

  • tm1 (str | datetime) –

    Start UTC time in 'YYYYMMDDHHmm' format or datetime object

  • tm2 (str | datetime) –

    End UTC time in 'YYYYMMDDHHmm' format or datetime object

  • stn (int | str, default: 0 ) –

    Station number (0 for all stations, default: 0)

Returns:

  • dict[str, Any] –

    Atmospheric stability indices

Example

async with AsyncRadiosondeClient('your_api_key') ... data = await client.get_stability_indices('202501010000', '202501020000')

Source code in python/src/kma_mcp/upper_air/async_radiosonde_client.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def get_stability_indices(
    self, tm1: str | datetime, tm2: str | datetime, stn: int | str = 0
) -> dict[str, Any]:
    """Get atmospheric stability analysis data.

    Provides derived indices for convective forecasting including
    CAPE, K-index, lifted index, and cloud layer information.

    Args:
        tm1: Start UTC time in 'YYYYMMDDHHmm' format or datetime object
        tm2: End UTC time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations, default: 0)

    Returns:
        Atmospheric stability indices

    Example:
        >>> async with AsyncRadiosondeClient('your_api_key')
        >>> ...     data = await client.get_stability_indices('202501010000', '202501020000')
    """
    if isinstance(tm1, datetime):
        tm1 = tm1.strftime('%Y%m%d%H%M')
    if isinstance(tm2, datetime):
        tm2 = tm2.strftime('%Y%m%d%H%M')

    params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
    return await self._make_request('upp_idx.php', params)

get_upper_air_data(tm, stn=0, pa=None) async

Get upper-air (TEMP) radiosonde data.

Parameters:

  • tm (str | datetime) –

    UTC observation time in 'YYYYMMDDHHmm' format or datetime object

  • stn (int | str, default: 0 ) –

    Station number (0 for all stations, default: 0)

  • pa (float | None, default: None ) –

    Pressure level in hPa (optional, e.g., 850, 500, 250)

Returns:

  • dict[str, Any] –

    Upper-air observation data including temperature, humidity, wind

Example

async with AsyncRadiosondeClient('your_api_key') ... data = await client.get_upper_air_data('202501010000', stn=47122, pa=850)

Source code in python/src/kma_mcp/upper_air/async_radiosonde_client.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def get_upper_air_data(
    self, tm: str | datetime, stn: int | str = 0, pa: float | None = None
) -> dict[str, Any]:
    """Get upper-air (TEMP) radiosonde data.

    Args:
        tm: UTC observation time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations, default: 0)
        pa: Pressure level in hPa (optional, e.g., 850, 500, 250)

    Returns:
        Upper-air observation data including temperature, humidity, wind

    Example:
        >>> async with AsyncRadiosondeClient('your_api_key')
        >>> ...     data = await client.get_upper_air_data('202501010000', stn=47122, pa=850)
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params: dict[str, Any] = {'tm': tm, 'stn': str(stn), 'help': '0'}
    if pa is not None:
        params['pa'] = str(pa)

    return await self._make_request('upp_temp.php', params)