<commit_before>from __future__ import absolute_import

import os
import re
import socket
import struct
import subprocess
import sys
import time
from contextlib import contextmanager
from distutils.spawn import find_executable

from . import irnetbox, utils
from .config import ConfigurationError
from .logging import debug, scoped_debug_level

__all__ = ['uri_to_control', 'uri_to_control_recorder']

try:
    from .control_gpl import controls as gpl_controls
except ImportError:
    gpl_controls = None


# pylint: disable=abstract-method

class RemoteControl(object):
    """Base class for remote-control implementations."""

    def press(self, key):
        raise NotImplementedError(
            "%s: 'press' is not implemented" % self.__class__.__name__)

    def keydown(self, key):
        raise NotImplementedError(
            "%s: 'keydown' is not implemented" % self.__class__.__name__)

    def keyup(self, key):
        raise NotImplementedError(
            "%s: 'keyup' is not implemented" % self.__class__.__name__)


class UnknownKeyError(Exception):
    pass


def uri_to_control(uri, display=None):
    controls = [
        (r'adb(:(?P<address>.*))?', new_adb_device),
        (r'error(:(?P<message>.*))?', ErrorControl),
        (r'file(:(?P<filename>[^,]+))?', FileControl),
        (r'''irnetbox:
             (?P<hostname>[^:]+)
             (:(?P<port>\d+))?
             :(?P<output>\d+)
             :(?P<config>[^:]+)''', IRNetBoxControl),
        (r'lirc(:(?P<hostname>[^:/]+))?:(?P<port>\d+):(?P<control_name>.+)',
         new_tcp_lirc_control),
        (r'lirc:(?P<lircd_socket>[^:]+)?:(?P<control_name>.+)',
         new_local_lirc_control),
        (r'none', NullControl),
        (r'roku:(?P<hostname>[^:]+)', RokuHttpControl),
        (r'samsung:(?P<hostname>[^:/]+)(:(?P<port>\d+))?',
         _new_samsung_tcp_control),
        (r'test', lambda: VideoTestSrcControl(display)),
        (r'x11:(?P<display>[^,]+)?(,(?P<mapping>.+)?)?', X11Control),
        (r'rfb:(?P<hostname>[^:/]+)(:(?P<port>\d+))?', RemoteFrameBuffer),
    ]
    if gpl_controls is not None:
        controls += gpl_controls

    for regex, factory in controls:
        m = re.match(regex, uri, re.VERBOSE | re.IGNORECASE)
        if m:
            return factory(**m.groupdict())
    raise ConfigurationError('Invalid remote control URI: "%s"' % uri)


def uri_to_control_recorder(uri):
    controls = [
        ('file://(?P<filename>.+)', file_control_recorder),
        (r'lirc(:(?P<hostname>[^:/]+))?:(?P<port>\d+):(?P<control_name>.+)',
         lirc_control_listen_tcp),
        (r'lirc:(?P<lircd_socket>[^:]+)?:(?P<control_name>.+)',
         lirc_control_listen),
        (r'stbt-control(:(?P<keymap_file>.+))?', stbt_control_listen),
    ]

    for regex, factory in controls:
        m = re.match(regex, uri)
        if m:
            return factory(**m.groupdict())
    raise ConfigurationError('Invalid remote control recorder URI: "%s"' % uri)


def new_adb_device(address):
    from stbt.android import AdbDevice
    tcpip = bool(re.match(r"\d+\.\d+\.\d+\.\d+", address))
    return AdbDevice(adb_device=address, tcpip=tcpip)


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

    def keydown(self, key):
        debug('NullControl: Ignoring request to hold "%s"' % key)

    def keyup(self, key):
        debug('NullControl: Ignoring request to release "%s"' % key)


class ErrorControl(RemoteControl):
    def __init__(self, message):
        if message is None:
            message = "No remote control configured"
        self.message = message

    def press(self, key):  # pylint:disable=unused-argument
        raise RuntimeError(self.message)

    def keydown(self, key):
        raise RuntimeError(self.message)

    def keyup(self, key):
        raise RuntimeError(self.message)


