client.py

''' Top-level python bindings for the lircd socket interface. '''
 
 
 
 
#   pylint: disable=W0613
 
 
 
from abc import ABCMeta, abstractmethod
from enum import Enum
import configparser
import os
import os.path
import selectors
import socket
import time
 
import lirc.config
import _client
 
_DEFAULT_PROG = 'lircd-client'
 
 
def get_default_socket_path() -> str:
    ''' Get default value for the lircd socket path, using (falling priority):
 
       - The environment variable LIRC_SOCKET_PATH.
       - The 'output' value in the lirc_options.conf file if value and the
         corresponding file exists.
       - A hardcoded default lirc.config.VARRUNDIR/lirc/lircd, possibly
         non-existing.
    '''
 
    if 'LIRC_SOCKET_PATH' in os.environ:
        return os.environ['LIRC_SOCKET_PATH']
    path = lirc.config.SYSCONFDIR + '/lirc/lirc_options.conf'
    parser = configparser.SafeConfigParser()
    try:
        parser.read(path)
    except configparser.Error:
        pass
    else:
        if parser.has_section('lircd'):
            try:
                path = str(parser.get('lircd', 'output'))
                if os.path.exists(path):
                    return path
            except configparser.NoOptionError:
                pass
    return lirc.config.VARRUNDIR + '/lirc/lircd'
 
 
def get_default_lircrc_path() -> str:
    ''' Get default path to the lircrc file according to (falling priority):
 
       - $XDG_CONFIG_HOME/lircrc if environment variable and file exists.
       - ~/.config/lircrc if it exists.
       - ~/.lircrc if it exists
       - A hardcoded default lirc.config.SYSCONFDIR/lirc/lircrc, whether
         it exists or not.
    '''
    if 'XDG_CONFIG_HOME' in os.environ:
        path = os.path.join(os.environ['XDG_CONFIG_HOME'], 'lircrc')
        if os.path.exists(path):
            return path
    path = os.path.join(os.path.expanduser('~'), '.config' 'lircrc')
    if os.path.exists(path):
        return path
    path = os.path.join(os.path.expanduser('~'), '.lircrc')
    if os.path.exists(path):
        return path
    return os.path.join(lirc.config.SYSCONFDIR, 'lirc', 'lircrc')
 
 
class BadPacketException(Exception):
    ''' Malformed or otherwise unparsable packet received. '''
    pass
 
 
class TimeoutException(Exception):
    ''' Timeout receiving data from remote host.'''
    pass
 
 
 
 
 
class AbstractConnection(metaclass=ABCMeta):
    ''' Abstract interface for all connections. '''
 
    def __enter__(self):
        return self
 
    def __exit__(self, exc_type, exc, traceback):
        self.close()
 
    @abstractmethod
    def readline(self, timeout: float = None) -> str:
        ''' Read a buffered line
 
        Parameters:
          - timeout: seconds.
              - If set to 0 immediately return either a line or None.
              - If set to None (default mode) use blocking read.
 
        Returns: code string as described in lircd(8) without trailing
                 newline or None.
 
        Raises: TimeoutException  if timeout > 0 expires.
        '''
        pass
 
    @abstractmethod
    def fileno(self) -> int:
        ''' Return the file nr used for IO, suitable for select() etc. '''
        pass
 
    @abstractmethod
    def has_data(self) -> bool:
        ''' Return true if next readline(None) won't block . '''
        pass
 
    @abstractmethod
    def close(self):
        ''' Close/release all resources '''
        pass
 
 
