def _startTimer(self):
self._updateData()
self._showStatus()
sublime.set_timeout_async(lambda: self._startTimer(), self._updateInterval * 1e3)
python类set_timeout_async()的实例源码
def updateLanguage(self, index=None):
if index == -1 or index == None:
return
languageName = re.search('^\w+', self.languageList[index]).group(0)
self.languageName = languageName
sublime.set_timeout_async(self.checkoutLanguage, 0)
def on_selection_modified_async(self, view):
if not getSetting('auto'):
return
global currentView
currentView = view
self.prevTime = time.time()
if not self.delaying:
sublime.set_timeout_async(self.doAutoShow, getSetting('auto_delay') + 50)
self.delaying = True
def doAutoShow(self):
delayTime = getSetting('auto_delay')
if (time.time() - self.prevTime) * 1000 > delayTime:
self.delaying = False
if not currentView.is_popup_visible():
currentView.run_command('docphp_show_definition')
else:
sublime.set_timeout_async(self.doAutoShow, int(delayTime - (time.time() - self.prevTime) * 1000) + 50)
def run(self, **args):
async = args['async']
action_id = args['action_id']
action, delay = code_map_marshaler._actions.pop(action_id)
if async:
sublime.set_timeout_async(action, delay)
else:
sublime.set_timeout(action, delay)
# =============================================================================
def run(self, edit):
point = self.view.sel()[0].a
line_region = self.view.line(point)
self.view.sel().clear()
self.view.sel().add(line_region)
sublime.set_timeout_async(lambda: self.view.sel().add(line_region), 10)
# =============================================================================
def on_close(self, view):
if ACTIVE and view.file_name() == code_map_file:
reset_globals()
if settings().get('close_empty_group_on_closing_map', False):
reset_layout()
sublime.set_timeout_async(focus_source_code)
# -----------------
def init():
log("Initializing icons")
if not os.path.exists(path.get_overlay()):
_create_dirs()
sublime.set_timeout_async(provide, 0)
else:
sublime.set_timeout_async(_copy_specific, 0)
dump("All the necessary icons are provided")
def enable():
log("Enabling aliases")
if not _is_enabled():
if settings.is_package_archive():
sublime.set_timeout_async(_extract, 0)
else:
sublime.set_timeout_async(_copy, 0)
else:
dump("Aliases already enabled")
def disable():
log("Disabling aliases")
if _is_enabled():
sublime.set_timeout_async(_remove, 0)
else:
dump("Aliases already disabled")
def on_close(self, view):
if is_supported_syntax(view.settings().get("syntax")):
Events.publish("view.on_close", view)
sublime.set_timeout_async(check_window_unloaded, 500)
def run(self):
view = self.window.active_view()
available_config = get_scope_client_config(view, client_configs.defaults) or get_default_client_config(view)
if available_config:
client_configs.enable(available_config.name)
clear_window_client_configs(self.window)
sublime.set_timeout_async(lambda: start_view(view), 500)
self.window.status_message("{} enabled, starting server...".format(available_config.name))
return
self.window.status_message("No config available to enable")
def run(self):
view = self.window.active_view()
# if no default_config, nothing we can do.
default_config = get_default_client_config(view)
if default_config:
enable_in_project(self.window, default_config.name)
clear_window_client_configs(self.window)
sublime.set_timeout_async(lambda: start_view(view), 500)
self.window.status_message("{} enabled in project, starting server...".format(default_config.name))
else:
self.window.status_message("No config available to enable")
def run(self):
view = self.window.active_view()
global_config = get_scope_client_config(view, client_configs.defaults)
if global_config:
disable_in_project(self.window, global_config.name)
clear_window_client_configs(self.window)
sublime.set_timeout_async(lambda: unload_window_clients(self.window.id()), 500)
self.window.status_message("{} disabled in project, shutting down server...".format(global_config.name))
return
else:
self.window.status_message("No config available to disable")
def finish(self, proc):
super(PytestExecCommand, self).finish(proc)
view = self.output_view
output = get_whole_text(view).strip()
last_line = output[output.rfind('\n'):]
summary = ''
matches = re.finditer(r' ([\d]+) ', last_line)
if matches:
test_count = sum(int(m.group(1)) for m in matches)
summary = "Ran %s tests. " % (test_count)
summary += last_line.replace('=', '')
failures = proc.exit_code() != 0
if failures:
broadcast("pytest_will_fail")
broadcast('pytest_finished', {
"summary": summary,
"failures": failures
})
# This is a 'best' guess. Maybe we should parse the output for the
# `rootdir` pytest uses.
base_dir = view.settings().get('result_base_dir')
# For the 'line' go with output regex parsing bc the reporter
# isn't useful at all.
if self._tb_mode == 'line':
sublime.set_timeout_async(
functools.partial(
parse_output, output, base_dir, Matchers[self._tb_mode]))
else:
sublime.set_timeout_async(
functools.partial(
parse_result, base_dir, Matchers[self._tb_mode]))
def prevent_spam(func):
"""Prevent spamming in the logger
"""
_last_messages = {}
def _remove_from_cache(args):
m = _last_messages.pop(args)
if m == 1:
return
method = getattr(Log._logger, args[0])
method(
'{}\n ...last message repeated {} in the last 10s'.format(
args[1], 'one more time' if m == 2 else '{} times'.format(m)
)
)
@functools.wraps(func)
def wrapper(cls, *args, **kwargs):
if args in _last_messages:
_last_messages[args] += 1
return
_last_messages[args] = 1
sublime.set_timeout_async(lambda: _remove_from_cache(args), 10000)
func(cls, args[0], *args[1:], **kwargs)
return wrapper
def execute(cls, callback, **data):
"""Execute the given method remotely and call the callback with result
"""
def _start_worker(wk, cb, **d):
wk.start()
if wk.status == WorkerStatus.healthy:
wk._execute(cb, **d)
return
sublime.set_timeout_async(lambda: _start_worker(wk, cb, **d), 5000)
window_id = sublime.active_window().id()
worker = cls.get(cls, window_id)
if worker is None:
# hire a new worker
worker = cls.hire(cls)
cls.add(cls, window_id, worker)
if worker.status == WorkerStatus.faulty:
return
if worker.status == WorkerStatus.quiting:
cls.fire(cls, window_id)
return
if worker.client is not None:
if not worker.client.connected:
worker.reconnecting = True
worker.state = WorkerStatus.incomplete
_start_worker(worker, callback, **data)
else:
worker._append_context_data(data)
worker._execute(callback, **data)
if worker.status == WorkerStatus.quiting:
# that means that we need to let the worker go
cls.fire(cls, window_id)
else:
_start_worker(worker, callback, **data)
def generate_menus(**kwargs):
"""Asynchronously call generate_menus_async."""
sublime.set_timeout_async(generate_menus_async, 0)
def update_log_windows(restart_timer=True):
global project_env_map
try:
for (project_path, env) in list(project_env_map.items()):
# Clean up project windows first
if not ProjectManager.is_electric_imp_project_window(env.window):
# It's not a windows that corresponds to an EI project, remove it from the list
del project_env_map[project_path]
log_debug("Removing project window: " + str(env.window) + ", total #: " + str(len(project_env_map)))
continue
env.log_manager.update_logs()
finally:
if restart_timer:
sublime.set_timeout_async(update_log_windows, PL_LOGS_UPDATE_PERIOD)
def run(self,edit):
mname = getModuleName(self.view)
if not mname:
print('[VHDL.navigation] No entity/architecture found !')
return
txt = self.view.substr(sublime.Region(0, self.view.size()))
inst_l = vhdl_util.get_inst_list(txt,mname)
if not inst_l:
print('[VHDL.navigation] No hierarchy found !')
return
sublime.status_message("Show Hierarchy can take some time, please wait ...")
sublime.set_timeout_async(lambda inst_l=inst_l, w=self.view.window(), mname=mname : self.showHierarchy(w,inst_l,mname))
