Python: 연습 7 (공공 API 크롤링 + 시각화)

공공 데이터 대시보드 (공공 API 크롤링 + 시각화)

수자원공사의 공공 API (ex. 수위, 댐 운영, 홍수정보 등) 를 활용해, 실시간 수자원 데이터를 수집 → 시각화 → 웹 대시보드 형태로 제공하는 프로젝트를 진행해보려고 한다. 

그리고 나는 그중에서도 '댐 실시간 수위정보' 의 API 를 가지고 프로젝트를 진행해보고자 한다. 

 

 

Step 1. 공공 API 찾고 신청하기

1. 공공데이터포털 접속: https://www.data.go.kr/tcs/dss/selectApiDataDetailView.do?publicDataPk=15057371

한국수자원공사_실시간 수도정보 수위(시간) 조회 서비스(GW)

2. 활용신청 버튼 클릭: 페이지 우측 상단의 '활용신청' 버튼을 클릭하여 신청서를 작성한다. 

3. 신청서 작성 및 제출: 대부분 자동 승인되며, 마이페이지에서 인증키(API key)를 확인 가능하다. 

 

 

Step 2. API 호출해서 실시간 댐 수위 데이터 받아오기

1. 실습을 위한 설치

pip install requests

 

2. python 코드 

import requests
import urllib3
import pandas as pd
import xml.etree.ElementTree as ET

# 경고 제거
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# API 요청 URL과 파라미터
url = "https://apis.data.go.kr/B500001/rwis/waterLevel/list"

params = {
    'serviceKey': 'Aw7s7KHLZTj3NFSl7dFCCEql6ESZJomb/8PMV2hinsNUFqmhRpwpFdvuTyv1mEaUW5IP3y5N4J/TcZaa1GUqrw==',  # 디코딩된 인증키
    'stDt': '20240401',       # 시작일자 (yyyymmdd)
    'stTm': '00',             # 시작시간 (00~24)
    'edDt': '20240401',       # 종료일자 (같은 날도 가능)
    'edTm': '24',             # 종료시간
    'sujCode': '331',         # 사업장코드 (예: 연초정수장 → 문서 참고)
    'fcltyMngNo': '4831012331',  # 시설관리번호 (연초정수장 예시)
    'numOfRows': '10',
    'pageNo': '1',
    'type': 'xml'             # JSON 대신 XML로 받음 (기본값임)
}

# API 호출
response = requests.get(url, params=params, verify=False)
print("응답 코드:", response.status_code)

# XML 파싱
root = ET.fromstring(response.text)

items = []
for item in root.findall(".//item"):
    row = {
        '발생일시': item.findtext('occrrncDt'),
        '시설명': item.findtext('fcltyNm'),
        '수위': item.findtext('dataVal'),
        '단위': item.findtext('itemUnit'),
        '태그설명': item.findtext('dataItemDesc')
    }
    items.append(row)

# Pandas DataFrame으로 변환
df = pd.DataFrame(items)
print(df.head())

 

3. 코드 실행 결과

 

 

Step 3. 시각화(그래프)

- Plotly or Matplotlib 으로 수위 변화를 시각화

 

1. 실습을 위한 설치

pip install matplotlib

 

2. python 코드

import matplotlib.pyplot as plt
import matplotlib

# 한글 폰트 설정
matplotlib.rcParams['font.family'] = 'Malgun Gothic'
matplotlib.rcParams['axes.unicode_minus'] = False

import matplotlib.pyplot as plt

# 수위 숫자형 변환
df['수위'] = pd.to_numeric(df['수위'], errors='coerce')

# 발생일시에서 앞 10자리만 사용
df['발생일시'] = df['발생일시'].str[:10]

# datetime 변환 (YYYYMMDDHH 형식)
df['발생일시'] = pd.to_datetime(df['발생일시'], format='%Y%m%d%H', errors='coerce')


# 시간 순 정렬
df = df.sort_values('발생일시')

