Add video comparison element

New video/image comparison element, find images in the stream and post
metadata of comparisons of the video frames to the application.

6 files changed, 965 insertions, 2 deletions

diff --git a/video/videofx/Cargo.toml b/video/videofx/Cargo.toml
index 334364933..4ddba4894 100644
--- a/video/videofx/Cargo.toml
+++ b/video/videofx/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "gst-plugin-videofx"
 version = "0.9.0-alpha.1"
-authors = ["Sanchayan Maity <sanchayan@asymptotic.io>"]
+authors = ["Sanchayan Maity <sanchayan@asymptotic.io>", "Rafael Caricio <rafael@caricio.com>"]
 repository = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs"
 license = "MPL-2.0"
 description = "Video Effects Plugin"
@@ -14,6 +14,10 @@ atomic_refcell = "0.1"
 once_cell = "1.0"
 color-thief = "0.2.2"
 color-name = "1.0.0"
+image = "0.24.2"
+image_hasher = "1.0.0"
+dssim-core = { version = "3.2.3", optional = true }
+rgb = { version = "0.8", optional = true }
 
 [dependencies.gst]
 git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs"
@@ -46,6 +50,7 @@ gst-plugin-version-helper = { path="../../version-helper" }
 static = []
 capi = []
 doc = ["gst/v1_18"]
+dssim = ["dssim-core", "rgb"]
 
 [package.metadata.capi]
 min_version = "0.8.0"
diff --git a/video/videofx/src/lib.rs b/video/videofx/src/lib.rs
index 66fd55846..232c0b4f8 100644
--- a/video/videofx/src/lib.rs
+++ b/video/videofx/src/lib.rs
@@ -13,12 +13,22 @@
  *
  * Since: plugins-rs-0.8.0
  */
+#[cfg(feature = "doc")]
+use gst::prelude::*;
+
 mod border;
 mod colordetect;
