AI 모델을 개발하거나 데이터 분석을 공부하는 여러분, 혹시 오늘도 인터넷 날씨 사이트를 크롤링하느라 BeautifulSoup과 씨름하고 계신가요?
데이터 사이언스의 영원한 진리, Garbage In, Garbage Out(GIGO). 쓰레기 데이터를 넣으면 쓰레기 결과가 나옵니다. 특히 태양광 발전량 예측이나 스마트팜 에너지 관리 같은 정밀한 모델을 만든다면, 단순한 예보가 아니라 검증된 관측 데이터가 필수입니다.
오늘 포스팅에서는 기상청 API 허브의 보물창고, [산업특화 – 에너지 – 태양광] 섹션에 숨겨진 알짜배기 데이터인 ‘지상관측데이터 일통계(일사, 일조, 구름, 기온, 습도) 조회서비스’를 활용해 보겠습니다.
기상청 API 허브에는 수많은 데이터가 있지만, 우리가 주목할 것은 에너지 분야에 특화된 데이터입니다.
위치: 기상청 API 허브 > 산업특화 > 에너지 > 태양광
서비스명: 지상관측데이터 일통계(일사, 일조, 구름, 기온, 습도) 조회서비스
이 서비스는 일반 방재기상관측(AWS)과 달리 일사량(Solar Radiation)과 일조시간(Sunshine Duration) 정보를 포함하고 있어, 태양광 발전 효율 분석이나 농작물 생육 환경 분석에 최적화된 데이터셋입니다.
1-1. KMADataCollector 클래스: 재사용 가능한 데이터 수집 도구
매번 URL을 입력하는 번거로움을 줄이기 위해 클래스를 만들었습니다. 이 코드는 ‘지상관측데이터 일통계’ API 규격에 맞춰 제작되었습니다.
import requests
import pandas as pd
from io import StringIO
class KMADataCollector:
"""
기상청(KMA) [산업특화-에너지-태양광] 지상관측 일통계 데이터를 수집하는 클래스
"""
# 컬럼 정의 (15개) - 일사(SI), 일조(SS) 데이터 포함
COLUMNS = [
'YYMMDD', 'STN', 'TA_AVG', 'TA_MAX', 'TA_MAX_TM', 'TA_MIN', 'TA_MIN_TM',
'HM_AVG', 'HM_MIN', 'HM_MIN_TM', 'CA_TOT', 'SS_DAY', 'SS_DUR', 'SI_DAY', 'SI_60M_MAX'
]
NUMERIC_COLS = ['STN', 'TA_AVG', 'TA_MAX', 'TA_MIN', 'HM_AVG', 'HM_MIN', 'CA_TOT', 'SS_DAY', 'SS_DUR', 'SI_DAY', 'SI_60M_MAX']
# 산업특화 > 에너지 > 태양광 > 지상관측데이터 일통계 엔드포인트
BASE_URL = "https://apihub.kma.go.kr/api/typ01/url/sun_sfc_day.php"
def __init__(self, api_key):
self.api_key = api_key
def collect(self, start_date, end_date, station_id, timeout=10):
try:
# URL 생성
url = f"{self.BASE_URL}?tm1={start_date}&tm2={end_date}&stn={station_id}&help=0&authKey={self.api_key}"
# API 요청
response = requests.get(url, timeout=timeout)
response.raise_for_status()
data_text = response.text
# [핵심] usecols=range(len(self.COLUMNS))
# 데이터 끝의 쉼표로 인한 빈 컬럼 무시
df = pd.read_csv(
StringIO(data_text),
comment='#',
sep=',',
header=None,
names=self.COLUMNS,
encoding='utf-8',
usecols=range(len(self.COLUMNS))
)
df = df.dropna(how='all')
self._convert_types(df)
print(f"✓ 데이터 수집 완료: {len(df)}개 행")
return df
except requests.exceptions.Timeout:
print(f"✗ API 요청 타임아웃 ({timeout}초)")
return None
except requests.exceptions.RequestException as e:
print(f"✗ API 요청 실패: {e}")
return None
except Exception as e:
print(f"✗ 데이터 파싱 중 오류: {e}")
return None
def _convert_types(self, df):
"""데이터 타입 변환"""
df['YYMMDD'] = pd.to_datetime(df['YYMMDD'], format='%Y%m%d')
for col in self.NUMERIC_COLS:
df[col] = pd.to_numeric(df[col], errors='coerce')
# --- 사용 예시 ---
# 본인의 API KEY 입력
api_key = "YOUR_AUTH_KEY"
collector = KMADataCollector(api_key)
# 2023년 서울(108) 데이터 수집
df_weather = collector.collect("20230101", "20230102", "108")
if df_weather is not None:
print(df_weather[['YYMMDD', 'TA_AVG', 'SI_DAY', 'SS_DAY']].head())
✓ 데이터 수집 완료: 2개 행
YYMMDD TA_AVG SI_DAY SS_DAY
0 2023-01-01 -0.2 10.81 9.0
1 2023-01-02 -4.5 11.63 9.1
1-2. LSTM 모델 학습을 위한 1년치 데이터 수집 및 저장
LSTM 모델 학습을 위해 서울, 부산, 대구, 제주 지역의 2023년 전체 데이터를 수집하고 CSV로 저장합니다.
import time
import os
# 실행 로직
api_key = "YOUR_AUTH_KEY"
collector = KMADataCollector(api_key)
stations = {'108': '서울', '159': '부산', '183': '대구', '184': '제주'}
start_date = "20230101"
end_date = "20231231"
data_dict = {}
# 저장할 폴더 생성
save_dir = "weather_data"
if not os.path.exists(save_dir):
os.makedirs(save_dir)
print(f"=== 산업특화(태양광) 데이터 수집 시작 ({start_date} ~ {end_date}) ===")
for station_id, station_name in stations.items():
print(f"\n[{station_name}] 데이터 요청 중...")
df = collector.collect(start_date, end_date, station_id)
if df is not None:
data_dict[station_name] = df
# CSV 파일로 저장 (한글 깨짐 방지: utf-8-sig)
file_name = f"{save_dir}/{station_name}_2023.csv"
df.to_csv(file_name, index=False, encoding='utf-8-sig')
print(f" → 저장 완료: {file_name}")
else:
print(f" → 실패")
time.sleep(1) # 서버 부하 방지
# 서울 데이터 로드 (분석용)
if '서울' in data_dict:
df_seoul = data_dict['서울']
=== 산업특화(태양광) 데이터 수집 시작 (20230101 ~ 20231231) ===
[서울] 데이터 요청 중...
✓ 데이터 수집 완료: 365개 행
→ 저장 완료: weather_data/서울_2023.csv
[부산] 데이터 요청 중...
✓ 데이터 수집 완료: 365개 행
→ 저장 완료: weather_data/부산_2023.csv
[대구] 데이터 요청 중...
✓ 데이터 수집 완료: 0개 행
→ 저장 완료: weather_data/대구_2023.csv
[제주] 데이터 요청 중...
✓ 데이터 수집 완료: 365개 행
→ 저장 완료: weather_data/제주_2023.csv
2. 데이터 시각화: 일사량과 기온의 관계 확인
데이터의 중요성은 아무리 강조해도 지나치지 않습니다. 특히 태양광 발전의 핵심 변수인 일사량(SI_DAY)과 평균기온(TA_AVG)의 연간 패턴을 시각화해 보겠습니다.
주의: 기상청 데이터에서 관측 장비 점검 등으로 데이터가 없을 경우 -9 또는 -1000 같은 음수로 표기될 수 있습니다. 이를 이상치(Outlier)로 보고 전처리하는 과정이 필수입니다.
# 한글 폰트 설정을 위한 라이브러리 설치 (실행 후 런타임 재시작 필요 없음)
!pip install koreanize-matplotlib
import koreanize_matplotlib
import matplotlib.pyplot as plt
# 한글 폰트 설치 (실행 후 런타임 재시작 필요)
!sudo apt-get install -y fonts-nanum
!sudo fc-cache -fv
!rm ~/.cache/matplotlib -rf
# 그래프 설정 (마이너스 폰트 깨짐 방지 등은 라이브러리가 알아서 해줍니다)
#plt.style.use('seaborn-v0_8-whitegrid')
import koreanize_matplotlib
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
plt.style.use('seaborn-v0_8-whitegrid')
# 폰트 설정 (앞 단계에서 설치한 나눔바른고딕 적용)
plt.rc('font', family='NanumBarunGothic')
plt.rcParams['axes.unicode_minus'] = False
# 이상치 처리
if 'df_seoul' in locals() and df_seoul is not None:
df_plot = df_seoul.copy()
# 일사량(SI_DAY)이 0 미만인 경우 NaN 처리 (그래프 왜곡 방지)
outlier_mask = df_plot['SI_DAY']
일사량 이상치 제거: 1개
3. PyTorch LSTM 모델링: AI로 내일 기온 예측하기
이제 수집한 고품질 데이터를 바탕으로, LSTM 모델을 이용해 ‘과거 7일간의 패턴으로 다음날 기온을 예측’하는 AI를 만들어 보겠습니다. PyTorch를 사용합니다.
데이터 전처리: LSTM을 위한 밥상 차리기
LSTM 학습을 위해 데이터를 0~1 사이로 정규화하고, 시계열 시퀀스로 변환합니다. 이때 Data Leakage(데이터 누수)를 방지하기 위해 훈련 데이터 기준으로만 스케일러를 학습시키는 것이 중요합니다.
from sklearn.preprocessing import MinMaxScaler
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import TensorDataset, DataLoader
# 1. 데이터 분리 (최근 30일을 테스트용으로)
TEST_SIZE = 30
train_df = df_seoul[['TA_AVG']].iloc[:-TEST_SIZE]
test_df = df_seoul[['TA_AVG']].iloc[-TEST_SIZE:]
# 결측치 보간 (ffill 사용)
train_filled = train_df.ffill()
test_filled = test_df.ffill()
# 2. 정규화 (학습 데이터 기준으로만 수행!)
scaler = MinMaxScaler()
train_scaled = scaler.fit_transform(train_filled)
test_scaled = scaler.transform(test_filled)
# 3. 시퀀스 생성 함수
def create_sequences(data, seq_length):
xs, ys = [], []
for i in range(len(data) - seq_length):
xs.append(data[i:i+seq_length])
ys.append(data[i+seq_length])
return np.array(xs), np.array(ys)
SEQ_LENGTH = 7
X_train, y_train = create_sequences(train_scaled, SEQ_LENGTH)
X_test, y_test = create_sequences(test_scaled, SEQ_LENGTH)
# 4. 텐서 변환 및 DataLoader
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
X_train_tensor = torch.FloatTensor(X_train).to(device)
y_train_tensor = torch.FloatTensor(y_train).to(device)
X_test_tensor = torch.FloatTensor(X_test).to(device)
y_test_tensor = torch.FloatTensor(y_test).to(device)
train_loader = DataLoader(TensorDataset(X_train_tensor, y_train_tensor), batch_size=16, shuffle=True)
PyTorch LSTM 모델 정의 및 학습
class WeatherLSTM(nn.Module):
def __init__(self, input_size=1, hidden_size=50, num_layers=1, output_size=1):
super(WeatherLSTM, self).__init__()
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
lstm_out, _ = self.lstm(x)
last_hidden = lstm_out[:, -1, :]
return self.fc(last_hidden)
model = WeatherLSTM().to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# 학습 루프
num_epochs = 100
model.train()
for epoch in range(num_epochs):
for X_batch, y_batch in train_loader:
outputs = model(X_batch)
loss = criterion(outputs, y_batch) # 차원 불일치 주의 (y_batch unsqueeze 불필요)
optimizer.zero_grad()
loss.backward()
optimizer.step()
if (epoch + 1) % 20 == 0:
print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.6f}')
from sklearn.metrics import mean_absolute_error
import matplotlib.dates as mdates
model.eval()
with torch.no_grad():
test_predictions = model(X_test_tensor).cpu().numpy()
# 원래 스케일로 복원
test_pred_actual = scaler.inverse_transform(test_predictions)
# [수정] y_test는 이미 numpy 배열이므로 .cpu().numpy() 제거
# 만약 y_test가 1차원 배열이라면 reshape(-1, 1)이 필요할 수도 있습니다.
test_actual = scaler.inverse_transform(y_test)
# MAE 계산
mae = mean_absolute_error(test_actual, test_pred_actual)
print(f"테스트 데이터 MAE: {mae:.4f}°C")
# 시각화 (날짜 매핑 포함)
plt.figure(figsize=(15, 6))
# 테스트 데이터의 실제 날짜 가져오기 (시퀀스 길이 이후부터)
dates = df_seoul['YYMMDD'].iloc[-TEST_SIZE:].iloc[SEQ_LENGTH:]
plt.plot(dates, test_actual, label='실제 기온', marker='o')
plt.plot(dates, test_pred_actual, label='AI 예측', marker='x', linestyle='--')
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%m-%d'))
plt.title(f'LSTM 모델 예측 결과 (MAE: {mae:.2f}°C)', fontsize=16)
plt.legend()
plt.show()
테스트 데이터 MAE: 3.6396°C
4. 더 넓은 활용 분야: 태양광 발전 예측까지
오늘 실습한 내용은 단순한 기온 예측이었지만, 우리가 사용한 데이터는 ‘산업특화 – 태양광’ 데이터였습니다.
입력 변수(input_size)에 기온뿐만 아니라 일사량(SI_DAY), 일조시간(SS_DAY), 전운량(CA_TOT)을 함께 넣는다면 어떨까요? 이것이 바로 실제 태양광 발전소에서 사용하는 발전량 예측 AI의 기초가 됩니다.
맺음말
기상청 API 허브의 [산업특화-에너지-태양광] 섹션은 데이터 분석가들에게 보물과 같습니다. 오늘 배운 내용을 바탕으로 단순한 기온 예측을 넘어, 에너지 효율을 최적화하는 멋진 AI 모델을 만들어보시길 바랍니다!
