<commit_before># coding: utf-8
"""Main stb-tester python module. Intended to be used with `stbt run`.

See `man stbt` and http://stb-tester.com for documentation.

Copyright 2012-2013 YouView TV Ltd and contributors.
License: LGPL v2.1 or (at your option) any later version (see
https://github.com/stb-tester/stb-tester/blob/master/LICENSE for details).
"""

from __future__ import absolute_import

import argparse
import datetime
import functools
import glob
import inspect
import os
import Queue
import re
import subprocess
import threading
import time
import traceback
import warnings
from collections import deque, namedtuple
from contextlib import contextmanager
from distutils.version import LooseVersion

import cv2
import gi
import numpy
from enum import IntEnum
from kitchen.text.converters import to_bytes

from _stbt import logging, utils
from _stbt.config import ConfigurationError, get_config
from _stbt.gst_hacks import gst_iterate, map_gst_sample
from _stbt.logging import ddebug, debug, warn

gi.require_version("Gst", "1.0")
from gi.repository import GLib, GObject, Gst  # isort:skip pylint: disable=E0611

if getattr(gi, "version_info", (0, 0, 0)) < (3, 12, 0):
    GObject.threads_init()
Gst.init(None)

warnings.filterwarnings(
    action="always", category=DeprecationWarning, message='.*stb-tester')


# Functions available to stbt scripts
# ===========================================================================


class MatchParameters(object):
    """Parameters to customise the image processing algorithm used by
    `match`, `wait_for_match`, and `press_until_match`.

    You can change the default values for these parameters by setting a key
    (with the same name as the corresponding python parameter) in the `[match]`
    section of stbt.conf. But we strongly recommend that you don't change the
    default values from what is documented here.

    You should only need to change these parameters when you're trying to match
    a template image that isn't actually a perfect match -- for example if
    there's a translucent background with live TV visible behind it; or if you
    have a template image of a button's background and you want it to match even
    if the text on the button doesn't match.

    :param str match_method:
      The method to be used by the first pass of stb-tester's image matching
      algorithm, to find the most likely location of the "template" image
      within the larger source image.

      Allowed values are "sqdiff-normed", "ccorr-normed", and "ccoeff-normed".
      For the meaning of these parameters, see OpenCV's `cvMatchTemplate
      <http://docs.opencv.org/modules/imgproc/doc/object_detection.html>`_.

      We recommend that you don't change this from its default value of
      "sqdiff-normed".

    :param float match_threshold:
      How strong a result from the first pass must be, to be considered a
      match. Valid values range from 0 (anything is considered to match)
      to 1 (the match has to be pixel perfect). This defaults to 0.8.

    :param str confirm_method:
      The method to be used by the second pass of stb-tester's image matching
      algorithm, to confirm that the region identified by the first pass is a
      good match.

      The first pass often gives false positives (it reports a "match" for an
      image that shouldn't match). The second pass is more CPU-intensive, but
      it only checks the position of the image that the first pass identified.
      The allowed values are:

      :"none":
        Do not confirm the match. Assume that the potential match found is
        correct.

      :"absdiff":
        Compare the absolute difference of each pixel from the template image
        against its counterpart from the candidate region in the source video
        frame.

      :"normed-absdiff":
        Normalise the pixel values from both the template image and the
        candidate region in the source video frame, then compare the absolute
        difference as with "absdiff".

        This gives better results with low-contrast images. We recommend setting
        this as the default `confirm_method` in stbt.conf, with a
        `confirm_threshold` of 0.30.

    :param float confirm_threshold:
      The maximum allowed difference between any given pixel from the template
      image and its counterpart from the candidate region in the source video
      frame, as a fraction of the pixel's total luminance range.

      Valid values range from 0 (more strict) to 1.0 (less strict).
      Useful values tend to be around 0.16 for the "absdiff" method, and 0.30
      for the "normed-absdiff" method.

    :param int erode_passes:
      After the "absdiff" or "normed-absdiff" absolute difference is taken,
      stb-tester runs an erosion algorithm that removes single-pixel differences
      to account for noise. Useful values are 1 (the default) and 0 (to disable
      this step).

    """

    def __init__(self, match_method=None, match_threshold=None,
                 confirm_method=None, confirm_threshold=None,
                 erode_passes=None):
        if match_method is None:
            match_method = get_config('match', 'match_method')
        if match_threshold is None:
            match_threshold = get_config(
                'match', 'match_threshold', type_=float)
        if confirm_method is None:
            confirm_method = get_config('match', 'confirm_method')
        if confirm_threshold is None:
            confirm_threshold = get_config(
                'match', 'confirm_threshold', type_=float)
        if erode_passes is None:
            erode_passes = get_config('match', 'erode_passes', type_=int)

        if match_method not in (
                "sqdiff-normed", "ccorr-normed", "ccoeff-normed"):
            raise ValueError("Invalid match_method '%s'" % match_method)
        if confirm_method not in ("none", "absdiff", "normed-absdiff"):
            raise ValueError("Invalid confirm_method '%s'" % confirm_method)

        self.match_method = match_method
        self.match_threshold = match_threshold
        self.confirm_method = confirm_method
        self.confirm_threshold = confirm_threshold
        self.erode_passes = erode_passes


class Position(namedtuple('Position', 'x y')):
    """A point within the video frame.

    `x` and `y` are integer coordinates (measured in number of pixels) from the
    top left corner of the video frame.
    """
    pass


class Region(namedtuple('Region', 'x y right bottom')):
    u"""
    ``Region(x, y, width=width, height=height)`` or
    ``Region(x, y, right=right, bottom=bottom)``

    Rectangular region within the video frame.

    For example, given the following regions a, b, and c::

        - 01234567890123
        0 ░░░░░░░░
        1 ░a░░░░░░
        2 ░░░░░░░░
        3 ░░░░░░░░
        4 ░░░░▓▓▓▓░░▓c▓
        5 ░░░░▓▓▓▓░░▓▓▓
        6 ░░░░▓▓▓▓░░░░░
        7 ░░░░▓▓▓▓░░░░░
        8     ░░░░░░b░░
        9     ░░░░░░░░░

    >>> a = Region(0, 0, width=8, height=8)
    >>> b = Region(4, 4, right=13, bottom=10)
    >>> c = Region(10, 4, width=3, height=2)
    >>> a.right
    8
    >>> b.bottom
    10
    >>> b.contains(c), a.contains(b), c.contains(b)
    (True, False, False)
    >>> b.extend(x=6, bottom=-4) == c
    True
    >>> a.extend(right=5).contains(c)
    True
    >>> a.width, a.extend(x=3).width, a.extend(right=-3).width
    (8, 5, 5)
    >>> print c.replace(bottom=10)
    Region(x=10, y=4, width=3, height=6)
    >>> print Region.intersect(a, b)
    Region(x=4, y=4, width=4, height=4)
    >>> Region.intersect(a, b) == Region.intersect(b, a)
    True
    >>> Region.intersect(c, b) == c
    True
    >>> print Region.intersect(a, c)
    None
    >>> print Region.intersect(None, a)
    None
    >>> quadrant = Region(x=float("-inf"), y=float("-inf"), right=0, bottom=0)
    >>> quadrant.translate(2, 2)
    Region(x=-inf, y=-inf, right=2, bottom=2)
    >>> print c.translate(x=-9, y=-3)
    Region(x=1, y=1, width=3, height=2)
    >>> Region.intersect(Region.ALL, c) == c
    True
    >>> Region.ALL
    Region.ALL
    >>> print Region.ALL
    Region.ALL

    .. py:attribute:: x

        The x coordinate of the left edge of the region, measured in pixels
        from the left of the video frame (inclusive).

    .. py:attribute:: y

        The y coordinate of the top edge of the region, measured in pixels from
        the top of the video frame (inclusive).

    .. py:attribute:: right

        The x coordinate of the right edge of the region, measured in pixels
        from the left of the video frame (exclusive).

    .. py:attribute:: bottom

        The y coordinate of the bottom edge of the region, measured in pixels
        from the top of the video frame (exclusive).

    ``x``, ``y``, ``right``, and ``bottom`` can be infinite -- that is,
    ``float("inf")`` or ``-float("inf")``.
    """
    def __new__(cls, x, y, width=None, height=None, right=None, bottom=None):
        if (width is None) == (right is None):
            raise ValueError("You must specify either 'width' or 'right'")
        if (height is None) == (bottom is None):
            raise ValueError("You must specify either 'height' or 'bottom'")
        if right is None:
            right = x + width
        if bottom is None:
            bottom = y + height
        if right <= x:
            raise ValueError("'right' must be greater than 'x'")
        if bottom <= y:
            raise ValueError("'bottom' must be greater than 'y'")
        return super(Region, cls).__new__(cls, x, y, right, bottom)

    def __str__(self):
        if self == Region.ALL:
            return 'Region.ALL'
        else:
            return 'Region(x=%s, y=%s, width=%s, height=%s)' \
                % (self.x, self.y, self.width, self.height)

    def __repr__(self):
        if self == Region.ALL:
            return 'Region.ALL'
        else:
            return super(Region, self).__repr__()

    @property
    def width(self):
        """The width of the region, measured in pixels."""
        return self.right - self.x

    @property
    def height(self):
        """The height of the region, measured in pixels."""
        return self.bottom - self.y

    @staticmethod
    def from_extents(x, y, right, bottom):
        """Create a Region using right and bottom extents rather than width and
        height.

        Deprecated since we added ``right`` and ``bottom`` to Region
        constructor.

        >>> b = Region.from_extents(4, 4, 13, 10)
        >>> print b
        Region(x=4, y=4, width=9, height=6)
        """
        return Region(x, y, right=right, bottom=bottom)

    @staticmethod
    def intersect(a, b):
        """
        :returns: The intersection of regions ``a`` and ``b``, or ``None`` if
            the regions don't intersect.

        Either ``a`` or ``b`` can be ``None`` so intersect is commutative and
        associative.
        """
        if a is None or b is None:
            return None
        else:
            extents = (max(a.x, b.x), max(a.y, b.y),
                       min(a.right, b.right), min(a.bottom, b.bottom))
            if extents[0] < extents[2] and extents[1] < extents[3]:
                return Region.from_extents(*extents)
            else:
                return None

    def contains(self, other):
        """:returns: True if ``other`` is entirely contained within self."""
        return (other and self.x <= other.x and self.y <= other.y and
                self.right >= other.right and self.bottom >= other.bottom)

    def extend(self, x=0, y=0, right=0, bottom=0):
        """
        :returns: A new region with the edges of the region adjusted by the
            given amounts.
        """
        return Region.from_extents(
            self.x + x, self.y + y, self.right + right, self.bottom + bottom)

    def replace(self, x=None, y=None, width=None, height=None, right=None,
                bottom=None):
        """
        :returns: A new region with the edges of the region set to the given
            coordinates.

        This is similar to `extend`, but it takes absolute coordinates within
        the image instead of adjusting by a relative number of pixels.

        ``Region.replace`` was added in stb-tester v24.
        """
        def norm_coords(name_x, name_width, name_right,
                        x, width, right,  # or y, height, bottom
                        default_x, _default_width, default_right):
            if all(z is not None for z in (x, width, right)):
                raise ValueError(
                    "Region.replace: Argument conflict: you may only specify "
                    "two of %s, %s and %s.  You specified %s=%s, %s=%s and "
                    "%s=%s" % (name_x, name_width, name_right,
                               name_x, x, name_width, width, name_right, right))
            if x is None:
                if width is not None and right is not None:
                    x = right - width
                else:
                    x = default_x
            if right is None:
                right = x + width if width is not None else default_right
            return x, right

        x, right = norm_coords('x', 'width', 'right', x, width, right,
                               self.x, self.width, self.right)
        y, bottom = norm_coords('y', 'height', 'bottom', y, height, bottom,
                                self.y, self.height, self.bottom)

        return Region(x=x, y=y, right=right, bottom=bottom)

    def translate(self, x=0, y=0):
        """
        :returns: A new region with the position of the region adjusted by the
            given amounts.
        """
        return Region.from_extents(self.x + x, self.y + y,
                                   self.right + x, self.bottom + y)

