seongyooncho · July 22, 2025 13:09
diff --git a/apt.md b/apt.md
diff --git a/pip.md b/pip.md
diff --git a/viewer.py b/viewer.py
 import cv2
 import time
 import threading
 import os
 from onvif import ONVIFCamera
 from picamera2 import Picamera2

 # ================================
 # RTSP Camera Setup
 # ================================
 RTSP_URL = "rtsp://admin:EdgePTZ%[email protected]/Stream/Live/101?transportmode=unicast&profile=ONFProfileToken_101"
 cap = cv2.VideoCapture(RTSP_URL)

 # ================================
 # ONVIF Setup
 # ================================
 ip = '192.168.1.68'
 port = 80
 username = 'admin'
 password = 'EdgePTZ#2025'
 wsdl_path = 'wsdl'

 print("Connecting to ONVIF camera...")
 cam = ONVIFCamera(ip, port, username, password, wsdl_path)
 media = cam.create_media_service()
 ptz = cam.create_ptz_service()
 profile = media.GetProfiles()[0]

 def zoom(inout):
    request = ptz.create_type('ContinuousMove')
    request.ProfileToken = profile.token
    request.Velocity = {'Zoom': {'x': 0.9 * inout}}
    ptz.ContinuousMove(request)

 def stop_zoom():
    ptz.Stop({'ProfileToken': profile.token})


 # ================================
 # PiCamera2 Setup
 # ================================
 picam2 = Picamera2()
 config = picam2.create_preview_configuration(
    main={"size": (640, 480), "format": "RGB888"},
    controls={"FrameDurationLimits": (33333, 33333)}  # Target ~30 FPS
 )
 picam2.configure(config)
 picam2.start()


 # ================================
 # Threaded Capture Classes
 # ================================
 class CameraThread(threading.Thread):
    def __init__(self, name, capture_func):
        super().__init__(daemon=True)
        self.name = name
        self.capture_func = capture_func
        self.frame = None
        self.running = True
        self.lock = threading.Lock()

    def run(self):
        while self.running:
            frame = self.capture_func()
            if frame is not None:
                with self.lock:
                    self.frame = frame
            time.sleep(0.01)  # avoid CPU spinning

    def get_frame(self):
        with self.lock:
            return self.frame.copy() if self.frame is not None else None

    def stop(self):
        self.running = False


 # RTSP and PiCamera capture functions
 def capture_rtsp():
    ret, frame = cap.read()
    return frame if ret else None

 def capture_picam():
    return picam2.capture_array()

 # Initialize threads
 rtsp_thread = CameraThread("RTSP", capture_rtsp)
 picam_thread = CameraThread("PiCam", capture_picam)

 rtsp_thread.start()
 picam_thread.start()

 print("Press 'z' to zoom in, 'x' to zoom out, space to stop zoom, 'q' to quit.")

 # Prepare main and preview windows once
 cv2.namedWindow("RTSP View", cv2.WINDOW_NORMAL)
 cv2.setWindowProperty("RTSP View", cv2.WND_PROP_FULLSCREEN, cv2.WINDOW_FULLSCREEN)

 cv2.namedWindow("PiCam Preview", cv2.WINDOW_NORMAL)
 cv2.resizeWindow("PiCam Preview", 320, 240)
 cv2.moveWindow("PiCam Preview", 100, 100)

 try:
    while True:
        frame1 = rtsp_thread.get_frame()
        frame2 = picam_thread.get_frame()

        if frame1 is not None:
            frame1 = cv2.rotate(frame1, cv2.ROTATE_180)
            cv2.imshow("RTSP View", frame1)

        if frame2 is not None:
            frame2 = cv2.rotate(frame2, cv2.ROTATE_180)
            cv2.imshow("PiCam Preview", frame2)

        key = cv2.waitKey(1) & 0xFF
        if key == ord('q'):
            break
        elif key == ord('z'):
            zoom(+1)
        elif key == ord('x'):
            zoom(-1)
        elif key == ord(' '):
            stop_zoom()

        time.sleep(1 / 30)  # display loop throttle

 finally:
    # Cleanup
    print("Stopping threads...")
    rtsp_thread.stop()
    picam_thread.stop()
    cap.release()
    cv2.destroyAllWindows()
	import cv2
	import time
	import threading
	import os
	from onvif import ONVIFCamera
	from picamera2 import Picamera2

	# ================================
	# RTSP Camera Setup
	# ================================
	RTSP_URL = "rtsp://admin:EdgePTZ%[email protected]/Stream/Live/101?transportmode=unicast&profile=ONFProfileToken_101"
	cap = cv2.VideoCapture(RTSP_URL)

	# ================================
	# ONVIF Setup
	# ================================
	ip = '192.168.1.68'
	port = 80
	username = 'admin'
	password = 'EdgePTZ#2025'
	wsdl_path = 'wsdl'

	print("Connecting to ONVIF camera...")
	cam = ONVIFCamera(ip, port, username, password, wsdl_path)
	media = cam.create_media_service()
	ptz = cam.create_ptz_service()
	profile = media.GetProfiles()[0]

	def zoom(inout):
	request = ptz.create_type('ContinuousMove')
	request.ProfileToken = profile.token
	request.Velocity = {'Zoom': {'x': 0.9 * inout}}
	ptz.ContinuousMove(request)

	def stop_zoom():
	ptz.Stop({'ProfileToken': profile.token})


	# ================================
	# PiCamera2 Setup
	# ================================
	picam2 = Picamera2()
	config = picam2.create_preview_configuration(
	main={"size": (640, 480), "format": "RGB888"},
	controls={"FrameDurationLimits": (33333, 33333)} # Target ~30 FPS
	)
	picam2.configure(config)
	picam2.start()


	# ================================
	# Threaded Capture Classes
	# ================================
	class CameraThread(threading.Thread):
	def __init__(self, name, capture_func):
	super().__init__(daemon=True)
	self.name = name
	self.capture_func = capture_func
	self.frame = None
	self.running = True
	self.lock = threading.Lock()

	def run(self):
	while self.running:
	frame = self.capture_func()
	if frame is not None:
	with self.lock:
	self.frame = frame
	time.sleep(0.01) # avoid CPU spinning

	def get_frame(self):
	with self.lock:
	return self.frame.copy() if self.frame is not None else None

	def stop(self):
	self.running = False


	# RTSP and PiCamera capture functions
	def capture_rtsp():
	ret, frame = cap.read()
	return frame if ret else None

	def capture_picam():
	return picam2.capture_array()

	# Initialize threads
	rtsp_thread = CameraThread("RTSP", capture_rtsp)
	picam_thread = CameraThread("PiCam", capture_picam)

	rtsp_thread.start()
	picam_thread.start()

	print("Press 'z' to zoom in, 'x' to zoom out, space to stop zoom, 'q' to quit.")

	# Prepare main and preview windows once
	cv2.namedWindow("RTSP View", cv2.WINDOW_NORMAL)
	cv2.setWindowProperty("RTSP View", cv2.WND_PROP_FULLSCREEN, cv2.WINDOW_FULLSCREEN)

	cv2.namedWindow("PiCam Preview", cv2.WINDOW_NORMAL)
	cv2.resizeWindow("PiCam Preview", 320, 240)
	cv2.moveWindow("PiCam Preview", 100, 100)

	try:
	while True:
	frame1 = rtsp_thread.get_frame()
	frame2 = picam_thread.get_frame()

	if frame1 is not None:
	frame1 = cv2.rotate(frame1, cv2.ROTATE_180)
	cv2.imshow("RTSP View", frame1)

	if frame2 is not None:
	frame2 = cv2.rotate(frame2, cv2.ROTATE_180)
	cv2.imshow("PiCam Preview", frame2)

	key = cv2.waitKey(1) & 0xFF
	if key == ord('q'):
	break
	elif key == ord('z'):
	zoom(+1)
	elif key == ord('x'):
	zoom(-1)
	elif key == ord(' '):
	stop_zoom()

	time.sleep(1 / 30) # display loop throttle

	finally:
	# Cleanup
	print("Stopping threads...")
	rtsp_thread.stop()
	picam_thread.stop()
	cap.release()
	cv2.destroyAllWindows()
No results found