377 lines
14 KiB
Python
377 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# klammertext_ls.py
|
|
#
|
|
# The Klammertext language server: a small, dependency-free implementation of
|
|
# the Language Server Protocol (JSON-RPC over stdio) on top of the shared
|
|
# editor core, doc/edit/shared/klammertext_edit.py. One server, any LSP
|
|
# client — the VS Code extension (doc/edit/vscode/) spawns it, and Neovim's
|
|
# built-in LSP, Emacs eglot, or Sublime's LSP package can attach to it with a
|
|
# one-line configuration (see doc/edit/README.md).
|
|
#
|
|
# What it serves:
|
|
#
|
|
# textDocument/publishDiagnostics unclosed / mismatched delimiters, on
|
|
# every open and change
|
|
# textDocument/formatting reindent the whole document
|
|
# textDocument/rangeFormatting reindent the selected lines
|
|
# textDocument/documentHighlight the matching application delimiter for
|
|
# the cursor position (live match
|
|
# highlighting in clients that request it
|
|
# on cursor movement)
|
|
# textDocument/definition jump-to-match: on an @ delimiter,
|
|
# "go to definition" goes to its match
|
|
# workspace/executeCommand klammertext.alignTable — align the
|
|
# @table enclosing the given position
|
|
# (applied via workspace/applyEdit)
|
|
#
|
|
# Formatting is EXPLICIT-ONLY by design: the server does not implement
|
|
# on-type formatting, because whitespace is content in Klammertext; nothing
|
|
# reformats as a side effect of typing.
|
|
#
|
|
# Protocol notes: full-text document sync (documents are small); positions in
|
|
# UTF-16 code units per the LSP default. No external libraries — the framing
|
|
# and dispatch below are the whole protocol layer.
|
|
#
|
|
# Usage: python3 klammertext_ls.py (talks LSP on stdin/stdout)
|
|
#
|
|
# The shared core is located next to this file (the repository layout and the
|
|
# vendored layouts both put the two files side by side), or under
|
|
# $KLAMMERTEXT_HOME/doc/edit/shared.
|
|
#
|
|
# Python floor: 3.8.
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
|
|
def _import_shared():
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
candidates = [here]
|
|
kh = os.environ.get('KLAMMERTEXT_HOME')
|
|
if kh:
|
|
candidates.append(os.path.join(kh, 'doc', 'edit', 'shared'))
|
|
for d in candidates:
|
|
if os.path.isfile(os.path.join(d, 'klammertext_edit.py')):
|
|
if d not in sys.path:
|
|
sys.path.insert(0, d)
|
|
break
|
|
import klammertext_edit
|
|
return klammertext_edit
|
|
|
|
|
|
KE = _import_shared()
|
|
|
|
|
|
# --- positions: LSP {line, character} (UTF-16) <-> python string offsets ---
|
|
|
|
def _utf16_len(s):
|
|
n = 0
|
|
for ch in s:
|
|
n += 2 if ord(ch) > 0xFFFF else 1
|
|
return n
|
|
|
|
|
|
def pos_to_offset(text, pos):
|
|
"""Offset in TEXT of LSP position POS ({'line', 'character'}, UTF-16)."""
|
|
line = pos.get('line', 0)
|
|
character = pos.get('character', 0)
|
|
off = 0
|
|
for _ in range(line):
|
|
nl = text.find('\n', off)
|
|
if nl == -1:
|
|
return len(text)
|
|
off = nl + 1
|
|
units = 0
|
|
n = len(text)
|
|
while off < n and text[off] != '\n' and units < character:
|
|
units += 2 if ord(text[off]) > 0xFFFF else 1
|
|
off += 1
|
|
return off
|
|
|
|
|
|
def offset_to_pos(text, offset):
|
|
"""LSP position of character OFFSET in TEXT."""
|
|
offset = max(0, min(offset, len(text)))
|
|
line = text.count('\n', 0, offset)
|
|
bol = text.rfind('\n', 0, offset) + 1
|
|
return {'line': line, 'character': _utf16_len(text[bol:offset])}
|
|
|
|
|
|
def offsets_to_range(text, start, end):
|
|
return {'start': offset_to_pos(text, start),
|
|
'end': offset_to_pos(text, end)}
|
|
|
|
|
|
# --- the server ------------------------------------------------------------
|
|
|
|
class Server:
|
|
|
|
def __init__(self, instream, outstream):
|
|
self.instream = instream
|
|
self.outstream = outstream
|
|
self.docs = {} # uri -> text
|
|
self.shutdown_received = False
|
|
self.running = True
|
|
self.next_server_id = 1 # ids for server->client requests
|
|
|
|
# -- transport --
|
|
|
|
def read_message(self):
|
|
length = None
|
|
while True:
|
|
line = self.instream.readline()
|
|
if not line:
|
|
return None # EOF
|
|
line = line.strip()
|
|
if not line:
|
|
break # end of headers
|
|
if line.lower().startswith(b'content-length:'):
|
|
length = int(line.split(b':', 1)[1])
|
|
if length is None:
|
|
return None
|
|
body = self.instream.read(length)
|
|
if len(body) < length:
|
|
return None
|
|
return json.loads(body.decode('utf-8'))
|
|
|
|
def send(self, message):
|
|
body = json.dumps(message, ensure_ascii=False).encode('utf-8')
|
|
self.outstream.write(b'Content-Length: ' + str(len(body)).encode()
|
|
+ b'\r\n\r\n' + body)
|
|
self.outstream.flush()
|
|
|
|
def reply(self, msg_id, result):
|
|
self.send({'jsonrpc': '2.0', 'id': msg_id, 'result': result})
|
|
|
|
def reply_error(self, msg_id, code, message):
|
|
self.send({'jsonrpc': '2.0', 'id': msg_id,
|
|
'error': {'code': code, 'message': message}})
|
|
|
|
def notify(self, method, params):
|
|
self.send({'jsonrpc': '2.0', 'method': method, 'params': params})
|
|
|
|
def request(self, method, params):
|
|
"""Server->client request (fire and forget: the client's response is
|
|
consumed, and ignored, by the main loop)."""
|
|
self.send({'jsonrpc': '2.0', 'id': 'server-%d' % self.next_server_id,
|
|
'method': method, 'params': params})
|
|
self.next_server_id += 1
|
|
|
|
# -- diagnostics --
|
|
|
|
def publish_diagnostics(self, uri):
|
|
text = self.docs.get(uri, '')
|
|
diags = []
|
|
for p in KE.diagnostics(text):
|
|
diags.append({
|
|
'range': offsets_to_range(text, p['start'], p['end']),
|
|
'severity': 1 if p['severity'] == 'error' else 2,
|
|
'source': 'klammertext',
|
|
'message': p['message'],
|
|
})
|
|
self.notify('textDocument/publishDiagnostics',
|
|
{'uri': uri, 'diagnostics': diags})
|
|
|
|
# -- dispatch --
|
|
|
|
def handle(self, msg):
|
|
method = msg.get('method')
|
|
msg_id = msg.get('id')
|
|
params = msg.get('params') or {}
|
|
|
|
if method is None:
|
|
return # a response to a server->client request
|
|
|
|
handler = getattr(self, 'on_' + method.replace('/', '_')
|
|
.replace('$', 'dollar'), None)
|
|
if handler is not None:
|
|
handler(msg_id, params)
|
|
elif msg_id is not None: # unknown request: MethodNotFound
|
|
self.reply_error(msg_id, -32601, 'method not found: ' + method)
|
|
# unknown notification: ignored
|
|
|
|
# -- lifecycle --
|
|
|
|
def on_initialize(self, msg_id, params):
|
|
self.reply(msg_id, {
|
|
'capabilities': {
|
|
'textDocumentSync': 1, # full
|
|
'documentFormattingProvider': True,
|
|
'documentRangeFormattingProvider': True,
|
|
'documentHighlightProvider': True,
|
|
'definitionProvider': True,
|
|
'executeCommandProvider': {
|
|
'commands': ['klammertext.alignTable'],
|
|
},
|
|
},
|
|
'serverInfo': {'name': 'klammertext-ls'},
|
|
})
|
|
|
|
def on_initialized(self, msg_id, params):
|
|
pass
|
|
|
|
def on_shutdown(self, msg_id, params):
|
|
self.shutdown_received = True
|
|
self.reply(msg_id, None)
|
|
|
|
def on_exit(self, msg_id, params):
|
|
self.running = False
|
|
|
|
def on_dollar_cancelRequest(self, msg_id, params):
|
|
pass # requests here are all fast
|
|
|
|
# -- document sync (full text) --
|
|
|
|
def on_textDocument_didOpen(self, msg_id, params):
|
|
doc = params['textDocument']
|
|
self.docs[doc['uri']] = doc.get('text', '')
|
|
self.publish_diagnostics(doc['uri'])
|
|
|
|
def on_textDocument_didChange(self, msg_id, params):
|
|
uri = params['textDocument']['uri']
|
|
changes = params.get('contentChanges') or []
|
|
if changes:
|
|
self.docs[uri] = changes[-1].get('text', '')
|
|
self.publish_diagnostics(uri)
|
|
|
|
def on_textDocument_didClose(self, msg_id, params):
|
|
uri = params['textDocument']['uri']
|
|
self.docs.pop(uri, None)
|
|
self.notify('textDocument/publishDiagnostics',
|
|
{'uri': uri, 'diagnostics': []})
|
|
|
|
def on_textDocument_didSave(self, msg_id, params):
|
|
pass
|
|
|
|
# -- formatting (reindentation) --
|
|
|
|
def _format_edits(self, text, bols):
|
|
edits = []
|
|
for a, b, new in KE.reindent_lines(text, bols):
|
|
edits.append({'range': offsets_to_range(text, a, b),
|
|
'newText': new})
|
|
return edits
|
|
|
|
def on_textDocument_formatting(self, msg_id, params):
|
|
text = self.docs.get(params['textDocument']['uri'], '')
|
|
self.reply(msg_id, self._format_edits(text, KE.line_beginnings(text)))
|
|
|
|
def on_textDocument_rangeFormatting(self, msg_id, params):
|
|
text = self.docs.get(params['textDocument']['uri'], '')
|
|
rng = params['range']
|
|
start = pos_to_offset(text, rng['start'])
|
|
end = pos_to_offset(text, rng['end'])
|
|
bols = [b for b in KE.line_beginnings(text)
|
|
if b <= end and (text.find('\n', b) == -1
|
|
or text.find('\n', b) >= start)]
|
|
self.reply(msg_id, self._format_edits(text, bols))
|
|
|
|
# -- matching (highlight + jump) --
|
|
|
|
def on_textDocument_documentHighlight(self, msg_id, params):
|
|
text = self.docs.get(params['textDocument']['uri'], '')
|
|
m = KE.match_at(text, pos_to_offset(text, params['position']))
|
|
if m is None:
|
|
self.reply(msg_id, None)
|
|
return
|
|
highlights = [{'range': offsets_to_range(text, *m['token']),
|
|
'kind': 1}]
|
|
if m['match_token'] is not None:
|
|
highlights.append({'range': offsets_to_range(text,
|
|
*m['match_token']),
|
|
'kind': 1})
|
|
self.reply(msg_id, highlights)
|
|
|
|
def on_klammertext_matchInfo(self, msg_id, params):
|
|
"""Custom request: the full matching story for a cursor position —
|
|
token, matching token, and whether the pair mismatches. The VS Code
|
|
extension draws its live match/mismatch decorations from this
|
|
(documentHighlight is word-gated in VS Code, so a bare @ close would
|
|
never trigger it; and it cannot carry the mismatch flag)."""
|
|
uri = params['textDocument']['uri']
|
|
text = self.docs.get(uri, '')
|
|
m = KE.match_at(text, pos_to_offset(text, params['position']))
|
|
if m is None:
|
|
self.reply(msg_id, None)
|
|
return
|
|
self.reply(msg_id, {
|
|
'token': offsets_to_range(text, *m['token']),
|
|
'matchToken': (offsets_to_range(text, *m['match_token'])
|
|
if m['match_token'] is not None else None),
|
|
'mismatch': m['mismatch'],
|
|
'message': m['message'],
|
|
})
|
|
|
|
def on_textDocument_definition(self, msg_id, params):
|
|
uri = params['textDocument']['uri']
|
|
text = self.docs.get(uri, '')
|
|
m = KE.match_at(text, pos_to_offset(text, params['position']))
|
|
if m is None or m['match_token'] is None:
|
|
self.reply(msg_id, None)
|
|
return
|
|
self.reply(msg_id, {'uri': uri,
|
|
'range': offsets_to_range(text,
|
|
*m['match_token'])})
|
|
|
|
# -- commands --
|
|
|
|
def on_workspace_executeCommand(self, msg_id, params):
|
|
command = params.get('command')
|
|
args = params.get('arguments') or []
|
|
if command != 'klammertext.alignTable' or not args:
|
|
self.reply_error(msg_id, -32602,
|
|
'unknown command or missing arguments')
|
|
return
|
|
arg = args[0]
|
|
uri = arg['uri']
|
|
text = self.docs.get(uri, '')
|
|
pos = pos_to_offset(text, arg['position'])
|
|
span = KE.enclosing_span(text, pos, KE.ALIGN_KLAMMERS)
|
|
if span is None:
|
|
self.reply(msg_id, None)
|
|
self.notify('window/showMessage',
|
|
{'type': 3, 'message':
|
|
'Klammertext: the cursor is not inside a table '
|
|
'klammer (%s)' % ', '.join(
|
|
'@' + n for n in sorted(KE.ALIGN_KLAMMERS))})
|
|
return
|
|
_name, cs, ce = span
|
|
edits, message = KE.compute_edits(text[cs:ce])
|
|
self.reply(msg_id, None)
|
|
if edits:
|
|
lsp_edits = [{'range': offsets_to_range(text, cs + a, cs + b),
|
|
'newText': new}
|
|
for a, b, new in edits]
|
|
self.request('workspace/applyEdit',
|
|
{'label': 'Klammertext: align table',
|
|
'edit': {'changes': {uri: lsp_edits}}})
|
|
self.notify('window/showMessage',
|
|
{'type': 3, 'message': 'Klammertext: ' + message})
|
|
|
|
# -- main loop --
|
|
|
|
def run(self):
|
|
while self.running:
|
|
msg = self.read_message()
|
|
if msg is None:
|
|
break # EOF or malformed stream
|
|
try:
|
|
self.handle(msg)
|
|
except Exception as e: # a bug must not kill the server
|
|
if msg.get('id') is not None and 'method' in msg:
|
|
self.reply_error(msg['id'], -32603,
|
|
'internal error: %s' % e)
|
|
sys.stderr.write('klammertext_ls: %s\n' % e)
|
|
sys.stderr.flush()
|
|
return 0 if self.shutdown_received else 1
|
|
|
|
|
|
def main():
|
|
server = Server(sys.stdin.buffer, sys.stdout.buffer)
|
|
sys.exit(server.run())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|