Skip to content

integrated_client

Client for KMA Integrated Meteorology API.

Provides access to integrated meteorological data combining various observation sources and specialized data products.

IntegratedClient

Client for accessing KMA Integrated Meteorology data.

Provides access to: - Lightning detection data - Wind profiler data - Integrated observation products

Source code in python/src/kma_mcp/integrated/integrated_client.py 
 12
 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
class IntegratedClient:
    """Client for accessing KMA Integrated Meteorology data.

    Provides access to:
    - Lightning detection data
    - Wind profiler data
    - Integrated observation products
    """

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

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

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

    def __enter__(self) -> 'IntegratedClient':
        """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 an API request."""
        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_lightning_data(
        self,
        tm1: str,
        tm2: str,
    ) -> dict[str, Any]:
        """Get lightning detection data.

        Provides lightning strike location and intensity data from
        the KMA lightning detection network.

        Args:
            tm1: Start time in 'YYYYMMDDHHmm' format
            tm2: End time in 'YYYYMMDDHHmm' format

        Returns:
            Lightning detection data including location and intensity

        Example:
            >>> with IntegratedClient('api_key') as client:
            >>>     data = client.get_lightning_data(
            >>>         '202501011200', '202501011500'
            >>>     )
        """
        params = {'tm1': tm1, 'tm2': tm2, 'help': '0'}
        return self._make_request('lgt_kma_np3.php', params)

    def get_wind_profiler_data(
        self,
        tm: str,
        stn: int = 0,
        mode: str = 'L',
    ) -> dict[str, Any]:
        """Get wind profiler data.

        Wind profilers provide vertical profiles of wind speed and
        direction at various altitudes.

        Args:
            tm: Observation time in 'YYYYMMDDHHmm' format
            stn: Station ID (0 for all stations, default: 0)
            mode: Data mode (default: 'L')
                  Options: 'L' (low mode), 'H' (high mode)

        Returns:
            Wind profiler vertical profile data

        Example:
            >>> with IntegratedClient('api_key') as client:
            >>>     data = client.get_wind_profiler_data(
            >>>         '202501011200', stn=0, mode='L'
            >>>     )
        """
        params = {'tm': tm, 'stn': stn, 'mode': mode, 'help': '0'}
        return self._make_request('kma_wpf.php', params)

close()

Close the HTTP client.

Source code in python/src/kma_mcp/integrated/integrated_client.py 
41
42
43
def close(self) -> None:
    """Close the HTTP client."""
    self._client.close()

get_lightning_data(tm1, tm2)

Get lightning detection data.

Provides lightning strike location and intensity data from the KMA lightning detection network.

Parameters:

  • tm1 (str) –

    Start time in 'YYYYMMDDHHmm' format

  • tm2 (str) –

    End time in 'YYYYMMDDHHmm' format

Returns:

  • dict[str, Any] –

    Lightning detection data including location and intensity

Example

with IntegratedClient('api_key') as client: data = client.get_lightning_data( '202501011200', '202501011500' )

Source code in python/src/kma_mcp/integrated/integrated_client.py 
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
def get_lightning_data(
    self,
    tm1: str,
    tm2: str,
) -> dict[str, Any]:
    """Get lightning detection data.

    Provides lightning strike location and intensity data from
    the KMA lightning detection network.

    Args:
        tm1: Start time in 'YYYYMMDDHHmm' format
        tm2: End time in 'YYYYMMDDHHmm' format

    Returns:
        Lightning detection data including location and intensity

    Example:
        >>> with IntegratedClient('api_key') as client:
        >>>     data = client.get_lightning_data(
        >>>         '202501011200', '202501011500'
        >>>     )
    """
    params = {'tm1': tm1, 'tm2': tm2, 'help': '0'}
    return self._make_request('lgt_kma_np3.php', params)

get_wind_profiler_data(tm, stn=0, mode='L')

Get wind profiler data.

Wind profilers provide vertical profiles of wind speed and direction at various altitudes.

Parameters:

  • tm (str) –

    Observation time in 'YYYYMMDDHHmm' format

  • stn (int, default: 0 ) –

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

  • mode (str, default: 'L' ) –

    Data mode (default: 'L') Options: 'L' (low mode), 'H' (high mode)

Returns:

  • dict[str, Any] –

    Wind profiler vertical profile data

Example

with IntegratedClient('api_key') as client: data = client.get_wind_profiler_data( '202501011200', stn=0, mode='L' )

Source code in python/src/kma_mcp/integrated/integrated_client.py 
 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
def get_wind_profiler_data(
    self,
    tm: str,
    stn: int = 0,
    mode: str = 'L',
) -> dict[str, Any]:
    """Get wind profiler data.

    Wind profilers provide vertical profiles of wind speed and
    direction at various altitudes.

    Args:
        tm: Observation time in 'YYYYMMDDHHmm' format
        stn: Station ID (0 for all stations, default: 0)
        mode: Data mode (default: 'L')
              Options: 'L' (low mode), 'H' (high mode)

    Returns:
        Wind profiler vertical profile data

    Example:
        >>> with IntegratedClient('api_key') as client:
        >>>     data = client.get_wind_profiler_data(
        >>>         '202501011200', stn=0, mode='L'
        >>>     )
    """
    params = {'tm': tm, 'stn': stn, 'mode': mode, 'help': '0'}
    return self._make_request('kma_wpf.php', params)