Skip to content

async_snow_client

Async KMA Snow Depth Observation API client.

This module provides a client for accessing the Korea Meteorological Administration's Snow Depth (적설관측) API for snow accumulation observations.

Snow depth observations monitor snow accumulation for winter weather analysis, transportation safety, and disaster prevention.

AsyncSnowClient

Async client for KMA Snow Depth Observation API.

The snow depth observation system monitors snow accumulation and provides critical data for winter weather analysis, transportation safety, and disaster prevention.

Snow depth types
  • tot: Total snow depth (적설)
  • day: Daily new snow (일신적설)
  • 3hr: 3-hour new snow (3시간 신적설)
  • 24h: 24-hour new snow (24시간 신적설)
Source code in python/src/kma_mcp/surface/async_snow_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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class AsyncSnowClient:
    """Async client for KMA Snow Depth Observation API.

    The snow depth observation system monitors snow accumulation
    and provides critical data for winter weather analysis,
    transportation safety, and disaster prevention.

    Snow depth types:
        - tot: Total snow depth (적설)
        - day: Daily new snow (일신적설)
        - 3hr: 3-hour new snow (3시간 신적설)
        - 24h: 24-hour new snow (24시간 신적설)
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0) -> None:
        """Initialize Snow Depth 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) -> 'AsyncSnowClient':
        """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 Snow Depth 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_snow_depth(
        self,
        tm: str | datetime,
        sd_type: Literal['tot', 'day', '3hr', '24h'] = 'tot',
    ) -> dict[str, Any]:
        """Get snow depth observation data.

        Documented endpoint: kma_snow1.php
        Reference: API_ENDPOINT_Surface.md line 1370-1393

        Args:
            tm: Time in 'YYYYMMDDHHmm' format or datetime object
            sd_type: Snow depth type:
                - 'tot': Total snow depth (적설)
                - 'day': Daily new snow (일신적설)
                - '3hr': 3-hour new snow
                - '24h': 24-hour new snow

        Returns:
            Snow depth observation data

        Example:
            >>> async with AsyncSnowClient('your_auth_key') as client:
            ...     # Get total snow depth
            ...     data = await client.get_snow_depth('201412051800', sd_type='tot')
            ...     # Get daily new snow
            ...     data = await client.get_snow_depth('201412051800', sd_type='day')
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params = {'sd': sd_type, 'tm': tm, 'help': '1'}
        return await self._make_request('kma_snow1.php', params)

    async def get_snow_period(
        self,
        tm: str | datetime,
        tm_st: str | datetime,
        snow: int = 0,
    ) -> dict[str, Any]:
        """Get snow depth data for a period.

        Documented endpoint: kma_snow2.php
        Reference: API_ENDPOINT_Surface.md line 1394-1399

        Args:
            tm: End time in 'YYYYMMDDHHmm' format or datetime object
            tm_st: Start time in 'YYYYMMDDHHmm' format or datetime object
            snow: Snow parameter (default: 0)

        Returns:
            Snow depth data for the period

        Example:
            >>> async with AsyncSnowClient('your_auth_key') as client:
            ...     data = await client.get_snow_period('201412051800', '201412040100')
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')
        if isinstance(tm_st, datetime):
            tm_st = tm_st.strftime('%Y%m%d%H%M')

        params = {'tm': tm, 'tm_st': tm_st, 'snow': str(snow), 'help': '1'}
        return await self._make_request('kma_snow2.php', params)

    async def get_max_snow_depth(
        self,
        tm: str | datetime,
        tm_st: str | datetime,
        sd_type: Literal['tot', 'day'] = 'tot',
        stn: int | str = 0,
        snow: int = 0,
    ) -> dict[str, Any]:
        """Get maximum snow depth for a period.

        Documented endpoint: kma_snow_day.php
        Reference: API_ENDPOINT_Surface.md line 1400-1417

        Args:
            tm: End date in 'YYYYMMDD' format or datetime object
            tm_st: Start date in 'YYYYMMDD' format or datetime object
            sd_type: Snow depth type:
                - 'tot': Maximum total snow depth (최심적설)
                - 'day': Maximum daily new snow (최심신적설)
            stn: Station number (0 for all stations)
            snow: Snow parameter (default: 0)

        Returns:
            Maximum snow depth data for the period

        Example:
            >>> async with AsyncSnowClient('your_auth_key') as client:
            ...     # Get maximum total snow depth
            ...     data = await client.get_max_snow_depth('20150131', '20150125', sd_type='tot')
            ...     # Get maximum daily new snow
            ...     data = await client.get_max_snow_depth('20150131', '20150125', sd_type='day')
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d')
        if isinstance(tm_st, datetime):
            tm_st = tm_st.strftime('%Y%m%d')

        params = {
            'sd': sd_type,
            'tm': tm,
            'tm_st': tm_st,
            'stn': str(stn),
            'snow': str(snow),
            'help': '1',
        }
        return await self._make_request('kma_snow_day.php', params)

