<commit_before>import argparse
import ConfigParser
import os
import re
import socket
import sys


class ArgvHider:
    """ For use with 'with' statement:  Unsets argv and resets it.

    This is used because otherwise gst-python will exit if '-h', '--help', '-v'
    or '--version' command line arguments are given.
    """
    def __enter__(self):
        self.argv = sys.argv[:]
        del sys.argv[1:]

    def __exit__(self, type, value, traceback):
        sys.argv = self.argv

with ArgvHider():
    import pygst  # gstreamer
    pygst.require("0.10")
    import gst
    import gtk  # for main loop


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

def press(*args, **keywords):
    return control.press(*args, **keywords)


def wait_for_match(*args, **keywords):
    if 'directory' not in keywords:
        keywords['directory'] = _caller_dir()
    return display.wait_for_match(*args, **keywords)


def press_until_match(key, image, interval_secs=3, max_presses=10):
    i = 0
    while True:
        try:
            wait_for_match(image, directory=_caller_dir(),
                           timeout_secs=interval_secs)
            return
        except MatchTimeout:
            if i < max_presses:
                press(key)
                i += 1
            else:
                raise


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

def argparser():
    parser = argparse.ArgumentParser(
        prog='stbt run', description='Run an stb-tester test script')
    parser.add_argument('--control',
        help='The remote control to control the stb (default: %(default)s)')
    parser.add_argument('--source-pipeline',
        help='A gstreamer pipeline to use for A/V input (default: '
             '%(default)s)')
    parser.add_argument('--sink-pipeline',
        help='A gstreamer pipeline to use for video output '
             '(default: %(default)s)')
    parser.set_defaults(**load_defaults('run'))
    return parser


def init_run(gst_source_pipeline, gst_sink_pipeline, control_uri):
    global display, control
    display = Display(gst_source_pipeline, gst_sink_pipeline)
    control = uri_to_remote(control_uri, display)


def teardown_run():
    display.teardown()


def save_frame(buf, filename):
    '''Save a gstreamer buffer to the specified file in png format.'''
    pipeline = gst.parse_launch(" ! ".join([
                'appsrc name="src" caps="%s"' % buf.get_caps(),
                'ffmpegcolorspace',
                'pngenc',
                'filesink location="%s"' % filename,
                ]))
    src = pipeline.get_by_name("src")
    # This is actually a (synchronous) method call to push-buffer:
    src.emit('push-buffer', buf)
    src.emit('end-of-stream')
    pipeline.set_state(gst.STATE_PLAYING)
    msg = pipeline.get_bus().poll(
        gst.MESSAGE_ERROR | gst.MESSAGE_EOS, 25 * gst.SECOND)
    pipeline.set_state(gst.STATE_NULL)
    if msg.type == gst.MESSAGE_ERROR:
        (e, debug) = msg.parse_error()
        raise RuntimeError(e.message)


class UITestFailure(Exception):
    pass


class MatchTimeout(UITestFailure):
    def __init__(self, screenshot, expected, timeout_secs):
        self.screenshot = screenshot
        self.expected = expected
        self.timeout_secs = timeout_secs


class ConfigurationError(Exception):
    pass


# Internal
#===========================================================================

