feat(parser): parse constraint statements

core/ast/midas.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Generic, Optional, TypeVar
+
+from lexer.token import Token
+
+T = TypeVar("T")
+
+
+# Statements
+
+
+@dataclass(frozen=True)
+class Stmt(ABC):
+    @abstractmethod
+    def accept(self, visitor: Visitor[T]) -> T: ...
+
+    class Visitor(ABC, Generic[T]):
+        @abstractmethod
+        def visit_type_stmt(self, stmt: TypeStmt) -> T: ...
+
+        @abstractmethod
+        def visit_property_stmt(self, stmt: PropertyStmt) -> T: ...
+
+        @abstractmethod
+        def visit_op_stmt(self, stmt: OpStmt) -> T: ...
+
+        @abstractmethod
+        def visit_constraint_stmt(self, stmt: ConstraintStmt) -> T: ...
+
+
+@dataclass(frozen=True)
+class TypeStmt(Stmt):
+    name: Token
+    bases: list[TypeExpr]
+    body: Optional[TypeBodyExpr]
+
+    def accept(self, visitor: Stmt.Visitor[T]) -> T:
+        return visitor.visit_type_stmt(self)
+
+
+@dataclass(frozen=True)
+class PropertyStmt(Stmt):
+    name: Token
+    type: TypeExpr
+
+    def accept(self, visitor: Stmt.Visitor[T]) -> T:
+        return visitor.visit_property_stmt(self)
+
+
+@dataclass(frozen=True)
+class OpStmt(Stmt):
+    left: TypeExpr
+    op: Token
+    right: TypeExpr
+    result: TypeExpr
+
+    def accept(self, visitor: Stmt.Visitor[T]) -> T:
+        return visitor.visit_op_stmt(self)
+
+
+@dataclass(frozen=True)
+class ConstraintStmt(Stmt):
+    name: Token
+    constraint: ConstraintExpr
+
+    def accept(self, visitor: Stmt.Visitor[T]) -> T:
+        return visitor.visit_constraint_stmt(self)
+
+
+# Expressions
+
+
+@dataclass(frozen=True)
+class Expr(ABC):
+    @abstractmethod
+    def accept(self, visitor: Visitor[T]) -> T: ...
+
+    class Visitor(ABC, Generic[T]):
+        @abstractmethod
+        def visit_type_expr(self, expr: TypeExpr) -> T: ...
+
+        @abstractmethod
+        def visit_constraint_expr(self, expr: ConstraintExpr) -> T: ...
+
+        @abstractmethod
+        def visit_type_body_expr(self, expr: TypeBodyExpr) -> T: ...
+
+
+@dataclass(frozen=True)
+class TypeExpr(Expr):
+    name: Token
+    constraints: list[ConstraintExpr]
+
+    def accept(self, visitor: Expr.Visitor[T]) -> T:
+        return visitor.visit_type_expr(self)
+
+
+@dataclass(frozen=True)
+class ConstraintExpr(Expr):
+    def accept(self, visitor: Expr.Visitor[T]) -> T:
+        return visitor.visit_constraint_expr(self)
+
+
+@dataclass(frozen=True)
+class TypeBodyExpr(Expr):
+    properties: list[PropertyStmt]
+
+    def accept(self, visitor: Expr.Visitor[T]) -> T:
+        return visitor.visit_type_body_expr(self)

core/ast/printer.py

@@ -1,9 +1,12 @@
+from __future__ import annotations
+
 from contextlib import contextmanager
 from enum import Enum, auto
 import io
-from typing import Generator, Optional
+from typing import Generator, Generic, Optional, Protocol, TypeVar
 
-from core.ast.annotations import Expr, TypeExpr, SchemaExpr, SchemaElementExpr
+import core.ast.annotations as a
+import core.ast.midas as m
 
 
 class _Level(Enum):
@@ -12,7 +15,14 @@ class _Level(Enum):
     LAST = auto()
 
 
-class AnnotationAstPrinter(Expr.Visitor[None]):
+class Expr(Protocol):
+    def accept(self, printer: AstPrinter) -> None: ...
+
+
+T = TypeVar("T", bound=Expr)
+
+
+class AstPrinter(Generic[T]):
     LAST_CHILD = "└── "
     CHILD = "├── "
     VERTICAL = "│   "
@@ -23,7 +33,7 @@ class AnnotationAstPrinter(Expr.Visitor[None]):
         self._idx: Optional[int] = None
         self._buf: io.StringIO = io.StringIO()
 
-    def print(self, expr: Expr):
+    def print(self, expr: T):
         self._buf = io.StringIO()
         expr.accept(self)
         return self._buf.getvalue()