close() async

Close the HTTP client.

Source code in python/src/kma_mcp/surface/async_snow_client.py 
51
52
53
async def close(self) -> None:
    """Close the HTTP client."""
    await self._client.aclose()

get_max_snow_depth(tm, tm_st, sd_type='tot', stn=0, snow=0) async

Get maximum snow depth for a period.

Documented endpoint: kma_snow_day.php Reference: API_ENDPOINT_Surface.md line 1400-1417

Parameters:

  • tm (str | datetime) –

    End date in 'YYYYMMDD' format or datetime object

  • tm_st (str | datetime) –

    Start date in 'YYYYMMDD' format or datetime object

  • sd_type (Literal['tot', 'day'], default: 'tot' ) –

    Snow depth type: - 'tot': Maximum total snow depth (최심적설) - 'day': Maximum daily new snow (최심신적설)

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

    Station number (0 for all stations)

  • snow (int, default: 0 ) –

    Snow parameter (default: 0)

Returns:

  • dict[str, Any] –

    Maximum snow depth data for the period

Example

async with AsyncSnowClient('your_auth_key') as client: ... # Get maximum total snow depth ... data = await client.get_max_snow_depth('20150131', '20150125', sd_type='tot') ... # Get maximum daily new snow ... data = await client.get_max_snow_depth('20150131', '20150125', sd_type='day')