# 그래프 그리기
plt.figure(figsize=(10, 5))
plt.plot(df['발생일시'], df['수위'], marker='o')
plt.title("시간대별 수위 변화")
plt.xlabel("시간")
plt.ylabel("수위 (m)")
plt.xticks(rotation=45)
plt.grid(True)
plt.tight_layout()
plt.show()


# 발생일시 값 출력해서 확인
print("발생일시 원본 값 확인:")
print(df['발생일시'])

# format 없이 파싱
df['발생일시'] = pd.to_datetime(df['발생일시'], errors='coerce')

# NaT로 변환된 문제 데이터 확인
print("날짜 파싱 실패한 행:")
print(df[df['발생일시'].isna()])

 

3. 코드 실행 결과 

 

Step 4.Plotly 로 그래프 다듬기 

- 인터랙티브 차트 (줌인/줌아웃, 툴팁, 범례 클릭 등)

 

1. 실습을 위한 설치 

pip install plotly

 

2. python 코드

import plotly.express as px

# Plotly 그래프 그리기
fig = px.line(df, x='발생일시', y='수위', title='시간대별 수위 변화', markers=True)
fig.update_layout(xaxis_title='시간', yaxis_title='수위 (m)', template='plotly_white')
fig.show()
fig.write_html("plot.html")

 

3. 코드 실행 결과

실행 후 그래프 확인을 하기 위해서는 아래와 같은 과정을 거쳐야 한다. 

파일 탐색기에서 바탕화면으로 이동

[plot.html] 더블클릭

Chrome 브라우저가 파일을 정상적으로 열어줄 것

 

 

 

Step 5. 대시보드 구현 (Streamlit)

- streamlit 으로 웹 형태 대시보드 구현

- 관측일자, 시설명 필터 추가

- 실시간 수위 변화 + 데이터 테이블 제공 

 

1. 실습을 위한 설치 

pip install streamlit

 

2. python 코드

새로운 Python 파일 (app.py) 생성 후 아래 코드 붙여넣기

import streamlit as st
import pandas as pd
import plotly.express as px
import requests
import urllib3
import xml.etree.ElementTree as ET

# ======================
# 💥 화면 폭 넓히기
# ======================
st.markdown(
    """
    <style>
    .main {
        max-width: 1200px;
    }
    </style>
    """,
    unsafe_allow_html=True
)

# ======================
# 💥 제목 줄바꿈 방지
# ======================
st.markdown("<h1 style='white-space:nowrap;'>💧 실시간 정수장·배수지 수위 대시보드</h1>", unsafe_allow_html=True)

# 📍 시설 유형 선택
facility_type = st.selectbox("시설 유형 선택", ["정수장", "배수지"])
fcltyDivCode = '2' if facility_type == "정수장" else '4'

# 📅 날짜 범위 선택
start_date, end_date = st.date_input('조회할 날짜 범위 선택', [])
if isinstance(start_date, tuple):
    start_date, end_date = start_date
start_date_str = start_date.strftime('%Y%m%d')
end_date_str = end_date.strftime('%Y%m%d')

# ✅ 시설 코드 목록 조회
url_code = "http://apis.data.go.kr/B500001/rwis/waterLevel/fcltylist/codelist"
params_code = {
    'serviceKey': 'Aw7s7KHLZTj3NFSl7dFCCEql6ESZJomb/8PMV2hinsNUFqmhRpwpFdvuTyv1mEaUW5IP3y5N4J/TcZaa1GUqrw==',
    'fcltyDivCode': fcltyDivCode,
    'numOfRows': '100',
    'pageNo': '1',
    'type': 'xml'
}

response_code = requests.get(url_code, params=params_code, verify=False)
root_code = ET.fromstring(response_code.text)

facility_list = []
for item in root_code.findall(".//item"):
    name = item.findtext('fcltyMngNm')
    suj_code = item.findtext('sujCode')
    facility_list.append({'시설관리명': name, '사업장코드': suj_code})

df_code = pd.DataFrame(facility_list)

