'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる
[Details]
Let's try first.
まずはやってみる
[Library install]
cv2: pip install opencv-python
'''
import cv2
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")
while True:
try:
ret, frame = cap.read()
if ret == True:
# Please modify the value to fit your PC screen size.
frame2 = cv2.resize(frame, (1280, 720))
# Display video.
cv2.imshow(winname, frame2)
cv2.waitKey(1) # necessary to display the video by imshow ()
except KeyboardInterrupt:
# Press '[ctrl] + [c]' on the console to exit the program.
print("KeyboardInterrupt")
break
cap.release()
cv2.destroyAllWindows()
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる
[Details]
Let's improve the three issues of "connect_with_rtsp_1.py".
"connect_with_rtsp_1.py" で確認した下記３つの課題を改善してみます。
[Issues 1]
Specifies the position where the window is displayed.
ウィンドウを指定する場所に表示するようにします。
[Issues 2]
Modify the program so that you can exit the program by clicking the [x] button.
ウィンドウ右上の[x]ボタンでプログラムを終了できるようにします。
[Issues 3]
Modify the program so that you can exit the program by pressing the [q] key.
"q" キー押下でプログラムを終了できるようにします。
[Library install]
cv2: pip install opencv-python
'''
import cv2
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")
#windowInitialized = False# Exception 定義BackendError = type('BackendError', (Exception,), {})def IsWindowVisible(winname): ''' [Abstract] Check if the target window exists. 対象ウィンドウが存在するかを確認する。 [Param] winname : Window title [Return] True : exist 存在する False : not exist 存在しない [Exception] BackendError : ''' try: ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE) if ret == -1: raise BackendError('Use Qt as backend to check whether window is visible or not.') return bool(ret) except cv2.error: return False
while True:
try:
ret, frame = cap.read()
if ret == True:
# Please modify the value to fit your PC screen size.
frame2 = cv2.resize(frame, (1280, 720))
# Display video.
cv2.imshow(winname, frame2)
if windowInitialized==False: # Specify window position only once at startup. cv2.moveWindow(winname, 100, 100) windowInitialized = True # Press the "q" key to finish. k = cv2.waitKey(1) & 0xff # necessary to display the video by imshow () if k == ord("q"): break # Exit the program if there is no specified window. if not IsWindowVisible(winname): break
except KeyboardInterrupt:
# Press '[ctrl] + [c]' on the console to exit the program.
print("KeyboardInterrupt")
break
cap.release()
cv2.destroyAllWindows()
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる
[Details]
Let's add face detection using OpenCV.
OpenCV を使って顔検知を追加してみます
[Library install]
cv2: pip install opencv-python
[OpenCV]
Get the file "haarcascade_frontalface_alt2.xml" from the URL below.
下記URLからファイル "haarcascade_frontalface_alt2.xml" を入手するしてください。
https://github.com/opencv/opencv/tree/master/data/haarcascades
'''
import cv2
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
# Exception 定義
BackendError = type('BackendError', (Exception,), {})
def IsWindowVisible(winname):
'''
[Abstract]
Check if the target window exists.
対象ウィンドウが存在するかを確認する。
[Param]
winname : Window title
[Return]
True : exist
存在する
False : not exist
存在しない
[Exception]
BackendError :
'''
try:
ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
if ret == -1:
raise BackendError('Use Qt as backend to check whether window is visible or not.')
return bool(ret)
except cv2.error:
return False
def DetectFaces(cascade, image): ''' [Abstract] Detect faces and return recognition result. 顔検知して認識結果を返す [Param] cascade : CascadeClassifier object in OpenCV format. image : Image in OpenCV format. [Return] Detection result. ''' # Convert to grayscale image for face detection. img_gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) # Detect faces from image. face_list = cascade.detectMultiScale(img_gray, minSize=(100, 100)) # Return result. return face_listdef DrawFaceRectangles(image, face_list): ''' [Abstract] Draw red frames on the image according to the detection result face_list. 検出結果 face_list にしたがって、image 上に赤枠を描画する。 [Param] image : Image in OpenCV format. face_list : List of detected face frames. [Return] None ''' # Draw red frames for the number of detected faces. if len(face_list) != 0: for (pos_x, pos_y, w, h) in face_list: print(f"pos_x = {pos_x}, pos_y = {pos_y}, w = {w}, h = {h}") cv2.rectangle(image, (pos_x, pos_y), (pos_x + w, pos_y + h), (0,0,255), thickness=5)
if __name__ == '__main__':
'''
[Abstract]
main function
'''
cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")
#
windowInitialized = False
# haarcascade file for opencv cascade classification. cascade_file = "haarcascade_frontalface_alt2.xml" # face #cascade_file = "haarcascade_eye.xml" # eye ? #cascade_file = "haarcascade_eye_tree_eyeglasses.xml" # eye ? cascade = cv2.CascadeClassifier(cascade_file)
while True:
try:
ret, frame = cap.read()
if ret == True:
# Face detection. face_list = DetectFaces(cascade, frame) # Draw face rectangles. DrawFaceRectangles(frame, face_list)
# Please modify the value to fit your PC screen size.
frame2 = cv2.resize(frame, (1280, 720))
# Display video.
cv2.imshow(winname, frame2)
if windowInitialized==False:
# Specify window position only once at startup.
cv2.moveWindow(winname, 100, 100)
windowInitialized = True
# Press the "q" key to finish.
k = cv2.waitKey(1) & 0xff # necessary to display the video by imshow ()
if k == ord("q"):
break
# Exit the program if there is no specified window.
if not IsWindowVisible(winname):
break
except KeyboardInterrupt:
# Press '[ctrl] + [c]' on the console to exit the program.
print("KeyboardInterrupt")
break
cap.release()
cv2.destroyAllWindows()
結果：
私のゲーミングPCではこれでもそこそこ動作しました。思ったより動く、という感想です。
が、それでもだんだん映像が遅れていきます。 顔検知処理と描画の部分をコメントアウトすると、映像表示の遅れはなくなります。
やはり顔検知処理は PC にとって結構重たい処理のようです。
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる
[Details]
Let's add face detection using OpenCV.
Improve performace of "connect_with_rtsp_3_1.py" by creating a face detection process.
OpenCV を使って顔検知を追加してみます
"connect_with_rtsp_3_1.py" で確認したパフォーマンス問題を、顔検知処理を別プロセスにすることで改善してみます。
[Library install]
cv2: pip install opencv-python
[OpenCV]
Get the file "haarcascade_frontalface_alt2.xml" from the URL below.
下記URLからファイル "haarcascade_frontalface_alt2.xml" を入手するしてください。
https://github.com/opencv/opencv/tree/master/data/haarcascades
'''
import cv2
import multiprocessing as mpfrom queue import Empty
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
# haarcascade file for opencv cascade classification.cascade_file = "haarcascade_frontalface_alt2.xml" # face#cascade_file = "haarcascade_eye.xml" # eye ?#cascade_file = "haarcascade_eye_tree_eyeglasses.xml" # eye ?cascade = cv2.CascadeClassifier(cascade_file)
# Exception definition
BackendError = type('BackendError', (Exception,), {})
def IsWindowVisible(winname):
'''
[Abstract]
Check if the target window exists.
対象ウィンドウが存在するかを確認する。
[Param]
winname : Window title
[Return]
True : exist
存在する
False : not exist
存在しない
[Exception]
BackendError :
'''
try:
ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
if ret == -1:
raise BackendError('Use Qt as backend to check whether window is visible or not.')
return bool(ret)
except cv2.error:
return False
def DetectFaces(cascade, image):
'''
[Abstract]
Detect faces and return recognition result.
顔検知して認識結果を返す
[Param]
cascade : CascadeClassifier object in OpenCV format.
image : Image in OpenCV format.
[Return]
Detection result.
'''
# Convert to grayscale image for face detection.
img_gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
# Detect faces from image.
face_list = cascade.detectMultiScale(img_gray, minSize=(100, 100))
# Return result.
return face_list
def DetectFacesProcess(q1, q2): ''' [Abstract] Face detection process. [Param] q1 : [i] Queue to save the image to detect the face. q2 : [o] Queue to save the result of face detection. [Return] 無し ''' while True: try: image = q1.get(True, 10) # Termination check: If type(image) is "int" and the value is -1, it ends. if type(image) == int: if image == -1: break # Face detection. face_list = DetectFaces(cascade, image) q2.put(face_list) except Empty: # timeout of q1.get() print("Timeout happen.(3)") print("Finish DetectFacesProcess()")
def DrawFaceRectangles(image, face_list):
'''
[Abstract]
Draw red frames on the image according to the detection result face_list.
検出結果 face_list にしたがって、image 上に赤枠を描画する。
[Param]
image : Image in OpenCV format.
face_list : List of detected face frames.
[Return]
無し
'''
# Draw red frames for the number of detected faces.
if len(face_list) != 0:
for (pos_x, pos_y, w, h) in face_list:
print(f"pos_x = {pos_x}, pos_y = {pos_y}, w = {w}, h = {h}")
cv2.rectangle(image, (pos_x, pos_y), (pos_x + w, pos_y + h), (0,0,255), thickness=5)
if __name__ == '__main__':
'''
[Abstract]
main function
'''
cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")
#
windowInitialized = False
q1 = mp.Queue() q2 = mp.Queue() p = mp.Process(target=DetectFacesProcess, args=(q1, q2)) p.daemon = True p.start()
init = False
while True:
try:
ret, frame = cap.read()
if ret == True:
# Pass the image to the face detection process. if (q1.qsize() <= 1) and (q2.qsize() <= 1): q1.put(frame) # Receive results from face detection processing. if q2.qsize() != 0: face_list = q2.get() init = True if init == True: # Draw face rectangles. DrawFaceRectangles(frame, face_list)
# Please modify the value to fit your PC screen size.
frame2 = cv2.resize(frame, (1280, 720))
# Display video.
cv2.imshow(winname, frame2)
if windowInitialized==False:
# Specify window position only once at startup.
cv2.moveWindow(winname, 100, 100)
windowInitialized = True
# Press the "q" key to finish.
k = cv2.waitKey(1) & 0xff # necessary to display the video by imshow ()
if k == ord("q"):
break
# Exit the program if there is no specified window.
if not IsWindowVisible(winname):
break
except KeyboardInterrupt:
# Press '[ctrl] + [c]' on the console to exit the program.
print("KeyboardInterrupt")
break
# Terminate process p q1.put(-1) # Waiting for process p to finish p.join()
print("Finish main()")
cap.release()
cv2.destroyAllWindows()
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる
[Details]
Save the received image as a JPEG file.
Add a 6-digit number to the end of the file name and save it as a serial number.
Exit the program after saving 100 files.
プログラム起動から10秒経過後から、受信した映像を JPEG ファイル保存します。
ファイル名の末尾に６ケタの番号を付けて連番で保存します。
100枚 保存したらプログラムを終了します。
[Note]
If you set a high number of frame rates or high resolution for the camera,
the program may not save the file because the program's file saving process may not be in time.
カメラ側の設定でフレームレート数を高くする、解像度を高くする、などした場合は
ファイル保存の処理が追い付かなくなってバッファが満杯になり、
ファイル保存が間引かれる場合があります。
[Library install]
cv2: pip install opencv-python
'''
import cv2
import multiprocessing as mp
from queue import Empty
import os
import datetime
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
pathOut = 'image' # Image file save folder namequeue_max = 30 # Maximum number of queuessave_max = 100 # Number of files to save
# Exception 定義
BackendError = type('BackendError', (Exception,), {})
def IsWindowVisible(winname):
'''
[Abstract]
Check if the target window exists.
対象ウィンドウが存在するかを確認する。
[Param]
winname : Window title
[Return]
True : exist
存在する
False : not exist
存在しない
[Exception]
BackendError :
'''
try:
ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
if ret == -1:
raise BackendError('Use Qt as backend to check whether window is visible or not.')
return bool(ret)
except cv2.error:
return False
def SaveImageProcess(imageQueue): ''' [Abstract] Image file save task [Param] imageQueue : [i] Queue to store images to save. [Return] None ''' while True: try: image, filename = imageQueue.get(True, 10) # timeout 10 sec. # Termination check: If type(image) is "int" and the value is -1, it ends. if type(image) == int: if image == -1: break # Save image file. cv2.imwrite(filename, image) except Empty: # timeout of q1.get() print("Timeout happen.") print("Finish SaveImageProcess()")
if __name__ == '__main__':
'''
[Abstract]
main function
'''
cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")
windowInitialized = False
count = 0 if not os.path.exists(pathOut): os.mkdir(pathOut)
imageQueue = mp.Queue()
starttime = datetime.datetime.now()
p = mp.Process(target=SaveImageProcess, args=(imageQueue,))
p.start()
while True:
try:
ret, frame = cap.read()
if ret == True:
# File saving starts after 10 seconds or more have passed since the program started. if (datetime.datetime.now() - starttime).seconds > 10: # Save file if queue size is within the upper limit. if imageQueue.qsize() < queue_max: # Save jpeg file. count += 1 filename = os.path.join(pathOut, 'image_{:06d}.jpg'.format(count)) print(filename) imageQueue.put([frame, filename]) if count >= save_max: break
# Please modify the value to fit your PC screen size.
frame2 = cv2.resize(frame, (1280, 720))
# Display video.
cv2.imshow(winname, frame2)
if windowInitialized==False:
# Specify window position only once at startup.
cv2.moveWindow(winname, 100, 100)
windowInitialized = True
# Press the "q" key to finish.
k = cv2.waitKey(1) & 0xff # necessary to display the video by imshow ()
if k == ord("q"):
break
# Exit the program if there is no specified window.
if not IsWindowVisible(winname):
break
except KeyboardInterrupt:
# Press '[ctrl] + [c]' on the console to exit the program.
print("KeyboardInterrupt")
break
# Terminate process p imageQueue.put([-1,-1]) print("Wait for process p to finish") p.join()
print("Finish main()")
cap.release()
cv2.destroyAllWindows()
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる。
[Details]
Add reconnection when video is disconnected.
映像切断時の再接続処理を追加する。
[Library install]
cv2: pip install opencv-python
'''
import cv2
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
url = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"
#
windowInitialized = False
# Exception definition.
BackendError = type('BackendError', (Exception,), {})
def IsWindowVisible(winname):
'''
[Abstract]
Check if the target window exists.
対象ウィンドウが存在するかを確認する。
[Param]
winname : Window title
[Return]
True : exist
存在する
False : not exist
存在しない
[Exception]
BackendError :
'''
try:
ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
if ret == -1:
raise BackendError('Use Qt as backend to check whether window is visible or not.')
return bool(ret)
except cv2.error:
return False
if __name__ == '__main__':
'''
[Abstract]
main function.
'''
cap = cv2.VideoCapture(url)
while True:
try:
ret, frame = cap.read()
if ret == True:
# Please modify the value to fit your PC screen size.
frame2 = cv2.resize(frame, (1280, 720))
# Display video.
cv2.imshow(winname, frame2)
if windowInitialized==False:
# Specify window position only once at startup.
cv2.moveWindow(winname, 100, 100)
windowInitialized = True
else: print("cap.read() return False.") # The timeout period seems to be 30 seconds. # And there seems to be no API to change the timeout value. # Reconnect cap.release() cap = cv2.VideoCapture(url)
# Press the "q" key to finish.
k = cv2.waitKey(1) & 0xff # necessary to display the video by imshow ()
if k == ord("q"):
break
# Exit the program if there is no specified window.
if not IsWindowVisible(winname):
break
except KeyboardInterrupt:
# Press'[ctrl] + [c]' on the console to exit the program.
print("KeyboardInterrupt")
break
cap.release()
cv2.destroyAllWindows()
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる。
[Details]
Display the video with GUI using tkinter.
tkinter を使ったGUIで映像を表示します。
BGR → RGB
numpy.ndarray → PIL.Image → ImageTk.PhotoImage
(1) BGR → RGB
(2) numpy.ndarray → PIL.Image
(3) PIL.Image → ImageTk.PhotoImage
[Library install]
cv2: pip install opencv-python
PIL : pip install pillow
'''
import cv2
import time
import tkinter as tk
from tkinter import messagebox
from PIL import Image, ImageTk, ImageOps
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
url = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"
class Application(tk.Frame):
def __init__(self, master = None):
super().__init__(master)
self.pack()
# Window settings.
self.master.title("Display i-PRO camera with tkinter") # Window title
self.master.geometry("800x600+100+100") # Window size, position
# Event registration for window termination.
self.master.protocol("WM_DELETE_WINDOW", self.on_closing_window)
# Create canvas.
self.canvas = tk.Canvas(self.master)
# Add mouse click event to canvas.
self.canvas.bind('<Button-1>', self.canvas_click)
# Place canvas.
self.canvas.pack(expand = True, fill = tk.BOTH)
# Raise a video display event (disp_image) after 500m
self.cap = None
self.disp_id = self.after(500, self.disp_image)
def on_closing_window(self):
''' Window closing event. '''
if messagebox.askokcancel("QUIT", "Do you want to quit?"):
self.master.destroy()
print("Finish Application.")
def canvas_click(self, event):
''' Event handling with mouse clicks on canvas '''
if self.disp_id is None:
# Connect camera.
self.cap = cv2.VideoCapture(url)
# Display image.
self.disp_image()
else:
# Release camera.
self.after_cancel(self.disp_id)
self.disp_id = None
self.cap.release()
def disp_image(self):
''' Display image on Canvas '''
if self.cap == None:
# Connect camera.
self.cap = cv2.VideoCapture(url)
# Get frame.
ret, frame = self.cap.read()
if ret == True:
# (1) Convert image from BGR to RGB.
cv_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
# (2) Convert image from ndarray to PIL.Image.
pil_image = Image.fromarray(cv_image)
# Get canvas size.
canvas_width = self.canvas.winfo_width()
canvas_height = self.canvas.winfo_height()
# Resize the image to the size of the canvas without changing the aspect ratio.
# アスペクトを維持したまま画像を Canvas と同じサイズにリサイズ
pil_image = ImageOps.pad(pil_image, (canvas_width, canvas_height))
# (3) Convert image from PIL.Image to PhotoImage
self.photo_image = ImageTk.PhotoImage(image=pil_image)
# Display image on the canvas.
self.canvas.create_image(
canvas_width / 2, # Image display position (center of the canvas)
canvas_height / 2,
image=self.photo_image # image data
)
else:
print("cap.read() return False.")
# The timeout period seems to be 30 seconds.
# And there seems to be no API to change the timeout value.
time.sleep(1)
# Reconnect
self.cap.release()
self.cap = cv2.VideoCapture(url)
# Raise a video display event (disp_image) after 1ms.
self.disp_id = self.after(1, self.disp_image)
if __name__ == "__main__":
root = tk.Tk()
app = Application(master = root)
app.mainloop()
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる。
[Details]
Display the video with GUI using tkinter.
Try to improve performance by creating a video receiving process.
tkinter を使ったGUIで映像を表示します。
映像受信プロセスを作成することでパフォーマンス改善を試みます。
BGR → RGB
numpy.ndarray → PIL.Image → ImageTk.PhotoImage
(1) BGR → RGB
(2) numpy.ndarray → PIL.Image
(3) PIL.Image → ImageTk.PhotoImage
[Library install]
cv2: pip install opencv-python
PIL : pip install pillow
'''
import cv2
import time
import tkinter as tk
from tkinter import messagebox
from PIL import Image, ImageTk, ImageOps
import multiprocessing as mp
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
url = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"
class Application(tk.Frame):
def __init__(self, master = None):
super().__init__(master)
self.pack()
# Window settings.
self.master.title("Display i-PRO camera with tkinter") # Window title
self.master.geometry("800x600+100+100") # Window size, position
# Event registration for window termination.
self.master.protocol("WM_DELETE_WINDOW", self.on_closing_window)
# Create canvas.
self.canvas = tk.Canvas(self.master)
# Add mouse click event to canvas.
self.canvas.bind('<Button-1>', self.canvas_click)
# Place canvas.
self.canvas.pack(expand = True, fill = tk.BOTH)
# Create image receiving process and queue self.imageQueue = mp.Queue() self.request = mp.Value('i', 0) # -1 : Exit ReceiveImageProcess. # 0 : Normal. # 1 : Connect camera. # 2 : Release camera. self.p = mp.Process(target=ReceiveImageProcess, args=(self.imageQueue, self.request)) self.p.start()
# Raise a video display event (disp_image) after 500m
self.disp_id = self.after(500, self.disp_image)
def on_closing_window(self):
''' Window closing event. '''
if messagebox.askokcancel("QUIT", "Do you want to quit?"):
# Request terminate process self.p.
self.request.value = -1
# Waiting for process p to finish
time.sleep(1)
# Flash buffer.
# The program cannot complete p.join() unless the imageQueue is emptied.
for i in range(self.imageQueue.qsize()):
pil_image = self.imageQueue.get()
# Wait for process p to be terminated.
self.p.join()
self.master.destroy()
print("Finish Application.")
def canvas_click(self, event):
''' Event handling with mouse clicks on canvas '''
if self.disp_id is None:
# Connect camera.
self.request.value = 1
# Display image.
self.disp_image()
else:
# Release camera.
self.request.value = 2
# Cancel scheduling
self.after_cancel(self.disp_id)
self.disp_id = None
def disp_image(self):
''' Display image on Canvas '''
# If there is data in the imageQueue, the program receives the data and displays the video.
num = self.imageQueue.qsize()
if num > 0:
if (num > 5):
num -= 1
for i in range(num):
cv_image = self.imageQueue.get()
# (2) Convert image from ndarray to PIL.Image.
pil_image = Image.fromarray(cv_image)
# Get canvas size.
canvas_width = self.canvas.winfo_width()
canvas_height = self.canvas.winfo_height()
# Resize the image to the size of the canvas without changing the aspect ratio.
# アスペクトを維持したまま画像を Canvas と同じサイズにリサイズ
pil_image = ImageOps.pad(pil_image, (canvas_width, canvas_height))
# (3) Convert image from PIL.Image to PhotoImage
# PIL.Image から PhotoImage へ変換する
self.photo_image = ImageTk.PhotoImage(image=pil_image)
# Display image on the canvas.
self.canvas.create_image(
canvas_width / 2, # Image display position (center of the canvas)
canvas_height / 2,
image=self.photo_image # image data
)
else:
pass
# Raise a video display event (disp_image) after 1ms.
self.disp_id = self.after(1, self.disp_image)
def ReceiveImageProcess(imageQueue, request): """ Receive Image Process. Args: imageQueue [o] This process stores the received image data in the imageQueue. request [i] Shared memory for receiving requests from the main process. -1: Terminate process. 0: Nothing. 1: Connect camera. 2: Release camera connection. Returns: None Raises None """ # Connect camera. cap = cv2.VideoCapture(url) while True: if cap != None: # Get frame. ret, frame = cap.read() if ret == True: # (1) Convert image from BGR to RGB. cv_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) if imageQueue.qsize() < 10: imageQueue.put(cv_image) else: print("cap.read() return False.") # The timeout period seems to be 30 seconds. # And there seems to be no API to change the timeout value. time.sleep(1) # Reconnect cap.release() cap = cv2.VideoCapture(url) else: time.sleep(0.1) # Check process termination request. if request.value == -1: # Terminate process. cap.release() request.value = 0 break # Check connect request. if request.value == 1: cap = cv2.VideoCapture(url) request.value = 0 # Check release request. if request.value == 2: cap.release() cap = None request.value = 0 print("Terminate SaveImageProcess().")
if __name__ == "__main__":
root = tk.Tk()
app = Application(master = root)
app.mainloop()
'''
[Abstract]
Try connecting to an i-PRO camera with RTSP.
RTSP で i-PRO カメラと接続してみる。
[Details]
Display the video with GUI using tkinter.
Add menus and buttons to make it look like a GUI app.
tkinter を使ったGUIで映像を表示します。
メニューとボタンを追加してGUIアプリらしくします。
BGR → RGB
numpy.ndarray → PIL.Image → ImageTk.PhotoImage
(1) BGR → RGB
(2) numpy.ndarray → PIL.Image
(3) PIL.Image → ImageTk.PhotoImage
[Library install]
cv2: pip install opencv-python
PIL : pip install pillow
'''
import cv2
import time
import tkinter as tk
from tkinter import messagebox
from PIL import Image, ImageTk, ImageOps
import multiprocessing as mp
user_id = "user-id" # Change to match your camera setting
user_pw = "password" # Change to match your camera setting
host = "192.168.0.10" # Change to match your camera setting
winname = "VIDEO" # Window title
url = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"
class Application(tk.Frame):
def __init__(self, master = None):
super().__init__(master)
self.pack()
# Window settings.
self.master.title("Display i-PRO camera with tkinter") # Window title
self.master.geometry("800x600+100+100") # Window size, position
# Event registration for window termination.
self.master.protocol("WM_DELETE_WINDOW", self.on_closing_window)
# Create menu. menubar = tk.Menu(self.master) self.master.configure(menu=menubar) filemenu = tk.Menu(menubar) menubar.add_cascade(label='File', menu=filemenu) filemenu.add_command(label='Quit', command = self.on_closing_window) # Create button_frame self.button_frame = tk.Frame(self.master, padx=10, pady=10, relief=tk.RAISED, bd=2) self.button_frame.pack(side = tk.BOTTOM, fill=tk.X) # Create quit_button self.quit_button = tk.Button(self.button_frame, text='Quit', width=10, command = self.on_closing_window) self.quit_button.pack(side=tk.RIGHT)
# Create canvas.
self.canvas = tk.Canvas(self.master)
# Add mouse click event to canvas.
self.canvas.bind('<Button-1>', self.canvas_click)
# Place canvas.
self.canvas.pack(expand = True, fill = tk.BOTH)
# Create image receiving process and queue
self.imageQueue = mp.Queue()
self.request = mp.Value('i', 0) # -1 : Exit ReceiveImageProcess.
# 0 : Normal.
# 1 : Connect camera.
# 2 : Release camera.
self.p = mp.Process(target=ReceiveImageProcess, args=(self.imageQueue, self.request))
self.p.start()
# Raise a video display event (disp_image) after 500m
self.disp_id = self.after(500, self.disp_image)
def on_closing_window(self):
''' Window closing event. '''
if messagebox.askokcancel("QUIT", "Do you want to quit?"):
# Request terminate process self.p.
self.request.value = -1
# Waiting for process p to finish
time.sleep(1)
# Flash buffer.
# The program cannot complete p.join() unless the imageQueue is emptied.
for i in range(self.imageQueue.qsize()):
pil_image = self.imageQueue.get()
# Wait for process p to be terminated.
self.p.join()
self.master.destroy()
print("Finish Application.")
def canvas_click(self, event):
''' Event handling with mouse clicks on canvas '''
if self.disp_id is None:
# Connect camera.
self.request.value = 1
# Display image.
self.disp_image()
else:
# Release camera.
self.request.value = 2
# Cancel scheduling
self.after_cancel(self.disp_id)
self.disp_id = None
def disp_image(self):
''' Display image on Canvas '''
# If there is data in the imageQueue, the program receives the data and displays the video.
num = self.imageQueue.qsize()
if num > 0:
if (num > 5):
num -= 1
for i in range(num):
cv_image = self.imageQueue.get()
# (2) Convert image from ndarray to PIL.Image.
pil_image = Image.fromarray(cv_image)
# Get canvas size.
canvas_width = self.canvas.winfo_width()
canvas_height = self.canvas.winfo_height()
# Resize the image to the size of the canvas without changing the aspect ratio.
# アスペクトを維持したまま画像を Canvas と同じサイズにリサイズ
pil_image = ImageOps.pad(pil_image, (canvas_width, canvas_height))
# (3) Convert image from PIL.Image to PhotoImage
# PIL.Image から PhotoImage へ変換する
self.photo_image = ImageTk.PhotoImage(image=pil_image)
# Display image on the canvas.
self.canvas.create_image(
canvas_width / 2, # Image display position (center of the canvas)
canvas_height / 2,
image=self.photo_image # image data
)
else:
pass
# Raise a video display event (disp_image) after 1ms.
self.disp_id = self.after(1, self.disp_image)
def ReceiveImageProcess(imageQueue, request):
'''
Receive Image Process.
Args:
imageQueue [o] This process stores the received image data in the imageQueue.
request [i] Shared memory for receiving requests from the main process.
-1: Terminate process.
0: Nothing.
1: Connect camera.
2: Release camera connection.
Returns:
None
Raises
None
'''
# Connect camera.
cap = cv2.VideoCapture(url)
while True:
if cap != None:
# Get frame.
ret, frame = cap.read()
if ret == True:
# (1) Convert image from BGR to RGB.
cv_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
if imageQueue.qsize() < 10:
imageQueue.put(cv_image)
else:
print("cap.read() return False.")
# The timeout period seems to be 30 seconds.
# And there seems to be no API to change the timeout value.
time.sleep(1)
# Reconnect
cap.release()
cap = cv2.VideoCapture(url)
else:
time.sleep(0.1)
# Check process termination request.
if request.value == -1:
# Terminate process.
cap.release()
request.value = 0
break
# Check connect request.
if request.value == 1:
cap = cv2.VideoCapture(url)
request.value = 0
# Check release request.
if request.value == 2:
cap.release()
cap = None
request.value = 0
print("Terminate SaveImageProcess().")
if __name__ == "__main__":
root = tk.Tk()
app = Application(master = root)
app.mainloop()
Licensed under the Apache License, Version
2.0 (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
Unless required by
applicable law or agreed to in writing, software distributed under the
License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for
the specific language governing permissions and limitations under the
License.