class RawConnection(AbstractConnection):
    ''' Interface to receive code strings as described in lircd(8).
 
    Parameters:
      - socket_path: lircd output socket path, see get_default_socket_path()
        for defaults.
      - prog: Program name used in lircrc decoding, see ircat(1). Could be
        omitted if only raw keypresses should be read.
 
    '''
    # pylint: disable=no-member
 
    def __init__(self, socket_path: str = None, prog: str = _DEFAULT_PROG):
        if socket_path:
            os.environ['LIRC_SOCKET_PATH'] = socket_path
        else:
            os.environ['LIRC_SOCKET_PATH'] = get_default_socket_path()
        _client.lirc_deinit()
        fd = _client.lirc_init(prog)
        self._socket = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)
        self._select = selectors.DefaultSelector()
        self._select.register(self._socket, selectors.EVENT_READ)
        self._buffer = bytearray(0)
 
    def readline(self, timeout: float = None) -> str:
        ''' Implements AbstractConnection.readline(). '''
        if timeout:
            start = time.clock()
        while b'\n' not in self._buffer:
            ready = self._select.select(
                start + timeout - time.clock() if timeout else timeout)
            if ready == []:
                if timeout:
                    raise TimeoutException(
                        "readline: no data within %f seconds" % timeout)
                else:
                    return None
            recv = self._socket.recv(4096)
            if len(recv) == 0:
                raise ConnectionResetError('Connection lost')
            self._buffer += recv
        line, self._buffer = self._buffer.split(b'\n', 1)
        return line.decode('ascii', 'ignore')
 
    def fileno(self) -> int:
        ''' Implements AbstractConnection.fileno(). '''
        return self._socket.fileno()
 
    def has_data(self) -> bool:
        ''' Implements AbstractConnection.has_data() '''
        return b'\n' in self._buffer
 
    def close(self):
        ''' Implements AbstractConnection.close() '''
        self._socket.close()
        _client.lirc_deinit()
 
 
AbstractConnection.register(RawConnection)          # pylint:disable=no-member
 
 
class LircdConnection(AbstractConnection):
    ''' Interface to receive lircrc-translated keypresses. This is basically
    built on top of lirc_code2char() and as such supporting centralized
    translations using lircrc_class. See lircrcd(8).
 
    Parameters:
      - program: string, used to identify client. See ircat(1)
      - lircrc: lircrc file path. See get_default_lircrc_path() for defaults.
      - socket_path: lircd output socket path,  see get_default_socket_path()
        for defaults.
    '''
    # pylint: disable=no-member
 
    def __init__(self, program: str,
                 lircrc_path: str = None,
                 socket_path: str = None):
        if not lircrc_path:
            lircrc_path = get_default_lircrc_path()
        if not lircrc_path:
            raise FileNotFoundError('Cannot find lircrc config file.')
        self._connection = RawConnection(socket_path, program)
        self._lircrc = _client.lirc_readconfig(lircrc_path)
        self._program = program
        self._buffer = []
 
    def readline(self, timeout: float = None):
        ''' Implements AbstractConnection.readline(). '''
        while len(self._buffer) <= 0:
            code = self._connection.readline(timeout)
            if code is None:
                return None
            strings = \
                _client.lirc_code2char(self._lircrc, self._program, code)
            if not strings or len(strings) == 0:
                if timeout == 0:
                    return None
                continue
            self._buffer.extend(strings)
        return self._buffer.pop(0)
 
    def has_data(self) -> bool:
        ''' Implements AbstractConnection.has_data() '''
        return len(self._buffer) > 0
 
    def fileno(self) -> int:
        ''' Implements AbstractConnection.fileno(). '''
        return self._connection.fileno()
 
    def close(self):
        ''' Implements AbstractConnection.close() '''
        self._connection.close()
        _client.lirc_freeconfig(self._lircrc)
 
 
AbstractConnection.register(LircdConnection)     # pylint: disable=no-member
 
 
 
 
 
 
 
class CommandConnection(RawConnection):
    ''' Extends the parent with a send() method. '''
 
    def __init__(self, socket_path: str = None):
        RawConnection.__init__(self, socket_path)
 
    def send(self, command: (bytearray, str)):
        ''' Send  single line over socket '''
        if not isinstance(command, bytearray):
            command = command.encode('ascii')
        while len(command) > 0:
            sent = self._socket.send(command)
            command = command[sent:]
 
 
class Result(Enum):
    ''' Public reply parser result, available when completed. '''
    OK = 1
    FAIL = 2
    INCOMPLETE = 3
 
 
class Command(object):
    ''' Command, parser and connection container with a run() method. '''
 
    def __init__(self, cmd: str,
                 connection: AbstractConnection,
                 timeout: float = 0.4):
        self._conn = connection
        self._cmd_string = cmd
        self._parser = ReplyParser()
 
    def run(self, timeout: float = None):
        ''' Run the command and return a Reply. Timeout as of
        AbstractConnection.readline()
        '''
        self._conn.send(self._cmd_string)
        while not self._parser.is_completed():
            line = self._conn.readline(timeout)
            if not line:
                raise TimeoutException('No data from lircd host.')
            self._parser.feed(line)
        return self._parser
 
 