Region.ALL = Region(x=float("-inf"), y=float("-inf"),
                    right=float("inf"), bottom=float("inf"))


def _bounding_box(a, b):
    """Find the bounding box of two regions.  Returns the smallest region which
    contains both regions a and b.

    >>> print _bounding_box(Region(50, 20, 10, 20), Region(20, 30, 10, 20))
    Region(x=20, y=20, width=40, height=30)
    >>> print _bounding_box(Region(20, 30, 10, 20), Region(20, 30, 10, 20))
    Region(x=20, y=30, width=10, height=20)
    >>> print _bounding_box(None, Region(20, 30, 10, 20))
    Region(x=20, y=30, width=10, height=20)
    >>> print _bounding_box(Region(20, 30, 10, 20), None)
    Region(x=20, y=30, width=10, height=20)
    >>> print _bounding_box(None, None)
    None
    """
    if a is None:
        return b
    if b is None:
        return a
    return Region.from_extents(min(a.x, b.x), min(a.y, b.y),
                               max(a.right, b.right), max(a.bottom, b.bottom))


class MatchResult(object):
    """The result from `match`.

    * ``timestamp``: Video stream timestamp.
    * ``match``: Boolean result, the same as evaluating `MatchResult` as a bool.
      That is, ``if match_result:`` will behave the same as
      ``if match_result.match:``.
    * ``region``: The `Region` in the video frame where the image was found.
    * ``first_pass_result``: Value between 0 (poor) and 1.0 (excellent match)
      from the first pass of stb-tester's two-pass image matching algorithm
      (see `MatchParameters` for details).
    * ``frame``: The video frame that was searched, in OpenCV format.
    * ``image``: The template image that was searched for, as given to `match`.
    """
    # pylint: disable=W0621
    def __init__(
            self, timestamp, match, region, first_pass_result, frame=None,
            image=None):
        self.timestamp = timestamp
        self.match = match
        self.region = region
        self.first_pass_result = first_pass_result
        if frame is None:
            warnings.warn(
                "Creating a 'MatchResult' without specifying 'frame' is "
                "deprecated. In a future release of stb-tester the 'frame' "
                "parameter will be mandatory.",
                DeprecationWarning, stacklevel=2)
        self.frame = frame
        if image is None:
            warnings.warn(
                "Creating a 'MatchResult' without specifying 'image' is "
                "deprecated. In a future release of stb-tester the 'image' "
                "parameter will be mandatory.",
                DeprecationWarning, stacklevel=2)
            image = ""
        self.image = image

    def __str__(self):
        return (
            "MatchResult(timestamp=%s, match=%s, region=%s, "
            "first_pass_result=%s, frame=%s, image=%s)" % (
                self.timestamp,
                self.match,
                self.region,
                self.first_pass_result,
                "None" if self.frame is None else "%dx%dx%d" % (
                    self.frame.shape[1], self.frame.shape[0],
                    self.frame.shape[2]),
                "<Custom Image>" if isinstance(self.image, numpy.ndarray)
                else repr(self.image)))

    @property
    def position(self):
        return Position(self.region.x, self.region.y)

    def __nonzero__(self):
        return self.match


class _AnnotatedTemplate(namedtuple('_AnnotatedTemplate',
                                    'image name filename')):
    @property
    def friendly_name(self):
        return self.filename or '<Custom Image>'


def _load_template(template):
    if isinstance(template, _AnnotatedTemplate):
        return template
    if isinstance(template, numpy.ndarray):
        return _AnnotatedTemplate(template, None, None)
    else:
        template_name = _find_path(template)
        if not template_name or not os.path.isfile(template_name):
            raise UITestError("No such template file: %s" % template)
        image = cv2.imread(template_name, cv2.CV_LOAD_IMAGE_COLOR)
        if image is None:
            raise UITestError("Failed to load template file: %s" %
                              template_name)
        return _AnnotatedTemplate(image, template, template_name)


def _crop(frame, region):
    if not _image_region(frame).contains(region):
        raise ValueError("'frame' doesn't contain 'region'")
    return frame[region.y:region.bottom, region.x:region.right]


def _image_region(image):
    return Region(0, 0, image.shape[1], image.shape[0])


class MotionResult(namedtuple('MotionResult', 'timestamp motion')):
    """The result from `detect_motion`.

    * `timestamp`: Video stream timestamp.
    * `motion`: Boolean result.
    """
    pass


class OcrMode(IntEnum):
    """Options to control layout analysis and assume a certain form of image.

    For a (brief) description of each option, see the `tesseract(1)
    <http://tesseract-ocr.googlecode.com/svn/trunk/doc/tesseract.1.html#_options>`_
    man page.
    """
    ORIENTATION_AND_SCRIPT_DETECTION_ONLY = 0
    PAGE_SEGMENTATION_WITH_OSD = 1
    PAGE_SEGMENTATION_WITHOUT_OSD_OR_OCR = 2
    PAGE_SEGMENTATION_WITHOUT_OSD = 3
    SINGLE_COLUMN_OF_TEXT_OF_VARIABLE_SIZES = 4
    SINGLE_UNIFORM_BLOCK_OF_VERTICALLY_ALIGNED_TEXT = 5
    SINGLE_UNIFORM_BLOCK_OF_TEXT = 6
    SINGLE_LINE = 7
    SINGLE_WORD = 8
    SINGLE_WORD_IN_A_CIRCLE = 9
    SINGLE_CHARACTER = 10

    # For nicer formatting of `ocr` signature in generated API documentation:
    def __repr__(self):
        return str(self)


class TextMatchResult(namedtuple(
        "TextMatchResult", "timestamp match region frame text")):

    """The result from `match_text`.

    * ``timestamp``: Video stream timestamp.
    * ``match``: Boolean result, the same as evaluating `TextMatchResult` as a
      bool. That is, ``if result:`` will behave the same as
      ``if result.match:``.
    * ``region``: The `Region` (bounding box) of the text found, or ``None`` if
      no text was found.
    * ``frame``: The video frame that was searched, in OpenCV format.
    * ``text``: The text (unicode string) that was searched for, as given to
      `match_text`.
    """
    # pylint: disable=E1101
    def __nonzero__(self):
        return self.match

    def __str__(self):
        return (
            "TextMatchResult(timestamp=%s, match=%s, region=%s, frame=%s, "
            "text=%s)" % (
                self.timestamp,
                self.match,
                self.region,
                "%dx%dx%d" % (self.frame.shape[1], self.frame.shape[0],
                              self.frame.shape[2]),
                repr(self.text)))


def new_device_under_test_from_config(
        gst_source_pipeline=None, gst_sink_pipeline=None, control_uri=None,
        save_video=False, restart_source=None, transformation_pipeline=None):
    from _stbt.control import uri_to_remote

    if gst_source_pipeline is None:
        gst_source_pipeline = get_config('global', 'source_pipeline')
    if gst_sink_pipeline is None:
        gst_sink_pipeline = get_config('global', 'sink_pipeline')
    if control_uri is None:
        control_uri = get_config('global', 'control')
    if restart_source is None:
        restart_source = get_config('global', 'restart_source')
    if transformation_pipeline is None:
        gst_sink_pipeline = get_config('global', 'transformation_pipeline')

    display = Display(
        gst_source_pipeline, gst_sink_pipeline, _mainloop(), save_video,
        restart_source, transformation_pipeline)
    return DeviceUnderTest(
        display=display, control=uri_to_remote(control_uri, display))


