#! /usr/bin/env python
# -*- coding: utf-8 -*-
"""
scripts.eww
~~~~~~~~~~~
The Eww client is the recommended frontend for connecting to a listening
Eww instance.
Before criticizing this implementation, it's important to understand why
this was done.
I consider :py:mod:`readline` support a **hard** requirement.
:py:mod:`readline` is currently only supported by :py:func:`raw_input`.
:py:mod:`readline` does expose a few calls into the underlying library, but
not nearly what we need to implement it ourselves without using
:py:func:`raw_input`.
There are a few solutions to this problem:
1. Implement readline-like support ourselves in Python
2. Call the underlying C library ourselves
3. Implement proper PTY support on both ends of the connection
4. Use :py:func:`raw_input`
Option 1 is a lot more work than it seems like. It would certainly be
valuable, and an interesting library to create. The scope of work required
to do it properly just for this is difficult to justify.
Option 2 would require a compilation step and complicate installation.
Option 3 would be ideal. However, implementing a cross-platform PTY over
a socket in Python is difficult, error-prone, and tough to debug. It's
absolutely on the table and would let us add a ton of features, but it's
not happening right off the bat.
Which leaves us with option 4, and results in our current (difficult to
test) implementation.
With that in mind, read on.
"""
# pylint: disable=invalid-name, unused-import
import optparse
try:
import readline
except ImportError: # pragma: no cover
# :(
pass
import socket
import sys
[docs]class ConnectionClosed(Exception):
"""Raised when a connection is closed."""
pass
[docs]class EwwClient(object):
"""Manages all client communication."""
[docs] def __init__(self, host, port):
"""Init.
Args:
host (str): A host to connect to.
port (int): A port to connect to.
"""
self.host = host
self.port = port
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.current_prompt = None
self.prompts = []
self.prompts.append('(eww) ')
self.prompts.append('>>> ')
self.prompts.append('... ')
[docs] def connect(self):
"""Connects to an Eww instance.
Returns:
None
"""
self.sock.connect((self.host, self.port))
[docs] def display_output(self):
"""Displays output from the Eww instance.
Returns:
None
"""
while True:
try:
msg = self.sock.recv(1024)
except socket.error:
raise ConnectionClosed
if not msg:
raise ConnectionClosed
# If the last line in the msg is a prompt, we've got
# the complete message. We'll want to reset current_prompt
# and print the message (minus the prompt). Otherwise
# we just print the part of the message we've received
# so far.
last_line = msg.split('\n')[-1]
if last_line in self.prompts:
self.current_prompt = last_line
sys.stdout.write(msg[:-len(last_line)])
sys.stdout.flush()
return
else:
sys.stdout.write(msg)
sys.stdout.flush()
[docs] def clientloop(self, debug=False, line=None):
"""Repeatedly loops through display_output and get_input.
Args:
debug (bool): Used for testing. If ``debug`` is True,
``clientloop`` only goes through one iteration.
line (str): If ``debug`` is True, this will be passed to get_input.
Useful for testing.
Returns:
None
"""
try:
while True:
self.display_output()
if debug:
self.get_input(line)
return # pragma: no cover
else: # pragma: no cover
self.get_input()
except ConnectionClosed:
pass
[docs]def main(debug=False, line=None, opt_args=None):
"""Main function.
Args:
debug (bool): Used for testing. If ``debug`` is True,
``clientloop`` only goes through one iteration.
line (str): If ``debug`` is True, this will be passed to get_input.
Useful for testing.
opt_args (list): If ``debug`` is True, this list is parsed by optparse
instead of sys.argv.
Returns:
None
"""
parser = optparse.OptionParser()
parser.add_option('-s', '--server',
action='store',
dest='host',
default='localhost',
type='str',
help='The server to connect to.')
parser.add_option('-p', '--port',
action='store',
dest='port',
default=10000,
type='int',
help='The port to connect on.')
if debug:
options, remainder = parser.parse_args(opt_args)
else:
options, remainder = parser.parse_args() # pragma: no cover
del remainder
options = vars(options)
client = EwwClient(options['host'], options['port'])
try:
client.connect()
except socket.error:
print 'Connection refused.'
sys.exit(1)
try:
client.clientloop(debug, line)
except KeyboardInterrupt: # pragma: no cover
pass
print "Shutting down..."
if __name__ == '__main__': # pragma: no cover -- we test the function directly
main()