IP Camera RTSP access with python

concept

1. RTSP : Real-Time Streaming Protocol
2. 실시간 응용 프로그램에 적합
   1. HTTP 라이브 스트리밍(HLS)이나 동적 적응 스트리밍(DASH)과 같은 다른 스트리밍 프로토콜과는 다르다.
   2. HLS와 DASH는 주로 점진적 다운로드를 통해 스트리밍을 구현

 

output

1. 데이터 수집
   1. 실시간으로 촬영되는 카메라의 프레임을 입력으로 이미지를 저장하고자 한다.
2. 인공지능 모델 테스트
   1. 추가적으로, 인공지능 모델에 해당 프레임을 입력 데이터로 사용해 추론 결과를 볼 수 있다.

 

method

1. 해당 내용은 IDIS IP Camera 에 준합니다.
2. IDIS 카메라 사이트에 접속해 해당 카메라 기종 자료 다운로드 페이지에 접근합니다.
3. 펌웨어를 설치합니다.
   1. install INIT4.6.1.exe.
4. 카메라 웹뷰에 접근해 해당 카메라의 카메라 ID 를 확인합니다.
   1. 별도로, 웹뷰 계정을 생성합니다.(카메라에 접속하기전에 요구됩니다, 보통은)
      1. RTSP url 에 사용됩니다.
5. 홈페이지로부터 해당 카메라의 RTSP url 를 확인합니다.
   1. rtsp://ID:PW@{}:554/trackID={}".format(cam_ip, CAMID)
6. python code

import os

import cv2
import time
import matplotlib.pyplot as plt
import logging
from datetime import datetime

from IPython.display import clear_output


class IPCam():

    def __init__(self,):
        self.set_logger()
        
    def set_logger(self,):
        
        simple_formatter = logging.Formatter("[%(name)s] %(message)s")
        complex_formatter = logging.Formatter(
            "%(asctime)s %(levelname)s [%(name)s] [%(filename)s:%(lineno)d] - %(message)s"
        )
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(simple_formatter)
        console_handler.setLevel(logging.DEBUG)
    
        # file_handler = logging.FileHandler(self.save_path_log)
        # file_handler.setFormatter(complex_formatter)
        # file_handler.setLevel(logging.DEBUG)
    
        self.sys_log = logging.getLogger('SPK')
    
        self.sys_log.handlers.clear()
        self.sys_log.addHandler(console_handler)
        # self.sys_log.addHandler(file_handler)
        self.sys_log.setLevel(logging.DEBUG)
        
        
       

def check_video_normality(self, term=1):

    while 1:
        self.cap = cv2.VideoCapture(self.param_dict['path_video'])
        self.cap.set(cv2.CAP_PROP_AUTOFOCUS, 0)
    
        cap_opened = self.cap.isOpened()
        if cap_opened:
            # self.cap_h_half = round(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) / 2)
            return self.cap
        else:
            self.sys_log.error("error in VideoCapture")
            time.sleep(term)

def get_video_writer(self, video_name):

    """
    v1.0
        rename to video_name

    last confirmed at, 20240331 1154.
    """
    
    w = round(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = round(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fps = self.cap.get(cv2.CAP_PROP_FPS)    
    fourcc = cv2.VideoWriter_fourcc(*'DIVX')
    # delay = round(1000 / fps)
    self.out = cv2.VideoWriter(os.path.join(self.param_dict['save_path_dir_video'], video_name), fourcc, fps, (w, h))   



cam_ip = "192.168.5.190"     
path_video = "rtsp://admin:123!@{}/main".format(cam_ip)
# rtsp://아이디:패스워드@ip/main 


ipcam.param_dict = \
{
    'path_video': path_video, 
    'save_path_dir_video': "./videos",
    'save_path_dir_image': "./images",
}



ipcam.check_video_normality = check_video_normality
ipcam.get_video_writer = get_video_writer


datetime_current = datetime.now().strftime('%Y%m%d_%H%M')
save_path_dir_image = os.path.join(ipcam.param_dict['save_path_dir_image'], datetime_current)
os.makedirs(save_path_dir_image, exist_ok=True)
# break


ipcam.check_video_normality(ipcam)


frame_cnt = 0
while ipcam.cap.isOpened():

    ret, frame = ipcam.cap.read()
    
    # plt.imshow(frame)
    # plt.show()
    # break
    
    try:
        save_abs_path_image = os.path.join(save_path_dir_image, "{}_{:06d}.jpg".format(datetime.now().strftime('%Y%m%d%H%M%S%f'), frame_cnt))
        cv2.imwrite(save_abs_path_image, frame)
        frame_cnt += 1
    
        print("frame_cnt : {}".format(frame_cnt), end='\r')
        # break
        # break
    except Exception as e:
        print("error in save imwrite : {}".format(e))
        time.sleep(1)
        clear_output()
        # pass

    if frame_cnt > 900:
        break

 

input

1. IP camera
   1. 기종
   2. 펌웨어
   3. IP 주소
2. IP camera 접속 PC
   1. Lenovo IdeaPad 530S