class DeviceUnderTest(object):
    def __init__(self, display=None, control=None):
        self._time_of_last_press = None
        self._display = display
        self._control = control

    def __enter__(self):
        if self._display:
            self._display.startup()
        return self

    def __exit__(self, _exc_type, _exc_value, _traceback):
        if self._display:
            self._display.teardown()
            self._display = None
        self._control = None

    def press(self, key, interpress_delay_secs=None):
        if interpress_delay_secs is None:
            interpress_delay_secs = get_config(
                "press", "interpress_delay_secs", type_=float)
        if self._time_of_last_press is not None:
            # `sleep` is inside a `while` loop because the actual suspension
            # time of `sleep` may be less than that requested.
            while True:
                seconds_to_wait = (
                    self._time_of_last_press - datetime.datetime.now() +
                    datetime.timedelta(seconds=interpress_delay_secs)
                ).total_seconds()
                if seconds_to_wait > 0:
                    time.sleep(seconds_to_wait)
                else:
                    break

        self._control.press(key)
        self._time_of_last_press = datetime.datetime.now()
        self.draw_text(key, duration_secs=3)

    def draw_text(self, text, duration_secs=3):
        self._display.draw(text, duration_secs)

    def match(self, image, frame=None, match_parameters=None,
              region=Region.ALL):

        if match_parameters is None:
            match_parameters = MatchParameters()

        template = _load_template(image)

        grabbed_from_live = (frame is None)
        if grabbed_from_live:
            frame = self._display.get_sample()

        with _numpy_from_sample(frame, readonly=True) as npframe:
            region = Region.intersect(_image_region(npframe), region)

            matched, match_region, first_pass_certainty = _match(
                _crop(npframe, region), template.image, match_parameters,
                template.friendly_name)

            match_region = match_region.translate(region.x, region.y)
            result = MatchResult(
                _get_frame_timestamp(frame), matched, match_region,
                first_pass_certainty, numpy.copy(npframe),
                (template.name or template.image))

        if grabbed_from_live:
            self._display.draw(result, None)

        if result.match:
            debug("Match found: %s" % str(result))
        else:
            debug("No match found. Closest match: %s" % str(result))

        return result

    def detect_match(self, image, timeout_secs=10, match_parameters=None,
                     region=Region.ALL):
        template = _load_template(image)

        debug("Searching for " + template.friendly_name)

        for sample in self._display.gst_samples(timeout_secs):
            result = self.match(
                template, frame=sample, match_parameters=match_parameters,
                region=region)
            self._display.draw(result, None)
            yield result

    def detect_motion(self, timeout_secs=10, noise_threshold=None, mask=None):
        if noise_threshold is None:
            noise_threshold = get_config(
                'motion', 'noise_threshold', type_=float)

        debug("Searching for motion")

        mask_image = None
        if mask:
            mask_image = _load_mask(mask)

        previous_frame_gray = None
        log = functools.partial(
            _log_image, directory="stbt-debug/detect_motion")

        for sample in self._display.gst_samples(timeout_secs):
            with _numpy_from_sample(sample, readonly=True) as frame:
                frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            log(frame_gray, "source")

            if previous_frame_gray is None:
                if (mask_image is not None and
                        mask_image.shape[:2] != frame_gray.shape[:2]):
                    raise UITestError(
                        "The dimensions of the mask '%s' %s don't match the "
                        "video frame %s" % (
                            mask, mask_image.shape, frame_gray.shape))
                previous_frame_gray = frame_gray
                continue

            absdiff = cv2.absdiff(frame_gray, previous_frame_gray)
            previous_frame_gray = frame_gray
            log(absdiff, "absdiff")

            if mask_image is not None:
                absdiff = cv2.bitwise_and(absdiff, mask_image)
                log(mask_image, "mask")
                log(absdiff, "absdiff_masked")

            _, thresholded = cv2.threshold(
                absdiff, int((1 - noise_threshold) * 255), 255,
                cv2.THRESH_BINARY)
            eroded = cv2.erode(
                thresholded,
                cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (3, 3)))
            log(thresholded, "absdiff_threshold")
            log(eroded, "absdiff_threshold_erode")

            motion = (cv2.countNonZero(eroded) > 0)

            # Visualisation: Highlight in red the areas where we detected motion
            if motion:
                with _numpy_from_sample(sample) as frame:
                    cv2.add(
                        frame,
                        numpy.multiply(
                            numpy.ones(frame.shape, dtype=numpy.uint8),
                            numpy.array((0, 0, 255), dtype=numpy.uint8),  # bgr
                            dtype=numpy.uint8),
                        mask=cv2.dilate(
                            thresholded,
                            cv2.getStructuringElement(
                                cv2.MORPH_ELLIPSE, (3, 3)),
                            iterations=1),
                        dst=frame)

            result = MotionResult(sample.get_buffer().pts, motion)
            debug("%s found: %s" % (
                "Motion" if motion else "No motion", str(result)))
            yield result

    def wait_for_match(self, image, timeout_secs=10, consecutive_matches=1,
                       match_parameters=None, region=Region.ALL):

        if match_parameters is None:
            match_parameters = MatchParameters()

        match_count = 0
        last_pos = Position(0, 0)
        image = _load_template(image)
        for res in self.detect_match(
                image, timeout_secs, match_parameters=match_parameters,
                region=region):
            if res.match and (match_count == 0 or res.position == last_pos):
                match_count += 1
            else:
                match_count = 0
            last_pos = res.position
            if match_count == consecutive_matches:
                debug("Matched " + image.friendly_name)
                return res

        raise MatchTimeout(res.frame, image.friendly_name, timeout_secs)  # pylint: disable=W0631,C0301

    def press_until_match(
            self,
            key,
            image,
            interval_secs=None,
            max_presses=None,
            match_parameters=None):

        if interval_secs is None:
            # Should this be float?
            interval_secs = get_config(
                "press_until_match", "interval_secs", type_=int)
        if max_presses is None:
            max_presses = get_config(
                "press_until_match", "max_presses", type_=int)

        if match_parameters is None:
            match_parameters = MatchParameters()

        i = 0

        while True:
            try:
                return self.wait_for_match(image, timeout_secs=interval_secs,
                                           match_parameters=match_parameters)
            except MatchTimeout:
                if i < max_presses:
                    self.press(key)
                    i += 1
                else:
                    raise

    def wait_for_motion(
            self, timeout_secs=10, consecutive_frames=None,
            noise_threshold=None, mask=None):

        if consecutive_frames is None:
            consecutive_frames = get_config('motion', 'consecutive_frames')

        consecutive_frames = str(consecutive_frames)
        if '/' in consecutive_frames:
            motion_frames = int(consecutive_frames.split('/')[0])
            considered_frames = int(consecutive_frames.split('/')[1])
        else:
            motion_frames = int(consecutive_frames)
            considered_frames = int(consecutive_frames)

        if motion_frames > considered_frames:
            raise ConfigurationError(
                "`motion_frames` exceeds `considered_frames`")

        debug("Waiting for %d out of %d frames with motion" % (
            motion_frames, considered_frames))

        matches = deque(maxlen=considered_frames)
        for res in self.detect_motion(timeout_secs, noise_threshold, mask):
            matches.append(res.motion)
            if matches.count(True) >= motion_frames:
                debug("Motion detected.")
                return res

        screenshot = self.get_frame()
        raise MotionTimeout(screenshot, mask, timeout_secs)

    def ocr(self, frame=None, region=Region.ALL,
            mode=OcrMode.PAGE_SEGMENTATION_WITHOUT_OSD,
            lang="eng", tesseract_config=None, tesseract_user_words=None,
            tesseract_user_patterns=None):

        if frame is None:
            frame = self._display.get_sample()

        if region is None:
            warnings.warn(
                "Passing region=None to ocr is deprecated since 0.21 and the "
                "meaning will change in a future version.  To OCR an entire "
                "video frame pass region=Region.ALL instead",
                DeprecationWarning, stacklevel=2)
            region = Region.ALL

        text, region = _tesseract(
            frame, region, mode, lang, tesseract_config,
            user_patterns=tesseract_user_patterns,
            user_words=tesseract_user_words)
        text = text.decode('utf-8').strip().translate(_ocr_transtab)
        debug(u"OCR in region %s read '%s'." % (region, text))
        return text

    def match_text(self, text, frame=None, region=Region.ALL,
                   mode=OcrMode.PAGE_SEGMENTATION_WITHOUT_OSD, lang="eng",
                   tesseract_config=None):

        import lxml.etree
        if frame is None:
            frame = self.get_frame()

        _config = dict(tesseract_config or {})
        _config['tessedit_create_hocr'] = 1
        if _tesseract_version() >= LooseVersion('3.04'):
            _config['tessedit_create_txt'] = 0

        ts = _get_frame_timestamp(frame)

        xml, region = _tesseract(frame, region, mode, lang, _config,
                                 user_words=text.split())
        if xml == '':
            return TextMatchResult(ts, False, None, frame, text)
        hocr = lxml.etree.fromstring(xml)
        p = _hocr_find_phrase(hocr, text.split())
        if p:
            # Find bounding box
            box = None
            for _, elem in p:
                box = _bounding_box(box, _hocr_elem_region(elem))
            # _tesseract crops to region and scales up by a factor of 3 so we
            # must undo this transformation here.
            box = Region.from_extents(
                region.x + box.x // 3, region.y + box.y // 3,
                region.x + box.right // 3, region.y + box.bottom // 3)
            return TextMatchResult(ts, True, box, frame, text)
        else:
            return TextMatchResult(ts, False, None, frame, text)

    def frames(self, timeout_secs=None):
        return self._display.frames(timeout_secs)

    def get_frame(self):
        with _numpy_from_sample(
                self._display.get_sample(), readonly=True) as frame:
            return frame.copy()

    def is_screen_black(self, frame=None, mask=None, threshold=None):
        if threshold is None:
            threshold = get_config('is_screen_black', 'threshold', type_=int)
        if mask:
            mask = _load_mask(mask)
        if frame is None:
            frame = self._display.get_sample()
        with _numpy_from_sample(frame, readonly=True) as f:
            greyframe = cv2.cvtColor(f, cv2.COLOR_BGR2GRAY)
        _, greyframe = cv2.threshold(
            greyframe, threshold, 255, cv2.THRESH_BINARY)
        _, maxVal, _, _ = cv2.minMaxLoc(greyframe, mask)
        if logging.get_debug_level() > 1:
            _log_image(frame, 'source', 'stbt-debug/is_screen_black')
            if mask is not None:
                _log_image(mask, 'mask', 'stbt-debug/is_screen_black')
                _log_image(numpy.bitwise_and(greyframe, mask),
                           'non-black-regions-after-masking',
                           'stbt-debug/is_screen_black')
            else:
                _log_image(greyframe, 'non-black-regions-after-masking',
                           'stbt-debug/is_screen_black')
        return maxVal == 0

# Utility functions
# ===========================================================================


def save_frame(image, filename):
    """Saves an OpenCV image to the specified file.

    Takes an image obtained from `get_frame` or from the `screenshot`
    property of `MatchTimeout` or `MotionTimeout`.
    """
    with _numpy_from_sample(image, readonly=True) as imagebuf:
        cv2.imwrite(filename, imagebuf)


def wait_until(callable_, timeout_secs=10, interval_secs=0):
    """Wait until a condition becomes true, or until a timeout.

    ``callable_`` is any python callable, such as a function or a lambda
    expression. It will be called repeatedly (with a delay of ``interval_secs``
    seconds between successive calls) until it succeeds (that is, it returns a
    truthy value) or until ``timeout_secs`` seconds have passed. In both cases,
    ``wait_until`` returns the value that ``callable_`` returns.

    After you send a remote-control signal to the system-under-test it usually
    takes a few frames to react, so a test script like this would probably
    fail::

        press("KEY_EPG")
        assert match("guide.png")

    Instead, use this::

        press("KEY_EPG")
        assert wait_until(lambda: match("guide.png"))

    Note that instead of the above `assert wait_until(...)` you could use
    `wait_for_match("guide.png")`. `wait_until` is a generic solution that
    also works with stbt's other functions, like `match_text` and
    `is_screen_black`.

    `wait_until` allows composing more complex conditions, such as::

        # Wait until something disappears:
        assert wait_until(lambda: not match("xyz.png"))

        # Assert that something doesn't appear within 10 seconds:
        assert not wait_until(lambda: match("xyz.png"))

        # Assert that two images are present at the same time:
        assert wait_until(lambda: match("a.png") and match("b.png"))

        # Wait but don't raise an exception:
        if not wait_until(lambda: match("xyz.png")):
            do_something_else()

    There are some drawbacks to using `assert` instead of `wait_for_match`:

    * The exception message won't contain the reason why the match failed
      (unless you specify it as a second parameter to `assert`, which is
      tedious and we don't expect you to do it), and
    * The exception won't have the offending video-frame attached (so the
      screenshot in the test-run artifacts will be a few frames later than the
      frame that actually caused the test to fail).

    We hope to solve both of the above drawbacks at some point in the future.

    ``wait_until`` was added in stb-tester v22.
    """
    expiry_time = time.time() + timeout_secs
    while True:
        e = callable_()
        if e:
            return e
        if time.time() > expiry_time:
            debug("wait_until timed out: %s" % _callable_description(callable_))
            return e
        time.sleep(interval_secs)


def _callable_description(callable_):
    """Helper to provide nicer debug output when `wait_until` fails.

    >>> _callable_description(wait_until)
    'wait_until'
    >>> _callable_description(
    ...     lambda: stbt.press("OK"))
    '    lambda: stbt.press("OK"))\\n'
    >>> _callable_description(functools.partial(int, base=2))
    'int'
    >>> _callable_description(functools.partial(functools.partial(int, base=2),
    ...                                         x='10'))
    'int'
    >>> class T(object):
    ...     def __call__(self): return True;
    >>> _callable_description(T())
    '<_stbt.core.T object at 0x...>'
    """
    if hasattr(callable_, "__name__"):
        name = callable_.__name__
        if name == "<lambda>":
            try:
                name = inspect.getsource(callable_)
            except IOError:
                pass
        return name
    elif isinstance(callable_, functools.partial):
        return _callable_description(callable_.func)
    else:
        return repr(callable_)


@contextmanager
def as_precondition(message):
    """Context manager that replaces test failures with test errors.

    Stb-tester's reports show test failures (that is, `UITestFailure` or
    `AssertionError` exceptions) as red results, and test errors (that is,
    unhandled exceptions of any other type) as yellow results. Note that
    `wait_for_match`, `wait_for_motion`, and similar functions raise a
    `UITestFailure` when they detect a failure. By running such functions
    inside an `as_precondition` context, any `UITestFailure` or
    `AssertionError` exceptions they raise will be caught, and a
    `PreconditionError` will be raised instead.

    When running a single testcase hundreds or thousands of times to reproduce
    an intermittent defect, it is helpful to mark unrelated failures as test
    errors (yellow) rather than test failures (red), so that you can focus on
    diagnosing the failures that are most likely to be the particular defect
    you are looking for. For more details see `Test failures vs. errors
    <http://stb-tester.com/preconditions>`_.

    :param str message:
        A description of the precondition. Word this positively: "Channels
        tuned", not "Failed to tune channels".

    :raises:
        `PreconditionError` if the wrapped code block raises a `UITestFailure`
        or `AssertionError`.

    Example::

        def test_that_the_on_screen_id_is_shown_after_booting():
            channel = 100

            with stbt.as_precondition("Tuned to channel %s" % channel):
                mainmenu.close_any_open_menu()
                channels.goto_channel(channel)
                power.cold_reboot()
                assert channels.is_on_channel(channel)

            stbt.wait_for_match("on-screen-id.png")

    """
    try:
        yield
    except (UITestFailure, AssertionError) as e:
        debug("stbt.as_precondition caught a %s exception and will "
              "re-raise it as PreconditionError.\nOriginal exception was:\n%s"
              % (type(e).__name__, traceback.format_exc(e)))
        exc = PreconditionError(message, e)
        if hasattr(e, 'screenshot'):
            exc.screenshot = e.screenshot  # pylint: disable=W0201
        raise exc


class UITestError(Exception):
    """The test script had an unrecoverable error."""
    pass


class UITestFailure(Exception):
    """The test failed because the system under test didn't behave as expected.

    Inherit from this if you need to define your own test-failure exceptions.
    """
    pass


class NoVideo(UITestFailure):
    """No video available from the source pipeline."""
    pass


class MatchTimeout(UITestFailure):
    """Exception raised by `wait_for_match`.

    * ``screenshot``: The last video frame that `wait_for_match` checked before
      timing out.
    * ``expected``: Filename of the image that was being searched for.
    * ``timeout_secs``: Number of seconds that the image was searched for.
    """
    def __init__(self, screenshot, expected, timeout_secs):
        super(MatchTimeout, self).__init__()
        self.screenshot = screenshot
        self.expected = expected
        self.timeout_secs = timeout_secs

    def __str__(self):
        return "Didn't find match for '%s' within %g seconds." % (
            self.expected, self.timeout_secs)


class MotionTimeout(UITestFailure):
    """Exception raised by `wait_for_motion`.

    * ``screenshot``: The last video frame that `wait_for_motion` checked before
      timing out.
    * ``mask``: Filename of the mask that was used, if any.
    * ``timeout_secs``: Number of seconds that motion was searched for.
    """
    def __init__(self, screenshot, mask, timeout_secs):
        super(MotionTimeout, self).__init__()
        self.screenshot = screenshot
        self.mask = mask
        self.timeout_secs = timeout_secs

    def __str__(self):
        return "Didn't find motion%s within %g seconds." % (
            " (with mask '%s')" % self.mask if self.mask else "",
            self.timeout_secs)


class PreconditionError(UITestError):
    """Exception raised by `as_precondition`."""
    def __init__(self, message, original_exception):
        super(PreconditionError, self).__init__()
        self.message = message
        self.original_exception = original_exception

    def __str__(self):
        return (
            "Didn't meet precondition '%s' (original exception was: %s)"
            % (self.message, self.original_exception))


# stbt-run initialisation and convenience functions
# (you will need these if writing your own version of stbt-run)
# ===========================================================================

def argparser():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--control',
        default=get_config('global', 'control'),
        help='The remote control to control the stb (default: %(default)s)')
    parser.add_argument(
        '--source-pipeline',
        default=get_config('global', 'source_pipeline'),
        help='A gstreamer pipeline to use for A/V input (default: '
             '%(default)s)')
    parser.add_argument(
        '--sink-pipeline',
        default=get_config('global', 'sink_pipeline'),
        help='A gstreamer pipeline to use for video output '
             '(default: %(default)s)')
    parser.add_argument(
        '--restart-source', action='store_true',
        default=(get_config('global', 'restart_source').lower() in
                 ("1", "yes", "true", "on")),
        help='Restart the GStreamer source pipeline when video loss is '
             'detected')

    logging.argparser_add_verbose_argument(parser)

    return parser


