python
[Python] 대기 오염 정보 조회 서비스 Open API 사용
오두기밥
2022. 4. 19. 18:44
Intro
항해 99 사전 프로젝트로 진행하는 출근 정보 도우미 대시보드에 필요한 미세먼지 데이터를 가져오기 위해 파라미터로 경도, 위도를 받아 해당 위치에 미세먼지 데이터를 받아오는 코드를 작성하며 정리해보았습니다.
1) IDE
- Pycharm
2) 라이브러리
- requests
- Pyproj
3) version
- python 3.9
4) API
- 근처 미세먼지 측정소 조회 [ https://www.data.go.kr/data/15073877/openapi.do ]
- 측정소별 실시간 측정정보 조회 [ https://www.data.go.kr/data/15073861/openapi.do ]
FlowChart
 |
코드
1. 근처 미세먼지 측정소 데이터 조회
# 위도 경도 >>> TM 좌표, 리턴 값(도,근처 측정소)
def get_nearby_station(lat, lng):
# pyproj를 이용한 좌표 변환
transformer = Transformer.from_crs("epsg:4326", "epsg:2097")
x, y = lat, lng
tmy, tmx = transformer.transform(x, y)
url = 'http://apis.data.go.kr/B552584/MsrstnInfoInqireSvc/getNearbyMsrstnList'
params = {'serviceKey': service_key, 'returnType': 'json', 'tmX': tmx, 'tmY': tmy,
}
response = requests.get(url, params=params)
stations = response.json().get('response').get('body').get('items')
# json 거리 순으로 정렬
stations = sorted(stations, key=lambda pos: pos['tm'], reverse=False)
target_station = stations[0]
sido_name = target_station['addr'][0:2]
station_name = target_station['stationName']
print(station_name)
return sido_name, station_name(1 - 1) 코드 설명
- transformer = Transformer.from_crs("epsg:4236", "epsg:2097"):
- pyproj 라이브러리의 Trasnsformer 객체의 from_crs 함수를 이용해 WGS84 타원체의 경위도 좌표계를 TM 직각 좌표계로 변환 객체 반환
- tmy,tmx = transformer.transform(x: 위도, y: 경도):
- 경위도 좌표의 위도, 경도 순서로 받아 tm 좌표의 tmy, tmx 좌표로 변환
- 근처 미세먼지 측정소 조회 요청:
- 요청 메시지 완성: 서비스 키 , 리턴 타입, tmx 좌표 , tmy 좌표 할당
- 요청 받은 데이터를 거리 순 [키 값: 'tm']으로 정렬( sort() 기본값 오름차순 )
- 해당 위치 측정소 명 , 도명 리턴
2. 측정소별 실시간 측정정보 조회
def get_mise_info(lat=37.29650670336803, lng=126.9850504453901):
sido_name, station_name = get_nearby_station(lat, lng)
url = 'http://apis.data.go.kr/B552584/ArpltnInforInqireSvc/getMsrstnAcctoRltmMesureDnsty'
params = {'serviceKey': service_key, 'returnType': 'json', 'pageNo': '1',
'stationName': station_name, 'dataTerm': 'DAILY', 'ver': '1.3'}
response = requests.get(url, params=params)
mise_data = response.json().get('response').get('body').get('items')
target_data = sorted(mise_data, key=lambda date: date['dataTime'], reverse=True) # dataTime을 기준 으로 내림 차순 정렬
target_data = target_data[0]
result = dict()
try:
pm10_grade = target_data['pm10Grade']
pm25_grade = target_data['pm25Grade']
except KeyError:
pm10_grade = target_data['pm10Grade1h']
pm25_grade = target_data['pm25Grade1h']
if pm10_grade == '1':
pm10_state = "좋음"
elif pm10_grade == '2':
pm10_state = "보통"
elif pm10_grade == '3':
pm10_state = "나쁨"
elif pm10_grade == '4':
pm10_state = "매우나쁨"
else:
pm10_state = "_"
pm10 = dict()
pm10['value'] = target_data['pm10Value']
pm10['grade'] = pm10_state
result['pm10'] = pm10
if pm25_grade == '1':
pm25_state = "좋음"
elif pm25_grade == '2':
pm25_state = "보통"
elif pm25_grade == '3':
pm25_state = "나쁨"
elif pm25_grade == '4':
pm25_state = "매우나쁨"
else:
pm25_state = "_"
pm25 = dict()
pm25['value'] = target_data['pm25Value']
pm25['grade'] = pm25_state
result['pm25'] = pm25
result['do'] = sido_name
return result(2 - 1) 코드 설명
- 요청 메시지 완성
- returnType: JSON
- stationName: 근처 측정소 조회 API를 통해 얻은 결과 값 할당
- ver 1.3: PM10,PM2.51시간 등급 자료가 포함된 결과 표출
- dataTerm: 'DAILY' ※요청 데이터 기간: 1일 :DAILY, 1개월: MONTH, 3개월: 3MONTH
- 근처 미세먼지 측정소 데이터로 해당 측정소 실시간 측정 정보 조회 요청
- target_data = sorted(mise_data, key=lamda date: date['dataTime'], reversed = True)
- 하루 데이터에서 가장 최신 데이터만 필요하므로 dataTime 기준으로 내림차순 정렬
- target_data[0]
- 내림 차순 정렬을 했으므로 가장 첫 번째 인덱스 데이터가 최신 데이터
- 필요 데이터 분류, 및 추출
- grade 에 따른 상태 변수 조건문을 통해 해당 코드에 맞는 미세먼지 상태 문자열 반환
- 미세먼지 등급 keyError 발생시 1시간 간격 등급 가져오기
GitHub( 전체 코드 )
https://github.com/biolkj28/MiseAPI