class Reply(object):
    ''' The status/result from parsing a command reply.
 
    Attributes:
        result: Enum Result, reflects parser state.
        success: bool, reflects SUCCESS/ERROR.
        data: List of lines, the command DATA payload.
        sighup: bool, reflects if a SIGHUP package has been received
                (these are otherwise ignored).
        last_line: str, last input line (for error messages).
    '''
    def __init__(self):
        self.result = Result.INCOMPLETE
        self.success = None
        self.data = []
        self.sighup = False
        self.last_line = ''
 
 
class ReplyParser(Reply):
    ''' Handles the actual parsing of a command reply.  '''
 
    def __init__(self):
        Reply.__init__(self)
        self._state = self._State.BEGIN
        self._lines_expected = None
        self._buffer = bytearray(0)
 
    def is_completed(self) -> bool:
        ''' Returns true if no more reply input is required. '''
        return self.result != Result.INCOMPLETE
 
    def feed(self, line: str):
        ''' Enter a line of data into parsing FSM, update state. '''
 
        fsm = {
            self._State.BEGIN: self._begin,
            self._State.COMMAND: self._command,
            self._State.RESULT: self._result,
            self._State.DATA: self._data,
            self._State.LINE_COUNT: self._line_count,
            self._State.LINES: self._lines,
            self._State.END: self._end,
            self._State.SIGHUP_END: self._sighup_end
        }
        line = line.strip()
        if not line:
            return
        self.last_line = line
        fsm[self._state](line)
        if self._state == self._State.DONE:
            self.result = Result.OK
 
 
 
    class _State(Enum):
        ''' Internal FSM state. '''
        BEGIN = 1
        COMMAND = 2
        RESULT = 3
        DATA = 4
        LINE_COUNT = 5
        LINES = 6
        END = 7
        DONE = 8
        NO_DATA = 9
        SIGHUP_END = 10
 
    def _bad_packet_exception(self, line):
        self.result = Result.FAIL
        raise BadPacketException(
            'Cannot parse: %s\nat state: %s\n' % (line, self._state))
 
    def _begin(self, line):
        if line == 'BEGIN':
            self._state = self._State.COMMAND
 
    def _command(self, line):
        if not line:
            self._bad_packet_exception(line)
        elif line == 'SIGHUP':
            self._state = self._State.SIGHUP_END
            self.sighup = True
        else:
            self._state = self._State.RESULT
 
    def _result(self, line):
        if line in ['SUCCESS', 'ERROR']:
            self.success = line == 'SUCCESS'
            self._state = self._State.DATA
        else:
            self._bad_packet_exception(line)
 
    def _data(self, line):
        if line == 'END':
            self._state = self._State.DONE
        elif line == 'DATA':
            self._state = self._State.LINE_COUNT
        else:
            self._bad_packet_exception(line)
 
    def _line_count(self, line):
        try:
            self._lines_expected = int(line)
        except ValueError:
            self._bad_packet_exception(line)
        if self._lines_expected == 0:
            self._state = self._State.END
        else:
            self._state = self._State.LINES
 
    def _lines(self, line):
        self.data.append(line)
        if len(self.data) >= self._lines_expected:
            self._state = self._State.END
 
    def _end(self, line):
        if line != 'END':
            self._bad_packet_exception(line)
        self._state = self._State.DONE
 
    def _sighup_end(self, line):
        if line == 'END':
            ReplyParser.__init__(self)
            self.sighup = True
        else:
            self._bad_packet_exception(line)
 
 
 
 
 
 
 
 
 
class SimulateCommand(Command):
    ''' Simulate a button press, see SIMULATE in lircd(8) manpage.  '''
    # pylint: disable=too-many-arguments
 
    def __init__(self, connection: AbstractConnection,
                 remote: str, key: str, repeat: int = 1, keycode: int = 0):
        cmd = 'SIMULATE %016d %02d %s %s\n' % \
            (int(keycode), int(repeat), key, remote)
        Command.__init__(self, cmd, connection)
 
 
class ListRemotesCommand(Command):
    ''' List available remotes, see LIST in lircd(8) manpage. '''
 
    def __init__(self, connection: AbstractConnection):
        Command.__init__(self, 'LIST\n', connection)
 
 
