'How to create a reusable screen recording thread?
I created a reusable screen recording thread by cv2. However, my computer crashed when I stop and restart this thread. Sometimes, it works but the video MP4 file crashed
Background:
for my code ,there is a main thread, it will call Video_Record.start_recording to record video until main thread calls Video_Record.stop_recording. The main thread can reuse this video thread and create new video anytime.
is there any suggestion to improve my code? I am thinking about the reason why my computer crashed is memory leak. I am not sure daemon thread is ok or not.
class Video_Record(threading.Thread):
def __init__(self):
threading.Thread.__init__(self, daemon=True)
self.running = True # must be true
self.recording_running = False
self.fps = 20
self.prev = 0
self.out = None
self.screen_size = tuple(pyautogui.size())
# self.screen_size=(1920,1080)
def run(self):
while self.running:
while self.recording_running:
try:
img = pyautogui.screenshot()
self.prev = time.time()
frame = np.array(img)
frame = cv2.resize(frame, self.screen_size)
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
self.out.write(frame)
# cv2.imshow("screen recorder",frame) # show screen recorder
except Exception as e:
logger.error('[MAIN] out={}'.format(self.out))
traceback.print_exc()
write_log(logpath=Processlogpath, company_name='RPA', function_name='Main RPA Trigger',
status='Fail', exception=e, spec_msg="video record,path={}".format(self.video_path))
# if cv2.waitKey(1)==ord("q"):
# break
def stop_recording(self):
try:
self.recording_running = False
self.out.release()
self.out = None
cv2.destroyAllWindows()
logger.info('[MAIN] video record end')
write_log(logpath=Processlogpath, company_name='RPA', function_name='Main RPA Trigger',
status='Success', spec_msg="video record success")
except Exception:
traceback.print_exc()
def start_recording(self, video_name=r'output'):
try:
if self.running:
logger.info('[MAIN] video record start')
if self.out == None:
#self.fourcc = cv2.VideoWriter_fourcc(*'XVID')
# self.video_path = "D:\\Desktop\\New folder\\{}-output3.avi".format(time.strftime("%Y%m%d-%H%M%S"))
self.fourcc = cv2.VideoWriter_fourcc(*'mp4v')
if production_platform:
self.video_path = "H:\\activity\\video_record\\{}-{}.mp4".format(
video_name, time.strftime("%Y%m%d-%H%M%S"))
else:
self.video_path = "H:\\RPA\\activity\\video_record\\{}-{}.mp4".format(
video_name, time.strftime("%Y%m%d-%H%M%S"))
self.out = cv2.VideoWriter(
self.video_path, self.fourcc, self.fps, (self.screen_size))
self.recording_running = True
except Exception:
traceback.print_exc()
def stop(self):
"""
pause the thread
"""
self.running = False
def resume(self):
"""
resume the thread
"""
self.running = True
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
| Solution | Source |
|---|