Source code in python/src/kma_mcp/surface/async_snow_client.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
async def get_max_snow_depth(
    self,
    tm: str | datetime,
    tm_st: str | datetime,
    sd_type: Literal['tot', 'day'] = 'tot',
    stn: int | str = 0,
    snow: int = 0,
) -> dict[str, Any]:
    """Get maximum snow depth for a period.

    Documented endpoint: kma_snow_day.php
    Reference: API_ENDPOINT_Surface.md line 1400-1417

    Args:
        tm: End date in 'YYYYMMDD' format or datetime object
        tm_st: Start date in 'YYYYMMDD' format or datetime object
        sd_type: Snow depth type:
            - 'tot': Maximum total snow depth (최심적설)
            - 'day': Maximum daily new snow (최심신적설)
        stn: Station number (0 for all stations)
        snow: Snow parameter (default: 0)

    Returns:
        Maximum snow depth data for the period

    Example:
        >>> async with AsyncSnowClient('your_auth_key') as client:
        ...     # Get maximum total snow depth
        ...     data = await client.get_max_snow_depth('20150131', '20150125', sd_type='tot')
        ...     # Get maximum daily new snow
        ...     data = await client.get_max_snow_depth('20150131', '20150125', sd_type='day')
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d')
    if isinstance(tm_st, datetime):
        tm_st = tm_st.strftime('%Y%m%d')

    params = {
        'sd': sd_type,
        'tm': tm,
        'tm_st': tm_st,
        'stn': str(stn),
        'snow': str(snow),
        'help': '1',
    }
    return await self._make_request('kma_snow_day.php', params)

get_snow_depth(tm, sd_type='tot') async

Get snow depth observation data.

Documented endpoint: kma_snow1.php Reference: API_ENDPOINT_Surface.md line 1370-1393

Parameters:

  • tm (str | datetime) –

    Time in 'YYYYMMDDHHmm' format or datetime object

  • sd_type (Literal['tot', 'day', '3hr', '24h'], default: 'tot' ) –

    Snow depth type: - 'tot': Total snow depth (적설) - 'day': Daily new snow (일신적설) - '3hr': 3-hour new snow - '24h': 24-hour new snow

Returns:

Example

async with AsyncSnowClient('your_auth_key') as client: ... # Get total snow depth ... data = await client.get_snow_depth('201412051800', sd_type='tot') ... # Get daily new snow ... data = await client.get_snow_depth('201412051800', sd_type='day')

Source code in python/src/kma_mcp/surface/async_snow_client.py 
 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
async def get_snow_depth(
    self,
    tm: str | datetime,
    sd_type: Literal['tot', 'day', '3hr', '24h'] = 'tot',
) -> dict[str, Any]:
    """Get snow depth observation data.

    Documented endpoint: kma_snow1.php
    Reference: API_ENDPOINT_Surface.md line 1370-1393

    Args:
        tm: Time in 'YYYYMMDDHHmm' format or datetime object
        sd_type: Snow depth type:
            - 'tot': Total snow depth (적설)
            - 'day': Daily new snow (일신적설)
            - '3hr': 3-hour new snow
            - '24h': 24-hour new snow

    Returns:
        Snow depth observation data

    Example:
        >>> async with AsyncSnowClient('your_auth_key') as client:
        ...     # Get total snow depth
        ...     data = await client.get_snow_depth('201412051800', sd_type='tot')
        ...     # Get daily new snow
        ...     data = await client.get_snow_depth('201412051800', sd_type='day')
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params = {'sd': sd_type, 'tm': tm, 'help': '1'}
    return await self._make_request('kma_snow1.php', params)

get_snow_period(tm, tm_st, snow=0) async

Get snow depth data for a period.

Documented endpoint: kma_snow2.php Reference: API_ENDPOINT_Surface.md line 1394-1399

Parameters:

  • tm (str | datetime) –

    End time in 'YYYYMMDDHHmm' format or datetime object

  • tm_st (str | datetime) –

    Start time in 'YYYYMMDDHHmm' format or datetime object

  • snow (int, default: 0 ) –

    Snow parameter (default: 0)

Returns:

  • dict[str, Any] –

    Snow depth data for the period

Example

async with AsyncSnowClient('your_auth_key') as client: ... data = await client.get_snow_period('201412051800', '201412040100')

Source code in python/src/kma_mcp/surface/async_snow_client.py 
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
async def get_snow_period(
    self,
    tm: str | datetime,
    tm_st: str | datetime,
    snow: int = 0,
) -> dict[str, Any]:
    """Get snow depth data for a period.

    Documented endpoint: kma_snow2.php
    Reference: API_ENDPOINT_Surface.md line 1394-1399

    Args:
        tm: End time in 'YYYYMMDDHHmm' format or datetime object
        tm_st: Start time in 'YYYYMMDDHHmm' format or datetime object
        snow: Snow parameter (default: 0)

    Returns:
        Snow depth data for the period

    Example:
        >>> async with AsyncSnowClient('your_auth_key') as client:
        ...     data = await client.get_snow_period('201412051800', '201412040100')
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')
    if isinstance(tm_st, datetime):
        tm_st = tm_st.strftime('%Y%m%d%H%M')

    params = {'tm': tm, 'tm_st': tm_st, 'snow': str(snow), 'help': '1'}
    return await self._make_request('kma_snow2.php', params)