54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
import regex
|
|
|
|
from parsimonious.grammar import Grammar
|
|
|
|
from .parser import ExecContext, Execuitor, ExecutorTerm
|
|
|
|
fragment_re = regex.compile(r"\$(?<br>{(?:[^{}]++|(?&br))*})")
|
|
|
|
|
|
class ExecClient:
|
|
def __init__(self):
|
|
self.grammer: Grammar = self.load_grammar()
|
|
self.execuitor: Execuitor = Execuitor()
|
|
|
|
self.globals = {}
|
|
|
|
def load_grammar(self):
|
|
with open("assets/grammar.ppeg") as f:
|
|
return Grammar(f.read())
|
|
|
|
def _parse_fragment(self, cmd: str) -> ExecutorTerm:
|
|
"""
|
|
Parse a provided fragment into an ExecutorTerm.
|
|
|
|
This could reasonably be LRU cached.
|
|
"""
|
|
tree = self.grammer.parse(cmd)
|
|
term = self.execuitor.visit(tree)
|
|
return term
|
|
|
|
async def run_command(self, ctx: ExecContext, command: str, dryrun=False) -> str:
|
|
"""
|
|
Process the given command string,
|
|
executing and replacing each fragment.
|
|
|
|
Executes in the given context,
|
|
which will generally be constructed through subclass methods.
|
|
"""
|
|
result = []
|
|
last_end = 0
|
|
|
|
for match in regex.finditer(fragment_re, command):
|
|
result.append(command[last_end : match.start()])
|
|
|
|
fragment = command[match.start() : match.end()]
|
|
parsed = self._parse_fragment(fragment)
|
|
if not dryrun:
|
|
fragment_result = await parsed.execute(ctx)
|
|
|
|
result.append(fragment_result)
|
|
last_end = match.end()
|
|
|
|
return "".join(result)
|