"""SimVX Language Server -- provides Property-aware completions and diagnostics."""
import json
import logging
import sys
from .protocol import decode_header, encode_message, response
log = logging.getLogger(__name__)
[docs]
class SimVXLSPServer:
"""Minimal LSP server with SimVX-specific features.
Provides:
- Node type completions (from ``Node._registry``)
- Property-aware completions (node attribute suggestions)
- Input action completions (from ``InputMap``)
"""
def __init__(self):
self._running = False
self._initialized = False
self._documents: dict[str, str] = {}
# ------------------------------------------------------------------
# Main loop
# ------------------------------------------------------------------
[docs]
def run(self):
"""Run the server on stdio, reading JSON-RPC messages."""
self._running = True
buf = b""
while self._running:
chunk = sys.stdin.buffer.read(1)
if not chunk:
break
buf += chunk
buf = self._try_parse(buf)
def _try_parse(self, buf: bytes) -> bytes:
"""Attempt to extract a complete message from *buf*."""
while True:
parsed = decode_header(buf)
if parsed is None:
return buf
content_length, offset = parsed
end = offset + content_length
if len(buf) < end:
return buf
body = buf[offset:end]
buf = buf[end:]
try:
msg = json.loads(body)
except json.JSONDecodeError:
log.warning("malformed JSON-RPC body")
continue
self._handle_message(msg)
return buf
# ------------------------------------------------------------------
# Dispatch
# ------------------------------------------------------------------
_HANDLERS: dict[str, str] = {
"initialize": "_handle_initialize",
"initialized": "_handle_initialized",
"shutdown": "_handle_shutdown",
"exit": "_handle_exit",
"textDocument/didOpen": "_handle_did_open",
"textDocument/didChange": "_handle_did_change",
"textDocument/didClose": "_handle_did_close",
"textDocument/completion": "_handle_completion",
}
def _handle_message(self, msg: dict):
method = msg.get("method", "")
handler_name = self._HANDLERS.get(method)
if handler_name:
getattr(self, handler_name)(msg)
elif "id" in msg and method:
# Unknown request -- respond with method-not-found
self._send(response(msg["id"], error={"code": -32601, "message": f"Method not found: {method}"}))
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
def _handle_initialize(self, msg: dict):
result = {
"capabilities": {
"completionProvider": {"triggerCharacters": ["."]},
"textDocumentSync": 1, # Full sync
},
"serverInfo": {"name": "simvx-lsp", "version": "0.1.0"},
}
self._send(response(msg["id"], result=result))
def _handle_initialized(self, _msg: dict):
self._initialized = True
def _handle_shutdown(self, msg: dict):
self._send(response(msg["id"], result=None))
self._running = False
def _handle_exit(self, _msg: dict):
self._running = False
# ------------------------------------------------------------------
# Document sync
# ------------------------------------------------------------------
def _handle_did_open(self, msg: dict):
params = msg.get("params", {})
doc = params.get("textDocument", {})
uri = doc.get("uri", "")
text = doc.get("text", "")
if uri:
self._documents[uri] = text
def _handle_did_change(self, msg: dict):
params = msg.get("params", {})
uri = params.get("textDocument", {}).get("uri", "")
changes = params.get("contentChanges", [])
if uri and changes:
# Full sync: last change has the full text
self._documents[uri] = changes[-1].get("text", "")
def _handle_did_close(self, msg: dict):
uri = msg.get("params", {}).get("textDocument", {}).get("uri", "")
self._documents.pop(uri, None)
# ------------------------------------------------------------------
# Completions
# ------------------------------------------------------------------
def _handle_completion(self, msg: dict):
items = self._build_completion_items()
self._send(response(msg["id"], result={"isIncomplete": False, "items": items}))
def _build_completion_items(self) -> list[dict]:
"""Build completion items from the Node registry and Property descriptors."""
items: list[dict] = []
try:
from ..node import Node
for name, cls in sorted(Node._registry.items()):
detail_parts: list[str] = []
# Collect Property descriptors for documentation
props = []
for attr_name in dir(cls):
try:
attr = getattr(cls, attr_name, None)
except Exception:
continue
if attr is not None and type(attr).__name__ == "Property":
props.append(attr_name)
if props:
detail_parts.append(f"Properties: {', '.join(props[:8])}")
if len(props) > 8:
detail_parts.append(f" (+{len(props) - 8} more)")
items.append({
"label": name,
"kind": 7, # CompletionItemKind.Class
"detail": "; ".join(detail_parts) if detail_parts else "SimVX node type",
})
except Exception:
log.debug("Could not load Node registry for completions", exc_info=True)
return items
# ------------------------------------------------------------------
# Transport
# ------------------------------------------------------------------
def _send(self, msg: dict):
data = encode_message(msg)
sys.stdout.buffer.write(data)
sys.stdout.buffer.flush()
[docs]
def send(self, msg: dict):
"""Public send for testing or external callers."""
self._send(msg)