File name
Commit message
Commit date
File name
Commit message
Commit date
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, type, cctvType, minX, maxX, minY, maxY, getType="json",
baseurl="https://openapi.its.go.kr:9443/cctvInfo"):
'''
국가교통정보센터 예제 실행
https://www.its.go.kr/opendata/openApiEx?service=cctv 참고
:param apiKey: ``str`` 국가교통정보센터에서 발급받은 api 키
:param type: ``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 type != "ex" or "its", 'Error! type 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={type}&"
f"cctvType={cctvType}&"
f"minX={minX}&maxX={maxX}&minY={minY}&maxY={maxY}&"
f"getType={getType}"
)
def gather_cctv_list(xmin, xmax, ymin, ymax, intervals, cctvtype):
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, 'its', 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}")
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.28, 129.35, 35.999, 36.07, 1, 1)
pass
# get_jpeg("http://cctvsec.ktict.co.kr:8090/74236/IM2NQs4/uHZcgnvJo3V/mjo3tswwgUj87kpcYZfR/BPxaQ4lk9agnl8ARIB9lhlgOD87VBx6RDHFl423kLkqHQ==")