docker-offlineimap/offlineimap/ui/Curses.py
Nicolas Sebrecht 8a34edc8ca cleanup import satements
- conform to PEP8
- explicitly define symbols instead of 'import *'
- remove unused import

Reviewed-by: Sebastian Spaeth <Sebastian@SSpaeth.de>
Signed-off-by: Nicolas Sebrecht <nicolas.s-dev@laposte.net>
2011-05-12 18:26:29 +02:00

610 lines
19 KiB
Python

# Curses-based interfaces
# Copyright (C) 2003 John Goerzen
# <jgoerzen@complete.org>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
from threading import RLock, Lock, Event
import time
import sys
import os
import signal
import curses
from Blinkenlights import BlinkenBase
from UIBase import UIBase
import offlineimap
acctkeys = '1234567890abcdefghijklmnoprstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-=;/.,'
class CursesUtil:
def __init__(self):
self.pairlock = Lock()
# iolock protects access to the
self.iolock = RLock()
self.start()
def initpairs(self):
self.pairlock.acquire()
try:
self.pairs = {self._getpairindex(curses.COLOR_WHITE,
curses.COLOR_BLACK): 0}
self.nextpair = 1
finally:
self.pairlock.release()
def lock(self):
"""Locks the Curses ui thread
Can be invoked multiple times from the owning thread. Invoking
from a non-owning thread blocks and waits until it has been
unlocked by the owning thread."""
self.iolock.acquire()
def unlock(self):
"""Unlocks the Curses ui thread
Decrease the lock counter by one and unlock the ui thread if the
counter reaches 0. Only call this method when the calling
thread owns the lock. A RuntimeError is raised if this method is
called when the lock is unlocked."""
self.iolock.release()
def locked(self, target, *args, **kwargs):
"""Perform an operation with full locking."""
self.lock()
try:
apply(target, args, kwargs)
finally:
self.unlock()
def refresh(self):
def lockedstuff():
curses.panel.update_panels()
curses.doupdate()
self.locked(lockedstuff)
def isactive(self):
return hasattr(self, 'stdscr')
def _getpairindex(self, fg, bg):
return '%d/%d' % (fg,bg)
def getpair(self, fg, bg):
if not self.has_color:
return 0
pindex = self._getpairindex(fg, bg)
self.pairlock.acquire()
try:
if self.pairs.has_key(pindex):
return curses.color_pair(self.pairs[pindex])
else:
self.pairs[pindex] = self.nextpair
curses.init_pair(self.nextpair, fg, bg)
self.nextpair += 1
return curses.color_pair(self.nextpair - 1)
finally:
self.pairlock.release()
def start(self):
self.stdscr = curses.initscr()
curses.noecho()
curses.cbreak()
self.stdscr.keypad(1)
try:
curses.start_color()
self.has_color = curses.has_colors()
except:
self.has_color = 0
self.oldcursor = None
try:
self.oldcursor = curses.curs_set(0)
except:
pass
self.stdscr.clear()
self.stdscr.refresh()
(self.height, self.width) = self.stdscr.getmaxyx()
self.initpairs()
def stop(self):
if not hasattr(self, 'stdscr'):
return
#self.stdscr.addstr(self.height - 1, 0, "\n",
# self.getpair(curses.COLOR_WHITE,
# curses.COLOR_BLACK))
if self.oldcursor != None:
curses.curs_set(self.oldcursor)
self.stdscr.refresh()
self.stdscr.keypad(0)
curses.nocbreak()
curses.echo()
curses.endwin()
del self.stdscr
def reset(self):
# dirty walkaround for bug http://bugs.python.org/issue7567 in python 2.6 to 2.6.5 (fixed since #83743)
if (sys.version_info[0:3] >= (2,6) and sys.version_info[0:3] <= (2,6,5)): return
self.stop()
self.start()
class CursesAccountFrame:
def __init__(s, master, accountname, ui):
s.c = master
s.children = []
s.accountname = accountname
s.ui = ui
def drawleadstr(s, secs = None):
if secs == None:
acctstr = '%s: [active] %13.13s: ' % (s.key, s.accountname)
else:
acctstr = '%s: [%3d:%02d] %13.13s: ' % (s.key,
secs / 60, secs % 60,
s.accountname)
s.c.locked(s.window.addstr, 0, 0, acctstr)
s.location = len(acctstr)
def setwindow(s, window, key):
s.window = window
s.key = key
s.drawleadstr()
for child in s.children:
child.update(window, 0, s.location)
s.location += 1
def getnewthreadframe(s):
tf = CursesThreadFrame(s.c, s.ui, s.window, 0, s.location)
s.location += 1
s.children.append(tf)
return tf
def startsleep(s, sleepsecs):
s.sleeping_abort = 0
def sleeping(s, sleepsecs, remainingsecs):
if remainingsecs:
s.c.lock()
try:
s.drawleadstr(remainingsecs)
s.window.refresh()
finally:
s.c.unlock()
time.sleep(sleepsecs)
else:
s.c.lock()
try:
s.drawleadstr()
s.window.refresh()
finally:
s.c.unlock()
return s.sleeping_abort
def syncnow(s):
s.sleeping_abort = 1
class CursesThreadFrame:
def __init__(s, master, ui, window, y, x):
"""master should be a CursesUtil object."""
s.c = master
s.ui = ui
s.window = window
s.x = x
s.y = y
s.colors = []
bg = curses.COLOR_BLACK
s.colormap = {'black': s.c.getpair(curses.COLOR_BLACK, bg),
'gray': s.c.getpair(curses.COLOR_WHITE, bg),
'white': curses.A_BOLD | s.c.getpair(curses.COLOR_WHITE, bg),
'blue': s.c.getpair(curses.COLOR_BLUE, bg),
'red': s.c.getpair(curses.COLOR_RED, bg),
'purple': s.c.getpair(curses.COLOR_MAGENTA, bg),
'cyan': s.c.getpair(curses.COLOR_CYAN, bg),
'green': s.c.getpair(curses.COLOR_GREEN, bg),
'orange': s.c.getpair(curses.COLOR_YELLOW, bg),
'yellow': curses.A_BOLD | s.c.getpair(curses.COLOR_YELLOW, bg),
'pink': curses.A_BOLD | s.c.getpair(curses.COLOR_RED, bg)}
#s.setcolor('gray')
s.setcolor('black')
def setcolor(self, color):
self.color = self.colormap[color]
self.colorname = color
self.display()
def display(self):
def lockedstuff():
if self.getcolor() == 'black':
self.window.addstr(self.y, self.x, ' ', self.color)
else:
self.window.addstr(self.y, self.x, self.ui.config.getdefault("ui.Curses.Blinkenlights", "statuschar", '.'), self.color)
self.c.stdscr.move(self.c.height - 1, self.c.width - 1)
self.window.refresh()
self.c.locked(lockedstuff)
def getcolor(self):
return self.colorname
def getcolorpair(self):
return self.color
def update(self, window, y, x):
self.window = window
self.y = y
self.x = x
self.display()
def setthread(self, newthread):
self.setcolor('black')
#if newthread:
# self.setcolor('gray')
#else:
# self.setcolor('black')
class InputHandler:
def __init__(s, util):
s.c = util
s.bgchar = None
s.inputlock = Lock()
s.lockheld = 0
s.statuslock = Lock()
s.startup = Event()
s.startthread()
def startthread(s):
s.thread = offlineimap.threadutil.ExitNotifyThread(target = s.bgreaderloop,
name = "InputHandler loop")
s.thread.setDaemon(1)
s.thread.start()
def bgreaderloop(s):
while 1:
s.statuslock.acquire()
if s.lockheld or s.bgchar == None:
s.statuslock.release()
s.startup.wait()
else:
s.statuslock.release()
ch = s.c.stdscr.getch()
s.statuslock.acquire()
try:
if s.lockheld or s.bgchar == None:
curses.ungetch(ch)
else:
s.bgchar(ch)
finally:
s.statuslock.release()
def set_bgchar(s, callback):
"""Sets a "background" character handler. If a key is pressed
while not doing anything else, it will be passed to this handler.
callback is a function taking a single arg -- the char pressed.
If callback is None, clears the request."""
s.statuslock.acquire()
oldhandler = s.bgchar
newhandler = callback
s.bgchar = callback
if oldhandler and not newhandler:
pass
if newhandler and not oldhandler:
s.startup.set()
s.statuslock.release()
def input_acquire(s):
"""Call this method when you want exclusive input control.
Make sure to call input_release afterwards!
"""
s.inputlock.acquire()
s.statuslock.acquire()
s.lockheld = 1
s.statuslock.release()
def input_release(s):
"""Call this method when you are done getting input."""
s.statuslock.acquire()
s.lockheld = 0
s.statuslock.release()
s.inputlock.release()
s.startup.set()
class Blinkenlights(BlinkenBase, UIBase):
def init_banner(s):
s.af = {}
s.aflock = Lock()
s.c = CursesUtil()
s.text = []
BlinkenBase.init_banner(s)
s.setupwindows()
s.inputhandler = InputHandler(s.c)
s.gettf().setcolor('red')
s._msg(offlineimap.banner)
s.inputhandler.set_bgchar(s.keypress)
signal.signal(signal.SIGWINCH, s.resizehandler)
s.resizelock = Lock()
s.resizecount = 0
def resizehandler(s, signum, frame):
s.resizeterm()
def resizeterm(s, dosleep = 1):
if not s.resizelock.acquire(0):
s.resizecount += 1
return
signal.signal(signal.SIGWINCH, signal.SIG_IGN)
s.aflock.acquire()
s.c.lock()
s.resizecount += 1
while s.resizecount:
s.c.reset()
s.setupwindows()
s.resizecount -= 1
s.c.unlock()
s.aflock.release()
s.resizelock.release()
signal.signal(signal.SIGWINCH, s.resizehandler)
if dosleep:
time.sleep(1)
s.resizeterm(0)
def isusable(s):
# Not a terminal? Can't use curses.
if not sys.stdout.isatty() and sys.stdin.isatty():
return 0
# No TERM specified? Can't use curses.
try:
if not len(os.environ['TERM']):
return 0
except: return 0
# ncurses doesn't want to start? Can't use curses.
# This test is nasty because initscr() actually EXITS on error.
# grr.
pid = os.fork()
if pid:
# parent
return not os.WEXITSTATUS(os.waitpid(pid, 0)[1])
else:
# child
curses.initscr()
curses.endwin()
# If we didn't die by here, indicate success.
sys.exit(0)
def keypress(s, key):
if key < 1 or key > 255:
return
if chr(key) == 'q':
# Request to quit.
s.terminate()
try:
index = acctkeys.index(chr(key))
except ValueError:
# Key not a valid one: exit.
return
if index >= len(s.hotkeys):
# Not in our list of valid hotkeys.
return
# Trying to end sleep somewhere.
s.getaccountframe(s.hotkeys[index]).syncnow()
def getpass(s, accountname, config, errmsg = None):
s.inputhandler.input_acquire()
# See comment on _msg for info on why both locks are obtained.
s.tflock.acquire()
s.c.lock()
try:
s.gettf().setcolor('white')
s._addline(" *** Input Required", s.gettf().getcolorpair())
s._addline(" *** Please enter password for account %s: " % accountname,
s.gettf().getcolorpair())
s.logwindow.refresh()
password = s.logwindow.getstr()
finally:
s.tflock.release()
s.c.unlock()
s.inputhandler.input_release()
return password
def setupwindows(s):
s.c.lock()
try:
s.bannerwindow = curses.newwin(1, s.c.width, 0, 0)
s.setupwindow_drawbanner()
s.logheight = s.c.height - 1 - len(s.af.keys())
s.logwindow = curses.newwin(s.logheight, s.c.width, 1, 0)
s.logwindow.idlok(1)
s.logwindow.scrollok(1)
s.logwindow.move(s.logheight - 1, 0)
s.setupwindow_drawlog()
accounts = s.af.keys()
accounts.sort()
accounts.reverse()
pos = s.c.height - 1
index = 0
s.hotkeys = []
for account in accounts:
accountwindow = curses.newwin(1, s.c.width, pos, 0)
s.af[account].setwindow(accountwindow, acctkeys[index])
s.hotkeys.append(account)
index += 1
pos -= 1
curses.doupdate()
finally:
s.c.unlock()
def setupwindow_drawbanner(s):
if s.c.has_color:
color = s.c.getpair(curses.COLOR_WHITE, curses.COLOR_BLUE) | \
curses.A_BOLD
else:
color = curses.A_REVERSE
s.bannerwindow.bkgd(' ', color) # Fill background with that color
s.bannerwindow.addstr("%s %s" % (offlineimap.__productname__,
offlineimap.__version__))
s.bannerwindow.addstr(0, s.bannerwindow.getmaxyx()[1] - len(offlineimap.__copyright__) - 1,
offlineimap.__copyright__)
s.bannerwindow.noutrefresh()
def setupwindow_drawlog(s):
if s.c.has_color:
color = s.c.getpair(curses.COLOR_WHITE, curses.COLOR_BLACK)
else:
color = curses.A_NORMAL
s.logwindow.bkgd(' ', color)
for line, color in s.text:
s.logwindow.addstr("\n" + line, color)
s.logwindow.noutrefresh()
def getaccountframe(s, accountname = None):
if accountname == None:
accountname = s.getthreadaccount()
s.aflock.acquire()
try:
if accountname in s.af:
return s.af[accountname]
# New one.
s.af[accountname] = CursesAccountFrame(s.c, accountname, s)
s.c.lock()
try:
s.c.reset()
s.setupwindows()
finally:
s.c.unlock()
finally:
s.aflock.release()
return s.af[accountname]
def _display(s, msg, color = None):
if "\n" in msg:
for thisline in msg.split("\n"):
s._msg(thisline)
return
# We must acquire both locks. Otherwise, deadlock can result.
# This can happen if one thread calls _msg (locking curses, then
# tf) and another tries to set the color (locking tf, then curses)
#
# By locking both up-front here, in this order, we prevent deadlock.
s.tflock.acquire()
s.c.lock()
try:
if not s.c.isactive():
# For dumping out exceptions and stuff.
print msg
return
if color:
s.gettf().setcolor(color)
elif s.gettf().getcolor() == 'black':
s.gettf().setcolor('gray')
s._addline(msg, s.gettf().getcolorpair())
s.logwindow.refresh()
finally:
s.c.unlock()
s.tflock.release()
def _addline(s, msg, color):
s.c.lock()
try:
s.logwindow.addstr("\n" + msg, color)
s.text.append((msg, color))
while len(s.text) > s.logheight:
s.text = s.text[1:]
finally:
s.c.unlock()
def terminate(s, exitstatus = 0, errortitle = None, errormsg = None):
s.c.stop()
UIBase.terminate(s, exitstatus = exitstatus, errortitle = errortitle, errormsg = errormsg)
def threadException(s, thread):
s.c.stop()
UIBase.threadException(s, thread)
def mainException(s):
s.c.stop()
UIBase.mainException(s)
def sleep(s, sleepsecs, account):
s.gettf().setcolor('red')
s._msg("Next sync in %d:%02d" % (sleepsecs / 60, sleepsecs % 60))
return BlinkenBase.sleep(s, sleepsecs, account)
if __name__ == '__main__':
x = Blinkenlights(None)
x.init_banner()
import time
time.sleep(5)
x.c.stop()
fgs = {'black': curses.COLOR_BLACK, 'red': curses.COLOR_RED,
'green': curses.COLOR_GREEN, 'yellow': curses.COLOR_YELLOW,
'blue': curses.COLOR_BLUE, 'magenta': curses.COLOR_MAGENTA,
'cyan': curses.COLOR_CYAN, 'white': curses.COLOR_WHITE}
x = CursesUtil()
win1 = curses.newwin(x.height, x.width / 4 - 1, 0, 0)
win1.addstr("Black/normal\n")
for name, fg in fgs.items():
win1.addstr("%s\n" % name, x.getpair(fg, curses.COLOR_BLACK))
win2 = curses.newwin(x.height, x.width / 4 - 1, 0, win1.getmaxyx()[1])
win2.addstr("Blue/normal\n")
for name, fg in fgs.items():
win2.addstr("%s\n" % name, x.getpair(fg, curses.COLOR_BLUE))
win3 = curses.newwin(x.height, x.width / 4 - 1, 0, win1.getmaxyx()[1] +
win2.getmaxyx()[1])
win3.addstr("Black/bright\n")
for name, fg in fgs.items():
win3.addstr("%s\n" % name, x.getpair(fg, curses.COLOR_BLACK) | \
curses.A_BOLD)
win4 = curses.newwin(x.height, x.width / 4 - 1, 0, win1.getmaxyx()[1] * 3)
win4.addstr("Blue/bright\n")
for name, fg in fgs.items():
win4.addstr("%s\n" % name, x.getpair(fg, curses.COLOR_BLUE) | \
curses.A_BOLD)
win1.refresh()
win2.refresh()
win3.refresh()
win4.refresh()
x.stdscr.refresh()
import time
time.sleep(5)
x.stop()
print x.has_color
print x.height
print x.width