Skip to content

dust_client

KMA Yellow Dust (PM10) Observation API client.

This module provides a client for accessing the Korea Meteorological Administration's Yellow Dust (황사관측) API for PM10 particle observations.

Yellow dust observations monitor Asian dust storms and particulate matter concentrations for air quality assessment and public health alerts.

DustClient

Client for KMA Yellow Dust (PM10) Observation API.

The Yellow Dust observation system monitors PM10 particulate matter concentrations from Asian dust storms and other sources, providing critical air quality data for public health protection.

Source code in python/src/kma_mcp/surface/dust_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
class DustClient:
    """Client for KMA Yellow Dust (PM10) Observation API.

    The Yellow Dust observation system monitors PM10 particulate matter
    concentrations from Asian dust storms and other sources, providing
    critical air quality data for public health protection.
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0) -> None:
        """Initialize Yellow Dust 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.Client(timeout=timeout)

    def __enter__(self) -> 'DustClient':
        """Context manager entry."""
        return self

    def __exit__(self, *args: object) -> None:
        """Context manager exit."""
        self.close()

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

    def _make_request(self, endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
        """Make HTTP request to Yellow Dust 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 = self._client.get(url, params=params)
        response.raise_for_status()
        return response.json()

    def get_hourly_data(
        self,
        tm: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get hourly PM10 observation data for a single time.

        Args:
            tm: Time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations)

        Returns:
            Hourly PM10 observation data

        Example:
            >>> client = DustClient('your_auth_key')
            >>> data = client.get_hourly_data('202501011200')
            >>> # Or using datetime
            >>> from datetime import datetime
            >>> data = client.get_hourly_data(datetime(2025, 1, 1, 12, 0))
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params = {'tm': tm, 'stn': str(stn), 'help': '0'}
        return self._make_request('kma_pm10.php', params)

    def get_hourly_period(
        self,
        tm1: str | datetime,
        tm2: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get hourly PM10 observation data for a time period.

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

        Returns:
            Hourly PM10 observation data for the period

        Example:
            >>> client = DustClient('your_auth_key')
            >>> data = client.get_hourly_period('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 self._make_request('kma_pm10_2.php', params)

    def get_daily_data(
        self,
        tm: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get daily PM10 observation data for a single day.

        Args:
            tm: Date in 'YYYYMMDD' format or datetime object
            stn: Station number (0 for all stations)

        Returns:
            Daily PM10 observation data

        Example:
            >>> client = DustClient('your_auth_key')
            >>> data = client.get_daily_data('20250101')
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d')

        params = {'tm': tm, 'stn': str(stn), 'help': '0'}
        return self._make_request('kma_pm10_day.php', params)

    def get_daily_period(
        self,
        tm1: str | datetime,
        tm2: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get daily PM10 observation data for a time period.

        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)

        Returns:
            Daily PM10 observation data for the period

        Example:
            >>> client = DustClient('your_auth_key')
            >>> data = client.get_daily_period('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 self._make_request('kma_pm10_day2.php', params)

close()

Close the HTTP client.

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

get_daily_data(tm, stn=0)

Get daily PM10 observation data for a single day.

Parameters:

  • tm (str | datetime) –

    Date in 'YYYYMMDD' format or datetime object

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

    Station number (0 for all stations)

Returns:

Example

client = DustClient('your_auth_key') data = client.get_daily_data('20250101')

Source code in python/src/kma_mcp/surface/dust_client.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def get_daily_data(
    self,
    tm: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get daily PM10 observation data for a single day.

    Args:
        tm: Date in 'YYYYMMDD' format or datetime object
        stn: Station number (0 for all stations)

    Returns:
        Daily PM10 observation data

    Example:
        >>> client = DustClient('your_auth_key')
        >>> data = client.get_daily_data('20250101')
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d')

    params = {'tm': tm, 'stn': str(stn), 'help': '0'}
    return self._make_request('kma_pm10_day.php', params)

get_daily_period(tm1, tm2, stn=0)

Get daily PM10 observation data for a time period.

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)

Returns:

  • dict[str, Any] –

    Daily PM10 observation data for the period

Example

client = DustClient('your_auth_key') data = client.get_daily_period('20250101', '20250131')

Source code in python/src/kma_mcp/surface/dust_client.py 
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
def get_daily_period(
    self,
    tm1: str | datetime,
    tm2: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get daily PM10 observation data for a time period.

    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)

    Returns:
        Daily PM10 observation data for the period

    Example:
        >>> client = DustClient('your_auth_key')
        >>> data = client.get_daily_period('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 self._make_request('kma_pm10_day2.php', params)

get_hourly_data(tm, stn=0)

Get hourly PM10 observation data for a single time.

Parameters:

  • tm (str | datetime) –

    Time in 'YYYYMMDDHHmm' format or datetime object

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

    Station number (0 for all stations)

Returns:

Example

client = DustClient('your_auth_key') data = client.get_hourly_data('202501011200')

Or using datetime

from datetime import datetime data = client.get_hourly_data(datetime(2025, 1, 1, 12, 0))

Source code in python/src/kma_mcp/surface/dust_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
93
def get_hourly_data(
    self,
    tm: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get hourly PM10 observation data for a single time.

    Args:
        tm: Time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations)

    Returns:
        Hourly PM10 observation data

    Example:
        >>> client = DustClient('your_auth_key')
        >>> data = client.get_hourly_data('202501011200')
        >>> # Or using datetime
        >>> from datetime import datetime
        >>> data = client.get_hourly_data(datetime(2025, 1, 1, 12, 0))
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params = {'tm': tm, 'stn': str(stn), 'help': '0'}
    return self._make_request('kma_pm10.php', params)

get_hourly_period(tm1, tm2, stn=0)

Get hourly PM10 observation data for 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

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

    Station number (0 for all stations)

Returns:

  • dict[str, Any] –

    Hourly PM10 observation data for the period

Example

client = DustClient('your_auth_key') data = client.get_hourly_period('202501010000', '202501020000')

Source code in python/src/kma_mcp/surface/dust_client.py 
 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
def get_hourly_period(
    self,
    tm1: str | datetime,
    tm2: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get hourly PM10 observation data for a time period.

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

    Returns:
        Hourly PM10 observation data for the period

    Example:
        >>> client = DustClient('your_auth_key')
        >>> data = client.get_hourly_period('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 self._make_request('kma_pm10_2.php', params)