本文整理汇总了Python中x84.bbs.session.getsession函数的典型用法代码示例。如果您正苦于以下问题:Python getsession函数的具体用法?Python getsession怎么用?Python getsession使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了getsession函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self, cmd='/bin/uname', args=(), env=None, cp437=False):
"""
Class initializer.
:param str cmd: full path of command to execute.
:param tuple args: command arguments as tuple.
:param bool cp437: When true, forces decoding of external program as
codepage 437. This is the most common encoding used
by DOS doors.
:param dict env: Environment variables to extend to the sub-process.
You should more than likely specify values for TERM,
PATH, HOME, and LANG.
"""
self._session, self._term = getsession(), getterminal()
self.cmd = cmd
if isinstance(args, tuple):
self.args = (self.cmd,) + args
elif isinstance(args, list):
self.args = [self.cmd, ] + args
else:
raise ValueError('args must be tuple or list')
self.log = logging.getLogger(__name__)
self.env = (env or {}).copy()
self.env.update(
{'LANG': env.get('LANG', 'en_US.UTF-8'),
'TERM': env.get('TERM', self._term.kind),
'PATH': env.get('PATH', get_ini('door', 'path')),
'HOME': env.get('HOME', os.getenv('HOME')),
'LINES': str(self._term.height),
'COLUMNS': str(self._term.width),
})
self.cp437 = cp437
self._utf8_decoder = codecs.getincrementaldecoder('utf8')()
开发者ID:adammendoza,项目名称:x84,代码行数:35,代码来源:door.py
示例2: echo
def echo(ucs):
"""
Display unicode terminal sequence.
"""
session = getsession()
if not isinstance(ucs, unicode):
warnings.warn('non-unicode: %r' % (ucs,), UnicodeWarning, 2)
return session.write(ucs.decode('iso8859-1'))
return session.write(ucs)
开发者ID:quastdog,项目名称:x84,代码行数:9,代码来源:output.py
示例3: _decode
def _decode(what):
# pylint: disable=C0111
# Missing function docstring (col 8)
session = getsession()
if session.encoding == 'utf8':
return what.decode(encoding)
elif session.encoding == 'cp437':
return what.decode('cp437')
else:
return what
开发者ID:tehmaze,项目名称:x84,代码行数:10,代码来源:output.py
示例4: read
def read(self):
"""
Reads input until ESCAPE key is pressed (Blocking). Returns None.
"""
from x84.bbs.session import getsession
from x84.bbs.output import echo
session = getsession()
self._quit = False
echo(self.refresh())
while not self.quit:
echo(self.process_keystroke(session.read_event('input')))
开发者ID:quastdog,项目名称:x84,代码行数:11,代码来源:pager.py
示例5: __init__
def __init__(self, schema, table='unnamed', use_session=True):
"""
Arguments:
schema: database key, to become basename of .sqlite3 files.
"""
from x84.bbs.session import getsession
self.log = logging.getLogger(__name__)
self.schema = schema
self.table = table
self._tap_db = should_tapdb()
self.session = use_session and getsession()
开发者ID:hick,项目名称:x84,代码行数:11,代码来源:dbproxy.py
示例6: __init__
def __init__(self, recipient=None, subject=u'', body=u''):
from x84.bbs.session import getsession
self._ctime = datetime.datetime.now()
self._stime = None
self.author = getsession().handle
self.recipient = recipient
self.subject = subject
self.body = body
self.tags = set()
# reply-to tracking
self.children = set()
self.parent = None
开发者ID:donfanning,项目名称:x84,代码行数:12,代码来源:msgbase.py
示例7: send_modem
def send_modem(stream, protocol='xmodem1k', retry=16, timeout=30,
callback=None):
"""
Send a file using 'xmodem1k' or 'xmodem' protocol.
Currently, these are the only protocols supported. Returns ``True`` upon
successful transmission, otherwise ``False``.
:param stream: The file-like stream object to send data from.
:param int retry: The maximum number of times to try to resend a failed
packet before failing.
:param int timeout: seconds to elapse for response before failing.
:param callable callback: Reference to a callback function that has the
following signature. This is useful for getting
status updates while a transfer is underway::
def callback(total_count, success_count, error_count)
"""
# get protocol implementation class
supported_protocols = ('xmodem', 'xmodem1k')
assert protocol in supported_protocols, (protocol, supported_protocols)
Modem = {
'xmodem': xmodem.XMODEM,
'xmodem1k': xmodem.XMODEM1k,
}[protocol]
# the session's 'input' event buffer is used for receiving
# transmissions. It arrives in raw bytes, and session.write
# is used, sending "unicode" data as encoding iso8859-1.
session = getsession()
def getc(size, timeout=10):
""" Callback function for (X)Modem interface. """
val = b''
while len(val) < size:
next_val = session.read_event('input', timeout=timeout)
if next_val is None:
break
val += next_val
if len(val) > size:
session.buffer_input(val[size:])
return val[:size] or None
def putc(data, timeout=10):
# pylint: disable=W0613
# Unused argument 'timeout'
""" Callback function for (X)Modem interface. """
session.write(data.decode('iso8859-1'), 'iso8859-1')
modem = Modem(getc, putc)
return modem.send(stream=stream, retry=retry, timeout=timeout,
quiet=True, callback=callback)
开发者ID:adammendoza,项目名称:x84,代码行数:52,代码来源:modem.py
示例8: init_theme
def init_theme(self):
"""
This initializer sets glyphs and colors appropriate for a "theme",
override this method to create a common color and graphic set.
"""
session = getsession()
term = getterminal()
self.colors['normal'] = term.normal
if term.number_of_colors != 0:
self.colors['border'] = term.cyan
# start with default 'ascii'
self.glyphs = GLYPHSETS['ascii'].copy()
# PC-DOS 'thin' on smart terminals
if session.env.get('TERM') != 'unknown':
self.glyphs = GLYPHSETS['thin'].copy()
开发者ID:donfanning,项目名称:x84,代码行数:15,代码来源:ansiwin.py
示例9: __init__
def __init__(self, recipient=None, subject=u'', body=u''):
self.author = None
session = getsession()
if session:
self.author = session.user.handle
self._ctime = datetime.datetime.now()
self._stime = None
self.recipient = recipient
self.subject = subject
self.body = body
self.tags = set()
self.children = set()
self.parent = None
self.idx = None
开发者ID:adammendoza,项目名称:x84,代码行数:15,代码来源:msgbase.py
示例10: __init__
def __init__(self, recipient=None, subject=u'', body=u''):
from x84.bbs.session import getsession
self.author = None
session = getsession()
if session:
self.author = session.handle
# msg attributes (todo: create method ..)
self._ctime = datetime.datetime.now()
self._stime = None
self.recipient = recipient
self.subject = subject
self.body = body
self.tags = set()
self.children = set()
self.parent = None
开发者ID:ztaylor,项目名称:x84,代码行数:16,代码来源:msgbase.py
示例11: read
def read(self):
"""
Reads input until the ENTER or ESCAPE key is pressed (Blocking).
Allows backspacing. Returns unicode text, or None when cancelled.
"""
from x84.bbs.session import getsession
from x84.bbs.output import echo
session = getsession()
self._selected = False
self._quit = False
echo(self.refresh())
while not (self.selected or self.quit):
echo(self.process_keystroke(session.read_event('input')) or u'')
if self.quit:
return None
return self.selection
开发者ID:quastdog,项目名称:x84,代码行数:16,代码来源:selector.py
示例12: __init__
def __init__(self, schema, table='unnamed', use_session=True):
"""
Class constructor.
:param str scheme: database key, becomes basename of .sqlite3 file.
:param str table: optional database table.
:param bool use_session: Whether iterable returns should be sent over
an IPC pipe (client is a
:class:`x84.bbs.session.Session` instance),
or returned directly (such as used by the main
thread engine components.)
"""
self.log = logging.getLogger(__name__)
self.schema = schema
self.table = table
self._tap_db = get_ini('session', 'tab_db', getter='getboolean')
self.session = use_session and getsession()
开发者ID:tehmaze,项目名称:x84,代码行数:17,代码来源:dbproxy.py
示例13: read
def read(self):
"""
Reads input until the ENTER or ESCAPE key is pressed (Blocking).
Allows backspacing. Returns unicode text, or None when cancelled.
"""
from x84.bbs.session import getsession
from x84.bbs.output import echo
session = getsession()
echo(self.refresh())
self._quit = False
self._carriage_returned = False
while not (self.quit or self.carriage_returned):
inp = session.read_event("input")
echo(self.process_keystroke(inp))
if not self.quit:
return self.content
return None
开发者ID:sweenzor,项目名称:x84,代码行数:18,代码来源:editor.py
示例14: emit
def emit(self, record):
""" Emit log record via IPC output queue. """
try:
e_inf = record.exc_info
if e_inf:
# a strange side-effect,
# sets record.exc_text
dummy = self.format(record) # NOQA
record.exc_info = None
record.handle = None
session = getsession()
if session:
record.handle = session.handle
self.oqueue.send(('logger', record))
except (KeyboardInterrupt, SystemExit):
raise
except Exception:
self.handleError(record)
开发者ID:tehmaze,项目名称:x84,代码行数:18,代码来源:ipc.py
示例15: __init__
def __init__(self, cmd='/bin/uname', args=(), env_lang='en_US.UTF-8',
env_term=None, env_path=None, env_home=None, cp437=False,
env=None):
"""
Class constructor.
:param str cmd: full path of command to execute.
:param tuple args: command arguments as tuple.
:param str env_lang: exported as environment variable ``LANG``.
:param str env_term: exported as environment variable ``TERM``. When
unspecified, it is determined by the same
TERM value the original ``blessed.Terminal``
instance is used.
:param str env_path: exported as environment variable ``PATH``.
When None (default), the .ini ``env_path``
value of section ``[door]`` is
:param str env_home: exported as environment variable ``HOME``.
When env_home is ``None``, the environment
value of the main process is used.
:param bool cp437: When true, forces decoding of external program as
codepage 437. This is the most common encoding used
by DOS doors.
:param dict env: Additional environment variables to extend to the
sub-process.
"""
self._session, self._term = getsession(), getterminal()
self.cmd = cmd
if isinstance(args, tuple):
self.args = (self.cmd,) + args
elif isinstance(args, list):
self.args = [self.cmd, ] + args
else:
raise ValueError('args must be tuple or list')
self.env_lang = env_lang
self.env_term = env_term or self._term.kind
self.env_path = env_path or get_ini('door', 'path')
self.env_home = env_home or os.getenv('HOME')
self.env = env or {}
self.cp437 = cp437
self._utf8_decoder = codecs.getincrementaldecoder('utf8')()
开发者ID:tehmaze,项目名称:x84,代码行数:40,代码来源:door.py
示例16: __len__
def __len__(self):
"""
Return the printed length of a string that contains (some types) of
ansi sequences. Although accounted for, strings containing sequences
such as cls() will not give accurate returns. backspace, delete, and
double-wide east-asian
"""
# 'nxt' points to first *ch beyond current ansi sequence, if any.
# 'width' is currently estimated display length.
nxt, width = 0, 0
# i regret the heavy re-instantiation of Ansi() ..
for idx in range(0, unicode.__len__(self)):
width += Ansi(self[idx:]).anspadd()
if idx == nxt:
nxt = idx + Ansi(self[idx:]).seqlen()
if nxt <= idx:
ucs = self[idx]
if getsession().encoding == 'cp437':
wide = 1
else:
# 'East Asian Fullwidth' and 'East Asian Wide' characters
# can take 2 cells, see
# http://www.unicode.org/reports/tr11/
# http://www.gossamer-threads.com/lists/python/bugs/972834
# we just use wcswidth, since that is what terminal
# client implementors seem to be using ..
wide = wcswidth(ucs)
# TODO
# my own NVT addition: allow -1 to be added to width when
# 127 and 8 are used (BACKSPACE, DEL); as well as \x0f .!?
# assert wide != -1 or ucs in (u'\b',
# unichr(127),
# unichr(15)), (
# 'indeterminate length %r in %r' % (self[idx], self))
width += wide if wide != -1 else 0
nxt = idx + Ansi(self[idx:]).seqlen() + 1
return width
开发者ID:quastdog,项目名称:x84,代码行数:39,代码来源:output.py
示例17: alias
def alias(self):
""" current session's handle. """
return getsession().user.handle
开发者ID:adammendoza,项目名称:x84,代码行数:3,代码来源:door.py
示例18: save
def save(self, send_net=True, ctime=None):
"""
Save message in 'Msgs' sqlite db, and record index in 'tags' db.
"""
from x84.bbs.ini import CFG
from x84.bbs import getsession
session = getsession()
use_session = bool(session is not None)
log = logging.getLogger(__name__)
new = self.idx is None or self._stime is None
# persist message record to MSGDB
db_msg = DBProxy(MSGDB, use_session=use_session)
with db_msg:
if new:
self.idx = max([int(key) for key in db_msg.keys()] or [-1]) + 1
if ctime is not None:
self._ctime = self._stime = ctime
else:
self._stime = datetime.datetime.now()
new = True
db_msg['%d' % (self.idx,)] = self
# persist message idx to TAGDB
db_tag = DBProxy(TAGDB, use_session=use_session)
with db_tag:
for tag in db_tag.keys():
msgs = db_tag[tag]
if tag in self.tags and self.idx not in msgs:
msgs.add(self.idx)
db_tag[tag] = msgs
log.debug("msg {self.idx} tagged '{tag}'"
.format(self=self, tag=tag))
elif tag not in self.tags and self.idx in msgs:
msgs.remove(self.idx)
db_tag[tag] = msgs
log.info("msg {self.idx} removed tag '{tag}'"
.format(self=self, tag=tag))
for tag in [_tag for _tag in self.tags if _tag not in db_tag]:
db_tag[tag] = set([self.idx])
# persist message as child to parent;
if not hasattr(self, 'parent'):
self.parent = None
assert self.parent not in self.children
if self.parent is not None:
parent_msg = get_msg(self.parent)
if self.idx != parent_msg.idx:
parent_msg.children.add(self.idx)
parent_msg.save()
else:
log.error('Parent idx same as message idx; stripping')
self.parent = None
with db_msg:
db_msg['%d' % (self.idx)] = self
if send_net and new and CFG.has_option('msg', 'network_tags'):
self.queue_for_network()
log.info(
u"saved {new}{public}msg {post}, addressed to '{self.recipient}'."
.format(new='new ' if new else '',
public='public ' if 'public' in self.tags else '',
post='post' if self.parent is None else 'reply',
self=self))
开发者ID:ztaylor,项目名称:x84,代码行数:66,代码来源:msgbase.py
示例19: lastcall_time
def lastcall_time(self):
""" Time of last call (format is ``%H:%M``). """
return time.strftime(
'%H:%M', time.localtime(getsession().user.lastcall))
开发者ID:adammendoza,项目名称:x84,代码行数:4,代码来源:door.py
示例20: lastcall_date
def lastcall_date(self):
""" Date of last call (format is ``%m/%d/%y``). """
return time.strftime(
'%m/%d/%y', time.localtime(getsession().user.lastcall))
开发者ID:adammendoza,项目名称:x84,代码行数:4,代码来源:door.py
注:本文中的x84.bbs.session.getsession函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论