import requests import pandas as pd import numpy as np import os import json import time from dotenv import load_dotenv import cv2 def create_url(apiKey, roadType, cctvType, minX, maxX, minY, maxY, getType="json", baseurl="https://openapi.its.go.kr:9443/cctvInfo"): ''' 국가교통정보센터 api 예제 실행 코드, 더 자세한 내용은 https://www.its.go.kr/opendata/openApiEx?service=cctv 참고 :param apiKey: ``str`` 국가교통정보센터에서 발급받은 api 키 :param roadType: ``str`` 도로 유형 ('ex' : 고속도로, 'its' : 국도) :param cctvType: ``int`` CCTV 유형 (1. 실시간 스트리밍(HLS) / 2. 동영상 파일(m3u8) / 3. 정지 영상(JPEG)) :param minX: 최소 경도 영역 :param maxX: 최대 경도 영역 :param minY: 최소 위도 영역 :param maxY: 최대 경도 영역 :param getType: 출력 결과 형식 ("xml" or "json") :return: api 요청 url ''' assert roadType != "ex" or "its", 'Error! roadType should be either "ex" or "its"' assert cctvType != 1 or 2 or 3, 'Error! cctvType should be one of 1, 2, 3!' assert getType != "json" or "xml", 'Error! gettype should be either "json" or "xml"!' return ( f"{baseurl}?" f"apiKey={apiKey}&" f"type={roadType}&" f"cctvType={cctvType}&" f"minX={minX}&maxX={maxX}&minY={minY}&maxY={maxY}&" f"getType={getType}" ) def gather_cctv_list(xmin, xmax, ymin, ymax, intervals, roadType, cctvType, depth=0, max_depth=5): ''' :param minX: 최소 경도 영역 :param maxX: 최대 경도 영역 :param minY: 최소 위도 영역 :param maxY: 최대 경도 영역 :param intervals: api를 통해서 cctv 목록을 불러올때 위경도 격자 간격 :param roadType: ``str`` 도로 유형 ('ex' : 고속도로, 'its' : 국도) :param cctvType: ``int`` CCTV 유형 (1. 실시간 스트리밍(HLS) / 2. 동영상 파일(m3u8) / 3. 정지 영상(JPEG)) :param depth: ``int`` 재귀적으로 불러온 함수의 깊이 (절대로 직접 넣지 말것.) :param max_depth: ``int`` 재귀 함수 깊이 제한 :return: pandas DataFrame 형태로 반환 주요 컬럼은 다음과 같음 coordx coordy cctvtype cctvformat cctvname cctvurl ''' dotenv = load_dotenv() apiKey = os.getenv("ITS_API") x_values = np.linspace(xmin, xmax, intervals + 1) y_values = np.linspace(ymin, ymax, intervals + 1) all_data_df = pd.DataFrame() if depth > max_depth: return all_data_df # Avoids excessive recursion depth for i in range(len(x_values) - 1): for j in range(len(y_values) - 1): x = x_values[i] y = y_values[j] x_next = x_values[i + 1] y_next = y_values[j + 1] url = create_url(apiKey, roadType, cctvType, x, x_next, y, y_next) try: response = requests.get(url) response_json = json.loads(response.text) if response_json['response']['datacount'] == 0: continue new_df = pd.json_normalize(response_json['response']['data']) all_data_df = pd.concat([all_data_df, new_df], ignore_index=True) print(f"x: {i}, y: {j}, d: {depth}") except Exception as e: # Splitting the area into four if an exception occurs (likely due to too many entries) if depth < max_depth: mid_x = (x + x_next) / 2 mid_y = (y + y_next) / 2 all_data_df = pd.concat([ all_data_df, gather_cctv_list(x, mid_x, y, mid_y, intervals, roadType, cctvType, depth + 1, max_depth), gather_cctv_list(mid_x, x_next, y, mid_y, intervals, roadType, cctvType, depth + 1, max_depth), gather_cctv_list(x, mid_x, mid_y, y_next, intervals, roadType, cctvType, depth + 1, max_depth), gather_cctv_list(mid_x, x_next, mid_y, y_next, intervals, roadType, cctvType, depth + 1, max_depth) ], ignore_index=True) time.sleep(1) # To prevent hitting the API rate limit return all_data_df if __name__ == "__main__": list2= [ [129.37762, 36.318485, 'its'], [129.381293, 36.339985, 'its'], [129.3459, 36.11983, 'its'], [129.200930555555, 35.772125, 'its'], [129.347244, 36.286135, 'its'], [128.03166, 36.011665, 'its'], [128.840556, 35.384722, 'ex'] ] delta = 0.01 all_data_df = pd.DataFrame() for i in list2: x = i[0] y = i[1] ty = i[2] df = gather_cctv_list(x-delta,x+delta, y-delta, y+delta, 1, ty, 1) all_data_df = pd.concat((all_data_df,df), ignore_index=True) # df = gather_cctv_list(127, 130, 33, 39, 8, "its", 1) all_data_df.to_csv("cctv_data_old.csv")