웹 크롤링 - [Python] 전기차 충전소 이용률 구하기

전기차 충전소 이용률에 대해 구해보도록 하겠습니다. 안타깝게도 충전소당 이용률을 알려주는 통계는 없고, 대신에 실시간 충전소 이용상태를 알려주는 환경공단의 API를 크롤링해서 이용률을 구해야 합니다. 시작해보겠습니다.

 

1. 오픈 API 활용 신청

https://www.data.go.kr/data/15076352/openapi.do 사이트에서 전기차 충전소 정보를 제공하는 API 사용신청을 해야합니다.

사이트 회원가입을 하시고 활용신청을 하게 되면 약 2시간 있다가 신청승인이 납니다. 따로 승인됐다고 메세지가 오는 건 아니고, 승인 전에 API 사용하려고 하면 사용이 안 됩니다. 적당히 2시간 있다가 하시면 될 것 같습니다.

 

2.  코딩을 시작하기 전에

제가 구할 데이터는 인천 전기차 충전소 실시간 이용상태를 20분 마다 수집을 해서 충전소마다의 이용률을 구하는 코딩을 작성할 거에요. 예를 들어서 1시간동안 코딩을 돌렸을 때 0분, 20분, 40분, 60분의 데이터가 수집될 것이고

인천시청 전기차 충전소	0분	20분	40분	60분
실시간 상태	미사용	사용	사용	사용

이렇게 데이터가 수집되었다면 1시간 이용률을 75%로 예측하는 코드를 만들 예정입니다.

3. 코딩 시작

import pandas as pd
import numpy as np
import requests
from bs4 import BeautifulSoup
from urllib.request import urlopen
import re
import time

url = 'http://apis.data.go.kr/B552584/EvCharger/getChargerInfo?'
## 본인 서비스키를 입력하세요!!!!!!!!!!!!!!!!!!!!!!!!!!!!
ServiceKey='SXFbrQnx0JUxP2h9REIqzA6W0w5YxkA3x8Fdt%2FTmpfeol5hnbVmr8Ah5fwyBnn%2FLtJp5IRP2A1cUED2KQasdfasdf'

위의 파이썬 라이브러리가 설치되어 있지 않다면 설치해주세요.

ServiceKey에 활용신청하면서 받은 서비스키를 복붙해주세요!!!!!!!!!!!!!!!!!

-----------------------------------------------------1----------------------------------------------------------------

######################################## 지역코드 입력하세요 
z = 42 ## 
###################################

names = []  # 이름
charge_types = []  # 충전기 타입
addresss = []  # 주소
latitudes = []  # 위도
longitudes = []  # 경도
states = []  # 충전기 상태
outputs = []  # 충전용량 3, 7, 50, 100, 200
methods = []  # 충전방식 단독 or 동시
zcodes = []  # 지역코드
limityns = []  # 이용자 제한 Y or N
limitdetails = []  # 이용자 제한 사유
parking_frees = []   # 주차료 무료 Y or N



page_num = 1
while 1:
    response = (url + 'ServiceKey=' + ServiceKey + '&numOfRows=1000' + '&pageNo='+ str(page_num) +'&zcode=' + str(z))
    url_new = response.encode('utf-8')
    req = requests.get(url_new)
    bs=BeautifulSoup(req.text, 'html.parser')

    name = bs.findAll('statnm')
    charge_type = bs.findAll('chgertype')
    address = bs.findAll('addr')
    latitude = bs.findAll('lat')
    longitude = bs.findAll('lng')
    state = bs.findAll('stat')
    output = bs.findAll('output')
    method = bs.findAll('method')
    zcode = bs.findAll('zcode')
    limityn = bs.findAll('limitYn')
    limitdetail = bs.findAll('limitDetail')
    parking_free = bs.findAll('parkingfree')

    total_num = str(bs.find('totalcount'))
    total = int(re.findall('\d+', total_num)[0])
    
    if page_num < total // 1000 + 1:
        for i in range(1000):
            names.append(name[i].text)
            charge_types.append(charge_type[i].text)
            addresss.append(address[i].text)
            latitudes.append(latitude[i].text)
            longitudes.append(longitude[i].text)
            states.append(state[i].text)
            outputs.append(output[i].text)
            methods.append(method[i].text)
            zcodes.append(zcode[i].text)
            try:
                limityns.append(limityn[i].text)
                limitdetails.append(limitdetail[i].text)
            except:
                limityns.append('')
                limitdetails.append('')
            parking_frees.append(parking_free[i].text)
            
        page_num += 1


    else:
        for i in range(total%1000):
            names.append(name[i].text)
            charge_types.append(charge_type[i].text)
            addresss.append(address[i].text)
            latitudes.append(latitude[i].text)
            longitudes.append(longitude[i].text)
            states.append(state[i].text)
            outputs.append(output[i].text)
            methods.append(method[i].text)
            zcodes.append(zcode[i].text)
            try:
                limityns.append(limityn[i].text)
                limitdetails.append(limitdetail[i].text)
            except:
                limityns.append('')
                limitdetails.append('')
            parking_frees.append(parking_free[i].text)
            
        break