@@ -40,7 +50,9 @@ class AnnotationAstPrinter(Expr.Visitor[None]):
         if self._levels:
             self._levels[-1] = _Level.LAST
 
-    def _write_line(self, text: str):
+    def _write_line(self, text: str, *, last: bool = False):
+        if last:
+            self._mark_last()
         indent: str = self._build_indent()
         if self._idx is not None:
             text = f"[{self._idx}] {text}"
@@ -60,7 +72,7 @@ class AnnotationAstPrinter(Expr.Visitor[None]):
         return "".join(parts)
 
     def _write_optional_child(
-        self, label: str, child: Optional[Expr], *, last: bool = False
+        self, label: str, child: Optional[T], *, last: bool = False
     ):
         if last:
             self._mark_last()
@@ -71,13 +83,15 @@ class AnnotationAstPrinter(Expr.Visitor[None]):
             with self._child_level(last=True):
                 child.accept(self)
 
-    def visit_type_expr(self, expr: TypeExpr):
+
+class AnnotationAstPrinter(AstPrinter, a.Expr.Visitor[None]):
+    def visit_type_expr(self, expr: a.TypeExpr):
         self._write_line("TypeExpr")
         with self._child_level():
             self._write_line(f'name: "{expr.name.lexeme}"')
             self._write_optional_child("schema", expr.schema, last=True)
 
-    def visit_schema_expr(self, expr: SchemaExpr):
+    def visit_schema_expr(self, expr: a.SchemaExpr):
         self._write_line("SchemaExpr")
         with self._child_level():
             for i, elmt in enumerate(expr.elements):
@@ -86,7 +100,7 @@ class AnnotationAstPrinter(Expr.Visitor[None]):
                     self._mark_last()
                 elmt.accept(self)
 
-    def visit_schema_element_expr(self, expr: SchemaElementExpr):
+    def visit_schema_element_expr(self, expr: a.SchemaElementExpr):
         self._write_line("SchemaElementExpr")
         with self._child_level():
             name_text: str = "None" if expr.name is None else f'"{expr.name.lexeme}"'
@@ -94,23 +108,23 @@ class AnnotationAstPrinter(Expr.Visitor[None]):
             self._write_optional_child("type", expr.type, last=True)
 
 
-class AnnotationPrinter(Expr.Visitor[str]):
-    def print(self, expr: Expr):
+class AnnotationPrinter(a.Expr.Visitor[str]):
+    def print(self, expr: a.Expr):
         return expr.accept(self)
 
-    def visit_type_expr(self, expr: TypeExpr) -> str:
+    def visit_type_expr(self, expr: a.TypeExpr) -> str:
         schema: str = ""
         if expr.schema is not None:
             schema = expr.schema.accept(self)
         return f"{expr.name.lexeme}{schema}"
 
-    def visit_schema_expr(self, expr: SchemaExpr) -> str:
+    def visit_schema_expr(self, expr: a.SchemaExpr) -> str:
         res: str = expr.left.lexeme
         res += ", ".join(elmt.accept(self) for elmt in expr.elements)
         res += expr.right.lexeme
         return res
 
-    def visit_schema_element_expr(self, expr: SchemaElementExpr) -> str:
+    def visit_schema_element_expr(self, expr: a.SchemaElementExpr) -> str:
         parts: list[str] = []
         if expr.name is not None:
             parts.append(expr.name.lexeme)
@@ -120,3 +134,82 @@ class AnnotationPrinter(Expr.Visitor[str]):
         else:
             parts.append(expr.type.accept(self))
         return ": ".join(parts)