class FileControl(RemoteControl):
    """Writes keypress events to file.  Mostly useful for testing.  Defaults to
    writing to stdout.
    """
    def __init__(self, filename):
        if filename is None:
            self.outfile = sys.stdout
        else:
            self.outfile = open(filename, 'w+')

    def press(self, key):
        self.outfile.write(key + '\n')
        self.outfile.flush()

    def keydown(self, key):
        self.outfile.write("Holding %s\n" % key)
        self.outfile.flush()

    def keyup(self, key):
        self.outfile.write("Released %s\n" % key)
        self.outfile.flush()


class VideoTestSrcControl(RemoteControl):
    """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.display = display

    @property
    def videosrc(self):
        videosrc = self.display.source_pipeline.get_by_name("videotestsrc0")
        if not videosrc:
            raise ConfigurationError('The "test" control can only be used '
                                     'with source-pipeline = "videotestsrc"')
        return videosrc

    def press(self, key):
        if key not in [
                0, "smpte",
                1, "snow",
                2, "black",
                3, "white",
                4, "red",
                5, "green",
                6, "blue",
                7, "checkers-1",
                8, "checkers-2",
                9, "checkers-4",
                10, "checkers-8",
                11, "circular",
                12, "blink",
                13, "smpte75",
                14, "zone-plate",
                15, "gamut",
                16, "chroma-zone-plate",
                17, "solid-color",
                18, "ball",
                19, "smpte100",
                20, "bar"]:
            raise RuntimeError(
                'Key "%s" not valid for the "test" control' % key)
        self.videosrc.props.pattern = key
        debug("Pressed %s" % key)


def _find_file(path, root=os.path.dirname(os.path.abspath(__file__))):
    return os.path.join(root, path)


class LircControl(RemoteControl):
    """Send a key-press via a LIRC-enabled infrared blaster.

    See http://www.lirc.org/html/technical.html#applications
    """

    def __init__(self, control_name, connect_fn):
        self.control_name = control_name
        self._connect = connect_fn

    def press(self, key):
        s = self._connect()
        command = "SEND_ONCE %s %s" % (self.control_name, key)
        s.sendall(command + "\n")
        _read_lircd_reply(s, command)
        debug("Pressed " + key)

    def keydown(self, key):
        s = self._connect()
        command = "SEND_START %s %s" % (self.control_name, key)
        s.sendall(command + "\n")
        _read_lircd_reply(s, command)
        debug("Holding " + key)

    def keyup(self, key):
        s = self._connect()
        command = "SEND_STOP %s %s" % (self.control_name, key)
        s.sendall(command + "\n")
        _read_lircd_reply(s, command)
        debug("Released " + key)


def _read_lircd_reply(stream, command):
    """Waits for lircd reply and checks if a LIRC send command was successful.

    Waits for a reply message from lircd (called "reply packet" in the LIRC
    reference) for the specified command; raises exception if it times out or
    the reply contains an error message.

    The structure of a lircd reply message for a SEND_ONCE|SEND_START|SEND_STOP
    command is the following:

    BEGIN
    <command>
    (SUCCESS|ERROR)
    [DATA
    <number-of-data-lines>
    <error-message>]
    END

    See: http://www.lirc.org/html/technical.html#applications
    """
    reply = []
    try:
        for line in read_records(stream, "\n"):
            if line == "BEGIN":
                reply = []
            reply.append(line)
            if line == "END" and reply[1] == command:
                break
    except socket.timeout:
        raise RuntimeError(
            "Timed out: No reply from LIRC remote control within %d seconds"
            % stream.gettimeout())
    if "SUCCESS" not in reply:
        if "ERROR" in reply and len(reply) >= 6 and reply[3] == "DATA":
            try:
                num_data_lines = int(reply[4])
                raise RuntimeError("LIRC remote control returned error: %s"
                                   % " ".join(reply[5:5 + num_data_lines]))
            except ValueError:
                pass
        raise RuntimeError("LIRC remote control returned unknown error")

DEFAULT_LIRCD_SOCKET = '/var/run/lirc/lircd'


def new_local_lirc_control(lircd_socket, control_name):
    if lircd_socket is None:
        lircd_socket = DEFAULT_LIRCD_SOCKET

    def _connect():
        try:
            s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            s.settimeout(3)
            s.connect(lircd_socket)
            return s
        except socket.error as e:
            e.args = (("Failed to connect to Lirc socket %s: %s" % (
                lircd_socket, e)),)
            e.strerror = e.args[0]
            raise

    # Connect once so that the test fails immediately if Lirc isn't running
    # (instead of failing at the first `press` in the script).
    debug("LircControl: Connecting to %s" % lircd_socket)
    _connect()
    debug("LircControl: Connected to %s" % lircd_socket)

    return LircControl(control_name, _connect)


def new_tcp_lirc_control(control_name, hostname=None, port=None):
    """Send a key-press via a LIRC-enabled device through a LIRC TCP listener.

        control = new_tcp_lirc_control("localhost", "8765", "humax")
        control.press("MENU")
    """
    if hostname is None:
        hostname = 'localhost'
    if port is None:
        port = 8765

    port = int(port)

    def _connect():
        return _connect_tcp_socket(hostname, port)

    # Connect once so that the test fails immediately if Lirc isn't running
    # (instead of failing at the first `press` in the script).
    debug("TCPLircControl: Connecting to %s:%d" % (hostname, port))
    _connect()
    debug("TCPLircControl: Connected to %s:%d" % (hostname, port))

    return LircControl(control_name, _connect)


class RemoteFrameBuffer(RemoteControl):
    """Send a key-press to a set-top box running a VNC Remote Frame Buffer
        protocol.
        Expected key press input:
            <KEY_LABEL>

        control = RemoteFrameBuffer("192.168.0.123")
        control.press("KEY_MENU")
    """

    # Map our recommended keynames (from linux input-event-codes.h) to the
    # equivalent RFB keyname.
    _KEYNAMES = {
        'KEY_BACK': 0xE002,
        'KEY_BLUE': 0xE203,
        'KEY_CHANNELDOWN': 0xE007,
        'KEY_CHANNELUP': 0xE006,
        'KEY_DOWN': 0xE101,
        'KEY_ELPS': 0xEF00,
        'KEY_FASTFORWARD': 0xE405,
        'KEY_GREEN': 0xE201,
        'KEY_GUIDE': 0xE00B,
        'KEY_HELP': 0xE00A,
        'KEY_HOME': 0xE015,
        'KEY_INFO': 0xE00E,
        'KEY_INPUTSELECT': 0xE010,
        'KEY_INTERACT': 0xE008,
        'KEY_0': 0xE300,
        'KEY_1': 0xE301,
        'KEY_2': 0xE302,
        'KEY_3': 0xE303,
        'KEY_4': 0xE304,
        'KEY_5': 0xE305,
        'KEY_6': 0xE306,
        'KEY_7': 0xE307,
        'KEY_8': 0xE308,
        'KEY_9': 0xE309,
        'KEY_LEFT': 0xE102,
        'KEY_MENU': 0xE00A,
        'KEY_MUTE': 0xE005,
        'KEY_MYTV': 0xE009,
        'KEY_PAUSE': 0xE401,
        'KEY_PLAY': 0xE400,
        'KEY_PLAYPAUSE': 0xE40A,
        'KEY_POWER': 0xE000,
        'KEY_PRIMAFILA': 0xEF00,
        'KEY_RECORD': 0xE403,
        'KEY_RED': 0xE200,
        'KEY_REWIND': 0xE407,
        'KEY_RIGHT': 0xE103,
        'KEY_SEARCH': 0xEF03,
        'KEY_SELECT': 0xE001,
        'KEY_SKY': 0xEF01,
        'KEY_STOP': 0xE402,
        'KEY_TEXT': 0xE00F,
        'KEY_UP': 0xE100,
        'KEY_VOLUMEDOWN': 0xE004,
        'KEY_VOLUMEUP': 0xE003,
        'KEY_YELLOW': 0xE202
    }

    def __init__(self, hostname, port=None):
        self.hostname = hostname
        self.port = int(port or 5900)
        self.timeout = 3
        self.socket = None

    def press(self, key):
        self._connect_socket()
        self._handshake()
        self._press_down(key)
        self._release(key)
        self._close()

    def _connect_socket(self):
        self.socket = socket.socket()
        s = self.socket
        if self.timeout:
            s.settimeout(self.timeout)
        s.connect((self.hostname, self.port))
        debug(
            "RemoteFrameBuffer: connected to %s:%d"
            % (self.hostname, self.port))

    def _handshake(self):
        s = self.socket
        prot_info = s.recv(20)
        if prot_info != b'RFB 003.008\n':
            raise socket.error("wrong RFB protocol info")
        s.send(b"RFB 003.003\n")
        s.recv(4)
        s.send(b'\0')
        s.recv(24)
        debug("RemoteFrameBuffer: handshake completed")

    def _press_down(self, key):
        key_code = self._get_key_code(key)
        self.socket.send(struct.pack('!BBxxI', 4, 1, key_code))
        debug(
            "RemoteFrameBuffer: pressed down (0x%04x)"
            % key_code)

    def _release(self, key):
        key_code = self._get_key_code(key)
        self.socket.send(struct.pack('!BBxxI', 4, 0, key_code))
        debug("RemoteFrameBuffer: release (0x%04x)" % key_code)

    def _close(self):
        self.socket.shutdown(socket.SHUT_RDWR)
        self.socket.close()
        debug("RemoteFrameBuffer: socket connection closed")

    def _get_key_code(self, key):
        key_code = self._KEYNAMES.get(key, key)
        return key_code


class IRNetBoxControl(RemoteControl):
    """Send a key-press via the network-controlled RedRat IRNetBox IR emitter.

    See http://www.redrat.co.uk/products/irnetbox.html

    """

    def __init__(self, hostname, port, output, config):  # pylint:disable=redefined-outer-name
        self.hostname = hostname
        self.port = int(port or 10001)
        self.output = int(output)
        self.config = irnetbox.RemoteControlConfig(config)
        # Connect once so that the test fails immediately if irNetBox not found
        # (instead of failing at the first `press` in the script).
        debug("IRNetBoxControl: Connecting to %s" % hostname)
        with self._connect() as irnb:
            irnb.power_on()
        time.sleep(0.5)
        debug("IRNetBoxControl: Connected to %s" % hostname)

    def press(self, key):
        with self._connect() as irnb:
            irnb.irsend_raw(
                port=self.output, power=100, data=self.config[key])
        debug("Pressed " + key)

    def _connect(self):
        try:
            return irnetbox.IRNetBox(self.hostname, self.port)
        except socket.error as e:
            e.args = (("Failed to connect to IRNetBox %s: %s" % (
                self.hostname, e)),)
            e.strerror = e.args[0]
            raise


class RokuHttpControl(object):
    """Send a key-press via Roku remote control protocol.

    See https://sdkdocs.roku.com/display/sdkdoc/External+Control+API
    """

    # Map our recommended keynames (from linux input-event-codes.h) to the
    # equivalent Roku keyname.
    _KEYNAMES = {
        "KEY_HOME": "Home",
        "KEY_REWIND": "Rev",
        "KEY_FASTFORWARD": "Fwd",
        "KEY_PLAY": "Play",
        "KEY_PAUSE": "Play",
        "KEY_PLAYPAUSE": "Play",
        "KEY_OK": "Select",<commit_msg>RokuHttpControl: Cleanup: Inherit from RemoteControl<commit_after>from __future__ import absolute_import

import os
import re
import socket
import struct
import subprocess
import sys
import time
from contextlib import contextmanager
from distutils.spawn import find_executable

from . import irnetbox, utils
from .config import ConfigurationError
from .logging import debug, scoped_debug_level

__all__ = ['uri_to_control', 'uri_to_control_recorder']

try:
    from .control_gpl import controls as gpl_controls
except ImportError:
    gpl_controls = None


# pylint: disable=abstract-method

class RemoteControl(object):
    """Base class for remote-control implementations."""

    def press(self, key):
        raise NotImplementedError(
            "%s: 'press' is not implemented" % self.__class__.__name__)

    def keydown(self, key):
        raise NotImplementedError(
            "%s: 'keydown' is not implemented" % self.__class__.__name__)

    def keyup(self, key):
        raise NotImplementedError(
            "%s: 'keyup' is not implemented" % self.__class__.__name__)


class UnknownKeyError(Exception):
    pass


def uri_to_control(uri, display=None):
    controls = [
        (r'adb(:(?P<address>.*))?', new_adb_device),
        (r'error(:(?P<message>.*))?', ErrorControl),
        (r'file(:(?P<filename>[^,]+))?', FileControl),
        (r'''irnetbox:
             (?P<hostname>[^:]+)
             (:(?P<port>\d+))?
             :(?P<output>\d+)
             :(?P<config>[^:]+)''', IRNetBoxControl),
        (r'lirc(:(?P<hostname>[^:/]+))?:(?P<port>\d+):(?P<control_name>.+)',
         new_tcp_lirc_control),
        (r'lirc:(?P<lircd_socket>[^:]+)?:(?P<control_name>.+)',
         new_local_lirc_control),
        (r'none', NullControl),
        (r'roku:(?P<hostname>[^:]+)', RokuHttpControl),
        (r'samsung:(?P<hostname>[^:/]+)(:(?P<port>\d+))?',
         _new_samsung_tcp_control),
        (r'test', lambda: VideoTestSrcControl(display)),
        (r'x11:(?P<display>[^,]+)?(,(?P<mapping>.+)?)?', X11Control),
        (r'rfb:(?P<hostname>[^:/]+)(:(?P<port>\d+))?', RemoteFrameBuffer),
    ]
    if gpl_controls is not None:
        controls += gpl_controls

    for regex, factory in controls:
        m = re.match(regex, uri, re.VERBOSE | re.IGNORECASE)
        if m:
            return factory(**m.groupdict())
    raise ConfigurationError('Invalid remote control URI: "%s"' % uri)


def uri_to_control_recorder(uri):
    controls = [
        ('file://(?P<filename>.+)', file_control_recorder),
        (r'lirc(:(?P<hostname>[^:/]+))?:(?P<port>\d+):(?P<control_name>.+)',
         lirc_control_listen_tcp),
        (r'lirc:(?P<lircd_socket>[^:]+)?:(?P<control_name>.+)',
         lirc_control_listen),
        (r'stbt-control(:(?P<keymap_file>.+))?', stbt_control_listen),
    ]

    for regex, factory in controls:
        m = re.match(regex, uri)
        if m:
            return factory(**m.groupdict())
    raise ConfigurationError('Invalid remote control recorder URI: "%s"' % uri)


def new_adb_device(address):
    from stbt.android import AdbDevice
    tcpip = bool(re.match(r"\d+\.\d+\.\d+\.\d+", address))
    return AdbDevice(adb_device=address, tcpip=tcpip)


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

    def keydown(self, key):
        debug('NullControl: Ignoring request to hold "%s"' % key)

    def keyup(self, key):
        debug('NullControl: Ignoring request to release "%s"' % key)


class ErrorControl(RemoteControl):
    def __init__(self, message):
        if message is None:
            message = "No remote control configured"
        self.message = message

    def press(self, key):  # pylint:disable=unused-argument
        raise RuntimeError(self.message)

    def keydown(self, key):
        raise RuntimeError(self.message)

    def keyup(self, key):
        raise RuntimeError(self.message)


class FileControl(RemoteControl):
    """Writes keypress events to file.  Mostly useful for testing.  Defaults to
    writing to stdout.
    """
    def __init__(self, filename):
        if filename is None:
            self.outfile = sys.stdout
        else:
            self.outfile = open(filename, 'w+')

    def press(self, key):
        self.outfile.write(key + '\n')
        self.outfile.flush()

    def keydown(self, key):
        self.outfile.write("Holding %s\n" % key)
        self.outfile.flush()

    def keyup(self, key):
        self.outfile.write("Released %s\n" % key)
        self.outfile.flush()


class VideoTestSrcControl(RemoteControl):
    """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.display = display

    @property
    def videosrc(self):
        videosrc = self.display.source_pipeline.get_by_name("videotestsrc0")
        if not videosrc:
            raise ConfigurationError('The "test" control can only be used '
                                     'with source-pipeline = "videotestsrc"')
        return videosrc

    def press(self, key):
        if key not in [
                0, "smpte",
                1, "snow",
                2, "black",
                3, "white",
                4, "red",
                5, "green",
                6, "blue",
                7, "checkers-1",
                8, "checkers-2",
                9, "checkers-4",
                10, "checkers-8",
                11, "circular",
                12, "blink",
                13, "smpte75",
                14, "zone-plate",
                15, "gamut",
                16, "chroma-zone-plate",
                17, "solid-color",
                18, "ball",
                19, "smpte100",
                20, "bar"]:
            raise RuntimeError(
                'Key "%s" not valid for the "test" control' % key)
        self.videosrc.props.pattern = key
        debug("Pressed %s" % key)


