(prototype, unix-only) a REPL with a staging buffer to validate the input as it's being typed

intellishell.py

#!/usr/bin/env -S python -B

import abc
import contextlib
import copy
import enum
import functools
import io
import os
import sys
import termios
import tty
import typing

@contextlib.contextmanager
def cbreak():
    attr = tty.setcbreak(sys.stdin.fileno())

    try:
        yield
    finally:
        termios.tcsetattr(sys.stdin.fileno(), termios.TCSAFLUSH, attr)

def minmax(min_value, value, max_value):
    return max(min_value, min(value, max_value))

def wrap(ctx):
    def decorator(f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            with ctx():
                ret = f(*args, **kwargs)

            return ret
        return wrapper
    return decorator

def read_key_bytes():
    try:
        key_bytes = os.read(sys.stdin.fileno(), 16)
    except KeyboardInterrupt:
        key_bytes = b"\x03"

    return key_bytes

class TokenType(enum.Enum):
    IDENT = enum.auto()
    INTEGER = enum.auto()
    SPACE = enum.auto()

class Token(typing.NamedTuple):
    type: TokenType
    lexeme: str
    position: int
    length: int

class AbstractCstShellExpr(abc.ABC):
    @abc.abstractmethod
    def accept(self, visitor):
        pass

class AbstractCstLangExpr(abc.ABC):
    @abc.abstractmethod
    def accept(self, visitor):
        pass

class CstIdentLangExpr(AbstractCstLangExpr):
    def __init__(self, lexeme):
        self.lexeme = lexeme

    def accept(self, visitor):
        return visitor.visit_ident_lang_expr(self)

class CstIntegerLangExpr(AbstractCstLangExpr):
    def __init__(self, lexeme):
        self.lexeme = lexeme

    def accept(self, visitor):
        return visitor.visit_integer_lang_expr(self)

class CstEmptyShellExpr(AbstractCstShellExpr):
    def accept(self, visitor):
        return visitor.visit_empty_shell_expr(self)

class CstLangExprShellExpr(AbstractCstShellExpr):
    def __init__(self, expr):
        self.expr = expr

    def accept(self, visitor):
        return visitor.visit_lang_expr_shell_expr(self)

class Tokenizer:
    def __init__(self, contents):
        self.contents = contents
        self.start = self.length = 0

    @property
    def position(self):
        return self.start + self.length

    def is_at_end(self):
        return self.position >= len(self.contents)

    def peek(self):
        if self.is_at_end():
            return b"\x00"

        return self.contents[self.position:self.position + 1]

    def advance(self):
        self.length += 1

    def consume(self):
        byte = self.peek()
        self.advance()

        return byte

    def finalize_token(self, type):
        token = Token(
            type,
            self.contents[self.start:self.position],
            self.start,
            self.length,
        )

        self.start += self.length
        self.length = 0

        return token

    def finish_ident(self):
        while (
            b"a" <= (byte := self.peek()) <= b"z"
            or b"0" <= byte <= b"9"
            or byte == b"_"
        ):
            self.advance()

        return self.finalize_token(TokenType.IDENT)

    def finish_integer(self):
        while b"0" <= (byte := self.peek()) <= b"9" or byte == b"_":
            self.advance()

        return self.finalize_token(TokenType.INTEGER)

    def finish_space(self):
        while (byte := self.peek()) == b" ":
            self.advance()

        return self.finalize_token(TokenType.SPACE)

    def scan_token(self):
        match self.consume():
            case byte if b"a" <= byte <= b"z":
                return self.finish_ident()
            case byte if b"0" <= byte <= b"9":
                return self.finish_integer()
            case b" ":
                return self.finish_space()
            case _:
                return None

    def tokenize(self):
        tokens = []

        while not self.is_at_end():
            token = self.scan_token()

            if token is None:
                return None

            tokens.append(token)

        return tokens

class Parser:
    def __init__(self, tokens):
        self.tokens = tokens
        self.position = 0

    def is_at_end(self):
        return self.position >= len(self.tokens)

    def peek(self):
        if self.is_at_end():
            return None

        return self.tokens[self.position]

    def advance(self):
        self.position += 1

    def consume(self, predicate):
        token = self.peek()

        if token is not None and predicate(token):
            self.advance()
            return token

        return None

    def consume_type(self, type):
        return self.consume(lambda token: token.type == type)

    def consume_any(self):
        token = self.peek()
        self.advance()

        return token

    def skip_space(self):
        consume_type(TokenType.SPACE)

    def parse_ident_lang_expr(self):
        token = self.consume_type(TokenType.IDENT)

        if token is None:
            return None

        return CstIdentLangExpr(token.lexeme)

    def parse_integer_lang_expr(self):
        token = self.consume_type(TokenType.INTEGER)

        if token is None:
            return None

        return CstIntegerLangExpr(token.lexeme)

    def get_lang_expr_parsers(self):
        return [
            self.parse_ident_lang_expr,
            self.parse_integer_lang_expr,
        ]

    def parse_lang_expr(self):
        for parser in self.get_lang_expr_parsers():
            expr = parser()

            if expr is not None:
                return expr

        return None

    def parse_empty_shell_expr(self):
        if self.tokens:
            return None

        return CstEmptyShellExpr()

    def parse_lang_expr_shell_expr(self):
        expr = self.parse_lang_expr()

        if expr is None or not self.is_at_end():
            return None

        return CstLangExprShellExpr(expr)

    def get_shell_expr_parsers(self):
        return [
            self.parse_empty_shell_expr,
            self.parse_lang_expr_shell_expr,
        ]

    def parse_shell_expr(self):
        for parser in self.get_shell_expr_parsers():
            expr = parser()

            if expr is not None:
                return expr

        return None

    def parse(self):
        return self.parse_shell_expr()

class Buffer:
    def __init__(self):
        self.raw = io.BytesIO()
        self.position = self.length = 0

    def is_empty(self):
        return self.length == 0

    def tell(self):
        return self.position

    def seek(self, position):
        self.raw.seek(position)
        self.position = position

    def seek_offset(self, offset):
        self.seek(minmax(0, self.position + offset, self.length))

    def read(self, n=-1):
        data = self.raw.read(n)
        self.position = self.raw.tell()

        return data

    def peek(self, n=-1):
        data = self.raw.read(n)
        self.seek(self.position)

        return data

    def write(self, data):
        n = self.raw.write(data)
        self.length += n - (self.length - self.position)
        self.position += n

    def insert(self, data):
        position = self.position + len(data)
        rest = self.peek()

        self.truncate()
        self.write(data + rest)
        self.seek(position)

    def extract(self, n):
        extracted = self.peek(abs(n))
        self.position = minmax(0, self.position - abs(n), self.length)
        
        self.insert(b"")

        return extracted

    def truncate(self):
        self.length = self.raw.truncate()

    def clear(self):
        self.seek(0)
        self.truncate()

    def get_contents(self):
        self.raw.seek(0)
        return self.peek()

class StagingBuffer:
    def __init__(self):
        self.stage = Buffer()
        self.back = Buffer()

        self.pre_commit_predicates = []

    @property
    def position(self):
        return self.back.position

    def is_empty(self):
        return self.back.length == 0

    def tell(self):
        return self.position

    def seek(self, position):
        self.stage.seek(position)

    def seek_offset(self, offset):
        self.stage.seek_offset(offset)

    def read(self, n=-1):
        return self.back.peek(n)

    def peek(self, n=-1):
        return self.back.peek(n)

    def write(self, data):
        self.stage.write(data)

    def insert(self, data):
        self.stage.insert(data)

    def extract(self, n):
        return self.stage.extract(n)

    def truncate(self):
        self.stage.truncate()

    def clear(self):
        self.stage.clear()

    def get_contents(self):
        return self.back.get_contents()

    def commit(self):
        new_contents = self.stage.get_contents()

        for predicate in self.pre_commit_predicates:
            if not predicate(new_contents):
                self.rollback()
                return False

        self.back = copy.deepcopy(self.stage)

        return True

    def rollback(self):
        self.stage = copy.deepcopy(self.back)

    def add_pre_commit_predicate(self, predicate):
        self.pre_commit_predicates.append(predicate)

class ShellState:
    def __init__(self, prompt):
        self.buffer = StagingBuffer()
        self.should_exit = False
        self.prompt = prompt

def clear_input(state):
    state.buffer.clear()

def insert_char(state, char):
    state.buffer.insert(char)

def delete_left_char(state):
    _ = state.buffer.extract(-1)

def move_by(state, offset):
    state.buffer.seek_offset(offset)

def tokenize(contents):
    tokenizer = Tokenizer(contents)
    return tokenizer.tokenize()

def parse(contents):
    tokens = tokenize(contents)

    if tokens is None:
        return None

    parser = Parser(tokens)
    return parser.parse()

def parsing_predicate(contents):
    return parse(contents) is not None

def process_input(state):
    contents = state.buffer.get_contents()

    state.buffer.clear()
    state.buffer.commit()

    cst = parse(contents)

    sys.stdout.buffer.raw.write(b"\n" + contents + b"\n")
    print(cst)

@wrap(cbreak)
def main(prompt=b">>> "):
    print("Welcome to Intellishell!")

    state = ShellState(prompt)
    
    state.buffer.add_pre_commit_predicate(parsing_predicate)

    os.write(sys.stdout.fileno(), state.prompt)
    sys.stdout.flush()

    while not state.should_exit:
        match read_key_bytes():
            case b"\x04":
                state.should_exit = True
            case b"\x03":
                clear_input(state)
            case b"\x7f":
                delete_left_char(state)
            case b"\x1b[C":
                move_by(state, +1)
            case b"\x1b[D":
                move_by(state, -1)
            case b"\n" if not state.buffer.is_empty():
                process_input(state)
            case key if key.isascii():
                insert_char(state, key)

        state.buffer.commit()

        os.write(sys.stdout.fileno(), b"\x1b[G%s\x1b[K" % state.prompt)
        os.write(sys.stdout.fileno(), state.buffer.get_contents().replace(b"echo ", b"\x1b[1mecho\x1b[22m "))
        os.write(sys.stdout.fileno(),b"\x1b[%dG" % (len(state.prompt) + state.buffer.position + 1))
        sys.stdout.flush()

    print("\nGoodbye!")

if __name__ == "__main__":
    main()