class _FrameObjectMeta(type):
    def __init__(cls, name, parents, dct):
        property_names = sorted([
            p for p in dir(cls)
            if isinstance(getattr(cls, p), property)])
        assert 'is_visible' in property_names
        cls._FrameObject__attrs = ["is_visible"] + sorted(
            x for x in property_names
            if x != "is_visible" and not x.startswith('_'))
        super(_FrameObjectMeta, cls).__init__(name, parents, dct)


class FrameObject(object):
    __metaclass__ = _FrameObjectMeta

    def __init__(self, frame):
        if frame is None:
            raise ValueError("FrameObject: frame must not be None")
        self._frame = frame

    def __repr__(self):
        args = ", ".join(("%s=%r" % x) for x in self._iter_attrs())
        return "%s(%s)" % (self.__class__.__name__, args)

    def _iter_attrs(self):
        if self:
            # pylint: disable=protected-access,no-member<commit_msg>FrameObject: Make properties return `None` if is_visible is `False`<commit_after># coding: utf-8
"""Main stb-tester python module. Intended to be used with `stbt run`.

See `man stbt` and http://stb-tester.com for documentation.

Copyright 2012-2013 YouView TV Ltd and contributors.
License: LGPL v2.1 or (at your option) any later version (see
https://github.com/stb-tester/stb-tester/blob/master/LICENSE for details).
"""

from __future__ import absolute_import

