[Python] Trouble writing images and videos using OpenCV

I wrote a yolo based detection program that detects humans as shown here. It fails to write images and videos (both), not always but only some times. Since the issue is occassional, I can not resolve the issue. I am running the code on ubuntu with OpenCV installed using apt repository. The logic is like observing 50 frames after the first detection of humans for 15 detections and confirming the detection. Twentyone frames after the confirmation is recorded with 4 frames per second to generate a 5 second long video. The sever I have been running the code was over loaded in terms of memmory and cpu usage. Does this make any impact on OpenCV writing both?

import os
from ultralytics import YOLO
from ultralytics.utils.plotting import Annotator
import datetime
import cv2
import time
import logging

# Defining the class for making inferences.
class Detector:
def __init__(self):
os.makedirs("Log",exist_ok=True)
logging.basicConfig(filename=f"Log/log.log", format='%(asctime)s %(message)s', filemode='a')
self.logger = logging.getLogger()
self.logger.setLevel(logging.CRITICAL)
self.model=YOLO('yolov8n.pt',task='detect')
self.cap=cv2.VideoCapture(self.camera_input_url)
if self.cap.isOpened:
self.logger.critical("Camera Connected.")
else:
self.logger.critical("Failed to connect to the camera.")
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, int(os.getenv("Frame_Width")))
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, int(os.getenv("Frame_Height")))
#Flag for logging only once that No data is receiving.
self.flag=False

def make_detection(self):

# Reading the video input.
ret, self.frame = self.cap.read()
# Checking whether there is a frame.
if not ret:
if self.flag == False:
self.logger.critical("No data is receiving from the camera.")
self.cap=cv2.VideoCapture(self.camera_input_url)
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, int(os.getenv("Frame_Width")))
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, int(os.getenv("Frame_Height")))
self.flag=True
time.sleep(60)
else:
self.flag=False
# Making predictions on the frame using the model
self.results=self.model.predict(self.frame,imgsz=(int(os.getenv("Frame_Width")),int(os.getenv("Frame_Height"))))

class Annotation_and_flagger(Detector):
def __init__(self):
super().__init__()
self.k,self.k1=1,0
self.person_detection,self.person_detection_count=False,0

def annotation_and_flag_detection(self):

# Iterating over the results of the frame
for r in self.results:
# Creating an annotator object of the frame
annotator = Annotator(self.frame)
boxes = r.boxes
# Iterating over the boxes of the frame
for box in boxes:
b = box.xyxy[0]
c = box.cls
#Checking whether the class is 2 (of person)
if int(c)==2 and float(box.conf[0])>float(os.getenv("Person_Confidence")):
# Assigning the annotator object with the respective box.
annotator.box_label(b, self.model.names[int(c)],color=(0,128,128))
# Flagging person detection and setting the count of person detection as 1
if self.k==1:
if self.k1==0:
self.person_detection=True
self.person_detection_count=1
self.frame=cv2.resize(self.frame,(636,360))

class Confirm_and_pusher(Annotation_and_flagger):

def __init__(self):
super().__init__()
self.cameraId=1
self.cameraName="cam1"
self.time_person_push=None

def confirm_and_push_person_detection(self):
if self.person_detection:
self.k+=1
# Loop to count the frames that has detected with a person
for r in self.results:
boxes=r.boxes
for box in boxes:
if int(box.cls)==2 and float(box.conf[0])>float(os.getenv("Person_Confidence")):
self.person_detection_count+=1
if self.person_detection_count==int(os.getenv("Person_Confirm_Detection_Count")):
self.logger.critical("A person is detected.")
# Writing the thumbnail for tiger.
cv2.imwrite(f"Person_{self.cameraName}.jpeg",self.frame)
# Initializing the video writter object.
self.out_person=cv2.VideoWriter(f"Person_{self.cameraName}.mp4", cv2.VideoWriter_fourcc(*'avc1'), 4.0, (636,360))
# Setting k1=1
self.k1=1
break
break

# Resetting the value of k when k=50 and setting the detection of person to False
if self.k==50:
self.person_detection,self.k=False,1
# Recording the time of person detection
if self.k1==1:
self.time_person_push=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# Writting 27 frames after person detection to the videowriter object and releasing the object
if self.k1>0:
self.k1+=1
self.out_person.write(self.frame)
if self.k1==21:
self.k1=0
self.out_person.release()

class Runner(Confirm_and_pusher):

def __init__(self,camera_input_url):
self.camera_input_url=camera_input_url
super().__init__()
self.cam_flag=True

def run(self):
while True:
try:
self.make_detection()
self.annotation_and_flag_detection()
self.confirm_and_push_person_detection()
except:
continue

if __name__=='__main__':
R=Runner(str(os.getenv("RTSP_URL_OF_THE_CAMERA")))
R.run()


I tried the code on a VM with limited resources. But, I could not reproduce the issue. I want to reproduce the issue and resolve it. It could be because the machine I have been running was high on cache memmroy and the testing one is high on actual RAM usage. Any ways, CPU usage was normal everywhere. The concern is about memory. Since the harddrive of the server is a server grade one, such issues can also be ruled out. I want a help with this issue of memroy. The actual question is whether high cache memmory cause any impact on application behaviour as mentioned here.

Continue reading...