# ✅ Streamlit 선택 박스
selected_facility = st.selectbox('시설 선택', df_code['시설관리명'])
selected_suj_code = df_code[df_code['시설관리명'] == selected_facility]['사업장코드'].values[0]

# ✅ 수위 데이터 API 호출
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
url_data = "https://apis.data.go.kr/B500001/rwis/waterLevel/list"
params_data = {
    'serviceKey': 'Aw7s7KHLZTj3NFSl7dFCCEql6ESZJomb/8PMV2hinsNUFqmhRpwpFdvuTyv1mEaUW5IP3y5N4J/TcZaa1GUqrw==',
    'stDt': start_date_str,
    'stTm': '00',
    'edDt': end_date_str,
    'edTm': '24',
    'sujCode': selected_suj_code,
    'numOfRows': '100',
    'pageNo': '1',
    'type': 'xml'
}

response_data = requests.get(url_data, params=params_data, verify=False)
root_data = ET.fromstring(response_data.text)

items = []
for item in root_data.findall(".//item"):
    row = {
        '발생일시': item.findtext('occrrncDt'),
        '시설명': item.findtext('fcltyNm'),
        '수위': item.findtext('dataVal'),
        '단위': item.findtext('itemUnit'),
        '태그설명': item.findtext('dataItemDesc')
    }
    items.append(row)

df = pd.DataFrame(items)

# 날짜·수위 전처리
if not df.empty:
    df['발생일시'] = df['발생일시'].str[:10]
    df['발생일시'] = pd.to_datetime(df['발생일시'], format='%Y%m%d%H', errors='coerce')
    df['수위'] = pd.to_numeric(df['수위'], errors='coerce')
    df = df.dropna()

# ✅ 데이터 확인 및 시각화
st.subheader(f"✅ {selected_facility} ({selected_suj_code}) 데이터")
if df.empty:
    st.warning("⚠️ 선택한 시설의 데이터가 없습니다.")
else:
    st.write("데이터 개수 확인:", len(df))
    st.write(df.head())

    # 📈 Plotly 그래프
    fig = px.line(df, x='발생일시', y='수위', title=f"{selected_facility} 수위 변화", markers=True)
    st.plotly_chart(fig)

    # 📋 데이터 테이블
    st.subheader("원본 데이터")
    st.dataframe(df)

 

3. 코드 실행 결과 

streamlit 코드는 terminal 에서 아래와 같이 실행시켜야 한다. 

streamlit run "c:/Users/kwater/Desktop/Python 연습/댐 실시간 수위정보/app.py"

 

코드를 실행시키면 아래와 같은 대시보드가 만들어진다.

 

Step 6. 대시보드의 핵심역할 이해

전국 정수장·배수지 실시간 수위 데이터를 원하는 날짜/시설별로 한눈에 조회·분석할 수 있는 웹 대시보드 

 

1. 실시간 수위 데이터 조회

- 한국수자원공사 API 에서 정수장 또는 배수지의 실시간 수위 데이터를 가져옴

- 조회할 날짜 또는 날짜 범위를 선택할 수 있음

 

2. 시설 목록 자동 불러오기 

- "정수장 코드 조회" 또는 "배수지 코드 조회" API 를 통해 현재 서비스 중인 시설 (ex. 연초정수장, 반월정수장, 덕소정수장, 배수지 등) 목록을 자동으로 가져옴

 

3. 사용자 선택형 대시보드

- 사용자가 [날짜 범위 선택, 시설 유형(정수장/배수지) 선택, 특정 시설 선택] 을 하면, 선택 결과에 맞춰 해당 시설의 실시간 수위 데이터를 조회해서 보여줌

 

4. 데이터 시각화

- 수위 데이터 시계열 그래프로 시각화 (Plotly line chart)

- 데이터 표로 원본 확인 가능 (Streamlit 데이터 테이블)

 

5. 데이터 품질 및 상태 확인

- 조회된 데이터 개수, 샘플 확인

- 데이터가 없거나 오류 발생 시 경고 메시지로 사용자 안내