1. 코드 맨 위에 보이는 z = 42를 본인이 찾고자 하는 지역으로 바꾸어주세요. 전국이 아마 00일 거고, 나머지 도, 광역시 번호는 https://www.code.go.kr/stdcode/regCodeL.do 여기서 확인할 수 있어요

예를 들어 제가 제주도 지역을 검색하고 싶으면, 지역선택에서 제주도를 선택하고, 조회를 누르면 아래에 결과가 나옵니다. 거기서 앞 두자리가 지역zcode입니다. 예를들어 의정부시, 양산시, 광양시 같은 시 단위로는 찾을 수 없고, 도나 광역시 단위로만 검색이 가능합니다. 만약에 "시" 하나만 보고 싶다면 "시"가 포함된 "도"로 검색하고, 결과값에서 본인이 찾고자 하는 "시"에 속한 전기차 충전소 데이터만 파이썬 코드를 따로 짜서 할 수 있겠네요.

 

아무튼 이 코드 맨 위에 z = 42를 본인의 지역코드로 바꾸어주세요. ex)제주도라면 z = 50 이렇게 ㅇㅇㅇㅇ.

 

저는 충전소마다 실시간 이용상태를 수집하기 위해, 개별 충전소에 대한 정보를 가져오려고 했어요.

그래서 충전소 이름, 충전타입, 주소, 위도, 경도, 충전방식, 지역코드, 이용자제한, 이용자제한사유, 무료주차여부 등의 정보를 가져오려고 합니다. 위 사이트에 API 사용 방법 워드 파일을 보시면 이 외에도 충전소 정보를 더 가져올 수 있으니 필요하시면 더 가져다 쓰시면 됩니다.

 

-----------------------------------------------------2----------------------------------------------------------------

def store_one_hour_state():
    states = []
    page_num = 1
    while 1:
        response = (url + 'ServiceKey=' + ServiceKey + '&numOfRows=1000' + '&pageNo='+ str(page_num) +'&zcode=' + str(z)) ## zcode 50이 제주도
        url_new = response.encode('utf-8')
        req = requests.get(url_new)
        bs=BeautifulSoup(req.text, 'html.parser')

        state = bs.findAll('stat')

        total_num = str(bs.find('totalcount'))
        total = int(re.findall('\d+', total_num)[0])

        if page_num >= total // 1000 + 1:
            for i in range(total%1000):
                states.append(state[i].text)
            break
            

        else:
            for i in range(1000):
                states.append(state[i].text)
            page_num += 1
    states
    return states

이 코드는 시간마다 충전소 상태를 가져오게 만드는 함수입니다.

 

-----------------------------------------------------3----------------------------------------------------------------

first_lists_data = pd.DataFrame([names, charge_types, addresss, latitudes, longitudes, outputs, methods, zcodes, limityns, limitdetails, parking_frees, states]).T
first_lists_data.columns=['이름','충전타입','주소', '위도','경도','출력', '충전방식', '지역코드', '이용자제한', '이용자제한사유', '무료 주차','실시간 상태(20분)']
first_lists_data

초기 데이터를 잘 받아왔는지 확인해보는 코드입니다. 보니까 이용자제한, 이용제제한사유 같은 칼럼은 데이터가 하나도 없네요. 애초에 왜 환경공단 사람들이 만들어놨는지 모르겠어요

 

-----------------------------------------------------4----------------------------------------------------------------

cumulative_data0_1 = first_lists_data
for i in range(1,4): 
    time.sleep(1195)     ## 1200초 = 20분, for문 안에 코딩 실행하는데 5초 걸림. 1195초 마다 실행시키면 1195초마다 for문 실행 됨
    now_state_data = store_one_hour_state()
    minutes = (i*20)%60
    hours = (i*20)//60
    now = str(hours) + ' : ' + str(minutes)
    now_state_data = pd.DataFrame(now_state_data,columns = [now])
    cumulative_data0_1 = pd.concat([cumulative_data0_1,now_state_data],axis=1)
    cumulative_data0_1

이제 충전소 실시간 상태를 수집해봅시다. 윗 줄에 cumulative_data0_1은 0시간부터 1시간까지의 데이터를 저장해놓는다는 뜻입니다. 저는 20분마다 데이터를 수집할 것이기 때문에 i * 20으로 해놓았고, for문도 1시간에 3번만 돌게 만들어놓았습니다. 1분마다 수집하셔도 되고, 10분마다, 혹은 1시간마다 수집하셔도 되기 때문에, 필요하시다면 알아서 이 부분만 코드 수정하시면 될 것 같습니다. 

 