def _find_file(path, root=os.path.dirname(os.path.abspath(__file__))):
    return os.path.join(root, path)


class LircControl(RemoteControl):
    """Send a key-press via a LIRC-enabled infrared blaster.

    See http://www.lirc.org/html/technical.html#applications
    """

    def __init__(self, control_name, connect_fn):
        self.control_name = control_name
        self._connect = connect_fn

    def press(self, key):
        s = self._connect()
        command = "SEND_ONCE %s %s" % (self.control_name, key)
        s.sendall(command + "\n")
        _read_lircd_reply(s, command)
        debug("Pressed " + key)

    def keydown(self, key):
        s = self._connect()
        command = "SEND_START %s %s" % (self.control_name, key)
        s.sendall(command + "\n")
        _read_lircd_reply(s, command)
        debug("Holding " + key)

    def keyup(self, key):
        s = self._connect()
        command = "SEND_STOP %s %s" % (self.control_name, key)
        s.sendall(command + "\n")
        _read_lircd_reply(s, command)
        debug("Released " + key)


def _read_lircd_reply(stream, command):
    """Waits for lircd reply and checks if a LIRC send command was successful.

    Waits for a reply message from lircd (called "reply packet" in the LIRC
    reference) for the specified command; raises exception if it times out or
    the reply contains an error message.

    The structure of a lircd reply message for a SEND_ONCE|SEND_START|SEND_STOP
    command is the following:

    BEGIN
    <command>
    (SUCCESS|ERROR)
    [DATA
    <number-of-data-lines>
    <error-message>]
    END

    See: http://www.lirc.org/html/technical.html#applications
    """
    reply = []
    try:
        for line in read_records(stream, "\n"):
            if line == "BEGIN":
                reply = []
            reply.append(line)
            if line == "END" and reply[1] == command:
                break
    except socket.timeout:
        raise RuntimeError(
            "Timed out: No reply from LIRC remote control within %d seconds"
            % stream.gettimeout())
    if "SUCCESS" not in reply:
        if "ERROR" in reply and len(reply) >= 6 and reply[3] == "DATA":
            try:
                num_data_lines = int(reply[4])
                raise RuntimeError("LIRC remote control returned error: %s"
                                   % " ".join(reply[5:5 + num_data_lines]))
            except ValueError:
                pass
        raise RuntimeError("LIRC remote control returned unknown error")

