import json import time import cv2 import requests import pandas as pd import os import numpy as np from dotenv import load_dotenv 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): ''' :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)) :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() 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) response = requests.get(url) response_json = json.loads(response.text) if response_json['response']['datacount'] == 1: new_df = pd.json_normalize(response_json['response']['data']) all_data_df = pd.concat([all_data_df, new_df], ignore_index=True) # df.to_csv(f"result/pohang/listofcctv_포항_{x}_{y}.csv", index=False) elif response_json['response']['datacount'] != 0: new_df = pd.DataFrame(response_json['response']['data']) all_data_df = pd.concat([all_data_df, new_df], ignore_index=True) # df.to_csv(f"result/pohang/listofcctv_포항_{x}_{y}.csv", index=False) time.sleep(1) print(f"{i}, {j}") # all_data_df.to_csv(f"result/{xmin}_{xmax}_y{ymin}_{ymax}.csv") return all_data_df def get_jpeg(url): response = requests.get(url) encoded_img = np.frombuffer(response.content, dtype=np.uint8) img = cv2.imdecode(encoded_img, cv2.IMREAD_COLOR) return img if __name__ == "__main__": df = gather_cctv_list(129.2, 129.3, 35.9, 36.07, 1, "its", 1) pass # get_jpeg("http://cctvsec.ktict.co.kr:8090/74236/IM2NQs4/uHZcgnvJo3V/mjo3tswwgUj87kpcYZfR/BPxaQ4lk9agnl8ARIB9lhlgOD87VBx6RDHFl423kLkqHQ==")