Display Watermark

Display a watermark on the video stream


The Display Watermark node allows you to display a watermark on the video stream.

Inputs & Outputs

  • Inputs : 1, Media Format : Raw Video
  • Outputs : 1, Media Format: Raw Video
  • Output Metadata : None


image_urlURL for the image to use as a watermark. This image is superimposed on the right bottom corner, and must be smaller than the video resolution. If not specified, Lumeo's logo is added as the watermark. Use a PNG file of approximately 100x100px for best results.stringnullYes


This node does not add any new metadata to the frame.

Metadata PropertyDescription


In order to customize this node for your use case, use the code below inside a Function Node.

from lumeopipeline import VideoFrame  # Auto added by Lumeo
import cv2                    # Auto added by Lumeo
import numpy                  # Auto added by Lumeo
from urllib.request import urlopen

first_run = True
watermark_overlay = None
LUMEO_LOGO_URL = "https://s.gravatar.com/avatar/f8066a0ab6f5ffa7d121aa210d988980?s=80"

def process_frame(frame: VideoFrame, image_url = LUMEO_LOGO_URL, **kwargs) -> bool:

    global watermark_overlay
    global first_run

    with frame.data() as mat:
            if first_run == True:
                # Get pipeline resolution
                width = frame.video_info().width
                height = frame.video_info().height            
                # Get watermark image from URL
                (wH, wW, watermark) = download_watermark_from_url(image_url)

                # If we have one, set the watermark overlay. 
                if watermark is not None:
                    watermark_overlay = numpy.zeros((height,width,4), dtype="uint8")
                    watermark_overlay[height-wH:height, width-wW:width] = watermark
                first_run = False

            if watermark_overlay is not None:
                cv2.addWeighted(watermark_overlay, 1.0, mat, 1.0, 0, mat)

        except Exception as error:
            first_run = False

    return True

def download_watermark_from_url(image_url):
    global LUMEO_LOGO_URL
    if not image_url or len(image_url) == 0:
        image_url = LUMEO_LOGO_URL
    wH = None
    wW = None
    watermark = None

        watermarkimage = numpy.asarray(bytearray(urlopen(image_url).read()), dtype="uint8")
        watermark = cv2.imdecode(watermarkimage, cv2.IMREAD_UNCHANGED)
        if watermark.shape[2] != 4:
            watermark = cv2.cvtColor(watermark,cv2.COLOR_BGR2BGRA)
        (wH, wW) = watermark.shape[:2]
    except Exception as e:
        print("Error extracting watermark from url ({}) : {}".format(image_url,str(e)))
        wH = None
        wW = None
        watermark = None
    return (wH, wW, watermark)