+
+
+class MidasAstPrinter(AstPrinter, m.Expr.Visitor[None], m.Stmt.Visitor[None]):
+    def visit_type_stmt(self, stmt: m.TypeStmt):
+        self._write_line("TypeStmt")
+        with self._child_level():
+            self._write_line(f'name: "{stmt.name.lexeme}"')
+            self._write_line("bases")
+            with self._child_level():
+                for i, base in enumerate(stmt.bases):
+                    self._idx = i
+                    if i == len(stmt.bases) - 1:
+                        self._mark_last()
+                    base.accept(self)
+            self._write_optional_child("body", stmt.body, last=True)
+
+    def visit_property_stmt(self, stmt: m.PropertyStmt):
+        self._write_line("PropertyStmt")
+        with self._child_level():
+            self._write_line(f'name: "{stmt.name.lexeme}"')
+            self._write_line("type", last=True)
+            with self._child_level():
+                self._mark_last()
+                stmt.type.accept(self)
+
+    def visit_op_stmt(self, stmt: m.OpStmt) -> None:
+        self._write_line("OpStmt")
+        with self._child_level():
+            self._write_line("left")
+            with self._child_level():
+                self._mark_last()
+                stmt.left.accept(self)
+
+            self._write_line(f'op: "{stmt.op.lexeme}"')
+
+            self._write_line("right")
+            with self._child_level():
+                self._mark_last()
+                stmt.right.accept(self)
+
+            self._write_line("result", last=True)
+            with self._child_level():
+                self._mark_last()
+                stmt.result.accept(self)
+
+    def visit_constraint_stmt(self, stmt: m.ConstraintStmt):
+        self._write_line("ConstraintStmt")
+        with self._child_level():
+            self._write_line(f'name: "{stmt.name.lexeme}"')
+            self._write_line("constraint", last=True)
+            with self._child_level():
+                self._mark_last()
+                stmt.constraint.accept(self)
+
+    def visit_type_expr(self, expr: m.TypeExpr):
+        self._write_line("TypeExpr")
+        with self._child_level():
+            self._write_line(f'name: "{expr.name.lexeme}"')
+            self._write_line("constraints", last=True)
+            with self._child_level():
+                for i, constraint in enumerate(expr.constraints):
+                    self._idx = i
+                    if i == len(expr.constraints) - 1:
+                        self._mark_last()
+                    constraint.accept(self)
+
+    def visit_constraint_expr(self, expr: m.ConstraintExpr):
+        self._write_line("ConstraintExpr")
+
+    def visit_type_body_expr(self, expr: m.TypeBodyExpr):
+        self._write_line("TypeBodyExpr")
+        with self._child_level():
+            self._write_line("properties", last=True)
+            with self._child_level():
+                for i, property in enumerate(expr.properties):
+                    self._idx = i
+                    if i == len(expr.properties) - 1:
+                        self._mark_last()
+                    property.accept(self)

parser/midas.py

@@ -0,0 +1,118 @@
+from typing import Optional
+
+from core.ast.midas import (
+    ConstraintExpr,
+    ConstraintStmt,
+    OpStmt,
+    PropertyStmt,
+    Stmt,
+    TypeBodyExpr,
+    TypeExpr,
+    TypeStmt,
+)
+from lexer.token import Token, TokenType
+from parser.base import Parser
+from parser.errors import ParsingError
+
+
+class MidasParser(Parser):
+    SYNC_BOUNDARY: set[TokenType] = {TokenType.TYPE, TokenType.OP, TokenType.CONSTRAINT}
+
+    def parse(self) -> list[Stmt]:
+        statements: list[Stmt] = []
+        while not self.is_at_end():
+            stmt: Optional[Stmt] = self.declaration()
+            if stmt is None:
+                print("Early stop")
+                break
+            statements.append(stmt)
+        return statements
+
+    def synchronize(self):
+        self.advance()
+        while not self.is_at_end():
+            if self.previous().type == TokenType.NEWLINE:
+                return
+            if self.peek().type in self.SYNC_BOUNDARY:
+                return
+            self.advance()
+
+    def declaration(self) -> Optional[Stmt]:
+        try:
+            if self.match(TokenType.TYPE):
+                return self.type_declaration()
+            if self.match(TokenType.OP):
+                return self.op_declaration()
+            if self.match(TokenType.CONSTRAINT):
+                return self.constraint_declaration()
+        except ParsingError:
+            self.synchronize()
+            return None
+
+    def type_declaration(self) -> TypeStmt:
+        name: Token = self.consume(TokenType.IDENTIFIER, "Expected type name")
+        self.consume(TokenType.LESS, "Expected '<' after type name")
+        bases: list[TypeExpr] = []
+        while not self.check(TokenType.GREATER) and not self.is_at_end():
+            bases.append(self.type_expr())
+            if not self.check(TokenType.GREATER):
+                self.consume(TokenType.COMMA, "Expected ',' between type bases")
+        self.consume(TokenType.GREATER, "Expected '>' after base type")
+
+        body: Optional[TypeBodyExpr] = None
+
+        if self.check(TokenType.LEFT_BRACE):
+            body = self.type_body_expr()
+        return TypeStmt(name=name, bases=bases, body=body)
+
+    def type_expr(self) -> TypeExpr:
+        name: Token = self.consume(TokenType.IDENTIFIER, "Expected type name")
+        constraints: list[ConstraintExpr] = []
+
+        while not self.is_at_end() and self.match(TokenType.PLUS):
+            constraints.append(self.constraint_expr())
+
+        return TypeExpr(name=name, constraints=constraints)
+
+    def constraint_expr(self) -> ConstraintExpr:
+        # TODO
+        return ConstraintExpr()
+
+    def type_body_expr(self) -> TypeBodyExpr:
+        self.consume(TokenType.LEFT_BRACE, "Expected '{' to start type body")
+        properties: list[PropertyStmt] = []
+        while not self.check(TokenType.RIGHT_BRACE) and not self.is_at_end():
+            properties.append(self.property_stmt())
+        self.consume(TokenType.RIGHT_BRACE, "Unclosed type body")
+        return TypeBodyExpr(properties=properties)
+
+    def property_stmt(self) -> PropertyStmt:
+        name: Token = self.consume(TokenType.IDENTIFIER, "Expected property name")
+        self.consume(TokenType.COLON, "Expected ':' after property name")
+        type: TypeExpr = self.type_expr()
+        return PropertyStmt(name=name, type=type)
+
+    def op_declaration(self) -> OpStmt:
+        self.consume(TokenType.LESS, "Expected '<' before first type")
+        left: TypeExpr = self.type_expr()
+        self.consume(TokenType.GREATER, "Expected '>' after first type")
+
+        op: Token = self.advance()
+
+        self.consume(TokenType.LESS, "Expected '<' before second type")
+        right: TypeExpr = self.type_expr()
+        self.consume(TokenType.GREATER, "Expected '>' after second type")
+
+        self.consume(TokenType.EQUAL, "Expected '=' after second type")
+
+        self.consume(TokenType.LESS, "Expected '<' before result type")
+        result: TypeExpr = self.type_expr()
+        self.consume(TokenType.GREATER, "Expected '>' after result type")
+
+        return OpStmt(left=left, op=op, right=right, result=result)
+
+    def constraint_declaration(self) -> ConstraintStmt:
+        name: Token = self.consume(TokenType.IDENTIFIER, "Expected constraint name")
+        self.consume(TokenType.EQUAL, "Expected '=' after constraint name")
+        constraint: ConstraintExpr = self.constraint_expr()
+        return ConstraintStmt(name=name, constraint=constraint)