import argparse
import datetime
import functools
import glob
import inspect
import os
import Queue
import re
import subprocess
import threading
import time
import traceback
import warnings
from collections import deque, namedtuple
from contextlib import contextmanager
from distutils.version import LooseVersion

import cv2
import gi
import numpy
from enum import IntEnum
from kitchen.text.converters import to_bytes

from _stbt import logging, utils
from _stbt.config import ConfigurationError, get_config
from _stbt.gst_hacks import gst_iterate, map_gst_sample
from _stbt.logging import ddebug, debug, warn

gi.require_version("Gst", "1.0")
from gi.repository import GLib, GObject, Gst  # isort:skip pylint: disable=E0611

if getattr(gi, "version_info", (0, 0, 0)) < (3, 12, 0):
    GObject.threads_init()
Gst.init(None)

warnings.filterwarnings(
    action="always", category=DeprecationWarning, message='.*stb-tester')


# Functions available to stbt scripts
# ===========================================================================


class MatchParameters(object):
    """Parameters to customise the image processing algorithm used by
    `match`, `wait_for_match`, and `press_until_match`.

    You can change the default values for these parameters by setting a key
    (with the same name as the corresponding python parameter) in the `[match]`
    section of stbt.conf. But we strongly recommend that you don't change the
    default values from what is documented here.

    You should only need to change these parameters when you're trying to match
    a template image that isn't actually a perfect match -- for example if
    there's a translucent background with live TV visible behind it; or if you
    have a template image of a button's background and you want it to match even
    if the text on the button doesn't match.

    :param str match_method:
      The method to be used by the first pass of stb-tester's image matching
      algorithm, to find the most likely location of the "template" image
      within the larger source image.

      Allowed values are "sqdiff-normed", "ccorr-normed", and "ccoeff-normed".
      For the meaning of these parameters, see OpenCV's `cvMatchTemplate
      <http://docs.opencv.org/modules/imgproc/doc/object_detection.html>`_.

      We recommend that you don't change this from its default value of
      "sqdiff-normed".

    :param float match_threshold:
      How strong a result from the first pass must be, to be considered a
      match. Valid values range from 0 (anything is considered to match)
      to 1 (the match has to be pixel perfect). This defaults to 0.8.

    :param str confirm_method:
      The method to be used by the second pass of stb-tester's image matching
      algorithm, to confirm that the region identified by the first pass is a
      good match.

      The first pass often gives false positives (it reports a "match" for an
      image that shouldn't match). The second pass is more CPU-intensive, but
      it only checks the position of the image that the first pass identified.
      The allowed values are:

      :"none":
        Do not confirm the match. Assume that the potential match found is
        correct.

      :"absdiff":
        Compare the absolute difference of each pixel from the template image
        against its counterpart from the candidate region in the source video
        frame.

      :"normed-absdiff":
        Normalise the pixel values from both the template image and the
        candidate region in the source video frame, then compare the absolute
        difference as with "absdiff".

        This gives better results with low-contrast images. We recommend setting
        this as the default `confirm_method` in stbt.conf, with a
        `confirm_threshold` of 0.30.

    :param float confirm_threshold:
      The maximum allowed difference between any given pixel from the template
      image and its counterpart from the candidate region in the source video
      frame, as a fraction of the pixel's total luminance range.

      Valid values range from 0 (more strict) to 1.0 (less strict).
      Useful values tend to be around 0.16 for the "absdiff" method, and 0.30
      for the "normed-absdiff" method.

    :param int erode_passes:
      After the "absdiff" or "normed-absdiff" absolute difference is taken,
      stb-tester runs an erosion algorithm that removes single-pixel differences
      to account for noise. Useful values are 1 (the default) and 0 (to disable
      this step).

    """

    def __init__(self, match_method=None, match_threshold=None,
                 confirm_method=None, confirm_threshold=None,
                 erode_passes=None):
        if match_method is None:
            match_method = get_config('match', 'match_method')
        if match_threshold is None:
            match_threshold = get_config(
                'match', 'match_threshold', type_=float)
        if confirm_method is None:
            confirm_method = get_config('match', 'confirm_method')
        if confirm_threshold is None:
            confirm_threshold = get_config(
                'match', 'confirm_threshold', type_=float)
        if erode_passes is None:
            erode_passes = get_config('match', 'erode_passes', type_=int)

        if match_method not in (
                "sqdiff-normed", "ccorr-normed", "ccoeff-normed"):
            raise ValueError("Invalid match_method '%s'" % match_method)
        if confirm_method not in ("none", "absdiff", "normed-absdiff"):
            raise ValueError("Invalid confirm_method '%s'" % confirm_method)

        self.match_method = match_method
        self.match_threshold = match_threshold
        self.confirm_method = confirm_method
        self.confirm_threshold = confirm_threshold
        self.erode_passes = erode_passes


class Position(namedtuple('Position', 'x y')):
    """A point within the video frame.

    `x` and `y` are integer coordinates (measured in number of pixels) from the
    top left corner of the video frame.
    """
    pass


class Region(namedtuple('Region', 'x y right bottom')):
    u"""
    ``Region(x, y, width=width, height=height)`` or
    ``Region(x, y, right=right, bottom=bottom)``

    Rectangular region within the video frame.

    For example, given the following regions a, b, and c::

        - 01234567890123
        0 ░░░░░░░░
        1 ░a░░░░░░
        2 ░░░░░░░░
        3 ░░░░░░░░
        4 ░░░░▓▓▓▓░░▓c▓
        5 ░░░░▓▓▓▓░░▓▓▓
        6 ░░░░▓▓▓▓░░░░░
        7 ░░░░▓▓▓▓░░░░░
        8     ░░░░░░b░░
        9     ░░░░░░░░░

    >>> a = Region(0, 0, width=8, height=8)
    >>> b = Region(4, 4, right=13, bottom=10)
    >>> c = Region(10, 4, width=3, height=2)
    >>> a.right
    8
    >>> b.bottom
    10
    >>> b.contains(c), a.contains(b), c.contains(b)
    (True, False, False)
    >>> b.extend(x=6, bottom=-4) == c
    True
    >>> a.extend(right=5).contains(c)
    True
    >>> a.width, a.extend(x=3).width, a.extend(right=-3).width
    (8, 5, 5)
    >>> print c.replace(bottom=10)
    Region(x=10, y=4, width=3, height=6)
    >>> print Region.intersect(a, b)
    Region(x=4, y=4, width=4, height=4)
    >>> Region.intersect(a, b) == Region.intersect(b, a)
    True
    >>> Region.intersect(c, b) == c
    True
    >>> print Region.intersect(a, c)
    None
    >>> print Region.intersect(None, a)
    None
    >>> quadrant = Region(x=float("-inf"), y=float("-inf"), right=0, bottom=0)
    >>> quadrant.translate(2, 2)
    Region(x=-inf, y=-inf, right=2, bottom=2)
    >>> print c.translate(x=-9, y=-3)
    Region(x=1, y=1, width=3, height=2)
    >>> Region.intersect(Region.ALL, c) == c
    True
    >>> Region.ALL
    Region.ALL
    >>> print Region.ALL
    Region.ALL

    .. py:attribute:: x

        The x coordinate of the left edge of the region, measured in pixels
        from the left of the video frame (inclusive).

    .. py:attribute:: y

        The y coordinate of the top edge of the region, measured in pixels from
        the top of the video frame (inclusive).

    .. py:attribute:: right

        The x coordinate of the right edge of the region, measured in pixels
        from the left of the video frame (exclusive).

    .. py:attribute:: bottom

        The y coordinate of the bottom edge of the region, measured in pixels
        from the top of the video frame (exclusive).

    ``x``, ``y``, ``right``, and ``bottom`` can be infinite -- that is,
    ``float("inf")`` or ``-float("inf")``.
    """
    def __new__(cls, x, y, width=None, height=None, right=None, bottom=None):
        if (width is None) == (right is None):
            raise ValueError("You must specify either 'width' or 'right'")
        if (height is None) == (bottom is None):
            raise ValueError("You must specify either 'height' or 'bottom'")
        if right is None:
            right = x + width
        if bottom is None:
            bottom = y + height
        if right <= x:
            raise ValueError("'right' must be greater than 'x'")
        if bottom <= y:
            raise ValueError("'bottom' must be greater than 'y'")
        return super(Region, cls).__new__(cls, x, y, right, bottom)

    def __str__(self):
        if self == Region.ALL:
            return 'Region.ALL'
        else:
            return 'Region(x=%s, y=%s, width=%s, height=%s)' \
                % (self.x, self.y, self.width, self.height)

    def __repr__(self):
        if self == Region.ALL:
            return 'Region.ALL'
        else:
            return super(Region, self).__repr__()

    @property
    def width(self):
        """The width of the region, measured in pixels."""
        return self.right - self.x

    @property
    def height(self):
        """The height of the region, measured in pixels."""
        return self.bottom - self.y

    @staticmethod
    def from_extents(x, y, right, bottom):
        """Create a Region using right and bottom extents rather than width and
        height.

        Deprecated since we added ``right`` and ``bottom`` to Region
        constructor.

        >>> b = Region.from_extents(4, 4, 13, 10)
        >>> print b
        Region(x=4, y=4, width=9, height=6)
        """
        return Region(x, y, right=right, bottom=bottom)

    @staticmethod
    def intersect(a, b):
        """
        :returns: The intersection of regions ``a`` and ``b``, or ``None`` if
            the regions don't intersect.

        Either ``a`` or ``b`` can be ``None`` so intersect is commutative and
        associative.
        """
        if a is None or b is None:
            return None
        else:
            extents = (max(a.x, b.x), max(a.y, b.y),
                       min(a.right, b.right), min(a.bottom, b.bottom))
            if extents[0] < extents[2] and extents[1] < extents[3]:
                return Region.from_extents(*extents)
            else:
                return None

    def contains(self, other):
        """:returns: True if ``other`` is entirely contained within self."""
        return (other and self.x <= other.x and self.y <= other.y and
                self.right >= other.right and self.bottom >= other.bottom)

    def extend(self, x=0, y=0, right=0, bottom=0):
        """
        :returns: A new region with the edges of the region adjusted by the
            given amounts.
        """
        return Region.from_extents(
            self.x + x, self.y + y, self.right + right, self.bottom + bottom)

    def replace(self, x=None, y=None, width=None, height=None, right=None,
                bottom=None):
        """
        :returns: A new region with the edges of the region set to the given
            coordinates.

        This is similar to `extend`, but it takes absolute coordinates within
        the image instead of adjusting by a relative number of pixels.

        ``Region.replace`` was added in stb-tester v24.
        """
        def norm_coords(name_x, name_width, name_right,
                        x, width, right,  # or y, height, bottom
                        default_x, _default_width, default_right):
            if all(z is not None for z in (x, width, right)):
                raise ValueError(
                    "Region.replace: Argument conflict: you may only specify "
                    "two of %s, %s and %s.  You specified %s=%s, %s=%s and "
                    "%s=%s" % (name_x, name_width, name_right,
                               name_x, x, name_width, width, name_right, right))
            if x is None:
                if width is not None and right is not None:
                    x = right - width
                else:
                    x = default_x
            if right is None:
                right = x + width if width is not None else default_right
            return x, right

        x, right = norm_coords('x', 'width', 'right', x, width, right,
                               self.x, self.width, self.right)
        y, bottom = norm_coords('y', 'height', 'bottom', y, height, bottom,
                                self.y, self.height, self.bottom)

        return Region(x=x, y=y, right=right, bottom=bottom)

    def translate(self, x=0, y=0):
        """
        :returns: A new region with the position of the region adjusted by the
            given amounts.
        """
        return Region.from_extents(self.x + x, self.y + y,
                                   self.right + x, self.bottom + y)