class Display:
    def __init__(self, gst_source_pipeline, gst_sink_pipeline):
        imageprocessing = " ! ".join([
                # Buffer the video stream, dropping frames if downstream
                # processors aren't fast enough:
                "queue leaky=2",
                # Convert to a colorspace that templatematch can handle:
                "ffmpegcolorspace",
                # OpenCV image-processing library:
                "templatematch name=templatematch method=1",
                ])
        xvideo = " ! ".join([
                # Convert to a colorspace that xvimagesink can handle:
                "ffmpegcolorspace",
                gst_sink_pipeline,
                ])
        screenshot = ("appsink name=screenshot max-buffers=1 drop=true "
                      "sync=false")
        pipe = " ".join([
                gst_source_pipeline,
                "!", imageprocessing,
                "! tee name=t",
                "! queue leaky=2 !", screenshot,
                "t. ! queue leaky=2 !", xvideo
                ])

        # Gstreamer loads plugin libraries on demand, when elements that need
        # those libraries are first mentioned. There is a bug in gst-opencv
        # where it erroneously claims to provide appsink, preventing the
        # loading of the real appsink -- so we load it first.
        # TODO: Fix gst-opencv so that it doesn't prevent appsink from being
        #       loaded.
        gst.parse_launch("appsink")

        self.pipeline = gst.parse_launch(pipe)
        self.templatematch = self.pipeline.get_by_name("templatematch")
        self.screenshot = self.pipeline.get_by_name("screenshot")
        self.bus = self.pipeline.get_bus()
        self.bus.connect("message::error", self.on_error)
        self.bus.connect("message::warning", self.on_warning)
        self.bus.add_signal_watch()
        self.pipeline.set_state(gst.STATE_PLAYING)

    def wait_for_match(self, image, directory,
                       timeout_secs=10, certainty=0.98, consecutive_matches=3):
        """Wait for a stable match of `image` in the source video stream.

        "Stable" means 3 consecutive frames with a match certainty > 98% at the
        same x,y position.

        "timeout_secs" is in seconds elapsed, as reported by the video stream.
        """

        if os.path.isabs(image):
            template = image
        else:
            # Image is relative to the script's own directory
            template = os.path.abspath(os.path.join(directory, image))
            if not os.path.isfile(template):
                # Fall back to image from cwd, for convenience of the selftests
                template = os.path.abspath(image)

        self.templatematch.props.template = template
        self.match_count, self.last_x, self.last_y = 0, 0, 0
        self.timeout_secs = timeout_secs
        self.start_timestamp = None
        self.certainty = certainty
        self.consecutive_matches = consecutive_matches

        debug("Searching for " + template)
        self.bus.connect("message::element", self.bus_message)
        gtk.main()
        if self.match_count == self.consecutive_matches:
            debug("MATCHED " + template)
            return
        else:
            buf = self.screenshot.get_property("last-buffer")
            raise MatchTimeout(buf, template, timeout_secs)

    def on_error(self, bus, message):
        assert message.type == gst.MESSAGE_ERROR
        err, dbg = message.parse_error()
        sys.stderr.write("Error: %s: %s\n%s\n" % (err, err.message, dbg))
        sys.exit(1)

    def on_warning(self, bus, message):
        assert message.type == gst.MESSAGE_WARNING
        err, dbg = message.parse_warning()
        sys.stderr.write("Warning: %s: %s\n%s\n" % (err, err.message, dbg))
        if (message.src == self.templatematch and
                err.message == "OpenCV failed to load template image"):
            sys.stderr.write("Error: %s\n" % err.message)
            sys.exit(1)

    def bus_message(self, bus, message):
        st = message.structure
        if st.get_name() == "template_match":

            buf = self.screenshot.get_property("last-buffer")
            if self.start_timestamp == None:
                self.start_timestamp = buf.timestamp

            certainty = st["result"]
            debug("Match %d found at %d,%d (last: %d,%d) with dimensions "
                  "%dx%d. Certainty: %d%%. Timestamp: %d."
                  % (self.match_count,
                     st["x"], st["y"], self.last_x, self.last_y,
                     st["width"], st["height"],
                     certainty * 100.0, buf.timestamp))

            if certainty > self.certainty and (
                    self.match_count == 0 or
                    (st["x"], st["y"]) == (self.last_x, self.last_y)):
                self.match_count += 1
            else:
                self.match_count = 0
            self.last_x, self.last_y = st["x"], st["y"]

            timed_out = (buf.timestamp - self.start_timestamp >
                         self.timeout_secs * 1000000000)
            if self.match_count == self.consecutive_matches or (
                    timed_out and self.match_count == 0):
                self.templatematch.props.template = None
                self.bus.disconnect_by_func(self.bus_message)
                gtk.main_quit()

    def teardown(self):
        if self.pipeline:
            self.pipeline.set_state(gst.STATE_NULL)


def uri_to_remote(uri, display):
    if uri.lower() == 'none':
        return NullRemote()
    if uri.lower() == 'test':
        return VideoTestSrcControl(display)
    vr = re.match(r'vr:(?P<hostname>[^:]*)(:(?P<port>\d+))?', uri)
    if vr:
        d = vr.groupdict()
        return VirtualRemote(d['hostname'], int(d['port'] or 2033))
    lirc = re.match(r'lirc:(?P<lircd_socket>[^:]*):(?P<control_name>.*)', uri)
    if lirc:
        d = lirc.groupdict()
        return LircRemote(d['lircd_socket'] or '/var/run/lirc/lircd',
                          d['control_name'])
    raise ConfigurationError('Invalid remote control URI: "%s"' % uri)


