Skip to content

async_aws_oa_client

Async KMA AWS Objective Analysis API client.

This module provides a client for accessing the Korea Meteorological Administration's AWS Objective Analysis (AWS 객관분석) API for objectively analyzed weather data.

AWS Objective Analysis provides gridded meteorological data derived from AWS observations through objective analysis techniques for improved spatial coverage.

AsyncAWSOAClient

Async client for KMA AWS Objective Analysis API.

The AWS Objective Analysis system provides gridded meteorological data derived from AWS observations through objective analysis techniques, offering improved spatial coverage and consistency for weather analysis.

Source code in python/src/kma_mcp/surface/async_aws_oa_client.py 
 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
class AsyncAWSOAClient:
    """Async client for KMA AWS Objective Analysis API.

    The AWS Objective Analysis system provides gridded meteorological data
    derived from AWS observations through objective analysis techniques,
    offering improved spatial coverage and consistency for weather analysis.
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0) -> None:
        """Initialize AWS Objective Analysis client.

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

    async def __aenter__(self) -> 'AsyncAWSOAClient':
        """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 HTTP request to AWS Objective Analysis API.

        Args:
            endpoint: API endpoint path
            params: Query parameters

        Returns:
            API response as dictionary

        Raises:
            httpx.HTTPError: If request fails
        """
        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_analysis_data(
        self,
        tm: str | datetime,
        x: float,
        y: float,
    ) -> dict[str, Any]:
        """Get AWS objective analysis data for a specific location and time.

        Args:
            tm: Time in 'YYYYMMDDHHmm' format or datetime object
            x: Longitude coordinate
            y: Latitude coordinate

        Returns:
            AWS objective analysis data

        Example:
            >>> async with AsyncAWSOAClient('your_auth_key')
            >>> ...     data = await client.get_analysis_data('202501011200', 127.0, 37.5)
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params = {'tm': tm, 'x': str(x), 'y': str(y), 'help': '0'}
        return await self._make_request('kma_awsoa.php', params)

    async def get_analysis_period(
        self,
        tm1: str | datetime,
        tm2: str | datetime,
        x: float,
        y: float,
    ) -> dict[str, Any]:
        """Get AWS objective analysis data for a location over a time period.

        Args:
            tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
            tm2: End time in 'YYYYMMDDHHmm' format or datetime object
            x: Longitude coordinate
            y: Latitude coordinate

        Returns:
            AWS objective analysis data for the period

        """
        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, 'x': str(x), 'y': str(y), 'help': '0'}
        return await self._make_request('kma_awsoa_2.php', params)

close() async

Close the HTTP client.

Source code in python/src/kma_mcp/surface/async_aws_oa_client.py 
45
46
47
async def close(self) -> None:
    """Close the HTTP client."""
    await self._client.aclose()

get_analysis_data(tm, x, y) async

Get AWS objective analysis data for a specific location and time.

Parameters:

  • tm (str | datetime) –

    Time in 'YYYYMMDDHHmm' format or datetime object

  • x (float) –

    Longitude coordinate

  • y (float) –

    Latitude coordinate

Returns:

Example

async with AsyncAWSOAClient('your_auth_key') ... data = await client.get_analysis_data('202501011200', 127.0, 37.5)

Source code in python/src/kma_mcp/surface/async_aws_oa_client.py 
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
async def get_analysis_data(
    self,
    tm: str | datetime,
    x: float,
    y: float,
) -> dict[str, Any]:
    """Get AWS objective analysis data for a specific location and time.

    Args:
        tm: Time in 'YYYYMMDDHHmm' format or datetime object
        x: Longitude coordinate
        y: Latitude coordinate

    Returns:
        AWS objective analysis data

    Example:
        >>> async with AsyncAWSOAClient('your_auth_key')
        >>> ...     data = await client.get_analysis_data('202501011200', 127.0, 37.5)
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params = {'tm': tm, 'x': str(x), 'y': str(y), 'help': '0'}
    return await self._make_request('kma_awsoa.php', params)

get_analysis_period(tm1, tm2, x, y) async

Get AWS objective analysis data for a location over a time period.

Parameters:

  • tm1 (str | datetime) –

    Start time in 'YYYYMMDDHHmm' format or datetime object

  • tm2 (str | datetime) –

    End time in 'YYYYMMDDHHmm' format or datetime object

  • x (float) –

    Longitude coordinate

  • y (float) –

    Latitude coordinate

Returns:

  • dict[str, Any] –

    AWS objective analysis data for the period

Source code in python/src/kma_mcp/surface/async_aws_oa_client.py 
 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
async def get_analysis_period(
    self,
    tm1: str | datetime,
    tm2: str | datetime,
    x: float,
    y: float,
) -> dict[str, Any]:
    """Get AWS objective analysis data for a location over a time period.

    Args:
        tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
        tm2: End time in 'YYYYMMDDHHmm' format or datetime object
        x: Longitude coordinate
        y: Latitude coordinate

    Returns:
        AWS objective analysis data for the period

    """
    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, 'x': str(x), 'y': str(y), 'help': '0'}
    return await self._make_request('kma_awsoa_2.php', params)