Region.ALL = Region(x=float("-inf"), y=float("-inf"),
                    right=float("inf"), bottom=float("inf"))


def _bounding_box(a, b):
    """Find the bounding box of two regions.  Returns the smallest region which
    contains both regions a and b.

    >>> print _bounding_box(Region(50, 20, 10, 20), Region(20, 30, 10, 20))
    Region(x=20, y=20, width=40, height=30)
    >>> print _bounding_box(Region(20, 30, 10, 20), Region(20, 30, 10, 20))
    Region(x=20, y=30, width=10, height=20)
    >>> print _bounding_box(None, Region(20, 30, 10, 20))
    Region(x=20, y=30, width=10, height=20)
    >>> print _bounding_box(Region(20, 30, 10, 20), None)
    Region(x=20, y=30, width=10, height=20)
    >>> print _bounding_box(None, None)
    None
    """
    if a is None:
        return b
    if b is None:
        return a
    return Region.from_extents(min(a.x, b.x), min(a.y, b.y),
                               max(a.right, b.right), max(a.bottom, b.bottom))


class MatchResult(object):
    """The result from `match`.

    * ``timestamp``: Video stream timestamp.
    * ``match``: Boolean result, the same as evaluating `MatchResult` as a bool.
      That is, ``if match_result:`` will behave the same as
      ``if match_result.match:``.
    * ``region``: The `Region` in the video frame where the image was found.
    * ``first_pass_result``: Value between 0 (poor) and 1.0 (excellent match)
      from the first pass of stb-tester's two-pass image matching algorithm
      (see `MatchParameters` for details).
    * ``frame``: The video frame that was searched, in OpenCV format.
    * ``image``: The template image that was searched for, as given to `match`.
    """
    # pylint: disable=W0621
    def __init__(
            self, timestamp, match, region, first_pass_result, frame=None,
            image=None):
        self.timestamp = timestamp
        self.match = match
        self.region = region
        self.first_pass_result = first_pass_result
        if frame is None:
            warnings.warn(
                "Creating a 'MatchResult' without specifying 'frame' is "
                "deprecated. In a future release of stb-tester the 'frame' "
                "parameter will be mandatory.",
                DeprecationWarning, stacklevel=2)
        self.frame = frame
        if image is None:
            warnings.warn(
                "Creating a 'MatchResult' without specifying 'image' is "
                "deprecated. In a future release of stb-tester the 'image' "
                "parameter will be mandatory.",
                DeprecationWarning, stacklevel=2)
            image = ""
        self.image = image

    def __str__(self):
        return (
            "MatchResult(timestamp=%s, match=%s, region=%s, "
            "first_pass_result=%s, frame=%s, image=%s)" % (
                self.timestamp,
                self.match,
                self.region,
                self.first_pass_result,
                "None" if self.frame is None else "%dx%dx%d" % (
                    self.frame.shape[1], self.frame.shape[0],
                    self.frame.shape[2]),
                "<Custom Image>" if isinstance(self.image, numpy.ndarray)
                else repr(self.image)))

    @property
    def position(self):
        return Position(self.region.x, self.region.y)

    def __nonzero__(self):
        return self.match


class _AnnotatedTemplate(namedtuple('_AnnotatedTemplate',
                                    'image name filename')):
    @property
    def friendly_name(self):
        return self.filename or '<Custom Image>'


def _load_template(template):
    if isinstance(template, _AnnotatedTemplate):
        return template
    if isinstance(template, numpy.ndarray):
        return _AnnotatedTemplate(template, None, None)
    else:
        template_name = _find_path(template)
        if not template_name or not os.path.isfile(template_name):
            raise UITestError("No such template file: %s" % template)
        image = cv2.imread(template_name, cv2.CV_LOAD_IMAGE_COLOR)
        if image is None:
            raise UITestError("Failed to load template file: %s" %
                              template_name)
        return _AnnotatedTemplate(image, template, template_name)


def _crop(frame, region):
    if not _image_region(frame).contains(region):
        raise ValueError("'frame' doesn't contain 'region'")
    return frame[region.y:region.bottom, region.x:region.right]


def _image_region(image):
    return Region(0, 0, image.shape[1], image.shape[0])


class MotionResult(namedtuple('MotionResult', 'timestamp motion')):
    """The result from `detect_motion`.

    * `timestamp`: Video stream timestamp.
    * `motion`: Boolean result.
    """
    pass


class OcrMode(IntEnum):
    """Options to control layout analysis and assume a certain form of image.

    For a (brief) description of each option, see the `tesseract(1)
    <http://tesseract-ocr.googlecode.com/svn/trunk/doc/tesseract.1.html#_options>`_
    man page.
    """
    ORIENTATION_AND_SCRIPT_DETECTION_ONLY = 0
    PAGE_SEGMENTATION_WITH_OSD = 1
    PAGE_SEGMENTATION_WITHOUT_OSD_OR_OCR = 2
    PAGE_SEGMENTATION_WITHOUT_OSD = 3
    SINGLE_COLUMN_OF_TEXT_OF_VARIABLE_SIZES = 4
    SINGLE_UNIFORM_BLOCK_OF_VERTICALLY_ALIGNED_TEXT = 5
    SINGLE_UNIFORM_BLOCK_OF_TEXT = 6
    SINGLE_LINE = 7
    SINGLE_WORD = 8
    SINGLE_WORD_IN_A_CIRCLE = 9
    SINGLE_CHARACTER = 10

    # For nicer formatting of `ocr` signature in generated API documentation:
    def __repr__(self):
        return str(self)


class TextMatchResult(namedtuple(
        "TextMatchResult", "timestamp match region frame text")):

    """The result from `match_text`.

    * ``timestamp``: Video stream timestamp.
    * ``match``: Boolean result, the same as evaluating `TextMatchResult` as a
      bool. That is, ``if result:`` will behave the same as
      ``if result.match:``.
    * ``region``: The `Region` (bounding box) of the text found, or ``None`` if
      no text was found.
    * ``frame``: The video frame that was searched, in OpenCV format.
    * ``text``: The text (unicode string) that was searched for, as given to
      `match_text`.
    """
    # pylint: disable=E1101
    def __nonzero__(self):
        return self.match

    def __str__(self):
        return (
            "TextMatchResult(timestamp=%s, match=%s, region=%s, frame=%s, "
            "text=%s)" % (
                self.timestamp,
                self.match,
                self.region,
                "%dx%dx%d" % (self.frame.shape[1], self.frame.shape[0],
                              self.frame.shape[2]),
                repr(self.text)))


def new_device_under_test_from_config(
        gst_source_pipeline=None, gst_sink_pipeline=None, control_uri=None,
        save_video=False, restart_source=None, transformation_pipeline=None):
    from _stbt.control import uri_to_remote

    if gst_source_pipeline is None:
        gst_source_pipeline = get_config('global', 'source_pipeline')
    if gst_sink_pipeline is None:
        gst_sink_pipeline = get_config('global', 'sink_pipeline')
    if control_uri is None:
        control_uri = get_config('global', 'control')
    if restart_source is None:
        restart_source = get_config('global', 'restart_source')
    if transformation_pipeline is None:
        gst_sink_pipeline = get_config('global', 'transformation_pipeline')

    display = Display(
        gst_source_pipeline, gst_sink_pipeline, _mainloop(), save_video,
        restart_source, transformation_pipeline)
    return DeviceUnderTest(
        display=display, control=uri_to_remote(control_uri, display))


