[python, html/css] 안개 속보 화면 만들기: 2. 자료 다운로드와 처리

 

- 개요

기상청 API허브에서 필요한 자료를 다운로드 받습니다.

먼저 API로 받은 AWS 시정거리 자료를 pandas.DataFrame 형식으로 저장합니다.

 

다음으로 AWS가 어디 권역에 속하는지 알아야합니다.

구역 4를 보시면 "권역: 관측지점 시정거리" 이런 형식의 글이 있습니다.

 

안개 속보 화면의 일부

 

시정거리가 작은 지역을 권역별로 알려주는 듯합니다.

AWS의 위치 정보는 마찬가지로 기상청 API 허브에 있습니다.

다만 이 자료를 pandas.DataFrame에 넣기 위해 파싱(parsing)하는 과정이 생각보다 까다롭습니다.

 

마지막으로 시정거리와 위치 정보를 이용해서 구역4에 넣을 문자열을 생성합니다.

 

 

- 시정거리 자료 다운로드 및 처리

기상청 API허브 홈페이지로 들어가서  좌측 지상관측탭 > 상단 방재기상관측(AWS)탭에서 "AWS2 시정자료"를 찾습니다.

안개 속보에서는 10분 평균 시정을 쓰는데 이 자료에서는 10분 시정은 결측이고 1분 평균 시정값만 제대로 있습니다.

어쩔 수 없이 1분 평균 시정으로 시각화를 해보죠. 참고로 이 자료의 관측소의 수는 안개 속보의 관측소 수보다 적습니다.

다른 자료가 더 있는듯한데 안개 속보 화면을 만들기엔 충분합니다.

 

그럼 무슨 함수를 만들지 생각해보죠. 전 아래의 2가지 함수를 만들겁니다.

1. 요청 시간을 기준으로 url을 만들고 자료를 다운 받는 함수

2. API로 받은 자료를 pandas.DataFrame 형식으로 저장하는 함수

 

함수 이름: download_file(): 실시간 혹은 원하는 시간의 특정 자료를 API로 다운로드

입력값: APIkey(본인 기상청 API허브의 API키), base_url(원하는 자료랑 관련한 url값), time(실시간 혹은 원하는 시간 설정)

가끔 자료가 받아지지 않는데 코드를 다시 실행해보세요.

from datetime import datetime, timedelta
import requests

APIkey = {본인 API 입력}
"""
time 변수가 "realtime"인 경우 10분 단위에서 최근 시간 자료를 받게 설정합니다.
time 변수에 datetime 형식으로 시간을 넣으면 해당 시간 자료를 받습니다.
API로 자료를 다운로드 받을 때 어떤 자료를 받을지에 대한 정보는
url의 앞부분에 있습니다. 이 정보를 base_url 변수에 넣습니다.
"""
def download_file(APIkey, base_url, time="realtime"):
    if time=="realtime":
        current_time = datetime.now()
    else:
        current_time = time
            
    trimmed_time = current_time - timedelta(minutes=current_time.minute % 10, seconds=current_time.second, microseconds=current_time.microsecond)
    url = f"{base_url}{trimmed_time.strftime("%Y%m%d%H%M")}&stn=0&disp=0&help=1&authKey={APIkey}"
    
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        file_content = response.content.decode("cp949")
        lines = file_content.splitlines()
        return lines
    except requests.exceptions.RequestException as e:
        print(f"에러: {e}")
        
"""
예시로 안개 속보 날짜가 2024년 12월 2일 00시이므로 해당 시간값을 받았습니다.
base_url_vis는 시정자료를 받을 때 쓰는 url 앞부분입니다.
"""
time = datetime(2024, 12, 2, 0, 0, 0)
base_url_vis = "https://apihub.kma.go.kr/api/typ01/cgi-bin/url/nph-aws2_min_vis?tm2="
data_API = download_file(APIkey, base_url_vis, time=time)

 

