socket_console.py

python s60 1.4.5版本的源代码

#
# This is a simple console class, which tries to handle a VT100-workalike
# terminal at the other end of a given socket connection.
#
# Limitations:
#  - currently the remote screen size is assumed to be 80x25
#  - sockets are assumed to be blocking (since the platform in mind didn't
#    have nonblocking sockets)
#
# Portions Copyright (c) 2005 Nokia Corporation 
#
# The code is derived from unix_console.py, which contained the
# following copyright notice:
#
#   Copyright 2000-2004 Michael Hudson [e-mail]
#
#                        All Rights Reserved
#
#
# Permission to use, copy, modify, and distribute this software and
# its documentation for any purpose is hereby granted without fee,
# provided that the above copyright notice appear in all copies and
# that both that copyright notice and this permission notice appear in
# supporting documentation.
#
# THE AUTHOR MICHAEL HUDSON DISCLAIMS ALL WARRANTIES WITH REGARD TO
# THIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
# AND FITNESS, IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL,
# INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
# RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
# CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

import sys
from pyrepl.console import Console, Event
from pyrepl import unix_eventqueue
import dumbcurses as curses
import socket

def _my_getstr(cap, optional=0):
    r = curses.tigetstr(cap)
    if not optional and r is None:
        raise RuntimeError, \
              "terminal doesn't have the required '%s' capability"%cap
    return r

class SocketConsole(Console):
    MAX_READ=16
    def __init__(self, socket, encoding=None):
#        if encoding is None:
#            encoding = sys.getdefaultencoding()
            
#        self.encoding = encoding
        #print >>sys.stderr,"Encoding "+encoding
        self.encoding='latin-1'

        self._socket=socket

        self.__buffer = []
        
        self._bel   = _my_getstr("bel")
        self._civis = _my_getstr("civis", optional=1)
        self._clear = _my_getstr("clear")
        self._cnorm = _my_getstr("cnorm", optional=1)
        self._cub   = _my_getstr("cub",   optional=1)
        self._cub1  = _my_getstr("cub1",  1)
        self._cud   = _my_getstr("cud",   1)
        self._cud1  = _my_getstr("cud1",  1)
        self._cuf   = _my_getstr("cuf",   1)
        self._cuf1  = _my_getstr("cuf1",  1)
        self._cup   = _my_getstr("cup")
        self._cuu   = _my_getstr("cuu",   1)
        self._cuu1  = _my_getstr("cuu1",  1)
        self._dch1  = _my_getstr("dch1",  1)
        self._dch   = _my_getstr("dch",   1)
        self._el    = _my_getstr("el")
        self._hpa   = _my_getstr("hpa",   1)
        self._ich   = _my_getstr("ich",   1)
        self._ich1  = _my_getstr("ich1",  1)
        self._ind   = _my_getstr("ind",   1)
        self._pad   = _my_getstr("pad",   1)
        self._ri    = _my_getstr("ri",    1)
        self._rmkx  = _my_getstr("rmkx",  1)
        self._smkx  = _my_getstr("smkx",  1)
        
        ## work out how we're going to sling the cursor around
        if 0 and self._hpa: # hpa don't work in windows telnet :-(
            self.__move_x = self.__move_x_hpa
        elif self._cub and self._cuf:
            self.__move_x = self.__move_x_cub_cuf
        elif self._cub1 and self._cuf1:
            self.__move_x = self.__move_x_cub1_cuf1
        else:
            raise RuntimeError, "insufficient terminal (horizontal)"

        if self._cuu and self._cud:
            self.__move_y = self.__move_y_cuu_cud
        elif self._cuu1 and self._cud1:
            self.__move_y = self.__move_y_cuu1_cud1
        else:
            raise RuntimeError, "insufficient terminal (vertical)"

        if self._dch1:
            self.dch1 = self._dch1
        elif self._dch:
            self.dch1 = curses.tparm(self._dch, 1)
        else:
            self.dch1 = None

        if self._ich1:
            self.ich1 = self._ich1
        elif self._ich:
            self.ich1 = curses.tparm(self._ich, 1)
        else:
            self.ich1 = None

        self.__move = self.__move_short

        self.event_queue = unix_eventqueue.EventQueue()
        self.busy=False

    def _oswrite(self,str):
        try:
            self._socket.send(str)
        except socket.error:
            raise IOError("Socket error: %s %s"%(sys.exc_info()[0:2]))

    def _osread(self,n=1):
        try:
            out=self._socket.recv(n)
        except socket.error:
            raise EOFError("Socket error: %s %s"%(sys.exc_info()[0:2]))
        return out

    def write(self,str):
        self._oswrite(str.replace('\n','\n\r'))

    def flush(self):
        self.flushoutput()
        
    def read(self,n=1):        
        return self._osread(n)

    def readline(self,n=None):
        line=[]
        while 1:
            ch=self.read(1)
            line.append(ch)
            self.write(ch)
            if ch == '\n':
                break
            if n and len(line)>=n:
                break
        return ''.join(line)

    # No readlines() because reading until EOF doesn't make sense
    # for the console.
    
    def isatty(self):
        return True

    def writelines(self,seq):
        for k in seq:
            self.write(k)

    def change_encoding(self, encoding):
        self.encoding = encoding
    
    def refresh(self, screen, (cx, cy)):
        # this function is still too long (over 90 lines)
        self.__maybe_write_code(self._civis)

        if not self.__gone_tall:
            while len(self.screen) < min(len(screen), self.height):
                self.__move(0, len(self.screen) - 1)
                self.__write("\n")
                self.__posxy = 0, len(self.screen)
                self.screen.append("")
        else:
            while len(self.screen) < len(screen):
                self.screen.append("")            

        if len(screen) > self.height:
            self.__gone_tall = 1
            self.__move = self.__move_tall

        px, py = self.__posxy
        old_offset = offset = self.__offset
        height = self.height

        # we make sure the cursor is on the screen, and that we're
        # using all of the screen if we can
        if cy < offset:
            offset = cy
        elif cy >= offset + height:
            offset = cy - height + 1
        elif offset > 0 and len(screen) < offset + height:
            offset = max(len(screen) - height, 0)
            screen.append([])

        oldscr = self.screen[old_offset:old_offset + height]
        newscr = screen[offset:offset + height]

        # use hardware scrolling if we have it.
        if old_offset > offset and self._ri:
            self.__write_code(self._cup, 0, 0)
            self.__posxy = 0, old_offset
            for i in range(old_offset - offset):
                self.__write_code(self._ri)
                oldscr.pop(-1)
                oldscr.insert(0, "")
        elif old_offset < offset and self._ind:
            self.__write_code(self._cup, self.height - 1, 0)
            self.__posxy = 0, old_offset + self.height - 1
            for i in range(offset - old_offset):
                self.__write_code(self._ind)
                oldscr.pop(0)
                oldscr.append("")

        self.__offset = offset

        for y, oldline, newline, in zip(range(offset, offset + height),
                                        oldscr,
                                        newscr):
            if oldline != newline:
                self.write_changed_line(y, oldline, newline, px)
                
        y = len(newscr)
        while y < len(oldscr):
            self.__move(0, y)
            self.__posxy = 0, y