DEFAULT_LIRCD_SOCKET = '/var/run/lirc/lircd'


def new_local_lirc_control(lircd_socket, control_name):
    if lircd_socket is None:
        lircd_socket = DEFAULT_LIRCD_SOCKET

    def _connect():
        try:
            s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            s.settimeout(3)
            s.connect(lircd_socket)
            return s
        except socket.error as e:
            e.args = (("Failed to connect to Lirc socket %s: %s" % (
                lircd_socket, e)),)
            e.strerror = e.args[0]
            raise

    # Connect once so that the test fails immediately if Lirc isn't running
    # (instead of failing at the first `press` in the script).
    debug("LircControl: Connecting to %s" % lircd_socket)
    _connect()
    debug("LircControl: Connected to %s" % lircd_socket)

    return LircControl(control_name, _connect)


def new_tcp_lirc_control(control_name, hostname=None, port=None):
    """Send a key-press via a LIRC-enabled device through a LIRC TCP listener.

        control = new_tcp_lirc_control("localhost", "8765", "humax")
        control.press("MENU")
    """
    if hostname is None:
        hostname = 'localhost'
    if port is None:
        port = 8765

    port = int(port)

    def _connect():
        return _connect_tcp_socket(hostname, port)

    # Connect once so that the test fails immediately if Lirc isn't running
    # (instead of failing at the first `press` in the script).
    debug("TCPLircControl: Connecting to %s:%d" % (hostname, port))
    _connect()
    debug("TCPLircControl: Connected to %s:%d" % (hostname, port))

    return LircControl(control_name, _connect)


