.config/coc/extensions/node_modules/coc-python/pythonFiles/completion.py

   1 import os
   2 import os.path
   3 import io
   4 import re
   5 import sys
   6 import json
   7 import traceback
   8 import platform
   9
  10 jediPreview = False
  11
  12
  13 class RedirectStdout(object):
  14     def __init__(self, new_stdout=None):
  15         """If stdout is None, redirect to /dev/null"""
  16         self._new_stdout = new_stdout or open(os.devnull, "w")
  17
  18     def __enter__(self):
  19         sys.stdout.flush()
  20         self.oldstdout_fno = os.dup(sys.stdout.fileno())
  21         os.dup2(self._new_stdout.fileno(), 1)
  22
  23     def __exit__(self, exc_type, exc_value, traceback):
  24         self._new_stdout.flush()
  25         os.dup2(self.oldstdout_fno, 1)
  26         os.close(self.oldstdout_fno)
  27
  28
  29 class JediCompletion(object):
  30     basic_types = {
  31         "module": "import",
  32         "instance": "variable",
  33         "statement": "value",
  34         "param": "variable",
  35     }
  36
  37     def __init__(self):
  38         self.default_sys_path = sys.path
  39         self.environment = jedi.api.environment.create_environment(
  40             sys.executable, safe=False
  41         )
  42         self._input = io.open(sys.stdin.fileno(), encoding="utf-8")
  43         if (os.path.sep == "/") and (platform.uname()[2].find("Microsoft") > -1):
  44             # WSL; does not support UNC paths
  45             self.drive_mount = "/mnt/"
  46         elif sys.platform == "cygwin":
  47             # cygwin
  48             self.drive_mount = "/cygdrive/"
  49         else:
  50             # Do no normalization, e.g. Windows build of Python.
  51             # Could add additional test: ((os.path.sep == '/') and os.path.isdir('/mnt/c'))
  52             # However, this may have more false positives trying to identify Windows/*nix hybrids
  53             self.drive_mount = ""
  54
  55     def _get_definition_type(self, definition):
  56         # if definition.type not in ['import', 'keyword'] and is_built_in():
  57         #    return 'builtin'
  58         try:
  59             if definition.type in ["statement"] and definition.name.isupper():
  60                 return "constant"
  61             return self.basic_types.get(definition.type, definition.type)
  62         except Exception:
  63             return "builtin"
  64
  65     def _additional_info(self, completion):
  66         """Provide additional information about the completion object."""
  67         if not hasattr(completion, "_definition") or completion._definition is None:
  68             return ""
  69         if completion.type == "statement":
  70             nodes_to_display = ["InstanceElement", "String", "Node", "Lambda", "Number"]
  71             return "".join(
  72                 c.get_code()
  73                 for c in completion._definition.children
  74                 if type(c).__name__ in nodes_to_display
  75             ).replace("\n", "")
  76         return ""
  77
  78     @classmethod
  79     def _get_top_level_module(cls, path):
  80         """Recursively walk through directories looking for top level module.
  81
  82         Jedi will use current filepath to look for another modules at same
  83         path, but it will not be able to see modules **above**, so our goal
  84         is to find the higher python module available from filepath.
  85         """
  86         _path, _ = os.path.split(path)
  87         if os.path.isfile(os.path.join(_path, "__init__.py")):
  88             return cls._get_top_level_module(_path)
  89         return path
  90
  91     def _generate_signature(self, completion):
  92         """Generate signature with function arguments.
  93         """
  94         if completion.type in ["module"] or not hasattr(completion, "params"):
  95             return ""
  96         return "%s(%s)" % (
  97             completion.name,
  98             ", ".join(p.description[6:] for p in completion.params if p),
  99         )
 100
 101     def _get_call_signatures(self, script, line, column):
 102         """Extract call signatures from jedi.api.Script object in failsafe way.
 103
 104         Returns:
 105             Tuple with original signature object, name and value.
 106         """
 107         _signatures = []
 108         try:
 109             call_signatures = script.get_signatures(line, column)
 110         except KeyError:
 111             call_signatures = []
 112         except:
 113             call_signatures = []
 114         for signature in call_signatures:
 115             for pos, param in enumerate(signature.params):
 116                 if not param.name:
 117                     continue
 118
 119                 name = self._get_param_name(param)
 120                 if param.name == "self" and pos == 0:
 121                     continue
 122                 if name.startswith("*"):
 123                     continue
 124
 125                 value = self._get_param_value(param)
 126                 _signatures.append((signature, name, value))
 127         return _signatures
 128
 129     def _get_param_name(self, p):
 130         if p.name.startswith("param "):
 131             return p.name[6:]  # drop leading 'param '
 132         return p.name
 133
 134     def _get_param_value(self, p):
 135         pair = p.description.split("=")
 136         if len(pair) > 1:
 137             return pair[1]
 138         return None
 139
 140     def _get_call_signatures_with_args(self, script, line, column):
 141         """Extract call signatures from jedi.api.Script object in failsafe way.
 142
 143         Returns:
 144             Array with dictionary
 145         """
 146         _signatures = []
 147         try:
 148             call_signatures = script.get_signatures(line, column)
 149         except KeyError:
 150             call_signatures = []
 151         for signature in call_signatures:
 152             sig = {
 153                 "name": "",
 154                 "description": "",
 155                 "docstring": "",
 156                 "paramindex": 0,
 157                 "params": [],
 158                 "bracketstart": [],
 159             }
 160             sig["description"] = signature.description
 161             try:
 162                 sig["docstring"] = signature.docstring()
 163                 sig["raw_docstring"] = signature.docstring(raw=True)
 164             except Exception:
 165                 sig["docstring"] = ""
 166                 sig["raw_docstring"] = ""
 167
 168             sig["name"] = signature.name
 169             sig["paramindex"] = signature.index
 170             sig["bracketstart"].append(signature.index)
 171
 172             _signatures.append(sig)
 173             for pos, param in enumerate(signature.params):
 174                 if not param.name:
 175                     continue
 176
 177                 name = self._get_param_name(param)
 178                 if param.name == "self" and pos == 0:
 179                     continue
 180
 181                 value = self._get_param_value(param)
 182                 paramDocstring = ""
 183                 try:
 184                     paramDocstring = param.docstring()
 185                 except Exception:
 186                     paramDocstring = ""
 187
 188                 sig["params"].append(
 189                     {
 190                         "name": name,
 191                         "value": value,
 192                         "docstring": paramDocstring,
 193                         "description": param.description,
 194                     }
 195                 )
 196         return _signatures
 197
 198     def _serialize_completions(self, script, line, column, identifier=None, prefix=""):
 199         """Serialize response to be read from VSCode.
 200
 201         Args:
 202             script: Instance of jedi.api.Script object.
 203             identifier: Unique completion identifier to pass back to VSCode.
 204             prefix: String with prefix to filter function arguments.
 205                 Used only when fuzzy matcher turned off.
 206
 207         Returns:
 208             Serialized string to send to VSCode.
 209         """
 210         _completions = []
 211
 212         for signature, name, value in self._get_call_signatures(script, line, column):
 213             if not self.fuzzy_matcher and not name.lower().startswith(prefix.lower()):
 214                 continue
 215             _completion = {
 216                 "type": "property",
 217                 "raw_type": "",
 218                 "rightLabel": self._additional_info(signature),
 219             }
 220             _completion["description"] = ""
 221             _completion["raw_docstring"] = ""
 222
 223             # we pass 'text' here only for fuzzy matcher
 224             if value:
 225                 _completion["snippet"] = "%s=${1:%s}$0" % (name, value)
 226                 _completion["text"] = "%s=" % (name)
 227             else:
 228                 _completion["snippet"] = "%s=$1$0" % name
 229                 _completion["text"] = name
 230                 _completion["displayText"] = name
 231             _completions.append(_completion)
 232
 233         try:
 234             completions = script.complete(line, column)
 235         except KeyError:
 236             completions = []
 237         except:
 238             completions = []
 239         for completion in completions:
 240             try:
 241                 _completion = {
 242                     "text": completion.name,
 243                     "type": self._get_definition_type(completion),
 244                     "raw_type": completion.type,
 245                     "rightLabel": self._additional_info(completion),
 246                 }
 247             except Exception:
 248                 continue
 249
 250             for c in _completions:
 251                 if c["text"] == _completion["text"]:
 252                     c["type"] = _completion["type"]
 253                     c["raw_type"] = _completion["raw_type"]
 254
 255             if any(
 256                 [c["text"].split("=")[0] == _completion["text"] for c in _completions]
 257             ):
 258                 # ignore function arguments we already have
 259                 continue
 260             _completions.append(_completion)
 261         return json.dumps({"id": identifier, "results": _completions})
 262
 263     def _serialize_methods(self, script, line, column, identifier=None, prefix=""):
 264         _methods = []
 265         try:
 266             completions = script.complete(line, column)
 267         except KeyError:
 268             return []
 269
 270         for completion in completions:
 271             if completion.name == "__autocomplete_python":
 272                 instance = completion.parent().name
 273                 break
 274         else:
 275             instance = "self.__class__"
 276
 277         for completion in completions:
 278             params = []
 279             if hasattr(completion, "params"):
 280                 params = [p.description for p in completion.params if p]
 281             if completion.parent().type == "class":
 282                 _methods.append(
 283                     {
 284                         "parent": completion.parent().name,
 285                         "instance": instance,
 286                         "name": completion.name,
 287                         "params": params,
 288                         "moduleName": completion.module_name,
 289                         "fileName": completion.module_path,
 290                         "line": completion.line,
 291                         "column": completion.column,
 292                     }
 293                 )
 294         return json.dumps({"id": identifier, "results": _methods})
 295
 296     def _serialize_arguments(self, script, line, column, identifier=None):
 297         """Serialize response to be read from VSCode.
 298
 299         Args:
 300             script: Instance of jedi.api.Script object.
 301             identifier: Unique completion identifier to pass back to VSCode.
 302
 303         Returns:
 304             Serialized string to send to VSCode.
 305         """
 306         return json.dumps(
 307             {
 308                 "id": identifier,
 309                 "results": self._get_call_signatures_with_args(script, line, column),
 310             }
 311         )
 312
 313     def _top_definition(self, definition):
 314         for d in definition.goto_assignments():
 315             if d == definition:
 316                 continue
 317             if d.type == "import":
 318                 return self._top_definition(d)
 319             else:
 320                 return d
 321         return definition
 322
 323     def _extract_range_jedi_0_11_1(self, definition):
 324         from parso.utils import split_lines
 325
 326         # get the scope range
 327         try:
 328             if definition.type in ["class", "function"]:
 329                 tree_name = definition._name.tree_name
 330                 scope = tree_name.get_definition()
 331                 start_line = scope.start_pos[0] - 1
 332                 start_column = scope.start_pos[1]
 333                 # get the lines
 334                 code = scope.get_code(include_prefix=False)
 335                 lines = split_lines(code)
 336                 # trim the lines
 337                 lines = "\n".join(lines).rstrip().split("\n")
 338                 end_line = start_line + len(lines) - 1
 339                 end_column = len(lines[-1]) - 1
 340             else:
 341                 symbol = definition._name.tree_name
 342                 start_line = symbol.start_pos[0] - 1
 343                 start_column = symbol.start_pos[1]
 344                 end_line = symbol.end_pos[0] - 1
 345                 end_column = symbol.end_pos[1]
 346             return {
 347                 "start_line": start_line,
 348                 "start_column": start_column,
 349                 "end_line": end_line,
 350                 "end_column": end_column,
 351             }
 352         except Exception as e:
 353             return {
 354                 "start_line": definition.line - 1,
 355                 "start_column": definition.column,
 356                 "end_line": definition.line - 1,
 357                 "end_column": definition.column,
 358             }
 359
 360     def _extract_range(self, definition):
 361         """Provides the definition range of a given definition
 362
 363         For regular symbols it returns the start and end location of the
 364         characters making up the symbol.
 365
 366         For scoped containers it will return the entire definition of the
 367         scope.
 368
 369         The scope that jedi provides ends with the first character of the next
 370         scope so it's not ideal. For vscode we need the scope to end with the
 371         last character of actual code. That's why we extract the lines that
 372         make up our scope and trim the trailing whitespace.
 373         """
 374         return self._extract_range_jedi_0_11_1(definition)
 375
 376     def _get_definitionsx(self, definitions, identifier=None, ignoreNoModulePath=False):
 377         """Serialize response to be read from VSCode.
 378
 379         Args:
 380             definitions: List of jedi.api.classes.Definition objects.
 381             identifier: Unique completion identifier to pass back to VSCode.
 382
 383         Returns:
 384             Serialized string to send to VSCode.
 385         """
 386         _definitions = []
 387         for definition in definitions:
 388             try:
 389                 if definition.type == "import":
 390                     definition = self._top_definition(definition)
 391                 definitionRange = {
 392                     "start_line": 0,
 393                     "start_column": 0,
 394                     "end_line": 0,
 395                     "end_column": 0,
 396                 }
 397                 module_path = ""
 398                 if hasattr(definition, "module_path") and definition.module_path:
 399                     module_path = definition.module_path
 400                     definitionRange = self._extract_range(definition)
 401                 else:
 402                     if not ignoreNoModulePath:
 403                         continue
 404                 try:
 405                     parent = definition.parent()
 406                     container = parent.name if parent.type != "module" else ""
 407                 except Exception:
 408                     container = ""
 409
 410                 try:
 411                     docstring = definition.docstring()
 412                     rawdocstring = definition.docstring(raw=True)
 413                 except Exception:
 414                     docstring = ""
 415                     rawdocstring = ""
 416                 _definition = {
 417                     "text": definition.name,
 418                     "type": self._get_definition_type(definition),
 419                     "raw_type": definition.type,
 420                     "fileName": str(module_path),
 421                     "container": container,
 422                     "range": definitionRange,
 423                     "description": definition.description,
 424                     "docstring": docstring,
 425                     "raw_docstring": rawdocstring,
 426                     "signature": self._generate_signature(definition),
 427                 }
 428                 _definitions.append(_definition)
 429             except Exception as e:
 430                 pass
 431         return _definitions
 432
 433     def _serialize_definitions(self, definitions, identifier=None):
 434         """Serialize response to be read from VSCode.
 435
 436         Args:
 437             definitions: List of jedi.api.classes.Definition objects.
 438             identifier: Unique completion identifier to pass back to VSCode.
 439
 440         Returns:
 441             Serialized string to send to VSCode.
 442         """
 443         _definitions = []
 444         for definition in definitions:
 445             try:
 446                 if definition.module_path:
 447                     if definition.type == "import":
 448                         definition = self._top_definition(definition)
 449                     if not definition.module_path:
 450                         continue
 451                     try:
 452                         parent = definition.parent()
 453                         container = parent.name if parent.type != "module" else ""
 454                     except Exception:
 455                         container = ""
 456
 457                     try:
 458                         docstring = definition.docstring()
 459                         rawdocstring = definition.docstring(raw=True)
 460                     except Exception:
 461                         docstring = ""
 462                         rawdocstring = ""
 463                     _definition = {
 464                         "text": definition.name,
 465                         "type": self._get_definition_type(definition),
 466                         "raw_type": definition.type,
 467                         "fileName": definition.module_path,
 468                         "container": container,
 469                         "range": self._extract_range(definition),
 470                         "description": definition.description,
 471                         "docstring": docstring,
 472                         "raw_docstring": rawdocstring,
 473                     }
 474                     _definitions.append(_definition)
 475             except Exception as e:
 476                 pass
 477         return json.dumps({"id": identifier, "results": _definitions})
 478
 479     def _serialize_tooltip(self, definitions, identifier=None):
 480         _definitions = []
 481         for definition in definitions:
 482             signature = definition.name
 483             description = None
 484             if definition.type in ["class", "function"]:
 485                 signature = self._generate_signature(definition)
 486                 try:
 487                     description = definition.docstring(raw=True).strip()
 488                 except Exception:
 489                     description = ""
 490                 if not description and not hasattr(definition, "get_line_code"):
 491                     # jedi returns an empty string for compiled objects
 492                     description = definition.docstring().strip()
 493             if definition.type == "module":
 494                 signature = definition.full_name
 495                 try:
 496                     description = definition.docstring(raw=True).strip()
 497                 except Exception:
 498                     description = ""
 499                 if not description and hasattr(definition, "get_line_code"):
 500                     # jedi returns an empty string for compiled objects
 501                     description = definition.docstring().strip()
 502             _definition = {
 503                 "type": self._get_definition_type(definition),
 504                 "text": definition.name,
 505                 "description": description,
 506                 "docstring": description,
 507                 "signature": signature,
 508             }
 509             _definitions.append(_definition)
 510         return json.dumps({"id": identifier, "results": _definitions})
 511
 512     def _serialize_usages(self, usages, identifier=None):
 513         _usages = []
 514         for usage in usages:
 515             _usages.append(
 516                 {
 517                     "name": usage.name,
 518                     "moduleName": usage.module_name,
 519                     "fileName": usage.module_path,
 520                     "line": usage.line,
 521                     "column": usage.column,
 522                 }
 523             )
 524         return json.dumps({"id": identifier, "results": _usages})
 525
 526     def _deserialize(self, request):
 527         """Deserialize request from VSCode.
 528
 529         Args:
 530             request: String with raw request from VSCode.
 531
 532         Returns:
 533             Python dictionary with request data.
 534         """
 535         return json.loads(request)
 536
 537     def _set_request_config(self, config):
 538         """Sets config values for current request.
 539
 540         This includes sys.path modifications which is getting restored to
 541         default value on each request so each project should be isolated
 542         from each other.
 543
 544         Args:
 545             config: Dictionary with config values.
 546         """
 547         sys.path = self.default_sys_path
 548         self.use_snippets = config.get("useSnippets")
 549         self.show_doc_strings = config.get("showDescriptions", True)
 550         self.fuzzy_matcher = config.get("fuzzyMatcher", False)
 551         jedi.settings.case_insensitive_completion = config.get(
 552             "caseInsensitiveCompletion", True
 553         )
 554         for path in config.get("extraPaths", []):
 555             if path and path not in sys.path:
 556                 sys.path.insert(0, path)
 557
 558     def _normalize_request_path(self, request):
 559         """Normalize any Windows paths received by a *nix build of
 560            Python. Does not alter the reverse os.path.sep=='\\',
 561            i.e. *nix paths received by a Windows build of Python.
 562         """
 563         if "path" in request:
 564             if not self.drive_mount:
 565                 return
 566             newPath = request["path"].replace("\\", "/")
 567             if newPath[0:1] == "/":
 568                 # is absolute path with no drive letter
 569                 request["path"] = newPath
 570             elif newPath[1:2] == ":":
 571                 # is path with drive letter, only absolute can be mapped
 572                 request["path"] = self.drive_mount + newPath[0:1].lower() + newPath[2:]
 573             else:
 574                 # is relative path
 575                 request["path"] = newPath
 576
 577     def _process_request(self, request):
 578         """Accept serialized request from VSCode and write response.
 579         """
 580         request = self._deserialize(request)
 581
 582         self._set_request_config(request.get("config", {}))
 583
 584         self._normalize_request_path(request)
 585         path = self._get_top_level_module(request.get("path", ""))
 586         if len(path) > 0 and path not in sys.path:
 587             sys.path.insert(0, path)
 588         lookup = request.get("lookup", "completions")
 589
 590         project = jedi.Project(os.path.dirname(path), sys_path=sys.path)
 591
 592         if lookup == "names":
 593             return self._serialize_definitions(
 594                 jedi.Script(
 595                     code=request.get("source", None),
 596                     path=request.get("path", ""),
 597                     project=project,
 598                     environment=self.environment,
 599                 ).get_names(all_scopes=True),
 600                 request["id"],
 601             )
 602
 603         line = request["line"] + 1
 604         column = request["column"]
 605         script = jedi.Script(
 606             code=request.get("source", None),
 607             path=request.get("path", ""),
 608             project=project,
 609             environment=self.environment,
 610         )
 611
 612         if lookup == "definitions":
 613             defs = self._get_definitionsx(
 614                 script.goto(line, column, follow_imports=True), request["id"]
 615             )
 616             return json.dumps({"id": request["id"], "results": defs})
 617         if lookup == "tooltip":
 618             if jediPreview:
 619                 defs = []
 620                 try:
 621                     defs = self._get_definitionsx(
 622                         script.infer(line, column), request["id"], True
 623                     )
 624                 except:
 625                     pass
 626                 try:
 627                     if len(defs) == 0:
 628                         defs = self._get_definitionsx(
 629                             script.goto(line, column), request["id"], True
 630                         )
 631                 except:
 632                     pass
 633                 return json.dumps({"id": request["id"], "results": defs})
 634             else:
 635                 try:
 636                     return self._serialize_tooltip(
 637                         script.infer(line, column), request["id"]
 638                     )
 639                 except:
 640                     return json.dumps({"id": request["id"], "results": []})
 641         elif lookup == "arguments":
 642             return self._serialize_arguments(script, line, column, request["id"])
 643         elif lookup == "usages":
 644             return self._serialize_usages(
 645                 script.get_references(line, column), request["id"]
 646             )
 647         elif lookup == "methods":
 648             return self._serialize_methods(
 649                 script, line, column, request["id"], request.get("prefix", "")
 650             )
 651         else:
 652             return self._serialize_completions(
 653                 script, line, column, request["id"], request.get("prefix", "")
 654             )
 655
 656     def _write_response(self, response):
 657         sys.stdout.write(response + "\n")
 658         sys.stdout.flush()
 659
 660     def watch(self):
 661         while True:
 662             try:
 663                 rq = self._input.readline()
 664                 if len(rq) == 0:
 665                     # Reached EOF - indication our parent process is gone.
 666                     sys.stderr.write(
 667                         "Received EOF from the standard input,exiting" + "\n"
 668                     )
 669                     sys.stderr.flush()
 670                     return
 671                 with RedirectStdout():
 672                     response = self._process_request(rq)
 673                 self._write_response(response)
 674
 675             except Exception:
 676                 sys.stderr.write(traceback.format_exc() + "\n")
 677                 sys.stderr.flush()
 678
 679
 680 if __name__ == "__main__":
 681     cachePrefix = "v"
 682     modulesToLoad = ""
 683     if len(sys.argv) > 2 and sys.argv[1] == "custom":
 684         jediPath = sys.argv[2]
 685         jediPreview = True
 686         cachePrefix = "custom_v"
 687         if len(sys.argv) > 3:
 688             modulesToLoad = sys.argv[3]
 689     else:
 690         # release
 691         jediPath = os.path.join(os.path.dirname(__file__), "lib", "python")
 692         if len(sys.argv) > 1:
 693             modulesToLoad = sys.argv[1]
 694
 695     sys.path.insert(0, jediPath)
 696     import jedi
 697
 698     digits = jedi.__version__.split(".")
 699     if int(digits[0]) == 0 and int(digits[1]) < 17:
 700         raise RuntimeError("Jedi version %s too old, requires >= 0.17.0" % (jedi.__version__))
 701     else:
 702         if jediPreview:
 703             jedi.settings.cache_directory = os.path.join(
 704                 jedi.settings.cache_directory,
 705                 cachePrefix + jedi.__version__.replace(".", ""),
 706             )
 707         # remove jedi from path after we import it so it will not be completed
 708         sys.path.pop(0)
 709         if len(modulesToLoad) > 0:
 710             jedi.preload_module(*modulesToLoad.split(","))
 711         JediCompletion().watch()