Editor support generalized: shared core, language server, Vim and VS Code (from dev eb5baf9cbe59)
doc/edit/ now holds a shared Python implementation of the language's structural layer (klammertext_edit.py) and a dependency-free language server (klammertext_ls.py), with integrations for Emacs, Sublime Text, Vim, and Visual Studio Code. The editor test suite in tst/ covers the core's API and CLI, the language server protocol, the VS Code extension, headless Vim, and Emacs byte-equality. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
356
doc/edit/shared/klammertext_ls.py
Normal file
356
doc/edit/shared/klammertext_ls.py
Normal file
@@ -0,0 +1,356 @@
|
||||
#!/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_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()
|
||||
Reference in New Issue
Block a user