class RemoteFrameBuffer(RemoteControl):
    """Send a key-press to a set-top box running a VNC Remote Frame Buffer
        protocol.
        Expected key press input:
            <KEY_LABEL>

        control = RemoteFrameBuffer("192.168.0.123")
        control.press("KEY_MENU")
    """

    # Map our recommended keynames (from linux input-event-codes.h) to the
    # equivalent RFB keyname.
    _KEYNAMES = {
        'KEY_BACK': 0xE002,
        'KEY_BLUE': 0xE203,
        'KEY_CHANNELDOWN': 0xE007,
        'KEY_CHANNELUP': 0xE006,
        'KEY_DOWN': 0xE101,
        'KEY_ELPS': 0xEF00,
        'KEY_FASTFORWARD': 0xE405,
        'KEY_GREEN': 0xE201,
        'KEY_GUIDE': 0xE00B,
        'KEY_HELP': 0xE00A,
        'KEY_HOME': 0xE015,
        'KEY_INFO': 0xE00E,
        'KEY_INPUTSELECT': 0xE010,
        'KEY_INTERACT': 0xE008,
        'KEY_0': 0xE300,
        'KEY_1': 0xE301,
        'KEY_2': 0xE302,
        'KEY_3': 0xE303,
        'KEY_4': 0xE304,
        'KEY_5': 0xE305,
        'KEY_6': 0xE306,
        'KEY_7': 0xE307,
        'KEY_8': 0xE308,
        'KEY_9': 0xE309,
        'KEY_LEFT': 0xE102,
        'KEY_MENU': 0xE00A,
        'KEY_MUTE': 0xE005,
        'KEY_MYTV': 0xE009,
        'KEY_PAUSE': 0xE401,
        'KEY_PLAY': 0xE400,
        'KEY_PLAYPAUSE': 0xE40A,
        'KEY_POWER': 0xE000,
        'KEY_PRIMAFILA': 0xEF00,
        'KEY_RECORD': 0xE403,
        'KEY_RED': 0xE200,
        'KEY_REWIND': 0xE407,
        'KEY_RIGHT': 0xE103,
        'KEY_SEARCH': 0xEF03,
        'KEY_SELECT': 0xE001,
        'KEY_SKY': 0xEF01,
        'KEY_STOP': 0xE402,
        'KEY_TEXT': 0xE00F,
        'KEY_UP': 0xE100,
        'KEY_VOLUMEDOWN': 0xE004,
        'KEY_VOLUMEUP': 0xE003,
        'KEY_YELLOW': 0xE202
    }

    def __init__(self, hostname, port=None):
        self.hostname = hostname
        self.port = int(port or 5900)
        self.timeout = 3
        self.socket = None

    def press(self, key):
        self._connect_socket()
        self._handshake()
        self._press_down(key)
        self._release(key)
        self._close()

    def _connect_socket(self):
        self.socket = socket.socket()
        s = self.socket
        if self.timeout:
            s.settimeout(self.timeout)
        s.connect((self.hostname, self.port))
        debug(
            "RemoteFrameBuffer: connected to %s:%d"
            % (self.hostname, self.port))

    def _handshake(self):
        s = self.socket
        prot_info = s.recv(20)
        if prot_info != b'RFB 003.008\n':
            raise socket.error("wrong RFB protocol info")
        s.send(b"RFB 003.003\n")
        s.recv(4)
        s.send(b'\0')
        s.recv(24)
        debug("RemoteFrameBuffer: handshake completed")

    def _press_down(self, key):
        key_code = self._get_key_code(key)
        self.socket.send(struct.pack('!BBxxI', 4, 1, key_code))
        debug(
            "RemoteFrameBuffer: pressed down (0x%04x)"
            % key_code)

    def _release(self, key):
        key_code = self._get_key_code(key)
        self.socket.send(struct.pack('!BBxxI', 4, 0, key_code))
        debug("RemoteFrameBuffer: release (0x%04x)" % key_code)

    def _close(self):
        self.socket.shutdown(socket.SHUT_RDWR)
        self.socket.close()
        debug("RemoteFrameBuffer: socket connection closed")

    def _get_key_code(self, key):
        key_code = self._KEYNAMES.get(key, key)
        return key_code