class ListKeysCommand(Command):
    ''' List available keys in given remote, see LIST in lircd(8) manpage. '''
 
    def __init__(self, connection: AbstractConnection, remote: str):
        Command.__init__(self, 'LIST %s\n' % remote, connection)
 
 
class StartRepeatCommand(Command):
    ''' Start repeating given key, see SEND_START in lircd(8) manpage. '''
 
    def __init__(self, connection: AbstractConnection,
                 remote: str, key: str):
        cmd = 'SEND_START %s %s\n' % (remote, key)
        Command.__init__(self, cmd, connection)
 
 
class StopRepeatCommand(Command):
    ''' Stop repeating given key, see SEND_STOP in lircd(8) manpage. '''
 
    def __init__(self, connection: AbstractConnection,
                 remote: str, key: str):
        cmd = 'SEND_STOP %s %s\n' % (remote, key)
        Command.__init__(self, cmd, connection)
 
 
class SendCommand(Command):
    ''' Send given key, see SEND_ONCE in lircd(8) manpage. '''
 
    def __init__(self, connection: AbstractConnection,
                 remote: str, keys: str):
        if not len(keys):
            raise ValueError('No keys to send given')
        cmd = 'SEND_ONCE %s %s\n' % (remote, ' '.join(keys))
        Command.__init__(self, cmd, connection)
 
 
class SetTransmittersCommand(Command):
    ''' Set transmitters to use, see SET_TRANSMITTERS in lircd(8) manpage.
 
    Arguments:
        transmitter: Either a bitmask or a list of int describing active
            transmitter numbers.
    '''
 
    def __init__(self, connection: AbstractConnection,
                 transmitters: (int, list)):
        if isinstance(transmitters, list):
            mask = 0
            for transmitter in transmitters:
                mask |= (1 << (int(transmitter) - 1))
        else:
            mask = transmitters
        cmd = 'SET_TRANSMITTERS %d\n' % mask
        Command.__init__(self, cmd, connection)
 
 
class VersionCommand(Command):
    ''' Get lircd version, see VERSION in lircd(8) manpage. '''
 
    def __init__(self, connection: AbstractConnection):
        Command.__init__(self, 'VERSION\n', connection)
 
 
class DrvOptionCommand(Command):
    ''' Set a driver option value, see DRV_OPTION in lircd(8) manpage. '''
 
    def __init__(self, connection: AbstractConnection,
                 option: str, value: str):
        cmd = 'DRV_OPTION %s %s\n' % (option, value)
        Command.__init__(self, cmd, connection)
 
 
class SetLogCommand(Command):
    ''' Start/stop logging lircd output , see SET_INPUTLOG in lircd(8)
    manpage.
    '''
 
    def __init__(self, connection: AbstractConnection,
                 logfile: str = None):
        cmd = 'SET_INPUTLOG' + (' ' + logfile if logfile else '') + '\n'
        Command.__init__(self, cmd, connection)
 
 
 
 
 
 
 
class IdentCommand(Command):
    ''' Identify client using the prog token, see IDENT in lircrcd(8) '''
 
    def __init__(self, connection: AbstractConnection,
                 prog: str = None):
        if not prog:
            raise ValueError('The prog argument cannot be None')
        cmd = 'IDENT {}\n'.format(prog)
        Command.__init__(self, cmd, connection)
 
 
class CodeCommand(Command):
    '''Translate a keypress to application string, see CODE in lircrcd(8) '''
 
    def __init__(self, connection: AbstractConnection,
                 code: str = None):
        if not code:
            raise ValueError('The prog argument cannot be None')
        Command.__init__(self, 'CODE {}\n'.format(code), connection)
 
 
class GetModeCommand(Command):
    '''Get current translation mode, see GETMODE in lircrcd(8) '''
 
    def __init__(self, connection: AbstractConnection):
        Command.__init__(self, "GETMODE\n", connection)
 
 
class SetModeCommand(Command):
    '''Set current translation mode, see SETMODE in lircrcd(8) '''
 
    def __init__(self, connection: AbstractConnection,
                 mode: str = None):
        if not mode:
            raise ValueError('The mode argument cannot be None')
        Command.__init__(self, 'SETMODE {}\n'.format(mode), connection)