def _test_process_input(self):
"""
>>> w = _refl("widget")
>>> w.selectable_rval = True
>>> scr = _refl("screen")
>>> scr.get_cols_rows_rval = (10, 5)
>>> ml = MainLoop(w, [], scr)
>>> ml.process_input(['enter', ('mouse drag', 1, 14, 20)])
screen.get_cols_rows()
widget.selectable()
widget.keypress((10, 5), 'enter')
widget.mouse_event((10, 5), 'mouse drag', 1, 14, 20, focus=True)
True
"""
python类MainLoop()的实例源码
def __init__(self):
from gi.repository import GLib
self.GLib = GLib
self._alarms = []
self._watch_files = {}
self._idle_handle = 0
self._glib_idle_enabled = False # have we called glib.idle_add?
self._idle_callbacks = {}
self._loop = GLib.MainLoop()
self._exc_info = None
self._enable_glib_idle()
def _get_loop(seat):
"Return a tuple of (main loop, quit loop fn)"
loop = GLib.MainLoop()
def loopquit():
seat.ungrab()
loop.quit()
return loop, loopquit
def main():
'''Dispatches control to the appropriate parts of the app'''
args = parse_args()
config = get_config()
if not dependencies_are_satisfied():
return 'This program depends on xprintidle and xdotool being installed.'
config.acpi_available = battery.acpi_available()
if args.delete:
config = get_config()
for file_ in (config.config_file, config.desktop_filename):
try:
os.unlink(file_)
except FileNotFoundError:
print('{}: File not found'.format(file_))
return 0
elif args.mode == 'indicator':
if args.battery is not None:
config.battery = args.battery
if args.percent is not None:
config.batt_percent = args.percent
inhibitor = sleepinhibit.gui.main.SleepInhibitGUI()
if config.start_inhibited or args.inhibit:
inhibitor.on_toggle()
try:
GLib.MainLoop().run()
except KeyboardInterrupt:
inhibitor.on_quit(signal='SIGINT')
return 0
elif args.mode == 'inhibit-process':
from sleepinhibit.inhibitor import run
kw = {}
if args.battery is True:
kw['battery'] = True
if args.percent:
kw['percent'] = args.percent
try:
run(**kw)
except KeyboardInterrupt:
return 0
else:
return 'ERROR: Invalid value for --mode!'
def pomodoro_daemon():
pomodoro = get_pomodoro_proxy()
pomodoro.StateChanged.connect(handle_state)
GLib.MainLoop().run()
def run(self):
loop = GLib.MainLoop()
loop.run()
def main():
from gi.repository import GLib
from . import base
parser = argparse.ArgumentParser(description='Command Interest Tests')
parser.add_argument("--faceid", "-i", type=int)
parser.add_argument("--uri", "-u")
parser.add_argument("--local-control-feature", "-l")
parser.add_argument("--origin", "-o", type=int)
parser.add_argument("--cost", "-c", type=int)
parser.add_argument("--forwarding-flags", "-F", type=int)
parser.add_argument("--strategy", "-s")
parser.add_argument("--expiration-period", "-e", type=int)
parser.add_argument("command")
args = utils.parse_args(parser=parser)
controlParameters = ControlParameters()
name = None
if args.name:
name = Name(args.name)
if args.faceid:
controlParameters.setFaceId(args.faceid)
if args.uri:
controlParameters.setUri(args.uri)
if args.local_control_feature:
controlParameters.setLocalControlFeature(args.local_control_feature)
if args.origin:
controlParameters.setOrigin(args.origin)
if args.cost:
controlParameters.setCost(args.cost)
if args.forwarding_flags:
controlParameters.setForwardingFlags(args.forwarding_flags)
if args.strategy:
controlParameters.setStrategy(Name(args.strategy))
if args.expiration_period:
controlParameters.setExpirationPeriod(args.expiration_period)
loop = GLib.MainLoop()
def print_and_quit(*args, **kwargs):
logger.info(*args, **kwargs)
loop.quit()
logger.info('running command: %s on %s', args.command, name)
ndn = base.Base(name)
ndn.expressCommandInterest(
args.command, name, controlParameters=controlParameters,
interestLifeTime=10000,
onFailed=lambda *
a: print_and_quit('FAILED: %s', a),
onSuccess=lambda *a: print_and_quit('SUCCESS: %s', a))
loop.run()