0시간 부터 12시간까지 코드를 나열해보았습니다. 이대로 실행하면 됩니다. 정말 12시간 컴터를 켜놓아야 코드가 끝납니다...... 주피터 노트북에 코드 블럭마다 왼쪽에 있는 [*]표시가 1시간마다 지워질거에요.. ㅋㅋㅋㅋㅋ

 

cumulative_data0_2 = cumulative_data0_1
try:
    for i in range(1,4): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 1
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_2 = pd.concat([cumulative_data0_2,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', hours = (i*20)//60 + 1, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_3 = cumulative_data0_2
try:
    for i in range(1,4): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 2
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_3 = pd.concat([cumulative_data0_3,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 2, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_4 = cumulative_data0_3
try:
    for i in range(1,4): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 3
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_4 = pd.concat([cumulative_data0_4,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 3, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_5 = cumulative_data0_4
try:
    for i in range(1,4): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 4
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_5 = pd.concat([cumulative_data0_5,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 4, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_6 = cumulative_data0_5
try:
    for i in range(1,4): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 5
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_6 = pd.concat([cumulative_data0_6,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 5, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_7 = cumulative_data0_6
try:
    for i in range(1,4): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 6
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_7 = pd.concat([cumulative_data0_7,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 6, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_8 = cumulative_data0_7
try:
    for i in range(1,4): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 7
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_8 = pd.concat([cumulative_data0_8,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 7, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_10 = cumulative_data0_8
try:
    for i in range(1,7): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 8
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_10 = pd.concat([cumulative_data0_10,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 8, '시간대의 데이터를 수집하지 못했어요')

 

cumulative_data0_12 = cumulative_data0_10
try:
    for i in range(1,7): 
        time.sleep(1195)                                ## 1200초 = 20분
        now_state_data = store_one_hour_state()
        minutes = (i*20)%60
        hours = (i*20)//60 + 10
        now = str(hours) + ' : ' + str(minutes)
        now_state_data = pd.DataFrame(now_state_data,columns = [now])
        cumulative_data0_12 = pd.concat([cumulative_data0_12,now_state_data],axis=1)
except:
    print('오류가 발생했습니다 ㅠㅠ', (i*20)//60 + 10, '시간대의 데이터를 수집하지 못했어요')

 

-----------------------------------------------------5----------------------------------------------------------------

final_data = cumulative_data0_12 ## 여태껏 모은 데이터 저장
final_data.head()

자 이제 코드가 다 끝났습니다. 12시간이 너무 길면 4시간만 돌리게 짜도 됩니다. 예를 들어 6시간만 코드를 돌린다면

final_data = cumulative_data0_6으로 위 코드를 바꾸면 됩니다.

 

-----------------------------------------------------6----------------------------------------------------------------

## 이제 이용률 구해보자. 그 전에 칼럼 names에서 parking _fees까지 없애자. 즉 state 칼럼들만 남기기
only_state_data = final_data.drop(['이름','충전타입','주소','위도','경도','출력', '충전방식', '지역코드', '이용자제한', '이용자제한사유', '무료 주차장'], axis = 'columns')

## 이용률 구하기
availability = []
avail = 0
for i in range(len(only_state_data)):
    num2 = 0
    num3 = 0
    num5 = 0
    for j in range(len(only_state_data.loc[i])):
        if only_state_data.loc[i][j] == '2':
            num2 += 1
        elif only_state_data.loc[i][j] == '3':
            num3 += 1
        elif only_state_data.loc[i][j] == '5':
            num5 += 1
    if num2 + num3 + num5 == 0:
            avail = 0                       ## 충전 중, 충전 대기 중, 점검 중도 없는 경우는 망가진 충전소. == 이용률 0
    else:
        avail = num3 / (num3 + num2 + num5) ## 하나 충전소당 -> 충전중 / (충전중 + 충전 대기중 + 점검중)
    
    availability.append(avail)

availabililties = pd.DataFrame(availability,columns = ['avail']) 
final_data2 = pd.concat([final_data,availabililties],axis=1)
## avail 칼럼 잘 들어갔나 볼까?
final_data2

이제 실시간 상태 데이터와 개별 충전소 정보가 담긴 final_data에서 충전소 정보 칼럼을 제거하고 이용률 계산합시다. 이용률은 계산 방법은 사실 작성자 맘입니다. 저는 충전 중 + 충전 대기 중 + 점검 중을 분모에 놓고 충전 중을 분자에 두었는데요, 충전 중 / (충전 중 + 충전 대기 중)으로 할 수도 있습니다. 다른 방법도 있을 수 있겠죠 ㅇㅇ.

fianl_data2가 잘 나왔다면 성공입니다.

 

-----------------------------------------------------7----------------------------------------------------------------

final_data2.to_csv("live_elec_charging_data.csv", mode='w',encoding='utf-8-sig')

마지막은 CSV파일로 저장하면 성공입니다!