test.py

@@ -1,32 +1,52 @@
 import importlib
 from pathlib import Path
 
-from core.ast.printer import AnnotationAstPrinter
+from core.ast.printer import AnnotationAstPrinter, MidasAstPrinter
 from lexer.annotations import AnnotationLexer
 from lexer.midas import MidasLexer
 from lexer.token import Token
 from parser.annotations import AnnotationParser
+from parser.midas import MidasParser
 
-# Frame annotation
-mod = importlib.import_module("examples.00_syntax_prototype.01_simple_types")
 
-annotation: str = mod.__annotations__["df"]
-lexer: AnnotationLexer = AnnotationLexer(annotation, "01_simple_types.py")
-tokens: list[Token] = lexer.process()
-# print([f"{t.type.name}('{t.lexeme}')" for t in tokens])
+def test_annotation():
+    # Frame annotation
+    mod = importlib.import_module("examples.00_syntax_prototype.01_simple_types")
 
-parser = AnnotationParser(tokens)
-parsed = parser.parse()
-print(parsed)
-for err in parser.errors:
-    print(err.get_report())
-printer = AnnotationAstPrinter()
-if parsed is not None:
-    print(printer.print(parsed))
+    annotation: str = mod.__annotations__["df"]
+    lexer: AnnotationLexer = AnnotationLexer(annotation, "01_simple_types.py")
+    tokens: list[Token] = lexer.process()
+    # print([f"{t.type.name}('{t.lexeme}')" for t in tokens])
 
-# Midas type definitions
-path: Path = Path("examples") / "00_syntax_prototype" / "02_custom_types.midas"
-definitions: str = path.read_text()
-midas_lexer: MidasLexer = MidasLexer(definitions, path.name)
-tokens = midas_lexer.process()
-# print([f"{t.type.name}('{t.lexeme}')" for t in tokens])
+    parser = AnnotationParser(tokens)
+    parsed = parser.parse()
+    print(parsed)
+    for err in parser.errors:
+        print(err.get_report())
+    printer = AnnotationAstPrinter()
+    if parsed is not None:
+        print(printer.print(parsed))
+
+
+def test_midas():
+    # Midas type definitions
+    path: Path = Path("examples") / "00_syntax_prototype" / "02_custom_types.midas"
+    definitions: str = path.read_text()
+    midas_lexer: MidasLexer = MidasLexer(definitions, path.name)
+    tokens: list[Token] = midas_lexer.process()
+    # print([f"{t.type.name}('{t.lexeme}')" for t in tokens])
+
+    parser = MidasParser(tokens)
+    parsed = parser.parse()
+    print(parsed)
+    for err in parser.errors:
+        print(err.get_report())
+    printer = MidasAstPrinter()
+    for stmt in parsed:
+        if stmt is None:
+            print("None")
+            continue
+        print(printer.print(stmt))
+
+
+test_midas()