class NullRemote:
    def press(self, key):
        debug('NullRemote: Ignoring request to press "%s"' % key)


class VideoTestSrcControl:
    """Remote control used by selftests.

    Changes the videotestsrc image to the specified pattern ("0" to "20").
    See `gst-inspect videotestsrc`.
    """
    def __init__(self, display):
        self.videosrc = display.pipeline.get_by_name("videotestsrc0")
        if not self.videosrc:
            raise ConfigurationError('The "test" control can only be used'
                                     'with source-pipeline = "videotestsrc"')

    def press(self, key):
        if key not in [str(x) for x in range(21)]:
            raise UITestFailure('Key "%s" not valid for the "test" control'
                                ' (only "0" to "20" allowed)' % key)
        self.videosrc.props.pattern = int(key)
        debug("Pressed " + key)


class VirtualRemote:
    """Send a key-press to a set-top box running a VirtualRemote listener.

        control = VirtualRemote("192.168.0.123")
        control.press("MENU")
    """
    def __init__(self, hostname, port):
        self.stb = hostname
        self.port = port

    def press(self, key):
        s = socket.socket()
        s.connect((self.stb, self.port))
        s.send("D\t%s\n\0U\t%s\n\0" % (key, key))  # send key Down, then key Up
        debug("Pressed " + key)


class LircRemote:
    """Send a key-press via a LIRC-enabled infrared blaster.

    See http://www.lirc.org/html/technical.html#applications
    """
    def __init__(self, lircd_socket, control_name):
        self.lircd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.lircd.connect(lircd_socket)
        self.control_name = control_name

    def press(self, key):
        self.lircd.send("SEND_ONCE %s %s\n" % (self.control_name, key))
        debug("Pressed " + key)


def uri_to_remote_recorder(uri):
    vr = re.match(r'vr:(?P<hostname>[^:]*)(:(?P<port>\d+))?', uri)<commit_msg>VirtualRemote: Avoid opening a new socket on each key-press<commit_after>import argparse
import ConfigParser
import os
import re
import socket
import sys


class ArgvHider:
    """ For use with 'with' statement:  Unsets argv and resets it.

    This is used because otherwise gst-python will exit if '-h', '--help', '-v'
    or '--version' command line arguments are given.
    """
    def __enter__(self):
        self.argv = sys.argv[:]
        del sys.argv[1:]

    def __exit__(self, type, value, traceback):
        sys.argv = self.argv

with ArgvHider():
    import pygst  # gstreamer
    pygst.require("0.10")
    import gst
    import gtk  # for main loop


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

def press(*args, **keywords):
    return control.press(*args, **keywords)


def wait_for_match(*args, **keywords):
    if 'directory' not in keywords:
        keywords['directory'] = _caller_dir()
    return display.wait_for_match(*args, **keywords)


def press_until_match(key, image, interval_secs=3, max_presses=10):
    i = 0
    while True:
        try:
            wait_for_match(image, directory=_caller_dir(),
                           timeout_secs=interval_secs)
            return
        except MatchTimeout:
            if i < max_presses:
                press(key)
                i += 1
            else:
                raise


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

def argparser():
    parser = argparse.ArgumentParser(
        prog='stbt run', description='Run an stb-tester test script')
    parser.add_argument('--control',
        help='The remote control to control the stb (default: %(default)s)')
    parser.add_argument('--source-pipeline',
        help='A gstreamer pipeline to use for A/V input (default: '
             '%(default)s)')
    parser.add_argument('--sink-pipeline',
        help='A gstreamer pipeline to use for video output '
             '(default: %(default)s)')
    parser.set_defaults(**load_defaults('run'))
    return parser


def init_run(gst_source_pipeline, gst_sink_pipeline, control_uri):
    global display, control
    display = Display(gst_source_pipeline, gst_sink_pipeline)
    control = uri_to_remote(control_uri, display)


def teardown_run():
    display.teardown()


