Public functions
open
Override. Create an opencv video capture from the given video.
Parameters
- video: (name of video file, image sequence, url or video stream, GStreamer pipeline, IP camera) The video to browse.
View Source
def open(self, video):
"""Override. Create an opencv video capture from the given video.
:param video: (name of video file, image sequence, url or video
stream, GStreamer pipeline, IP camera) The video to browse.
"""
self.reset()
sppasVideoReader.open(self, video)
close
Override. Release the flow taken by the reading of the video.
View Source
def close(self):
"""Override. Release the flow taken by the reading of the video."""
self.reset()
sppasVideoReader.close(self)
reset
Reset the buffer but does not change anything to the video.
View Source
def reset(self):
"""Reset the buffer but does not change anything to the video."""
self.__images = list()
self.__buffer_idx = (-1, -1)
reset_with_start_idx
Reset AND set the index of the 1st frame of the current buffer.
Returns
- (tuple) start index of the frames in the buffer
Parameters
View Source
def reset_with_start_idx(self, idx):
"""Reset AND set the index of the 1st frame of the current buffer.
:return: (tuple) start index of the frames in the buffer
"""
self.__images = list()
idx = int(idx)
self.__buffer_idx = (idx, idx)
get_buffer_size
Return the defined size of the buffer.
View Source
def get_buffer_size(self):
"""Return the defined size of the buffer."""
return self.__nb_img
set_buffer_size
Set the size of the buffer.
The new value is applied to the next buffer, it won't affect the currently in-use data.
A value of -1 will fix automatically the buffer to use a MAXMEMORYSIZE Gb of RAM.
Parameters
- value: (int) New size of the buffer.
Raises
- ValueError: invalid given value
View Source
def set_buffer_size(self, value=-1):
"""Set the size of the buffer.
The new value is applied to the next buffer, it won't affect the currently in-use data.
A value of -1 will fix automatically the buffer to use a MAX_MEMORY_SIZE Gb of RAM.
:param value: (int) New size of the buffer.
:raises: ValueError: invalid given value
"""
value = int(value)
if value == -1:
if self.is_opened() is False:
w, h = (1920, 1080)
else:
w, h = (self.get_width(), self.get_height())
nbytes = w * h * 3
value = sppasVideoReaderBuffer.MAX_MEMORY_SIZE // nbytes
if self.is_opened() is True and value > self.get_nframes():
value = self.get_nframes()
if value <= 0:
raise NegativeValueError(value)
if self.is_opened() is True and value > self.get_nframes():
value = self.get_nframes()
if self.__overlap >= value:
raise ValueError("The already defined overlap value {:d} can't be greater than the buffer size.")
self.__nb_img = value
logging.info('The video buffer is set to {:d} images'.format(self.__nb_img))
get_buffer_overlap
Return the overlap value of the buffer.
View Source
def get_buffer_overlap(self):
"""Return the overlap value of the buffer."""
return self.__overlap
set_buffer_overlap
Set the number of images to keep from the previous buffer.
The new value is applied to the next buffer, it won't affect the
currently in-use data.
Parameters
Raises
- ValueError: Invalid given value
View Source
def set_buffer_overlap(self, value):
"""Set the number of images to keep from the previous buffer.
The new value is applied to the next buffer, it won't affect the
currently in-use data.
:param value: (int) Nb of image
:raises: ValueError: Invalid given value
"""
overlap = int(value)
if overlap >= self.__nb_img or overlap < 0:
raise ValueError('Invalid buffer overlap')
self.__overlap = value
seek_buffer
Set the position of the frame for the next buffer to be read.
It won't change the current position in the video until "next" is
invoked. It invalidates the current buffer.
Parameters
- value: (int) Frame position
View Source
def seek_buffer(self, value):
"""Set the position of the frame for the next buffer to be read.
It won't change the current position in the video until "next" is
invoked. It invalidates the current buffer.
:param value: (int) Frame position
"""
value = self.check_frame(value)
self.reset()
self.__buffer_idx = (-1, value - 1)
tell_buffer
Return the frame position for the next buffer to be read.
Possibly, it can't match the current position in the stream, if
video.read() was invoked for example.
View Source
def tell_buffer(self):
"""Return the frame position for the next buffer to be read.
Possibly, it can't match the current position in the stream, if
video.read() was invoked for example.
"""
return self.__buffer_idx[1] + 1
get_buffer_range
Return the indexes of the frames of the current buffer.
Returns
- (tuple) start index, end index of the frames in the buffer
View Source
def get_buffer_range(self):
"""Return the indexes of the frames of the current buffer.
:return: (tuple) start index, end index of the frames in the buffer
"""
if -1 in self.__buffer_idx:
return (-1, -1)
return self.__buffer_idx
next
Fill in the buffer with the next sequence of images of the video.
Returns
- False if we reached the end of the video
View Source
def next(self):
"""Fill in the buffer with the next sequence of images of the video.
:return: False if we reached the end of the video
"""
if self.is_opened() is False:
return False
self.__images = list()
nb_frames = self.__nb_img - self.__overlap
if self.__buffer_idx[1] == -1:
nb_frames = self.__nb_img
start_frame = self.__buffer_idx[1] + 1
if start_frame >= self.get_nframes():
return False
result = self.__load_frames(start_frame, nb_frames)
next_frame = start_frame + len(result)
delta = self.__nb_img - self.__overlap
self.__images = self.__images[delta:]
self.__buffer_idx = (start_frame - len(self.__images), next_frame - 1)
self.__images.extend(result)
result.clear()
return next_frame < self.get_nframes()
check_buffer_index
Raise an exception if the given image index is not valid.
Parameters
Raises
NegativeValueError
IndexRangeException
View Source
def check_buffer_index(self, value):
"""Raise an exception if the given image index is not valid.
:param value: (int)
:raises: NegativeValueError
:raises: IndexRangeException
"""
value = int(value)
if value < 0:
raise NegativeValueError(value)
begin, end = self.get_buffer_range()
if value < self.get_buffer_size():
return value
raise IndexRangeException(value, 0, self.get_buffer_size())
append
Append an image into the buffer and pop the first if full queue.
Parameters
- image: (sppasImage) A new image to append to the list
View Source
def append(self, image):
"""Append an image into the buffer and pop the first if full queue.
:param image: (sppasImage) A new image to append to the list
"""
if isinstance(image, sppasImage) is False:
raise sppasTypeError(type(image), 'sppasImage')
self.__images.append(image)
if len(self.__images) > self.__nb_img:
self.__images.pop(0)
pop
Pop an image of the buffer.
Parameters
- img_idx: (int) Index of the image in the buffer
Raises
IndexRangeException
View Source
def pop(self, img_idx):
"""Pop an image of the buffer.
:param img_idx: (int) Index of the image in the buffer
:raise: IndexRangeException
"""
img_idx = int(img_idx)
if 0 <= img_idx < self.get_buffer_size():
self.__images.pop(img_idx)
else:
raise IndexRangeException(img_idx, 0, self.get_buffer_size())
set_at
Set an image of the buffer.
No verification is performed on the image. It should be the
same format (size, nb channels, etc) than the other ones.
Use this method with caution!
Parameters
- img: (sppasImage) Set the image in the buffer
- img_idx: (int) Index of the image in the buffer
Raises
IndexRangeException
View Source
def set_at(self, img, img_idx):
"""Set an image of the buffer.
No verification is performed on the image. It should be the
same format (size, nb channels, etc) than the other ones.
Use this method with caution!
:param img: (sppasImage) Set the image in the buffer
:param img_idx: (int) Index of the image in the buffer
:raises: IndexRangeException
"""
img_idx = int(img_idx)
if 0 <= img_idx < self.get_buffer_size():
self.__images[img_idx] = img
else:
raise IndexRangeException(img_idx, 0, self.get_buffer_size())