|
| 1 | +"""pyang LSP handling""" |
| 2 | + |
| 3 | +from __future__ import absolute_import |
| 4 | +import optparse |
| 5 | +from pathlib import Path |
| 6 | + |
| 7 | +from pyang import error |
| 8 | +from pyang import context |
| 9 | +from pyang import plugin |
| 10 | +from pyang import syntax |
| 11 | + |
| 12 | +from lsprotocol import types as lsp |
| 13 | + |
| 14 | +from pygls.server import LanguageServer |
| 15 | + |
| 16 | +SERVER_NAME = "pyangls" |
| 17 | +SERVER_VERSION = "v0.1" |
| 18 | + |
| 19 | +SERVER_MODE_IO = "io" |
| 20 | +SERVER_MODE_TCP = "tcp" |
| 21 | +SERVER_MODE_WS = "ws" |
| 22 | +supported_modes = [ |
| 23 | + SERVER_MODE_IO, |
| 24 | + SERVER_MODE_TCP, |
| 25 | + SERVER_MODE_WS, |
| 26 | +] |
| 27 | +default_mode = SERVER_MODE_IO |
| 28 | +default_host = "127.0.0.1" |
| 29 | +default_port = 2087 |
| 30 | + |
| 31 | +class PyangLanguageServer(LanguageServer): |
| 32 | + def __init__(self, *args): |
| 33 | + self.ctx : context.Context |
| 34 | + super().__init__(*args) |
| 35 | + |
| 36 | +pyangls = PyangLanguageServer(SERVER_NAME, SERVER_VERSION) |
| 37 | + |
| 38 | +def _validate(ls: LanguageServer, |
| 39 | + params: lsp.DidChangeTextDocumentParams | lsp.DidOpenTextDocumentParams): |
| 40 | + ls.show_message_log("Validating YANG...") |
| 41 | + |
| 42 | + text_doc = ls.workspace.get_text_document(params.text_document.uri) |
| 43 | + source = text_doc.source |
| 44 | + |
| 45 | + pyangls.ctx.errors = [] |
| 46 | + modules = [] |
| 47 | + diagnostics = [] |
| 48 | + if source: |
| 49 | + m = syntax.re_filename.search(Path(text_doc.filename).name) |
| 50 | + if m is not None: |
| 51 | + name, rev, in_format = m.groups() |
| 52 | + module = pyangls.ctx.get_module(name, rev) |
| 53 | + if module is not None: |
| 54 | + pyangls.ctx.del_module(module) |
| 55 | + module = pyangls.ctx.add_module(text_doc.path, source, |
| 56 | + in_format, name, rev, |
| 57 | + expect_failure_error=False, |
| 58 | + primary_module=True) |
| 59 | + else: |
| 60 | + module = pyangls.ctx.add_module(text_doc.path, source, |
| 61 | + primary_module=True) |
| 62 | + if module is not None: |
| 63 | + modules.append(module) |
| 64 | + p : plugin.PyangPlugin |
| 65 | + for p in plugin.plugins: |
| 66 | + p.pre_validate_ctx(pyangls.ctx, modules) |
| 67 | + |
| 68 | + pyangls.ctx.validate() |
| 69 | + module.prune() |
| 70 | + |
| 71 | + diagnostics = build_diagnostics() |
| 72 | + |
| 73 | + ls.publish_diagnostics(text_doc.uri, diagnostics) |
| 74 | + |
| 75 | +def build_diagnostics(): |
| 76 | + """Builds lsp diagnostics from pyang context""" |
| 77 | + diagnostics = [] |
| 78 | + |
| 79 | + for epos, etag, eargs in pyangls.ctx.errors: |
| 80 | + msg = error.err_to_str(etag, eargs) |
| 81 | + # pyang just stores line context, not keyword/argument context |
| 82 | + start_line = epos.line - 1 |
| 83 | + start_col = 0 |
| 84 | + end_line = epos.line - 1 |
| 85 | + end_col = 1 |
| 86 | + def level_to_severity(level): |
| 87 | + if level == 1 or level == 2: |
| 88 | + return lsp.DiagnosticSeverity.Error |
| 89 | + elif level == 3: |
| 90 | + return lsp.DiagnosticSeverity.Warning |
| 91 | + elif level == 4: |
| 92 | + return lsp.DiagnosticSeverity.Information |
| 93 | + else: |
| 94 | + return None |
| 95 | + d = lsp.Diagnostic( |
| 96 | + range=lsp.Range( |
| 97 | + start=lsp.Position(line=start_line, character=start_col), |
| 98 | + end=lsp.Position(line=end_line, character=end_col), |
| 99 | + ), |
| 100 | + message=msg, |
| 101 | + severity=level_to_severity(error.err_level(etag)), |
| 102 | + code=etag, |
| 103 | + source=SERVER_NAME, |
| 104 | + ) |
| 105 | + |
| 106 | + diagnostics.append(d) |
| 107 | + |
| 108 | + return diagnostics |
| 109 | + |
| 110 | + |
| 111 | +@pyangls.feature( |
| 112 | + lsp.TEXT_DOCUMENT_DIAGNOSTIC, |
| 113 | + lsp.DiagnosticOptions( |
| 114 | + identifier="pyangls", |
| 115 | + inter_file_dependencies=True, |
| 116 | + workspace_diagnostics=True, |
| 117 | + ), |
| 118 | +) |
| 119 | +def text_document_diagnostic( |
| 120 | + params: lsp.DocumentDiagnosticParams, |
| 121 | +) -> lsp.DocumentDiagnosticReport: |
| 122 | + """Returns diagnostic report.""" |
| 123 | + return lsp.RelatedFullDocumentDiagnosticReport( |
| 124 | + items=_validate(pyangls, params), |
| 125 | + kind=lsp.DocumentDiagnosticReportKind.Full, |
| 126 | + ) |
| 127 | + |
| 128 | + |
| 129 | +@pyangls.feature(lsp.WORKSPACE_DIAGNOSTIC) |
| 130 | +def workspace_diagnostic( |
| 131 | + params: lsp.WorkspaceDiagnosticParams, |
| 132 | +) -> lsp.WorkspaceDiagnosticReport: |
| 133 | + """Returns diagnostic report.""" |
| 134 | + documents = pyangls.workspace.text_documents.keys() |
| 135 | + |
| 136 | + if len(documents) == 0: |
| 137 | + items = [] |
| 138 | + else: |
| 139 | + first = list(documents)[0] |
| 140 | + document = pyangls.workspace.get_text_document(first) |
| 141 | + items = [ |
| 142 | + lsp.WorkspaceFullDocumentDiagnosticReport( |
| 143 | + uri=document.uri, |
| 144 | + version=document.version, |
| 145 | + items=_validate(pyangls, params), |
| 146 | + kind=lsp.DocumentDiagnosticReportKind.Full, |
| 147 | + ) |
| 148 | + ] |
| 149 | + |
| 150 | + return lsp.WorkspaceDiagnosticReport(items=items) |
| 151 | + |
| 152 | + |
| 153 | +@pyangls.feature(lsp.TEXT_DOCUMENT_DID_CHANGE) |
| 154 | +def did_change(ls: LanguageServer, params: lsp.DidChangeTextDocumentParams): |
| 155 | + """Text document did change notification.""" |
| 156 | + |
| 157 | + _validate(ls, params) |
| 158 | + |
| 159 | + |
| 160 | +@pyangls.feature(lsp.TEXT_DOCUMENT_DID_CLOSE) |
| 161 | +def did_close(ls: PyangLanguageServer, params: lsp.DidCloseTextDocumentParams): |
| 162 | + """Text document did close notification.""" |
| 163 | + ls.show_message("Text Document Did Close") |
| 164 | + |
| 165 | + |
| 166 | +@pyangls.feature(lsp.TEXT_DOCUMENT_DID_OPEN) |
| 167 | +async def did_open(ls: LanguageServer, params: lsp.DidOpenTextDocumentParams): |
| 168 | + """Text document did open notification.""" |
| 169 | + ls.show_message("Text Document Did Open") |
| 170 | + _validate(ls, params) |
| 171 | + |
| 172 | + |
| 173 | +@pyangls.feature(lsp.TEXT_DOCUMENT_INLINE_VALUE) |
| 174 | +def inline_value(params: lsp.InlineValueParams): |
| 175 | + """Returns inline value.""" |
| 176 | + return [lsp.InlineValueText(range=params.range, text="Inline value")] |
| 177 | + |
| 178 | + |
| 179 | +def add_opts(optparser: optparse.OptionParser): |
| 180 | + optlist = [ |
| 181 | + # use capitalized versions of std options help and version |
| 182 | + optparse.make_option("--lsp-mode", |
| 183 | + dest="pyangls_mode", |
| 184 | + default=default_mode, |
| 185 | + metavar="LSP_MODE", |
| 186 | + help="Provide LSP Service in this mode" \ |
| 187 | + "Supported LSP server modes are: " + |
| 188 | + ', '.join(supported_modes)), |
| 189 | + optparse.make_option("--lsp-host", |
| 190 | + dest="pyangls_host", |
| 191 | + default=default_host, |
| 192 | + metavar="LSP_HOST", |
| 193 | + help="Bind LSP Server to this address"), |
| 194 | + optparse.make_option("--lsp-port", |
| 195 | + dest="pyangls_port", |
| 196 | + type="int", |
| 197 | + default=default_port, |
| 198 | + metavar="LSP_PORT", |
| 199 | + help="Bind LSP Server to this port"), |
| 200 | + ] |
| 201 | + g = optparser.add_option_group("LSP Server specific options") |
| 202 | + g.add_options(optlist) |
| 203 | + |
| 204 | +def start_server(optargs, ctx: context.Context): |
| 205 | + pyangls.ctx = ctx |
| 206 | + if optargs.pyangls_mode == SERVER_MODE_TCP: |
| 207 | + pyangls.start_tcp(optargs.pyangls_host, optargs.pyangls_port) |
| 208 | + elif optargs.pyangls_mode == SERVER_MODE_WS: |
| 209 | + pyangls.start_ws(optargs.pyangls_host, optargs.pyangls_port) |
| 210 | + else: |
| 211 | + pyangls.start_io() |
0 commit comments