class DeviceUnderTest(object):
    def __init__(self, display=None, control=None):
        self._time_of_last_press = None
        self._display = display
        self._control = control

    def __enter__(self):
        if self._display:
            self._display.startup()
        return self

    def __exit__(self, _exc_type, _exc_value, _traceback):
        if self._display:
            self._display.teardown()
            self._display = None
        self._control = None

    def press(self, key, interpress_delay_secs=None):
        if interpress_delay_secs is None:
            interpress_delay_secs = get_config(
                "press", "interpress_delay_secs", type_=float)
        if self._time_of_last_press is not None:
            # `sleep` is inside a `while` loop because the actual suspension
            # time of `sleep` may be less than that requested.
            while True:
                seconds_to_wait = (
                    self._time_of_last_press - datetime.datetime.now() +
                    datetime.timedelta(seconds=interpress_delay_secs)
                ).total_seconds()
                if seconds_to_wait > 0:
                    time.sleep(seconds_to_wait)
                else:
                    break

        self._control.press(key)
        self._time_of_last_press = datetime.datetime.now()
        self.draw_text(key, duration_secs=3)

    def draw_text(self, text, duration_secs=3):
        self._display.draw(text, duration_secs)

    def match(self, image, frame=None, match_parameters=None,
              region=Region.ALL):

        if match_parameters is None:
            match_parameters = MatchParameters()

        template = _load_template(image)

        grabbed_from_live = (frame is None)
        if grabbed_from_live:
            frame = self._display.get_sample()

        with _numpy_from_sample(frame, readonly=True) as npframe:
            region = Region.intersect(_image_region(npframe), region)

            matched, match_region, first_pass_certainty = _match(
                _crop(npframe, region), template.image, match_parameters,
                template.friendly_name)

            match_region = match_region.translate(region.x, region.y)
            result = MatchResult(
                _get_frame_timestamp(frame), matched, match_region,
                first_pass_certainty, numpy.copy(npframe),
                (template.name or template.image))

        if grabbed_from_live:
            self._display.draw(result, None)

        if result.match:
            debug("Match found: %s" % str(result))
        else:
            debug("No match found. Closest match: %s" % str(result))

        return result

    def detect_match(self, image, timeout_secs=10, match_parameters=None,
                     region=Region.ALL):
        template = _load_template(image)

        debug("Searching for " + template.friendly_name)

        for sample in self._display.gst_samples(timeout_secs):
            result = self.match(
                template, frame=sample, match_parameters=match_parameters,
                region=region)
            self._display.draw(result, None)
            yield result

    def detect_motion(self, timeout_secs=10, noise_threshold=None, mask=None):
        if noise_threshold is None:
            noise_threshold = get_config(
                'motion', 'noise_threshold', type_=float)

        debug("Searching for motion")

        mask_image = None
        if mask:
            mask_image = _load_mask(mask)

        previous_frame_gray = None
        log = functools.partial(
            _log_image, directory="stbt-debug/detect_motion")

        for sample in self._display.gst_samples(timeout_secs):
            with _numpy_from_sample(sample, readonly=True) as frame:
                frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            log(frame_gray, "source")

            if previous_frame_gray is None:
                if (mask_image is not None and
                        mask_image.shape[:2] != frame_gray.shape[:2]):
                    raise UITestError(
                        "The dimensions of the mask '%s' %s don't match the "
                        "video frame %s" % (
                            mask, mask_image.shape, frame_gray.shape))
                previous_frame_gray = frame_gray
                continue

            absdiff = cv2.absdiff(frame_gray, previous_frame_gray)
            previous_frame_gray = frame_gray
            log(absdiff, "absdiff")

            if mask_image is not None:
                absdiff = cv2.bitwise_and(absdiff, mask_image)
                log(mask_image, "mask")
                log(absdiff, "absdiff_masked")

            _, thresholded = cv2.threshold(
                absdiff, int((1 - noise_threshold) * 255), 255,
                cv2.THRESH_BINARY)
            eroded = cv2.erode(
                thresholded,
                cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (3, 3)))
            log(thresholded, "absdiff_threshold")
            log(eroded, "absdiff_threshold_erode")

            motion = (cv2.countNonZero(eroded) > 0)

            # Visualisation: Highlight in red the areas where we detected motion
            if motion:
                with _numpy_from_sample(sample) as frame:
                    cv2.add(
                        frame,
                        numpy.multiply(
                            numpy.ones(frame.shape, dtype=numpy.uint8),
                            numpy.array((0, 0, 255), dtype=numpy.uint8),  # bgr
                            dtype=numpy.uint8),
                        mask=cv2.dilate(
                            thresholded,
                            cv2.getStructuringElement(
                                cv2.MORPH_ELLIPSE, (3, 3)),
                            iterations=1),
                        dst=frame)

            result = MotionResult(sample.get_buffer().pts, motion)
            debug("%s found: %s" % (
                "Motion" if motion else "No motion", str(result)))
            yield result

    def wait_for_match(self, image, timeout_secs=10, consecutive_matches=1,
                       match_parameters=None, region=Region.ALL):

        if match_parameters is None:
            match_parameters = MatchParameters()

        match_count = 0
        last_pos = Position(0, 0)
        image = _load_template(image)
        for res in self.detect_match(
                image, timeout_secs, match_parameters=match_parameters,
                region=region):
            if res.match and (match_count == 0 or res.position == last_pos):
                match_count += 1
            else:
                match_count = 0
            last_pos = res.position
            if match_count == consecutive_matches:
                debug("Matched " + image.friendly_name)
                return res

        raise MatchTimeout(res.frame, image.friendly_name, timeout_secs)  # pylint: disable=W0631,C0301

    def press_until_match(
            self,
            key,
            image,
            interval_secs=None,
            max_presses=None,
            match_parameters=None):

        if interval_secs is None:
            # Should this be float?
            interval_secs = get_config(
                "press_until_match", "interval_secs", type_=int)
        if max_presses is None:
            max_presses = get_config(
                "press_until_match", "max_presses", type_=int)

        if match_parameters is None:
            match_parameters = MatchParameters()

        i = 0

        while True:
            try:
                return self.wait_for_match(image, timeout_secs=interval_secs,
                                           match_parameters=match_parameters)
            except MatchTimeout:
                if i < max_presses:
                    self.press(key)
                    i += 1
                else:
                    raise

    def wait_for_motion(
            self, timeout_secs=10, consecutive_frames=None,
            noise_threshold=None, mask=None):

        if consecutive_frames is None:
            consecutive_frames = get_config('motion', 'consecutive_frames')

        consecutive_frames = str(consecutive_frames)
        if '/' in consecutive_frames:
            motion_frames = int(consecutive_frames.split('/')[0])
            considered_frames = int(consecutive_frames.split('/')[1])
        else:
            motion_frames = int(consecutive_frames)
            considered_frames = int(consecutive_frames)

        if motion_frames > considered_frames:
            raise ConfigurationError(
                "`motion_frames` exceeds `considered_frames`")

        debug("Waiting for %d out of %d frames with motion" % (
            motion_frames, considered_frames))

        matches = deque(maxlen=considered_frames)
        for res in self.detect_motion(timeout_secs, noise_threshold, mask):
            matches.append(res.motion)
            if matches.count(True) >= motion_frames:
                debug("Motion detected.")
                return res

        screenshot = self.get_frame()
        raise MotionTimeout(screenshot, mask, timeout_secs)

    def ocr(self, frame=None, region=Region.ALL,
            mode=OcrMode.PAGE_SEGMENTATION_WITHOUT_OSD,
            lang="eng", tesseract_config=None, tesseract_user_words=None,
            tesseract_user_patterns=None):

        if frame is None:
            frame = self._display.get_sample()

        if region is None:
            warnings.warn(
                "Passing region=None to ocr is deprecated since 0.21 and the "
                "meaning will change in a future version.  To OCR an entire "
                "video frame pass region=Region.ALL instead",
                DeprecationWarning, stacklevel=2)
            region = Region.ALL

        text, region = _tesseract(
            frame, region, mode, lang, tesseract_config,
            user_patterns=tesseract_user_patterns,
            user_words=tesseract_user_words)
        text = text.decode('utf-8').strip().translate(_ocr_transtab)
        debug(u"OCR in region %s read '%s'." % (region, text))
        return text

    def match_text(self, text, frame=None, region=Region.ALL,
                   mode=OcrMode.PAGE_SEGMENTATION_WITHOUT_OSD, lang="eng",
                   tesseract_config=None):

        import lxml.etree
        if frame is None:
            frame = self.get_frame()

        _config = dict(tesseract_config or {})
        _config['tessedit_create_hocr'] = 1
        if _tesseract_version() >= LooseVersion('3.04'):
            _config['tessedit_create_txt'] = 0

        ts = _get_frame_timestamp(frame)

        xml, region = _tesseract(frame, region, mode, lang, _config,
                                 user_words=text.split())
        if xml == '':
            return TextMatchResult(ts, False, None, frame, text)
        hocr = lxml.etree.fromstring(xml)
        p = _hocr_find_phrase(hocr, text.split())
        if p:
            # Find bounding box
            box = None
            for _, elem in p:
                box = _bounding_box(box, _hocr_elem_region(elem))
            # _tesseract crops to region and scales up by a factor of 3 so we
            # must undo this transformation here.
            box = Region.from_extents(
                region.x + box.x // 3, region.y + box.y // 3,
                region.x + box.right // 3, region.y + box.bottom // 3)
            return TextMatchResult(ts, True, box, frame, text)
        else:
            return TextMatchResult(ts, False, None, frame, text)

    def frames(self, timeout_secs=None):
        return self._display.frames(timeout_secs)

    def get_frame(self):
        with _numpy_from_sample(
                self._display.get_sample(), readonly=True) as frame:
            return frame.copy()

    def is_screen_black(self, frame=None, mask=None, threshold=None):
        if threshold is None:
            threshold = get_config('is_screen_black', 'threshold', type_=int)
        if mask:
            mask = _load_mask(mask)
        if frame is None:
            frame = self._display.get_sample()
        with _numpy_from_sample(frame, readonly=True) as f:
            greyframe = cv2.cvtColor(f, cv2.COLOR_BGR2GRAY)
        _, greyframe = cv2.threshold(
            greyframe, threshold, 255, cv2.THRESH_BINARY)
        _, maxVal, _, _ = cv2.minMaxLoc(greyframe, mask)
        if logging.get_debug_level() > 1:
            _log_image(frame, 'source', 'stbt-debug/is_screen_black')
            if mask is not None:
                _log_image(mask, 'mask', 'stbt-debug/is_screen_black')
                _log_image(numpy.bitwise_and(greyframe, mask),
                           'non-black-regions-after-masking',
                           'stbt-debug/is_screen_black')
            else:
                _log_image(greyframe, 'non-black-regions-after-masking',
                           'stbt-debug/is_screen_black')
        return maxVal == 0

# Utility functions
# ===========================================================================


def save_frame(image, filename):
    """Saves an OpenCV image to the specified file.

    Takes an image obtained from `get_frame` or from the `screenshot`
    property of `MatchTimeout` or `MotionTimeout`.
    """
    with _numpy_from_sample(image, readonly=True) as imagebuf:
        cv2.imwrite(filename, imagebuf)


def wait_until(callable_, timeout_secs=10, interval_secs=0):
    """Wait until a condition becomes true, or until a timeout.

    ``callable_`` is any python callable, such as a function or a lambda
    expression. It will be called repeatedly (with a delay of ``interval_secs``
    seconds between successive calls) until it succeeds (that is, it returns a
    truthy value) or until ``timeout_secs`` seconds have passed. In both cases,
    ``wait_until`` returns the value that ``callable_`` returns.

    After you send a remote-control signal to the system-under-test it usually
    takes a few frames to react, so a test script like this would probably
    fail::

        press("KEY_EPG")
        assert match("guide.png")

    Instead, use this::

        press("KEY_EPG")
        assert wait_until(lambda: match("guide.png"))

    Note that instead of the above `assert wait_until(...)` you could use
    `wait_for_match("guide.png")`. `wait_until` is a generic solution that
    also works with stbt's other functions, like `match_text` and
    `is_screen_black`.

    `wait_until` allows composing more complex conditions, such as::

        # Wait until something disappears:
        assert wait_until(lambda: not match("xyz.png"))

        # Assert that something doesn't appear within 10 seconds:
        assert not wait_until(lambda: match("xyz.png"))

        # Assert that two images are present at the same time:
        assert wait_until(lambda: match("a.png") and match("b.png"))

        # Wait but don't raise an exception:
        if not wait_until(lambda: match("xyz.png")):
            do_something_else()

    There are some drawbacks to using `assert` instead of `wait_for_match`:

    * The exception message won't contain the reason why the match failed
      (unless you specify it as a second parameter to `assert`, which is
      tedious and we don't expect you to do it), and
    * The exception won't have the offending video-frame attached (so the
      screenshot in the test-run artifacts will be a few frames later than the
      frame that actually caused the test to fail).

    We hope to solve both of the above drawbacks at some point in the future.

    ``wait_until`` was added in stb-tester v22.
    """
    expiry_time = time.time() + timeout_secs
    while True:
        e = callable_()
        if e:
            return e
        if time.time() > expiry_time:
            debug("wait_until timed out: %s" % _callable_description(callable_))
            return e
        time.sleep(interval_secs)


def _callable_description(callable_):
    """Helper to provide nicer debug output when `wait_until` fails.

    >>> _callable_description(wait_until)
    'wait_until'
    >>> _callable_description(
    ...     lambda: stbt.press("OK"))
    '    lambda: stbt.press("OK"))\\n'
    >>> _callable_description(functools.partial(int, base=2))
    'int'
    >>> _callable_description(functools.partial(functools.partial(int, base=2),
    ...                                         x='10'))
    'int'
    >>> class T(object):
    ...     def __call__(self): return True;
    >>> _callable_description(T())
    '<_stbt.core.T object at 0x...>'
    """
    if hasattr(callable_, "__name__"):
        name = callable_.__name__
        if name == "<lambda>":
            try:
                name = inspect.getsource(callable_)
            except IOError:
                pass
        return name
    elif isinstance(callable_, functools.partial):
        return _callable_description(callable_.func)
    else:
        return repr(callable_)


@contextmanager
def as_precondition(message):
    """Context manager that replaces test failures with test errors.

    Stb-tester's reports show test failures (that is, `UITestFailure` or
    `AssertionError` exceptions) as red results, and test errors (that is,
    unhandled exceptions of any other type) as yellow results. Note that
    `wait_for_match`, `wait_for_motion`, and similar functions raise a
    `UITestFailure` when they detect a failure. By running such functions
    inside an `as_precondition` context, any `UITestFailure` or
    `AssertionError` exceptions they raise will be caught, and a
    `PreconditionError` will be raised instead.

    When running a single testcase hundreds or thousands of times to reproduce
    an intermittent defect, it is helpful to mark unrelated failures as test
    errors (yellow) rather than test failures (red), so that you can focus on
    diagnosing the failures that are most likely to be the particular defect
    you are looking for. For more details see `Test failures vs. errors
    <http://stb-tester.com/preconditions>`_.

    :param str message:
        A description of the precondition. Word this positively: "Channels
        tuned", not "Failed to tune channels".

    :raises:
        `PreconditionError` if the wrapped code block raises a `UITestFailure`
        or `AssertionError`.

    Example::

        def test_that_the_on_screen_id_is_shown_after_booting():
            channel = 100

            with stbt.as_precondition("Tuned to channel %s" % channel):
                mainmenu.close_any_open_menu()
                channels.goto_channel(channel)
                power.cold_reboot()
                assert channels.is_on_channel(channel)

            stbt.wait_for_match("on-screen-id.png")

    """
    try:
        yield
    except (UITestFailure, AssertionError) as e:
        debug("stbt.as_precondition caught a %s exception and will "
              "re-raise it as PreconditionError.\nOriginal exception was:\n%s"
              % (type(e).__name__, traceback.format_exc(e)))
        exc = PreconditionError(message, e)
        if hasattr(e, 'screenshot'):
            exc.screenshot = e.screenshot  # pylint: disable=W0201
        raise exc


class UITestError(Exception):
    """The test script had an unrecoverable error."""
    pass


class UITestFailure(Exception):
    """The test failed because the system under test didn't behave as expected.

    Inherit from this if you need to define your own test-failure exceptions.
    """
    pass


class NoVideo(UITestFailure):
    """No video available from the source pipeline."""
    pass


class MatchTimeout(UITestFailure):
    """Exception raised by `wait_for_match`.

    * ``screenshot``: The last video frame that `wait_for_match` checked before
      timing out.
    * ``expected``: Filename of the image that was being searched for.
    * ``timeout_secs``: Number of seconds that the image was searched for.
    """
    def __init__(self, screenshot, expected, timeout_secs):
        super(MatchTimeout, self).__init__()
        self.screenshot = screenshot
        self.expected = expected
        self.timeout_secs = timeout_secs

    def __str__(self):
        return "Didn't find match for '%s' within %g seconds." % (
            self.expected, self.timeout_secs)


class MotionTimeout(UITestFailure):
    """Exception raised by `wait_for_motion`.

    * ``screenshot``: The last video frame that `wait_for_motion` checked before
      timing out.
    * ``mask``: Filename of the mask that was used, if any.
    * ``timeout_secs``: Number of seconds that motion was searched for.
    """
    def __init__(self, screenshot, mask, timeout_secs):
        super(MotionTimeout, self).__init__()
        self.screenshot = screenshot
        self.mask = mask
        self.timeout_secs = timeout_secs

    def __str__(self):
        return "Didn't find motion%s within %g seconds." % (
            " (with mask '%s')" % self.mask if self.mask else "",
            self.timeout_secs)


class PreconditionError(UITestError):
    """Exception raised by `as_precondition`."""
    def __init__(self, message, original_exception):
        super(PreconditionError, self).__init__()
        self.message = message
        self.original_exception = original_exception

    def __str__(self):
        return (
            "Didn't meet precondition '%s' (original exception was: %s)"
            % (self.message, self.original_exception))


# stbt-run initialisation and convenience functions
# (you will need these if writing your own version of stbt-run)
# ===========================================================================

def argparser():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--control',
        default=get_config('global', 'control'),
        help='The remote control to control the stb (default: %(default)s)')
    parser.add_argument(
        '--source-pipeline',
        default=get_config('global', 'source_pipeline'),
        help='A gstreamer pipeline to use for A/V input (default: '
             '%(default)s)')
    parser.add_argument(
        '--sink-pipeline',
        default=get_config('global', 'sink_pipeline'),
        help='A gstreamer pipeline to use for video output '
             '(default: %(default)s)')
    parser.add_argument(
        '--restart-source', action='store_true',
        default=(get_config('global', 'restart_source').lower() in
                 ("1", "yes", "true", "on")),
        help='Restart the GStreamer source pipeline when video loss is '
             'detected')

    logging.argparser_add_verbose_argument(parser)

    return parser