함수 이름: to_pandas_vis(): API로 받은 텍스트를 파싱하고 pandas.DataFrame 형태로 저장

입력값: lines(download_file 함수의 return을 넣으면 됨)

참고로 파싱하는 과정은 훨씬 효율적으로 만들 수 있을 겁니다. 이건 대충 만든 겁니다.

이렇게 df_vis에 시정거리 자료를 저장합니다.

import pandas as pd
def to_pandas_vis(lines):
    header = lines[10].split()
    columns = header[1:]

    contents = lines[12:-2]
    dfs = []
    for content in contents:
        dfs.append(content.split())
    return pd.DataFrame(dfs, columns=columns)
    
df_vis = to_pandas_vis(data_API)

 

 

- AWS의 지리 정보 다운로드 및 처리

여기서 지리정보는 관측소가 위치한 행정구역을 의미합니다.

강원권, 충청권, 경상권, 전라권처럼 특정 권역으로 묶기 위해 AWS가 무슨 도에 속하는지 알아야합니다.

그리고 앞선 시정자료에 관측소 번호가 나오므로 특정 번호의 관측소가 어떤 도에 속하는지 알고 이 도가 어떤 권역에 속하는지를 판별해야합니다.

 

기상청 API허브의 왼쪽 지상관측탭 > 상단 지상관측정보탭에서 1. 지상관측 지점정보 조회 부분의 AWS 자료를 다운 받아야합니다. 아래 빨간 네모에 적힌 url을 써야겠죠.

 

AWS 지리 정보

 

아까 만든 download_file 함수를 활용하되 base_url을 바꿔줍니다.

마찬가지로 이 자료도 pandas.DataFrame 형식으로 저장을 할 건데요.

앞서 작성 to_pandas_vis 함수와 to_pandas_geo 함수는 좀 다릅니다.

일단 열 이름 정보를 담은 행이 몇 번재 인지를 입력값(row_header)으로 넣어야 합니다.

 'STN'란 열이름이 2개로 중복이라 하나를 columns[7]='STN_'으로 바꿔줍니다.

 

공백을 기준으로 파싱을 하는 식으로 코드를 짯습니다만 특정 문자열에 *이 있고 이 * 앞에 공백이 있습니다. * 문자를 아무 것도 없게 ''로 대체합니다.

또한, 영어지명에서 'Kang Jin'이 있었는데 이 또한 공백이 있어 분리가 되며 그럼 배열의 크기가 1개 더 커져 14개 됩니다.

if (len_split) == 13:처럼 원하는대로 파싱되었을 때의 배열 크기는 13개이므로 그렇지 않을 때를 따로 처리를 했습니다.

전 이 파일만 읽으면 되니까 범용적이지 않은 코드를 짠 것이지 웬만하면 범용적으로 짜는 것을 추천합니다.

 

혹시 범용적으로 코드를 짤 분들을 위해 팁을 드리자면 이 자료에는 한글이 있어 칸수를 기준으로 읽으면 문제가 있습니다. 제 생각에는 bytes 수를 이용해서 구역을 나누면 좀 더 범용적인 코드를 짤 수 있을 것 같네요.

base_url_geo = "https://apihub.kma.go.kr/api/typ01/url/stn_inf.php?inf=AWS&stn=&tm="
data_geo = download_file(APIkey, base_url_geo, time=time)

def to_pandas_geo(lines, row_header):
    columns = lines[row_header].strip().split()
    columns = columns[1:]
    columns[7] = 'STN_'
    
    data = []
    lines = lines[row_header+2:-2]
    for line in lines:
        line_split = line.strip().replace("*", "").split()
        if len(line_split) == 13:
            data.append(line_split)
        else: 
            line_ = [None]*13
            line_[0:9] = line_split[0:9]
            line_[10:] = line_split[11:]
            line_[9] = line_split[9] + line_split[10]
            data.append(line_)
    return pd.DataFrame(data, columns=columns)
df_geo = to_pandas_geo(data_geo, 18)

