https://sppas.org/

SPPAS 4.16

Module sppas.src.videodata

Class sppasCoordsVideoWriter

Description

Write a video, a set of images and/or coordinates into files.

There are 4 main solutions to write the result (images+coords):

  1. CSV: coordinates into a spreadsheet
  1. XRA: coordinates into an XML annotation file
  1. video
  1. folder with images

For 3 and 4, there are 2 options - at least one has to be selected:

  1. tag: draw a square for each coord in the original image
  1. crop: create one image for each coord

Constructor

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.

Parameters
  • image_writer
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'

Public functions

get_csv_output

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()
get_xra_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()
get_tag_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()
get_crop_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()
get_output_width

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()
get_output_height

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_options

Set any/some/all of the options.

Parameters
  • csv
  • xra
  • tag
  • crop
  • width
  • height
  • folder
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)
get_xra_tiername
def get_xra_tiername(self):
    return self._xra_tiername
get_image_size

Return the size of the image depending on the image and options.

Parameters
  • image
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

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)
write

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.

Parameters

  • video_buffer: (sppasCoordsVideoBuffer) The images and results to write
  • out_name: (str) The output name for the folder and/or the video
  • opt_pattern: (str) Optional pattern to add to filename(s)

Returns

  • list of newly created file names
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_csv_coords

Write or append a list of coordinates in a CSV file.

Parameters

  • video_buffer: (sppasFacesVideoBuffer) The images and results to write
  • outcsvname: (str) The filename of the CSV file to write
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_coords

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

Parameters

  • fd: (Stream) File descriptor, String descriptor, stdout, etc
  • video_buffer: (sppasCoordsVideoBuffer)
  • buffer_idx: (int) Buffer number
  • idx: (int) An integer to write
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')
write_xra_coords

Append a list of coordinates in an XRA object.

Parameters

  • video_buffer: (sppasFacesVideoBuffer) The images and results to write
  • outxraname: (str) The filename of the XRA file to write
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. Cant 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)
write_xra_ann
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
write_video

Save the result in video format.

Parameters

  • video_buffer: (sppasImage) The image to write
  • out_name: (str) The filename of the output video file
  • pattern: (str) Pattern to add to cropped video filename(s)

Returns

  • list of newly created video file names
def write_video(self, video_buffer, out_name, pattern):
    """Save the result in video format.

        :param video_buffer: (sppasImage) The image 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 self._tag_video_writer is None and image is not 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)
    return new_files
write_folder

Save the result in image format into a folder.

Parameters

  • video_buffer: (sppasImage) The image to write
  • out_name: (str) The folder name of the output image files
  • pattern: (str) Pattern to add to a cropped image filename
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

Private functions

_tag_and_crop
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