Write a video, a set of images and/or coordinates into files.
There are 4 main solutions to write the result (images+coords):
For 3 and 4, there are 2 options - at least one has to be selected:
Create a new instance.
Parts of each image can be extracted in separate image files and/or
surrounded on the given image.
Output video and images can be resized.
def __init__(self, image_writer=None):
"""Create a new instance.
Parts of each image can be extracted in separate image files and/or
surrounded on the given image.
Output video and images can be resized.
"""
sppasBufferVideoWriter.__init__(self)
self._img_writer = sppasCoordsImageWriter()
if image_writer is not None:
if isinstance(image_writer, sppasCoordsImageWriter) is True:
self._img_writer = image_writer
self._tag_video_writer = None
self._xra = dict()
self._xra_tiername = 'VideoCoords'
Return True if coordinates will be saved in a CSV file.
def get_csv_output(self):
"""Return True if coordinates will be saved in a CSV file."""
return self._img_writer.options.get_csv_output()
Return True if coordinates will be saved in an XRA file.
def get_xra_output(self):
"""Return True if coordinates will be saved in an XRA file."""
return self._img_writer.options.get_xra_output()
Return True if faces of the images will be surrounded.
def get_tag_output(self):
"""Return True if faces of the images will be surrounded."""
return self._img_writer.options.get_tag_output()
Return True if the option to crop faces is enabled.
def get_crop_output(self):
"""Return True if the option to crop faces is enabled."""
return self._img_writer.options.get_crop_output()
Return the width of the output image files.
def get_output_width(self):
"""Return the width of the output image files."""
return self._img_writer.options.get_width()
Return the height of the outputs files.
def get_output_height(self):
"""Return the height of the outputs files."""
return self._img_writer.options.get_height()
Set any/some/all of the options.
def set_options(self, csv=None, xra=None, tag=None, crop=None, width=None, height=None, folder=None):
"""Set any/some/all of the options."""
self._img_writer.set_options(csv=csv, xra=xra, tag=tag, crop=crop, width=width, height=height)
if folder is not None:
self._folder = bool(folder)
def get_xra_tiername(self):
return self._xra_tiername
Return the size of the image depending on the image and options.
def get_image_size(self, image):
"""Return the size of the image depending on the image and options."""
return image.get_proportional_size(width=self._img_writer.options.get_width(), height=self._img_writer.options.get_height())
Close all currently used sppasVideoWriter().
It has to be invoked when writing buffers is finished in order to
release the video writers.
def close(self):
"""Close all currently used sppasVideoWriter().
It has to be invoked when writing buffers is finished in order to
release the video writers.
"""
if self._tag_video_writer is not None:
self._tag_video_writer.close()
self._tag_video_writer = None
if len(self._xra) > 0:
for out_xra_name in self._xra:
logging.info('Writing file {}'.format(out_xra_name))
trs = self._xra[out_xra_name]
trs.write(out_xra_name)
Save the result into file(s) depending on the options.
The out_name is a base name, its extension is ignored and replaced by
the one(s) defined in this class.
def write(self, video_buffer, out_name, opt_pattern=''):
"""Save the result into file(s) depending on the options.
The out_name is a base name, its extension is ignored and replaced by
the one(s) defined in this class.
:param video_buffer: (sppasCoordsVideoBuffer) The images and results to write
:param out_name: (str) The output name for the folder and/or the video
:param opt_pattern: (str) Optional pattern to add to filename(s)
:return: list of newly created file names
"""
new_files = list()
fn, _ = os.path.splitext(out_name)
out_name = fn
if self._img_writer.options.csv is True:
out_csv_name = out_name + opt_pattern + '.csv'
self.write_csv_coords(video_buffer, out_csv_name)
new_files.append(out_csv_name)
if self._img_writer.options.xra is True:
out_xra_name = out_name + opt_pattern + '.xra'
self.write_xra_coords(video_buffer, out_xra_name)
new_files.append(out_xra_name)
if self.get_video_output() is True or self._img_writer.options.get_tag_output() is True:
new_video_files = self.write_video(video_buffer, out_name, opt_pattern)
if len(new_video_files) > 1:
logging.info('{:d} video files created'.format(len(new_video_files)))
new_files.extend(new_video_files)
if self._folder is True:
new_image_files = self.write_folder(video_buffer, out_name, opt_pattern)
if len(new_image_files) > 1:
logging.info('{:d} image files created'.format(len(new_image_files)))
return new_files
Write or append a list of coordinates in a CSV file.
def write_csv_coords(self, video_buffer, out_csv_name):
"""Write or append a list of coordinates in a CSV file.
:param video_buffer: (sppasFacesVideoBuffer) The images and results to write
:param out_csv_name: (str) The filename of the CSV file to write
"""
begin_idx, end_idx = video_buffer.get_buffer_range()
buffer_nb = end_idx // video_buffer.get_buffer_size()
mode = 'w'
if os.path.exists(out_csv_name) is True:
if buffer_nb == 0:
trash_name = sppasTrash().put_file_into(out_csv_name)
logging.warning('A file with name {:s} is already existing.\nThis file is moved into the Trash of SPPAS with name: {:s}'.format(out_csv_name, trash_name))
else:
mode = 'a+'
with codecs.open(out_csv_name, mode, encoding='utf-8') as fd:
for i in range(video_buffer.__len__()):
self.write_coords(fd, video_buffer, buffer_nb, i)
Write the coords into the given stream.
def write_coords(self, fd, video_buffer, buffer_idx, idx):
"""Write the coords into the given stream.
- frame number
- the index of the coords
- timestamp
- confidence
- success
- x, y, w, h,
- buffer number
- index in the buffer
:param fd: (Stream) File descriptor, String descriptor, stdout, etc
:param video_buffer: (sppasCoordsVideoBuffer)
:param buffer_idx: (int) Buffer number
:param idx: (int) An integer to write
"""
sep = self._img_writer.get_csv_sep()
coords = video_buffer.get_coordinates(idx)
frame_idx = buffer_idx * video_buffer.get_buffer_size() + idx
if len(coords) == 0:
fd.write('{:d}{:s}'.format(frame_idx + 1, sep))
fd.write('0{:s}'.format(sep))
fd.write('{:.3f}{:s}'.format(float(frame_idx) / self._fps, sep))
fd.write('none{:s}'.format(sep))
fd.write('0{:s}'.format(sep))
fd.write('0{:s}0{:s}0{:s}0{:s}'.format(sep, sep, sep, sep))
fd.write('{:d}{:s}'.format(buffer_idx + 1, sep))
fd.write('{:d}{:s}'.format(idx, sep))
fd.write('\n')
else:
radius = 0.5 / float(self._fps)
mid_time = float(frame_idx) / float(self._fps) + radius
for j in range(len(coords)):
fd.write('{:d}{:s}'.format(frame_idx + 1, sep))
fd.write('{:d}{:s}'.format(j + 1, sep))
fd.write('{:.3f}{:s}'.format(mid_time, sep))
fd.write('{:f}{:s}'.format(coords[j].get_confidence(), sep))
fd.write('1{:s}'.format(sep))
fd.write('{:d}{:s}'.format(coords[j].x, sep))
fd.write('{:d}{:s}'.format(coords[j].y, sep))
fd.write('{:d}{:s}'.format(coords[j].w, sep))
fd.write('{:d}{:s}'.format(coords[j].h, sep))
fd.write('{:d}{:s}'.format(buffer_idx + 1, sep))
fd.write('{:d}{:s}'.format(idx, sep))
fd.write('\n')
Append a list of coordinates in an XRA object.
def write_xra_coords(self, video_buffer, out_xra_name):
"""Append a list of coordinates in an XRA object.
:param video_buffer: (sppasFacesVideoBuffer) The images and results to write
:param out_xra_name: (str) The filename of the XRA file to write
"""
if out_xra_name not in self._xra:
m = mimetypes.guess_type(out_xra_name)
if m[0] is None:
fn, fe = os.path.splitext(out_xra_name)
mime_type = 'video/' + fe[1:]
else:
mime_type = m[0]
media = sppasMedia(out_xra_name, mime_type=mime_type)
media.set_meta('fps', str(self._fps))
media.set_meta('width', str(self._img_writer.get_width()))
media.set_meta('height', str(self._img_writer.get_height()))
tier = sppasTier(self._xra_tiername)
tier.set_media(media)
trs = sppasXRA('VideoCoordinates')
trs.append(tier)
self._xra[out_xra_name] = trs
else:
trs = self._xra[out_xra_name]
tier = trs.find(self._xra_tiername)
if tier is None:
raise Exception("Invalid tier in XRA. Can't add coordinates.")
begin_idx, end_idx = video_buffer.get_buffer_range()
buffer_nb = end_idx // video_buffer.get_buffer_size()
for i in range(video_buffer.__len__()):
self.write_xra_ann(tier, video_buffer, buffer_nb, i)
def write_xra_ann(self, tier, video_buffer, buffer_idx, idx):
labels = list()
coords = video_buffer.get_coordinates(idx)
frame_idx = buffer_idx * video_buffer.get_buffer_size() + idx
for c in coords:
tag = sppasTag((c.x, c.y, c.w, c.h), tag_type='rect')
label = sppasLabel(tag, c.get_confidence())
labels.append(label)
radius = 0.5 / float(self._fps)
mid_time = float(frame_idx) / float(self._fps) + radius
frame_point = sppasPoint(mid_time, radius)
ann = tier.create_annotation(sppasLocation(frame_point), labels)
ann.set_meta('frame_index', str(frame_idx))
return ann
Save the result in video format.
def write_video(self, video_buffer, out_name, pattern):
"""Save the result in video format.
:param video_buffer: (sppasFacesVideoBuffer) The buffer with images to write
:param out_name: (str) The filename of the output video file
:param pattern: (str) Pattern to add to cropped video filename(s)
:return: list of newly created video file names
"""
new_files = list()
if pattern == '' and self._video is True and (self._img_writer.options.tag is True):
pattern = '-tag'
iter_images = video_buffer.__iter__()
for i in range(video_buffer.__len__()):
image = next(iter_images)
if self._video is True:
if self._video_writer is None:
self._video_writer, fn = self.create_video_writer(out_name, image, '')
new_files.append(fn)
self._video_writer.write(image)
if self._img_writer.options.tag is True:
coords = video_buffer.get_coordinates(i)
if image is not None:
if self._tag_video_writer is None:
self._tag_video_writer, fn = self.create_video_writer(out_name, image, pattern)
new_files.append(fn)
img = self._img_writer.tag_image(image, coords)
self._tag_video_writer.write(img)
else:
self._tag_video_writer.write(image)
return new_files
Save the result in image format into a folder.
def write_folder(self, video_buffer, out_name, pattern=''):
"""Save the result in image format into a folder.
:param video_buffer: (sppasImage) The image to write
:param out_name: (str) The folder name of the output image files
:param pattern: (str) Pattern to add to a cropped image filename
"""
new_files = list()
if os.path.exists(out_name) is False:
os.mkdir(out_name)
begin_idx, end_idx = video_buffer.get_buffer_range()
folder = os.path.join(out_name, '{:06d}'.format(begin_idx))
if os.path.exists(folder) is True:
shutil.rmtree(folder)
os.mkdir(folder)
iter_images = video_buffer.__iter__()
for i in range(video_buffer.__len__()):
image = next(iter_images)
img_name = self._image_name(begin_idx + i)
nf = self._tag_and_crop(video_buffer, image, i, img_name, folder, pattern)
new_files.extend(nf)
return new_files
def _tag_and_crop(self, video_buffer, image, idx, img_name, folder, pattern):
new_files = list()
coords = video_buffer.get_coordinates(idx)
if self._img_writer.options.tag is True:
img = self._img_writer.tag_image(image, coords)
out_iname = os.path.join(folder, img_name + self._image_ext)
img.write(out_iname)
new_files.append(out_iname)
if self._img_writer.options.crop is True:
for j, c in enumerate(coords):
iname = img_name + '_' + str(j) + pattern + self._image_ext
out_iname = os.path.join(folder, iname)
img = self._img_writer.crop_and_size_image(image, coords[j])
img.write(out_iname)
new_files.append(out_iname)
return new_files