def _noneify_property_fn(fn):
    @functools.wraps(fn)
    def inner(self):
        if self:
            return fn(self)
        else:
            return None
    return inner


class _FrameObjectMeta(type):
    def __new__(mcs, name, parents, dct):
        for k, v in dct.iteritems():
            if isinstance(v, property):
                if v.fset is not None:
                    raise Exception(
                        "FrameObjects must be immutable but this property has "
                        "a setter")
                f = v.fget
                if k != 'is_visible' and not k.startswith('_'):
                    f = _noneify_property_fn(f)
                dct[k] = property(f)

        return super(_FrameObjectMeta, mcs).__new__(mcs, name, parents, dct)

    def __init__(cls, name, parents, dct):
        property_names = sorted([
            p for p in dir(cls)
            if isinstance(getattr(cls, p), property)])
        assert 'is_visible' in property_names
        cls._FrameObject__attrs = ["is_visible"] + sorted(
            x for x in property_names
            if x != "is_visible" and not x.startswith('_'))
        super(_FrameObjectMeta, cls).__init__(name, parents, dct)


class FrameObject(object):
    __metaclass__ = _FrameObjectMeta

    def __init__(self, frame):
        if frame is None:
            raise ValueError("FrameObject: frame must not be None")
        self._frame = frame

    def __repr__(self):
        args = ", ".join(("%s=%r" % x) for x in self._iter_attrs())
        return "%s(%s)" % (self.__class__.__name__, args)

    def _iter_attrs(self):
        if self:
            # pylint: disable=protected-access,no-member
{'fraction_non_alphanumeric': 0.10316134280123991, 'fraction_numerical': 0.00817816079319366, 'mean_word_length': 2.2909492114021126, 'pattern_counts': {'":': 10, '<': 33, '<?xml version=': 0, '>': 239, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2300959', 'n_tokens_mistral': 26834, 'n_tokens_neox': 25775, 'n_words': 7879}