def save_frame(buf, filename):
    '''Save a gstreamer buffer to the specified file in png format.'''
    pipeline = gst.parse_launch(" ! ".join([
                'appsrc name="src" caps="%s"' % buf.get_caps(),
                'ffmpegcolorspace',
                'pngenc',
                'filesink location="%s"' % filename,
                ]))
    src = pipeline.get_by_name("src")
    # This is actually a (synchronous) method call to push-buffer:
    src.emit('push-buffer', buf)
    src.emit('end-of-stream')
    pipeline.set_state(gst.STATE_PLAYING)
    msg = pipeline.get_bus().poll(
        gst.MESSAGE_ERROR | gst.MESSAGE_EOS, 25 * gst.SECOND)
    pipeline.set_state(gst.STATE_NULL)
    if msg.type == gst.MESSAGE_ERROR:
        (e, debug) = msg.parse_error()
        raise RuntimeError(e.message)


class UITestFailure(Exception):
    pass


class MatchTimeout(UITestFailure):
    def __init__(self, screenshot, expected, timeout_secs):
        self.screenshot = screenshot
        self.expected = expected
        self.timeout_secs = timeout_secs


class ConfigurationError(Exception):
    pass


# Internal
#===========================================================================

class Display:
    def __init__(self, gst_source_pipeline, gst_sink_pipeline):
        imageprocessing = " ! ".join([
                # Buffer the video stream, dropping frames if downstream
                # processors aren't fast enough:
                "queue leaky=2",
                # Convert to a colorspace that templatematch can handle:
                "ffmpegcolorspace",
                # OpenCV image-processing library:
                "templatematch name=templatematch method=1",
                ])
        xvideo = " ! ".join([
                # Convert to a colorspace that xvimagesink can handle:
                "ffmpegcolorspace",
                gst_sink_pipeline,
                ])
        screenshot = ("appsink name=screenshot max-buffers=1 drop=true "
                      "sync=false")
        pipe = " ".join([
                gst_source_pipeline,
                "!", imageprocessing,
                "! tee name=t",
                "! queue leaky=2 !", screenshot,
                "t. ! queue leaky=2 !", xvideo
                ])

        # Gstreamer loads plugin libraries on demand, when elements that need
        # those libraries are first mentioned. There is a bug in gst-opencv
        # where it erroneously claims to provide appsink, preventing the
        # loading of the real appsink -- so we load it first.
        # TODO: Fix gst-opencv so that it doesn't prevent appsink from being
        #       loaded.
        gst.parse_launch("appsink")

        self.pipeline = gst.parse_launch(pipe)
        self.templatematch = self.pipeline.get_by_name("templatematch")
        self.screenshot = self.pipeline.get_by_name("screenshot")
        self.bus = self.pipeline.get_bus()
        self.bus.connect("message::error", self.on_error)
        self.bus.connect("message::warning", self.on_warning)
        self.bus.add_signal_watch()
        self.pipeline.set_state(gst.STATE_PLAYING)

    def wait_for_match(self, image, directory,
                       timeout_secs=10, certainty=0.98, consecutive_matches=3):
        """Wait for a stable match of `image` in the source video stream.

        "Stable" means 3 consecutive frames with a match certainty > 98% at the
        same x,y position.

        "timeout_secs" is in seconds elapsed, as reported by the video stream.
        """

        if os.path.isabs(image):
            template = image
        else:
            # Image is relative to the script's own directory
            template = os.path.abspath(os.path.join(directory, image))
            if not os.path.isfile(template):
                # Fall back to image from cwd, for convenience of the selftests
                template = os.path.abspath(image)

        self.templatematch.props.template = template
        self.match_count, self.last_x, self.last_y = 0, 0, 0
        self.timeout_secs = timeout_secs
        self.start_timestamp = None
        self.certainty = certainty
        self.consecutive_matches = consecutive_matches

        debug("Searching for " + template)
        self.bus.connect("message::element", self.bus_message)
        gtk.main()
        if self.match_count == self.consecutive_matches:
            debug("MATCHED " + template)
            return
        else:
            buf = self.screenshot.get_property("last-buffer")
            raise MatchTimeout(buf, template, timeout_secs)

    def on_error(self, bus, message):
        assert message.type == gst.MESSAGE_ERROR
        err, dbg = message.parse_error()
        sys.stderr.write("Error: %s: %s\n%s\n" % (err, err.message, dbg))
        sys.exit(1)

    def on_warning(self, bus, message):
        assert message.type == gst.MESSAGE_WARNING
        err, dbg = message.parse_warning()
        sys.stderr.write("Warning: %s: %s\n%s\n" % (err, err.message, dbg))
        if (message.src == self.templatematch and
                err.message == "OpenCV failed to load template image"):
            sys.stderr.write("Error: %s\n" % err.message)
            sys.exit(1)

    def bus_message(self, bus, message):
        st = message.structure
        if st.get_name() == "template_match":

            buf = self.screenshot.get_property("last-buffer")
            if self.start_timestamp == None:
                self.start_timestamp = buf.timestamp

            certainty = st["result"]
            debug("Match %d found at %d,%d (last: %d,%d) with dimensions "
                  "%dx%d. Certainty: %d%%. Timestamp: %d."
                  % (self.match_count,
                     st["x"], st["y"], self.last_x, self.last_y,
                     st["width"], st["height"],
                     certainty * 100.0, buf.timestamp))

            if certainty > self.certainty and (
                    self.match_count == 0 or
                    (st["x"], st["y"]) == (self.last_x, self.last_y)):
                self.match_count += 1
            else:
                self.match_count = 0
            self.last_x, self.last_y = st["x"], st["y"]

            timed_out = (buf.timestamp - self.start_timestamp >
                         self.timeout_secs * 1000000000)
            if self.match_count == self.consecutive_matches or (
                    timed_out and self.match_count == 0):
                self.templatematch.props.template = None
                self.bus.disconnect_by_func(self.bus_message)
                gtk.main_quit()

    def teardown(self):
        if self.pipeline:
            self.pipeline.set_state(gst.STATE_NULL)


