diff options
Diffstat (limited to '.emacs.d.back/.python-environments/default/lib/python3.7/site-packages/jediepcserver.py')
| -rw-r--r-- | .emacs.d.back/.python-environments/default/lib/python3.7/site-packages/jediepcserver.py | 496 |
1 files changed, 496 insertions, 0 deletions
diff --git a/.emacs.d.back/.python-environments/default/lib/python3.7/site-packages/jediepcserver.py b/.emacs.d.back/.python-environments/default/lib/python3.7/site-packages/jediepcserver.py new file mode 100644 index 00000000..8c63e9d8 --- /dev/null +++ b/.emacs.d.back/.python-environments/default/lib/python3.7/site-packages/jediepcserver.py @@ -0,0 +1,496 @@ +#!/usr/bin/env python + +""" +Jedi EPC server. + +Copyright (C) 2012 Takafumi Arakaki + +Author: Takafumi Arakaki <aka.tkf at gmail.com> + +This file is NOT part of GNU Emacs. + +Jedi EPC server 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 3 of the +License, or (at your option) any later version. + +Jedi EPC server 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 Jedi EPC server. +If not, see <http://www.gnu.org/licenses/>. + +""" + +import argparse +import glob +import itertools +import logging +import logging.handlers +import os +import re +import site +import sys +from collections import namedtuple + +import jedi +import jedi.api + +import epc +import epc.server +import sexpdata + +logger = logging.getLogger('jediepcserver') + + +parser = argparse.ArgumentParser( + formatter_class=argparse.RawTextHelpFormatter, + description=__doc__) +parser.add_argument( + '--address', default='localhost') +parser.add_argument( + '--port', default=0, type=int) +parser.add_argument( + '--port-file', '-f', default='-', type=argparse.FileType('wt'), + help='file to write port on. default is stdout.') +parser.add_argument( + '--sys-path', '-p', default=[], action='append', + help='paths to be inserted at the top of `sys.path`.') +parser.add_argument( + '--sys-path-append', default=[], action='append', + help='paths to be appended at the end of `sys.path`.') +parser.add_argument( + '--virtual-env', '-v', default=[], action='append', + help='paths to be used as if VIRTUAL_ENV is set to it.') +parser.add_argument( + '--log', help='Save server log to this file.') +parser.add_argument( + '--log-level', + choices=['CRITICAL', 'ERROR', 'WARN', 'INFO', 'DEBUG'], + help='Logging level for log file.') +parser.add_argument( + '--log-rotate-max-size', default=0, type=int, + help='Rotate log file after it reaches this size', +) +parser.add_argument( + '--log-rotate-max-count', default=3, type=int, + help='Max number of log rotations before removal', +) +parser.add_argument( + '--log-traceback', action='store_true', default=False, + help='Include traceback in logging output.') +parser.add_argument( + '--pdb', dest='debugger', const='pdb', action='store_const', + help='start pdb when error occurs.') +parser.add_argument( + '--ipdb', dest='debugger', const='ipdb', action='store_const', + help='start ipdb when error occurs.') + + +PY3 = (sys.version_info[0] >= 3) +NEED_ENCODE = not PY3 + + +LogSettings = namedtuple( + 'LogSettings', + [ + 'log_file', + 'log_level', + 'log_rotate_max_size', + 'log_rotate_max_count', + ], +) + + +try: + jedi.create_environment +except AttributeError: + jedi_create_environment = None +else: + _cached_jedi_environments = {} + + def jedi_create_environment(venv, safe=False): + """Cache jedi environments to avoid startup cost.""" + try: + return _cached_jedi_environments[venv] + except KeyError: + logger.info('Creating jedi environment: %s', venv) + if venv is None: + jedienv = jedi.api.environment.get_default_environment() + else: + jedienv = jedi.create_environment(venv, safe=safe) + _cached_jedi_environments[venv] = jedienv + return jedienv + + +def get_venv_sys_path(venv): + if jedi_create_environment is not None: + return jedi_create_environment(venv).get_sys_path() + from jedi.evaluate.sys_path import get_venv_path + return get_venv_path(venv) + + +class JediEPCHandler(object): + def __init__(self, sys_path=(), virtual_envs=(), sys_path_append=()): + self.script_kwargs = self._get_script_path_kwargs( + sys_path=sys_path, + virtual_envs=virtual_envs, + sys_path_append=sys_path_append, + ) + + def get_sys_path(self): + environment = self.script_kwargs.get('environment') + if environment is not None: + return environment.get_sys_path() + sys_path = self.script_kwargs.get('sys_path') + if sys_path is not None: + return sys_path + return sys.path + + @classmethod + def _get_script_path_kwargs(cls, sys_path, virtual_envs, sys_path_append): + result = {} + if jedi_create_environment: + # Need to specify some environment explicitly to workaround + # https://github.com/davidhalter/jedi/issues/1242. Otherwise jedi + # will create a lot of child processes. + if virtual_envs: + primary_env, virtual_envs = virtual_envs[0], virtual_envs[1:] + primary_env = path_expand_vars_and_user(primary_env) + else: + primary_env = None + try: + result['environment'] = jedi_create_environment(primary_env) + except Exception: + logger.warning( + 'Cannot create environment for %r', primary_env, exc_info=1 + ) + if primary_env is not None: + result['environment'] = jedi_create_environment(None) + + if not sys_path and not virtual_envs and not sys_path_append: + # No additional path customizations. + return result + + # Either multiple environments or custom sys_path extensions are + # specified, or jedi version doesn't support environments. + final_sys_path = [] + final_sys_path.extend(path_expand_vars_and_user(p) for p in sys_path) + for p in virtual_envs: + final_sys_path.extend(get_venv_sys_path(path_expand_vars_and_user(p))) + final_sys_path.extend( + path_expand_vars_and_user(p) for p in sys_path_append + ) + dupes = set() + + def not_seen_yet(val): + if val in dupes: + return False + dupes.add(val) + return True + result['sys_path'] = [p for p in final_sys_path if not_seen_yet(p)] + return result + + def jedi_script(self, source, line, column, source_path): + if NEED_ENCODE: + source = source.encode('utf-8') + source_path = source_path and source_path.encode('utf-8') + return jedi.Script( + source, line, column, source_path or '', **self.script_kwargs + ) + + def complete(self, *args): + reply = [] + for comp in self.jedi_script(*args).completions(): + try: + docstr = comp.docstring() + except KeyError: + docstr = "" + + reply.append(dict( + word=comp.name, + doc=docstr, + description=candidates_description(comp), + symbol=candidate_symbol(comp), + )) + return reply + + def get_in_function_call(self, *args): + sig = self.jedi_script(*args).call_signatures() + call_def = sig[0] if sig else None + + if not call_def: + return [] + + return dict( + # p.description should do the job. But jedi-vim use replace. + # So follow what jedi-vim does... + params=[PARAM_PREFIX_RE.sub('', p.description).replace('\n', '') + for p in call_def.params], + index=call_def.index, + call_name=call_def.name, + ) + + def _goto(self, method, *args): + """ + Helper function for `goto_assignments` and `usages`. + + :arg method: `jedi.Script.goto_assignments` or `jedi.Script.usages` + :arg args: Arguments to `jedi_script` + + """ + # `definitions` is a list. Each element is an instances of + # `jedi.api_classes.BaseOutput` subclass, i.e., + # `jedi.api_classes.RelatedName` or `jedi.api_classes.Definition`. + definitions = method(self.jedi_script(*args)) + return [dict( + column=d.column, + line_nr=d.line, + module_path=d.module_path if d.module_path != '__builtin__' else [], + module_name=d.module_name, + description=d.description, + ) for d in definitions] + + def goto(self, *args): + return self._goto(jedi.Script.goto_assignments, *args) + + def related_names(self, *args): + return self._goto(jedi.Script.usages, *args) + + def get_definition(self, *args): + definitions = self.jedi_script(*args).goto_definitions() + return [definition_to_dict(d) for d in definitions] + + def defined_names(self, *args): + # XXX: there's a bug in Jedi that returns returns definitions from inside + # classes or functions even though all_scopes=False is set by + # default. Hence some additional filtering is in order. + # + # See https://github.com/davidhalter/jedi/issues/1202 + top_level_names = [ + defn + for defn in jedi.api.names(*args) + if defn.parent().type == 'module' + ] + return list(map(get_names_recursively, top_level_names)) + + def get_jedi_version(self): + return [dict( + name=module.__name__, + file=getattr(module, '__file__', []), + version=get_module_version(module) or [], + ) for module in [sys, jedi, epc, sexpdata]] + + +def candidate_symbol(comp): + """ + Return a character representing completion type. + + :type comp: jedi.api.Completion + :arg comp: A completion object returned by `jedi.Script.completions`. + + """ + try: + return comp.type[0].lower() + except (AttributeError, TypeError): + return '?' + + +def candidates_description(comp): + """ + Return `comp.description` in an appropriate format. + + * Avoid return a string 'None'. + * Strip off all newlines. This is required for using + `comp.description` as candidate summary. + + """ + desc = comp.description + return _WHITESPACES_RE.sub(' ', desc) if desc and desc != 'None' else '' +_WHITESPACES_RE = re.compile(r'\s+') + + +PARAM_PREFIX_RE = re.compile(r'^param\s+') +"""RE to strip unwanted "param " prefix returned by param.description.""" + + +def definition_to_dict(d): + return dict( + doc=d.docstring(), + description=d.description, + desc_with_module=d.desc_with_module, + line_nr=d.line, + column=d.column, + module_path=d.module_path, + name=getattr(d, 'name', []), + full_name=getattr(d, 'full_name', []), + type=getattr(d, 'type', []), + ) + + +def get_names_recursively(definition, parent=None): + """ + Fetch interesting defined names in sub-scopes under `definition`. + + :type names: jedi.api_classes.Definition + + """ + d = definition_to_dict(definition) + try: + d['local_name'] = parent['local_name'] + '.' + d['name'] + except (AttributeError, TypeError): + d['local_name'] = d['name'] + if definition.type == 'class': + ds = definition.defined_names() + return [d] + [get_names_recursively(c, d) for c in ds] + else: + return [d] + + +def get_module_version(module): + notfound = object() + for key in ['__version__', 'version']: + version = getattr(module, key, notfound) + if version is not notfound: + return version + try: + from pkg_resources import get_distribution, DistributionNotFound + try: + return get_distribution(module.__name__).version + except DistributionNotFound: + pass + except ImportError: + pass + + +def path_expand_vars_and_user(p): + return os.path.expandvars(os.path.expanduser(p)) + + +def configure_logging(log_settings): + """ + :type log_settings: LogSettings + """ + if not log_settings.log_file: + return + + fmter = logging.Formatter('%(asctime)s:' + logging.BASIC_FORMAT) + if log_settings.log_rotate_max_size > 0: + handler = logging.handlers.RotatingFileHandler( + filename=log_settings.log_file, + mode='w', + maxBytes=log_settings.log_rotate_max_size, + backupCount=log_settings.log_rotate_max_count, + ) + else: + handler = logging.FileHandler(filename=log_settings.log_file, mode='w') + handler.setFormatter(fmter) + if log_settings.log_level: + logging.root.setLevel(log_settings.log_level.upper()) + logging.root.addHandler(handler) + + +def jedi_epc_server( + address='localhost', + port=0, + port_file=sys.stdout, + sys_path=[], + virtual_env=[], + sys_path_append=[], + debugger=None, + log_traceback=None, +): + """Start EPC server. + + :type log_settings: LogSettings + + """ + logger.debug( + 'jedi_epc_server: sys_path=%r virtual_env=%r sys_path_append=%r', + sys_path, virtual_env, sys_path_append, + ) + + if not virtual_env and os.getenv('VIRTUAL_ENV'): + logger.debug( + 'Taking virtual env from VIRTUAL_ENV: %r', + os.environ['VIRTUAL_ENV'], + ) + virtual_env = [os.environ['VIRTUAL_ENV']] + + handler = JediEPCHandler( + sys_path=sys_path, + virtual_envs=virtual_env, + sys_path_append=sys_path_append, + ) + logger.debug( + 'Starting Jedi EPC server with the following sys.path: %r', + handler.get_sys_path(), + ) + server = epc.server.EPCServer((address, port)) + server.register_function(handler.complete) + server.register_function(handler.get_in_function_call) + server.register_function(handler.goto) + server.register_function(handler.related_names) + server.register_function(handler.get_definition) + server.register_function(handler.defined_names) + server.register_function(handler.get_jedi_version) + + @server.register_function + def toggle_log_traceback(): + server.log_traceback = not server.log_traceback + return server.log_traceback + + port_file.write(str(server.server_address[1])) # needed for Emacs client + port_file.write("\n") + port_file.flush() + if port_file is not sys.stdout: + port_file.close() + + # This is not supported Python-EPC API, but I am using this for + # backward compatibility for Python-EPC < 0.0.4. In the future, + # it should be passed to the constructor. + server.log_traceback = bool(log_traceback) + + if debugger: + server.set_debugger(debugger) + handler = logging.StreamHandler() + fmter = logging.Formatter('%(asctime)s:' + logging.BASIC_FORMAT) + handler.setFormatter(fmter) + handler.setLevel(logging.DEBUG) + server.logger.addHandler(handler) + server.logger.setLevel(logging.DEBUG) + return server + + +# def add_virtualenv_path(venv): +# """Add virtualenv's site-packages to `sys.path`.""" +# venv = os.path.abspath(venv) +# paths = glob.glob(os.path.join( +# venv, 'lib', 'python*', 'site-packages')) +# if not paths: +# raise ValueError('Invalid venv: no site-packages found: %s' % venv) +# for path in paths: +# site.addsitedir(path) + + +def main(args=None): + ns = parser.parse_args(args) + + ns_vars = vars(ns).copy() + log_settings = LogSettings( + log_file=ns_vars.pop('log'), + log_level=ns_vars.pop('log_level'), + log_rotate_max_size=ns_vars.pop('log_rotate_max_size'), + log_rotate_max_count=ns_vars.pop('log_rotate_max_count'), + ) + configure_logging(log_settings) + server = jedi_epc_server(**ns_vars) + server.serve_forever() + server.logger.info('exit') + + +if __name__ == '__main__': + main() |