+mod videocompare;
+
+pub use videocompare::{HashAlgorithm, PadDistance, VideoCompareMessage};
 
 fn plugin_init(plugin: &gst::Plugin) -> Result<(), gst::glib::BoolError> {
+    #[cfg(feature = "doc")]
+    HashAlgorithm::static_type().mark_as_plugin_api(gst::PluginAPIFlags::empty());
+
     border::register(plugin)?;
-    colordetect::register(plugin)
+    colordetect::register(plugin)?;
+    videocompare::register(plugin)
 }
 
 gst::plugin_define!(
diff --git a/video/videofx/src/videocompare/hashed_image.rs b/video/videofx/src/videocompare/hashed_image.rs
new file mode 100644
index 000000000..c242c80c9
--- /dev/null
+++ b/video/videofx/src/videocompare/hashed_image.rs
@@ -0,0 +1,129 @@
+// Copyright (C) 2022 Rafael Caricio <rafael@caricio.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+
+use crate::HashAlgorithm;
+#[cfg(feature = "dssim")]
+use dssim_core::{Dssim, DssimImage};
+use gst_video::VideoFormat;
+use image_hasher::{HashAlg, Hasher, HasherConfig, ImageHash};
+#[cfg(feature = "dssim")]
+use rgb::FromSlice;
+
+pub enum HasherEngine {
+    ImageHasher(Hasher),
+    #[cfg(feature = "dssim")]
+    DssimHasher(Dssim),
+}
+
+impl HasherEngine {
+    pub fn hash_image(
+        &self,
+        frame: &gst_video::VideoFrameRef<&gst::BufferRef>,
+    ) -> Result<HashedImage, gst::FlowError> {
+        use HasherEngine::*;
+
+        let height = frame.height();
+        let width = frame.width();
+        let buf = tightly_packed_framebuffer(frame);
+
+        let res = match self {
+            ImageHasher(hasher) => match frame.format() {
+                VideoFormat::Rgb => {
+                    let frame_buf = image::RgbImage::from_raw(width, height, buf)
+                        .ok_or(gst::FlowError::Error)?;
+                    HashedImage::ImageHash(hasher.hash_image(&frame_buf))
+                }
+                VideoFormat::Rgba => {
+                    let frame_buf = image::RgbaImage::from_raw(width, height, buf)
+                        .ok_or(gst::FlowError::Error)?;
+                    HashedImage::ImageHash(hasher.hash_image(&frame_buf))
+                }
+                _ => unreachable!(),
+            },
+            #[cfg(feature = "dssim")]
+            DssimHasher(hasher) => {
+                let hashed_img = match frame.format() {
+                    VideoFormat::Rgb => hasher
+                        .create_image_rgb(buf.as_rgb(), width as usize, height as usize)
+                        .unwrap(),
+                    VideoFormat::Rgba => hasher
+                        .create_image_rgba(buf.as_rgba(), width as usize, height as usize)
+                        .unwrap(),
+                    _ => unreachable!(),
+                };
+                HashedImage::Dssim(hashed_img)
+            }
+        };
+
+        Ok(res)
+    }
+
+    pub fn compare(&self, img1: &HashedImage, img2: &HashedImage) -> f64 {
+        use HashedImage::*;
+
+        match (self, img1, img2) {
+            (_, ImageHash(left), ImageHash(right)) => left.dist(right) as f64,
+            #[cfg(feature = "dssim")]
+            (Self::DssimHasher(algo), Dssim(left), Dssim(right)) => {
+                let (val, _) = algo.compare(left, right);
+                val.into()
+            }
+            _ => unreachable!(),
+        }
+    }
+}
+
+pub enum HashedImage {
+    ImageHash(ImageHash),
+    #[cfg(feature = "dssim")]
+    Dssim(DssimImage<f32>),
+}
+
+impl From<super::HashAlgorithm> for HasherEngine {
+    fn from(algo: HashAlgorithm) -> Self {
+        use super::HashAlgorithm::*;
+
+        let algo = match algo {
+            Mean => HashAlg::Mean,
+            Gradient => HashAlg::Gradient,
+            VertGradient => HashAlg::VertGradient,
+            DoubleGradient => HashAlg::DoubleGradient,
+            Blockhash => HashAlg::Blockhash,
+            #[cfg(feature = "dssim")]
+            Dssim => {
+                return HasherEngine::DssimHasher(dssim_core::Dssim::new());
+            }
+        };
+
+        HasherEngine::ImageHasher(HasherConfig::new().hash_alg(algo).to_hasher())
+    }
+}
+
+/// Helper method that takes a gstreamer video-frame and copies it into a
+/// tightly packed rgb(a) buffer, ready for consumption.
+fn tightly_packed_framebuffer(frame: &gst_video::VideoFrameRef<&gst::BufferRef>) -> Vec<u8> {
+    assert_eq!(frame.n_planes(), 1); // RGB and RGBA are tightly packed
+    let line_size = (frame.width() * frame.n_components()) as usize;
+    let line_stride = frame.plane_stride()[0] as usize;
+
+    if line_size == line_stride {
+        return frame.plane_data(0).unwrap().to_vec();
+    }
+
+    let mut raw_frame = Vec::with_capacity(line_size * frame.info().height() as usize);
+
+    // copy gstreamer frame to tightly packed rgb(a) frame.
+    frame
+        .plane_data(0)
+        .unwrap()
+        .chunks_exact(line_stride)
+        .map(|padded_line| &padded_line[..line_size])
+        .for_each(|line| raw_frame.extend_from_slice(line));
+
+    raw_frame
+}
diff --git a/video/videofx/src/videocompare/imp.rs b/video/videofx/src/videocompare/imp.rs
new file mode 100644
index 000000000..f69b5cb23
--- /dev/null
+++ b/video/videofx/src/videocompare/imp.rs
@@ -0,0 +1,390 @@
+// Copyright (C) 2022 Rafael Caricio <rafael@caricio.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+
+use crate::videocompare::hashed_image::HasherEngine;
+use crate::videocompare::HashAlgorithm;
+use crate::{PadDistance, VideoCompareMessage};
+use gst::subclass::prelude::*;
+use gst::{glib, glib::prelude::*, prelude::*};
+use gst_base::prelude::*;
+use gst_base::AggregatorPad;
+use gst_video::prelude::VideoAggregatorPadExtManual;
+use gst_video::subclass::prelude::*;
+use gst_video::subclass::AggregateFramesToken;
+use gst_video::VideoFormat;
+use once_cell::sync::Lazy;
+use std::sync::{Arc, Mutex};
+
+static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
+    gst::DebugCategory::new(
+        "videocompare",
+        gst::DebugColorFlags::empty(),
+        Some("Video frames comparison"),
+    )
+});
+
+const DEFAULT_HASH_ALGO: HashAlgorithm = HashAlgorithm::Blockhash;
+const DEFAULT_MAX_DISTANCE_THRESHOLD: f64 = 0.0;
+
+struct Settings {
+    hash_algo: HashAlgorithm,
+    max_distance_threshold: f64,
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Settings {
+            hash_algo: DEFAULT_HASH_ALGO,
+            max_distance_threshold: DEFAULT_MAX_DISTANCE_THRESHOLD,
+        }
+    }
+}
+
+struct State {
+    hasher: HasherEngine,
+}
+
+impl Default for State {
+    fn default() -> Self {
+        Self {
+            hasher: DEFAULT_HASH_ALGO.into(),
+        }
+    }
+}
+
+#[derive(Default)]
+pub struct VideoCompare {
+    settings: Arc<Mutex<Settings>>,
+    state: Arc<Mutex<State>>,
+    reference_pad: Mutex<Option<gst::Pad>>,
+}
+
+#[glib::object_subclass]
+impl ObjectSubclass for VideoCompare {
+    const NAME: &'static str = "GstVideoCompare";
+    type Type = super::VideoCompare;
+    type ParentType = gst_video::VideoAggregator;
+}
+
+impl ObjectImpl for VideoCompare {
+    fn properties() -> &'static [glib::ParamSpec] {
+        static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
+            vec![
+                glib::ParamSpecEnum::builder::<HashAlgorithm>("hash-algo", DEFAULT_HASH_ALGO)
+                    .nick("Hashing Algorithm")
+                    .blurb("Which hashing algorithm to use for image comparisons")
+                    .mutable_ready()
+                    .build(),
+                glib::ParamSpecDouble::builder("max-dist-threshold")
+                    .nick("Maximum Distance Threshold")
+                    .blurb("Maximum distance threshold to emit messages when an image is detected, by default emits only on exact match")
+                    .minimum(0f64)
+                    .default_value(DEFAULT_MAX_DISTANCE_THRESHOLD)
+                    .mutable_ready()
+                    .build(),
+            ]
+        });
+
+        PROPERTIES.as_ref()
+    }
+
+    fn set_property(&self, _id: usize, value: &glib::Value, pspec: &glib::ParamSpec) {
+        let mut settings = self.settings.lock().unwrap();
+        match pspec.name() {
+            "hash-algo" => {
+                let hash_algo = value.get().expect("type checked upstream");
+                if settings.hash_algo != hash_algo {
+                    gst::info!(
+                        CAT,
+                        imp: self,
+                        "Changing hash-algo from {:?} to {:?}",
+                        settings.hash_algo,
+                        hash_algo
+                    );
+                    settings.hash_algo = hash_algo;
+
+                    let mut state = self.state.lock().unwrap();
+                    state.hasher = hash_algo.into();
+                }
+            }
+            "max-dist-threshold" => {
+                let max_distance_threshold = value.get().expect("type checked upstream");
+                if settings.max_distance_threshold != max_distance_threshold {
+                    gst::info!(
+                        CAT,
+                        imp: self,
+                        "Changing max-dist-threshold from {} to {}",
+                        settings.max_distance_threshold,
+                        max_distance_threshold
+                    );
+                    settings.max_distance_threshold = max_distance_threshold;
+                }
+            }
+            _ => unimplemented!(),
+        }
+    }
+
+    fn property(&self, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
+        let settings = self.settings.lock().unwrap();
+        match pspec.name() {
+            "hash-algo" => settings.hash_algo.to_value(),
+            "max-dist-threshold" => settings.max_distance_threshold.to_value(),
+            _ => unimplemented!(),
+        }
+    }
+}
+
+impl GstObjectImpl for VideoCompare {}
+
+impl ElementImpl for VideoCompare {
+    fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
+        static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
+            gst::subclass::ElementMetadata::new(
+                "Image comparison",
+                "Filter/Video",
+                "Compare similarity of video frames",
+                "Rafael Caricio <rafael@caricio.com>",
+            )
+        });
+
+        Some(&*ELEMENT_METADATA)
+    }
+
+    fn pad_templates() -> &'static [gst::PadTemplate] {
+        static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
+            let caps = gst_video::VideoCapsBuilder::new()
+                .format_list([VideoFormat::Rgb, VideoFormat::Rgba])
+                .build();
+
+            let sink_pad_template = gst::PadTemplate::with_gtype(
+                "sink_%u",
+                gst::PadDirection::Sink,
+                gst::PadPresence::Request,
+                &caps,
+                gst_video::VideoAggregatorPad::static_type(),
+            )
+            .unwrap();
+
+            let src_pad_template = gst::PadTemplate::with_gtype(
+                "src",
+                gst::PadDirection::Src,
+                gst::PadPresence::Always,
+                &caps,
+                gst_video::VideoAggregatorPad::static_type(),
+            )
+            .unwrap();
+
+            vec![sink_pad_template, src_pad_template]
+        });
+
+        PAD_TEMPLATES.as_ref()
+    }
+
+    fn release_pad(&self, pad: &gst::Pad) {
+        let mut reference_pad = self.reference_pad.lock().unwrap();
+        if let Some(current_reference_pad) = reference_pad.to_owned() {
+            if pad != &current_reference_pad {
+                // We don't worry if any other pads get released
+                return;
+            }
+
+            // Since we are releasing the reference pad, we need to select a new pad for the
+            // comparisons. At the moment we have no defined criteria to select the next
+            // reference sink pad, so we choose the first that comes.
+            for sink_pad in self.instance().sink_pads() {
+                if current_reference_pad != sink_pad {
+                    // Choose the first available left sink pad
+                    *reference_pad = Some(sink_pad);
+                }
+            }
+        }
+    }
+}
+
+impl AggregatorImpl for VideoCompare {
+    fn create_new_pad(
+        &self,
+        templ: &gst::PadTemplate,
+        req_name: Option<&str>,
+        caps: Option<&gst::Caps>,
+    ) -> Option<AggregatorPad> {
+        let pad = self.parent_create_new_pad(templ, req_name, caps);
+        if let Some(pad) = &pad {
+            let mut reference_pad = self.reference_pad.lock().unwrap();
+            // We store the first pad added to the element for later use, this way we guarantee that
+            // the first pad is used as the reference for comparisons
+            if reference_pad.is_none() && pad.direction() == gst::PadDirection::Sink {
+                let pad = pad.clone().upcast::<gst::Pad>();
+                gst::info!(
+                    CAT,
+                    imp: self,
+                    "Reference sink pad selected: {}",
+                    pad.name()
+                );
+                *reference_pad = Some(pad);
+            }
+        }
+        pad
+    }
+
+    fn update_src_caps(&self, caps: &gst::Caps) -> Result<gst::Caps, gst::FlowError> {
+        let reference_pad = self.reference_pad.lock().unwrap();
+        let sink_caps = reference_pad
+            .as_ref()
+            .and_then(|pad| pad.current_caps())
+            .unwrap_or_else(|| caps.to_owned()); // Allow any caps for now
+
+        if !sink_caps.can_intersect(caps) {
+            gst::error!(
+                CAT,
+                imp: self,
+                "Proposed src caps ({:?}) not supported, needs to intersect with the reference sink caps ({:?})",
+                caps,
+                sink_caps
+            );
+            return Err(gst::FlowError::NotNegotiated);
+        }
+
+        gst::info!(CAT, imp: self, "Caps for src pad: {:?}", sink_caps);
+        Ok(sink_caps)
+    }
+}
+
+impl VideoAggregatorImpl for VideoCompare {
+    fn aggregate_frames(
+        &self,
+        token: &AggregateFramesToken,
+        outbuf: &mut gst::BufferRef,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let state = self.state.lock().unwrap();
+
+        let reference_pad = {
+            let reference_pad = self.reference_pad.lock().unwrap();
+            reference_pad
+                .as_ref()
+                .map(|pad| {
+                    pad.clone()
+                        .downcast::<gst_video::VideoAggregatorPad>()
+                        .unwrap()
+                })
+                .ok_or_else(|| {
+                    gst::warning!(CAT, imp: self, "No reference sink pad exists");
+                    gst::FlowError::Eos
+                })?
+        };
+
+        let reference_frame = match reference_pad.prepared_frame(token) {
+            Some(f) => f,
+            None => {
+                return if reference_pad.is_eos() {
+                    Err(gst::FlowError::Eos)
+                } else {
+                    gst::warning!(
+                        CAT,
+                        imp: self,
+                        "The reference sink pad '{}' has not produced a buffer, image comparison not possible",
+                        reference_pad.name()
+                    );
+                    Ok(gst::FlowSuccess::Ok)
+                }
+            }
+        };
+
+        let mut message = VideoCompareMessage::default();
+        let buffer = reference_frame.buffer();
+
+        // Add running time to message when possible
+        message.running_time = reference_pad
+            .segment()
+            .downcast::<gst::ClockTime>()
+            .ok()
+            .and_then(|segment| buffer.pts().map(|pts| segment.to_running_time(pts)))
+            .flatten();
+
+        // output the reference buffer
+        outbuf.remove_all_memory();
+        buffer
+            .copy_into(outbuf, gst::BufferCopyFlags::all(), 0, None)
+            .map_err(|_| gst::FlowError::Error)?;
+
+        // Use current frame as the reference to the comparison
+        let reference_hash = state.hasher.hash_image(&reference_frame)?;
+
+        // Loop through all remaining sink pads and compare the latest available buffer
+        for pad in self
+            .instance()
+            .sink_pads()
+            .into_iter()
+            .map(|pad| pad.downcast::<gst_video::VideoAggregatorPad>().unwrap())
+            .collect::<Vec<_>>()
+        {
+            // Do not compare the reference pad with itself
+            if pad == reference_pad {
+                continue;
+            }
+
+            let frame = match pad.prepared_frame(token) {
+                Some(f) => f,
+                None => return Ok(gst::FlowSuccess::Ok),
+            };
+
+            // Make sure the sizes are the same
+            if reference_frame.width() != frame.width()
+                || reference_frame.height() != frame.height()
+            {
+                gst::error!(
+                    CAT,
+                    imp: self,
+                    "Video streams do not have the same sizes (add videoscale and force the sizes to be equal on all sink pads)",
+                );
+                return Err(gst::FlowError::NotNegotiated);
+            }
+
+            // compare current frame with the reference and add to the results to the structure
+            let frame_hash = state.hasher.hash_image(&frame)?;
+            let distance = state.hasher.compare(&reference_hash, &frame_hash);
+            message
+                .pad_distances
+                .push(PadDistance::new(pad.upcast(), distance));
+        }
+
+        let max_distance_threshold = {
+            let settings = self.settings.lock().unwrap();
+            settings.max_distance_threshold as f64
+        };
+
+        if message
+            .pad_distances
+            .iter()
+            .any(|p| p.distance <= max_distance_threshold)
+        {
+            gst::debug!(
+                CAT,
+                imp: self,
+                "Image detected {}",
+                message.running_time.unwrap().display()
+            );
+            let element = self.instance();
+            let _ = element.post_message(
+                gst::message::Element::builder(message.into())
+                    .src(element.as_ref())
+                    .build(),
+            );
+        } else {
+            gst::debug!(
+                CAT,
+                imp: self,
+                "Compared images and could not find any frame with distance lower than the threshold of {}: {:?}",
+                max_distance_threshold,
+                message
+            );
+        }
+
+        Ok(gst::FlowSuccess::Ok)
+    }
+}
diff --git a/video/videofx/src/videocompare/mod.rs b/video/videofx/src/videocompare/mod.rs
new file mode 100644
index 000000000..51c3df4d4
--- /dev/null
+++ b/video/videofx/src/videocompare/mod.rs
@@ -0,0 +1,255 @@
+// Copyright (C) 2022 Rafael Caricio <rafael@caricio.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+/**
+ * element-videocompare:
+ * @short_description: Compares multiple video feeds and post message to the application when frames are similar.
+ *
+ * Compare multiple video pad buffers, with the first added pad being the reference. All video
+ * streams must use the same geometry. A message is posted to the application when the measurement
+ * of distance between frames falls under the desired threshold.
+ *
+ * The application can decide what to do next with the comparison result. This element can be used
+ * to detect badges or slates in the video stream.
+ *
+ * The src pad passthrough buffers from the reference sink pad.
+ *
+ * ## Example pipeline
+ * ```bash
+ * gst-launch-1.0 videotestsrc pattern=red ! videocompare name=compare ! videoconvert \
+ *   ! autovideosink  videotestsrc pattern=red ! imagefreeze ! compare. -m
+ * ```
+ *
+ * The message posted to the application contains a structure that looks like:
+ *
+ * ```ignore
+ * videocompare,
+ *   pad-distances=(structure)< "pad-distance\,\ pad\=\(GstPad\)\"\\\(GstVideoAggregatorPad\\\)\\\ sink_1\"\,\ distance\=\(double\)0\;" >,
+ *   running-time=(guint64)0;
+ * ```
+ *
+ * Since: plugins-rs-0.9.0
+ */
+use gst::glib;
+use gst::prelude::*;
+use std::fmt::Debug;
+
+mod hashed_image;
+mod imp;
+
+glib::wrapper! {
+    pub struct VideoCompare(ObjectSubclass<imp::VideoCompare>) @extends gst_video::VideoAggregator, gst_base::Aggregator, gst::Element, gst::Object;
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "videocompare",
+        gst::Rank::None,
+        VideoCompare::static_type(),
+    )
+}
+
+#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash, Clone, Copy, glib::Enum)]
+#[repr(u32)]
+#[enum_type(name = "GstVideoCompareHashAlgorithm")]
+#[non_exhaustive]
+pub enum HashAlgorithm {
+    #[enum_value(name = "Mean: The Mean hashing algorithm.", nick = "mean")]
+    Mean = 0,
+
+    #[enum_value(name = "Gradient: The Gradient hashing algorithm.", nick = "gradient")]
+    Gradient = 1,
+
+    #[enum_value(
+        name = "VertGradient: The Vertical-Gradient hashing algorithm.",
+        nick = "vertgradient"
+    )]
+    VertGradient = 2,
+
+    #[enum_value(
+        name = "DoubleGradient: The Double-Gradient hashing algorithm.",
+        nick = "doublegradient"
+    )]
+    DoubleGradient = 3,
+
+    #[enum_value(
+        name = "Blockhash: The [Blockhash](https://github.com/commonsmachinery/blockhash-rfc) algorithm.",
+        nick = "blockhash"
+    )]
+    Blockhash = 4,
+
+    #[cfg(feature = "dssim")]
+    #[enum_value(
+        name = "Dssim: Image similarity comparison simulating human perception.",
+        nick = "dssim"
+    )]
+    Dssim = 5,
+}
+
+#[derive(Default, Debug, Clone, PartialEq)]
+pub struct VideoCompareMessage {
+    pad_distances: Vec<PadDistance>,
+    running_time: Option<gst::ClockTime>,
+}
+
+impl VideoCompareMessage {
+    pub fn pad_distances(&self) -> &[PadDistance] {
+        self.pad_distances.as_slice()
+    }
+
+    pub fn running_time(&self) -> Option<gst::ClockTime> {
+        self.running_time
+    }
+}
+
+impl From<VideoCompareMessage> for gst::Structure {
+    fn from(msg: VideoCompareMessage) -> Self {
+        gst::Structure::builder("videocompare")
+            .field(
+                "pad-distances",
+                gst::Array::new(msg.pad_distances.into_iter().map(|v| {
+                    let s: gst::Structure = v.into();
+                    s.to_send_value()
+                })),
+            )
+            .field("running-time", msg.running_time)
+            .build()
+    }
+}
+
+impl TryFrom<gst::Structure> for VideoCompareMessage {
+    type Error = Box<dyn std::error::Error>;
+
+    fn try_from(structure: gst::Structure) -> Result<Self, Self::Error> {
+        let mut pad_distances = Vec::<PadDistance>::new();
+        for value in structure.get::<gst::Array>("pad-distances")?.iter() {
+            let s = value.get::<gst::Structure>()?;
+            pad_distances.push(s.try_into()?)
+        }
+
+        Ok(VideoCompareMessage {
+            running_time: structure.get("running-time")?,
+            pad_distances,
+        })
+    }
+}
+
+#[derive(Clone, PartialEq, Debug)]
+pub struct PadDistance {
+    pad: gst::Pad,
+    distance: f64,
+}
+
+impl PadDistance {
+    pub fn new(pad: gst::Pad, distance: f64) -> Self {
+        Self { pad, distance }
+    }
+
+    pub fn pad(&self) -> &gst::Pad {
+        &self.pad
+    }
+
+    pub fn distance(&self) -> f64 {
+        self.distance
+    }
+}
+
+impl From<PadDistance> for gst::Structure {
+    fn from(pad_distance: PadDistance) -> Self {
+        gst::Structure::builder("pad-distance")
+            .field("pad", pad_distance.pad)
+            .field("distance", pad_distance.distance)
+            .build()
+    }
+}
+
+impl TryFrom<gst::Structure> for PadDistance {
+    type Error = Box<dyn std::error::Error>;
+
+    fn try_from(structure: gst::Structure) -> Result<Self, Self::Error> {
+        Ok(PadDistance {
+            pad: structure.get("pad")?,
+            distance: structure.get("distance")?,
+        })
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    fn init() {
+        use std::sync::Once;
+        static INIT: Once = Once::new();
+
+        INIT.call_once(|| {
+            gst::init().unwrap();
+        });
+    }
+
+    #[test]
+    fn reverse_order_serialization_of_structure() {
+        init();
+
+        let running_time = gst::ClockTime::from_seconds(2);
+
+        let mut messsage = VideoCompareMessage::default();
+        messsage.pad_distances.push(PadDistance {
+            pad: gst::Pad::new(Some("sink_0"), gst::PadDirection::Sink),
+            distance: 42_f64,
+        });
+        messsage.running_time = Some(running_time);
+
+        let structure: gst::Structure = messsage.into();
+
+        let pad_distances = structure.get::<gst::Array>("pad-distances").unwrap();
+        let first = pad_distances
+            .first()
+            .unwrap()
+            .get::<gst::Structure>()
+            .unwrap();
+        assert_eq!(first.get::<gst::Pad>("pad").unwrap().name(), "sink_0");
+        assert_eq!(first.get::<f64>("distance").unwrap(), 42_f64);
+
+        assert_eq!(
+            structure
+                .get_optional::<gst::ClockTime>("running-time")
+                .unwrap(),
+            Some(running_time)
+        );
+    }
+
+    #[test]
+    fn from_structure_to_message_struct() {
+        init();
+
+        let running_time = gst::ClockTime::from_seconds(2);
+
+        let structure = gst::Structure::builder("videocompare")
+            .field(
+                "pad-distances",
+                gst::Array::from_iter([gst::Structure::builder("pad-distance")
+                    .field(
+                        "pad",
+                        gst::Pad::new(Some("sink_0"), gst::PadDirection::Sink),
+                    )
+                    .field("distance", 42f64)
+                    .build()
+                    .to_send_value()]),
+            )
+            .field("running-time", running_time)
+            .build();
+
+        let message: VideoCompareMessage = structure.try_into().unwrap();
+        assert_eq!(message.running_time, Some(running_time));
+
+        let pad_distance = message.pad_distances.get(0).unwrap();
+        assert_eq!(pad_distance.pad.name().as_str(), "sink_0");
+        assert_eq!(pad_distance.distance, 42f64);
+    }
+}
diff --git a/video/videofx/tests/videocompare.rs b/video/videofx/tests/videocompare.rs
new file mode 100644
index 000000000..251697c1b
--- /dev/null
+++ b/video/videofx/tests/videocompare.rs
@@ -0,0 +1,174 @@
+// Copyright (C) 2022 Rafael Caricio <rafael@caricio.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+
+use gst::prelude::*;
+use gstvideofx::{HashAlgorithm, VideoCompareMessage};
+
+fn init() {
+    use std::sync::Once;
+    static INIT: Once = Once::new();
+
+    INIT.call_once(|| {
+        gst::init().unwrap();
+        gstvideofx::plugin_register_static().expect("Failed to register videofx plugin");
+    });
+}
+
+fn setup_pipeline(
+    pipeline: &gst::Pipeline,
+    pattern_a: &str,
+    pattern_b: &str,
+    max_distance_threshold: f64,
+    hash_algo: HashAlgorithm,
+) {
+    let videocompare = gst::ElementFactory::make("videocompare", None).unwrap();
+    videocompare.set_property("max-dist-threshold", max_distance_threshold);
+    videocompare.set_property("hash-algo", hash_algo);
+
+    let reference_src = gst::ElementFactory::make("videotestsrc", Some("reference_src")).unwrap();
+    reference_src.set_property_from_str("pattern", pattern_a);
+    reference_src.set_property("num-buffers", 1i32);
+
+    let secondary_src = gst::ElementFactory::make("videotestsrc", Some("secondary_src")).unwrap();
+    reference_src.set_property_from_str("pattern", pattern_b);
+
+    let sink = gst::ElementFactory::make("fakesink", None).unwrap();
+
+    pipeline
+        .add_many(&[&reference_src, &secondary_src, &videocompare, &sink])
+        .unwrap();
+    gst::Element::link_many(&[&reference_src, &videocompare, &sink]).expect("Link primary path");
+    gst::Element::link_many(&[&secondary_src, &videocompare]).expect("Link secondary path");
+}
+
+#[test]
+fn test_can_find_similar_frames() {
+    init();
+
+    // TODO: for some reason only in the tests, the distance is higher
+    //       than when running via gst-launch tool for the same pipeline. What is happening?
+    let max_distance = 32_f64;
+
+    let pipeline = gst::Pipeline::new(None);
+    setup_pipeline(
+        &pipeline,
+        "red",
+        "red",
+        max_distance,
+        HashAlgorithm::Blockhash,
+    );
+
+    pipeline.set_state(gst::State::Playing).unwrap();
+
+    let mut detection = None;
+    let bus = pipeline.bus().unwrap();
+    for msg in bus.iter_timed(gst::ClockTime::NONE) {
+        use gst::MessageView;
+        match msg.view() {
+            MessageView::Element(elt) => {
+                if let Some(s) = elt.structure() {
+                    if s.name() == "videocompare" {
+                        detection = Some(
+                            VideoCompareMessage::try_from(s.to_owned())
+                                .expect("Can convert message to struct"),
+                        );
+                    }
+                }
+            }
+            MessageView::Eos(..) => break,
+            _ => (),
+        }
+    }
+
+    pipeline.set_state(gst::State::Null).unwrap();
+
+    let detection = detection.expect("Has found similar images");
+    let pad_distance = detection
+        .pad_distances()
+        .iter()
+        .find(|pd| pd.pad().name() == "sink_1")
+        .unwrap();
+    assert!(pad_distance.distance() <= max_distance);
+}
+
+#[test]
+fn test_do_not_send_message_when_image_not_found() {
+    init();
+
+    let pipeline = gst::Pipeline::new(None);
+    setup_pipeline(&pipeline, "black", "red", 0f64, HashAlgorithm::Blockhash);
+
+    pipeline.set_state(gst::State::Playing).unwrap();
+
+    let mut detection = None;
+    let bus = pipeline.bus().unwrap();
+    for msg in bus.iter_timed(gst::ClockTime::NONE) {
+        use gst::MessageView;
+        match msg.view() {
+            MessageView::Element(elt) => {
+                if let Some(s) = elt.structure() {
+                    if s.name() == "videocompare" {
+                        detection = Some(
+                            VideoCompareMessage::try_from(s.to_owned())
+                                .expect("Can convert message to struct"),
+                        );
+                    }
+                }
+            }
+            MessageView::Eos(..) => break,
+            _ => (),
+        }
+    }
+
+    pipeline.set_state(gst::State::Null).unwrap();
+
+    assert!(detection.is_none());
+}
+
+#[cfg(feature = "dssim")]
+#[test]
+fn test_use_dssim_to_find_similar_frames() {
+    init();
+
+    let max_distance = 1_f64;
+
+    let pipeline = gst::Pipeline::new(None);
+    setup_pipeline(&pipeline, "red", "red", max_distance, HashAlgorithm::Dssim);
+
+    pipeline.set_state(gst::State::Playing).unwrap();
+
+    let mut detection = None;
+    let bus = pipeline.bus().unwrap();
+    for msg in bus.iter_timed(gst::ClockTime::NONE) {
+        use gst::MessageView;
+        match msg.view() {
+            MessageView::Element(elt) => {
+                if let Some(s) = elt.structure() {
+                    if s.name() == "videocompare" {
+                        detection = Some(
+                            VideoCompareMessage::try_from(s.to_owned())
+                                .expect("Can convert message to struct"),
+                        );
+                    }
+                }
+            }
+            MessageView::Eos(..) => break,
+            _ => (),
+        }
+    }
+
+    pipeline.set_state(gst::State::Null).unwrap();
+
+    let detection = detection.expect("Has found similar images");
+    let pad_distance = detection
+        .pad_distances()
+        .iter()
+        .find(|pd| pd.pad().name() == "sink_1")
+        .unwrap();
+    assert!(pad_distance.distance() <= max_distance);
+}