- 구역4의 "권역: 관측지점 이름 시정거리" 문자열 생성

위에서 관측소 번호로 어떤 행정구역에 속하는지 알아낸 다음 어떤 권역에 속하는지 알아야한다고 말했습니다.

이 작업을 거쳐야 시정거리가 적은 관측지점의 권역과 이름을 추출할 수 있습니다.

df_geo의 'LAW_ID'에 행정코드로 이를 통해 특정 관측소가 어떤 지역에 속하는지 알 수 있습니다.

'LAW_ID'의 앞 두자리 숫자가 어떤 도에 속하는지 알려주는 수입니다.

"""
권역: 행정코드 앞의 두자리 지역
수도권(3): 11 서울, 28 인천, 41 경기
강원도(1): 51 강원도
충청권(4): 30 대전 36 세종 43 충북 44 충남
전라권(3): 29 광주 52 전북 46 전남
경상권(5): 26 부산 27 대구 31 울산 47 경북 48 경남
제주도(1): 50 제주
"""
dict_AREA_to_LAW = {
    "수도권": [11, 28, 41],
    "강원도": [51],
    "충청권": [30, 36, 43, 44],
    "전라권": [29, 52, 46],
    "경상권": [26, 27, 31, 47, 48],
    "제주도": [50]
}

df_LAW_ID_2 = []
for i in df_geo['LAW_ID']:
    df_LAW_ID_2.append(int(i[0:2]))
    
df_geo['LAW_ID_2'] =df_LAW_ID_2

df_G = []
aa = 0
for i, law_id in enumerate(df_geo['LAW_ID_2']):
    for region, law_ids in dict_AREA_to_LAW.items():
        if law_id in law_ids:
            aa +=1
            df_G.append(region)
            break  

df_geo['G_NAME'] = df_G

"""
시정자료(df_vis)와 도 이름(df_geo)를 df_vis, 'STN'열을 기준으로 병합
"""
df = pd.merge(df_vis, df_geo, how="left", on="STN")

 

이제 시정거리가 낮은 지역을 찾은 뒤 권역('G_NAME' 열), 관측지점의 이름('STN_KO'열), 시정거리('VIS1' 열)를 표기하면 됩니다.

1000 m 보다 낮은 지역을 추출한 뒤 시정거리가 작은 순으로 최대 5개 지역까지만 표기합시다.

df['VIS1'] = df['VIS1'].astype(int) # string이라 int 변환
df_1000 = df[['VIS1', 'G_NAME', 'STN_KO']][df['VIS1'] < 1000] # 시정 1000 m 이하

top_5_by_region = (
    df_1000.sort_values(by='VIS1')  # VIS1 기준으로 정렬
    .groupby('G_NAME')                 # G_ONAME 그룹별
    .head(5)                           # 각 그룹에서 상위 5개 선택
)

"""
시정거리가 낮은 지역을 권역별 문장으로 정리
"""
sentences = []
for region in df_1000['G_NAME'].unique():
    ss = f'- {region}:'
    for index, row in top_5_by_region[["VIS1", "STN_KO"]][top_5_by_region['G_NAME'] == region].iterrows():
        ss += f' {row['STN_KO']} {int(row['VIS1']):.0f}'
    sentences.append(ss)

for sentence in sentences:
    print(sentence)
    
"""
print의 결과
- 수도권: 판문점 72 파주 339 시흥 365 오산 490 안산 660
- 전라권: 임실 150 임실강진 170 장성 180 월야 195 순창군 200
- 경상권: 함양군 961
- 충청권: 음성 760 세종연서 907

"""

 

전 1분 자료를 쓰고 기상청은 10분 자료를 써서 다를 수 밖에 없습니다만 print의 결과물은 기상청 안개 경보에 나오는 텍스트와 비슷합니다. sentences 변수는 html/css 파일을 작성할 때 사용

 

다음 포스트에서는 시정자료로 그림을 그려보겠습니다.