def uri_to_remote(uri, display):
    if uri.lower() == 'none':
        return NullRemote()
    if uri.lower() == 'test':
        return VideoTestSrcControl(display)
    vr = re.match(r'vr:(?P<hostname>[^:]*)(:(?P<port>\d+))?', uri)
    if vr:
        d = vr.groupdict()
        return VirtualRemote(d['hostname'], int(d['port'] or 2033))
    lirc = re.match(r'lirc:(?P<lircd_socket>[^:]*):(?P<control_name>.*)', uri)
    if lirc:
        d = lirc.groupdict()
        return LircRemote(d['lircd_socket'] or '/var/run/lirc/lircd',
                          d['control_name'])
    raise ConfigurationError('Invalid remote control URI: "%s"' % uri)


class NullRemote:
    def press(self, key):
        debug('NullRemote: Ignoring request to press "%s"' % key)


class VideoTestSrcControl:
    """Remote control used by selftests.

    Changes the videotestsrc image to the specified pattern ("0" to "20").
    See `gst-inspect videotestsrc`.
    """
    def __init__(self, display):
        self.videosrc = display.pipeline.get_by_name("videotestsrc0")
        if not self.videosrc:
            raise ConfigurationError('The "test" control can only be used'
                                     'with source-pipeline = "videotestsrc"')

    def press(self, key):
        if key not in [str(x) for x in range(21)]:
            raise UITestFailure('Key "%s" not valid for the "test" control'
                                ' (only "0" to "20" allowed)' % key)
        self.videosrc.props.pattern = int(key)
        debug("Pressed " + key)


class VirtualRemote:
    """Send a key-press to a set-top box running a VirtualRemote listener.

        control = VirtualRemote("192.168.0.123")
        control.press("MENU")
    """
    def __init__(self, hostname, port):
        self.s = socket.socket()
        self.s.connect((hostname, port))

    def press(self, key):
        self.s.send("D\t%s\n\0U\t%s\n\0" % (key, key))  # key Down, then key Up
        debug("Pressed " + key)


class LircRemote:
    """Send a key-press via a LIRC-enabled infrared blaster.

    See http://www.lirc.org/html/technical.html#applications
    """
    def __init__(self, lircd_socket, control_name):
        self.lircd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.lircd.connect(lircd_socket)
        self.control_name = control_name

    def press(self, key):
        self.lircd.send("SEND_ONCE %s %s\n" % (self.control_name, key))
        debug("Pressed " + key)


def uri_to_remote_recorder(uri):
    vr = re.match(r'vr:(?P<hostname>[^:]*)(:(?P<port>\d+))?', uri)
{'fraction_non_alphanumeric': 0.13375170765027322, 'fraction_numerical': 0.0062329234972677595, 'mean_word_length': 2.274391948560246, 'pattern_counts': {'":': 2, '<': 17, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6968323', 'n_tokens_mistral': 7241, 'n_tokens_neox': 6880, 'n_words': 1714}