class IRNetBoxControl(RemoteControl):
    """Send a key-press via the network-controlled RedRat IRNetBox IR emitter.

    See http://www.redrat.co.uk/products/irnetbox.html

    """

    def __init__(self, hostname, port, output, config):  # pylint:disable=redefined-outer-name
        self.hostname = hostname
        self.port = int(port or 10001)
        self.output = int(output)
        self.config = irnetbox.RemoteControlConfig(config)
        # Connect once so that the test fails immediately if irNetBox not found
        # (instead of failing at the first `press` in the script).
        debug("IRNetBoxControl: Connecting to %s" % hostname)
        with self._connect() as irnb:
            irnb.power_on()
        time.sleep(0.5)
        debug("IRNetBoxControl: Connected to %s" % hostname)

    def press(self, key):
        with self._connect() as irnb:
            irnb.irsend_raw(
                port=self.output, power=100, data=self.config[key])
        debug("Pressed " + key)

    def _connect(self):
        try:
            return irnetbox.IRNetBox(self.hostname, self.port)
        except socket.error as e:
            e.args = (("Failed to connect to IRNetBox %s: %s" % (
                self.hostname, e)),)
            e.strerror = e.args[0]
            raise


class RokuHttpControl(RemoteControl):
    """Send a key-press via Roku remote control protocol.

    See https://sdkdocs.roku.com/display/sdkdoc/External+Control+API
    """

    # Map our recommended keynames (from linux input-event-codes.h) to the
    # equivalent Roku keyname.
    _KEYNAMES = {
        "KEY_HOME": "Home",
        "KEY_REWIND": "Rev",
        "KEY_FASTFORWARD": "Fwd",
        "KEY_PLAY": "Play",
        "KEY_PAUSE": "Play",
        "KEY_PLAYPAUSE": "Play",
        "KEY_OK": "Select",
{'fraction_non_alphanumeric': 0.1286865012342471, 'fraction_numerical': 0.019682993374041835, 'mean_word_length': 2.291181186531267, 'pattern_counts': {'":': 18, '<': 63, '<?xml version=': 0, '>': 65, 'https://': 2, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, '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': '5238705', 'n_tokens_mistral': 10575, 'n_tokens_neox': 9876, 'n_words': 2361}