interpreter

AssemblyInterpreter

Bases: UnvalidatedForm, AssemblyLoader

An interpreter for FinchAssembly.

Source code in src/finch/finch_assembly/interpreter.py

class AssemblyInterpreter(UnvalidatedForm, AssemblyLoader):
    """
    An interpreter for FinchAssembly.
    """

    def __init__(
        self,
        bindings=None,
        slots=None,
        types=None,
        loop_state: HaltState | None = None,
        function_state: HaltState | None = None,
        stdout=None,
        mode: CompilerMode | None = None,
    ):
        if bindings is None:
            bindings = ScopedDict()
        if slots is None:
            slots = ScopedDict()
        if types is None:
            types = ScopedDict()
        self.bindings = bindings
        self.slots = slots
        self.types = types
        self.loop_state = loop_state
        self.function_state = function_state
        if stdout is None:
            stdout = sys.stdout
        self.stdout = stdout
        self.mode = mode if mode is not None else CompilerMode()

    def scope(
        self,
        bindings=None,
        slots=None,
        types=None,
        loop_state=None,
        function_state=None,
        stdout=None,
        mode=None,
    ):
        """
        Create a new scope for the interpreter.
        This allows for nested scopes and variable shadowing.
        """
        if bindings is None:
            bindings = self.bindings.scope()
        if slots is None:
            slots = self.slots.scope()
        if types is None:
            types = self.types.scope()
        if loop_state is None:
            loop_state = self.loop_state
        if function_state is None:
            function_state = self.function_state
        if stdout is None:
            stdout = self.stdout
        return AssemblyInterpreter(
            bindings=bindings,
            slots=slots,
            types=types,
            loop_state=loop_state,
            function_state=function_state,
            stdout=stdout,
            mode=self.mode if mode is None else mode,
        )

    def should_halt(self):
        """
        Check if the interpreter should halt execution.
        This is used to stop execution in loops or when a return
        statement is encountered.
        """
        return (
            self.loop_state
            and self.loop_state.should_halt
            or self.function_state
            and self.function_state.should_halt
        )

    def lower(self, prgm: asm.Module, *, mode: CompilerMode | None = None):
        ctx = self if mode is None else self.scope(mode=mode)
        return ctx._dispatch(prgm)

    @overload
    def __call__(
        self, prgm: asm.Module, *, mode: CompilerMode | None = None
    ) -> AssemblyLibrary: ...

    @overload
    def __call__(self, prgm: asm.AssemblyNode) -> Any: ...

    def __call__(self, prgm, *, mode=None):
        """
        Run the program.
        """
        if isinstance(prgm, asm.Module):
            return super().__call__(prgm, mode=mode)
        return self._dispatch(prgm)

    def _dispatch(self, prgm):
        match prgm:
            case asm.Assert(exp):
                if not self(exp):
                    raise AssertionError(f"Finch assertion failed: {exp}")
                return None
            case asm.Literal(value):
                return value
            case asm.Variable(var_n, var_t):
                if var_n in self.types:
                    def_t = self.types[var_n]
                    if def_t != var_t:
                        raise TypeError(
                            f"Variable '{var_n}' is declared as type {def_t}, "
                            f"but used as type {var_t}."
                        )
                if var_n in self.bindings:
                    return self.bindings[var_n]
                raise KeyError(
                    f"Variable '{var_n}' is not defined in the current context."
                )
            case asm.Assign(asm.Variable(var_n, var_t), val):
                val_e = self(val)
                if not fisinstance(val_e, var_t):
                    raise TypeError(
                        f"Assigned value {val_e} is not of type {var_t} for "
                        f"variable '{var_n}'."
                    )
                self.bindings[var_n] = val_e
                self.types[var_n] = var_t
                return None
            case asm.GetAttr(obj, attr):
                obj_e = self(obj)
                attr = attr.val
                return obj.result_type.struct_getattr(obj_e, attr)
            case asm.SetAttr(obj, attr, val):
                obj_e = self(obj)
                attr = attr.val
                val_e = self(val)
                obj.result_type.struct_setattr(obj_e, attr, val_e)
                return None
            case asm.Slot(var_n, var_t):
                if var_n in self.types:
                    def_t = self.types[var_n]
                    if def_t != var_t:
                        raise TypeError(
                            f"Slot '{var_n}' is declared as type {def_t}, "
                            f"but used as type {var_t}."
                        )
                if var_n in self.slots:
                    return self.slots[var_n]
                raise KeyError(f"Slot '{var_n}' is not defined in the current context.")
            case asm.Unpack(asm.Slot(var_n, var_t), val):
                val_e = self(val)
                if not fisinstance(val_e, var_t):
                    raise TypeError(
                        f"Assigned value {val_e} is not of type {var_t} for "
                        f"variable '{var_n}'."
                    )
                assert var_n not in self.types, (
                    f"Variable '{var_n}' is already defined in the current"
                    f" context, cannot overwrite with slot."
                )
                self.types[var_n] = var_t
                self.slots[var_n] = val_e
                return None
            case asm.Repack(slot):
                self(slot)
                return None
            case asm.Call(f, args):
                f_e = self(f)
                args_e = [self(arg) for arg in args]
                return f_e(*args_e)
            case asm.Load(buf, idx):
                buf_e = self(buf)
                idx_e = self(idx)
                if self.mode.debug and not 0 <= idx_e < buf_e.length():
                    raise AssertionError(f"Buffer index {idx_e} is out of bounds")
                return buf_e.load(idx_e)
            case asm.Store(buf, idx, val):
                buf_e = self(buf)
                idx_e = self(idx)
                val_e = self(val)
                if self.mode.debug and not 0 <= idx_e < buf_e.length():
                    raise AssertionError(f"Buffer index {idx_e} is out of bounds")
                buf_e.store(idx_e, val_e)
                return None
            case asm.Resize(buf, len_):
                buf_e = self(buf)
                len_e = self(len_)
                buf_e.resize(len_e)
                return None
            case asm.Length(buf):
                buf_e = self(buf)
                return buf_e.length()
            case asm.Block(bodies):
                for body in bodies:
                    if self.should_halt():
                        break
                    self(body)
                return None
            case asm.ForLoop(asm.Variable(var_n, var_t) as var, start, end, body):
                start_e = self(start)
                end_e = self(end)
                if not fisinstance(start_e, var_t):
                    raise TypeError(
                        f"Start value {start_e} is not of type {var_t} for "
                        f"variable '{var_n}'."
                    )
                ctx_2 = self.scope(loop_state=HaltState())
                var_e = start_e
                while var_e < end_e:
                    if ctx_2.should_halt():
                        break
                    ctx_3 = self.scope()
                    ctx_3(asm.Block((asm.Assign(var, asm.Literal(var_e)), body)))
                    var_e = type(var_e)(var_e + 1)  # type: ignore[call-arg,operator]
                return None
            case asm.BufferLoop(buf, var, body):
                ctx_2 = self.scope(loop_state=HaltState())
                buf_e = self(buf)
                for i in range(buf_e.length()):
                    if ctx_2.should_halt():
                        break
                    ctx_3 = ctx_2.scope()
                    ctx_3(
                        asm.Block(
                            (asm.Assign(var, asm.Load(buf, asm.Literal(i))), body)
                        )
                    )
                return None
            case asm.WhileLoop(cond, body):
                ctx_2 = self.scope(loop_state=HaltState())
                while self(cond):
                    ctx_3 = ctx_2.scope()
                    if ctx_3.should_halt():
                        break
                    ctx_3(body)
                return None
            case asm.If(cond, body):
                if self(cond):
                    ctx_2 = self.scope()
                    ctx_2(body)
                return None
            case asm.IfElse(cond, body, else_body):
                if not self(cond):
                    body = else_body
                ctx_2 = self.scope()
                ctx_2(body)
                return None
            case asm.Function(
                asm.Variable(
                    func_n, asm.AssemblyKernelFType(result_type=ret_t) as func_type
                ),
                args,
                body,
            ):

                def my_func(*args_e):
                    ctx_2 = self.scope(function_state=HaltState())
                    if len(args_e) != len(args):
                        raise ValueError(
                            f"Function '{func_n}' expects {len(args)} arguments, "
                            f"but got {len(args_e)}."
                        )
                    for arg, arg_e in zip(args, args_e, strict=False):
                        match arg:
                            case asm.Variable(arg_n, arg_t):
                                if not fisinstance(arg_e, arg_t):
                                    raise TypeError(
                                        f"Argument '{arg_n}' is expected to be of type "
                                        f"{arg_t}, but got {type(arg_e)}."
                                    )
                                ctx_2.bindings[arg_n] = arg_e
                            case _:
                                raise NotImplementedError(
                                    f"Unrecognized argument type: {arg}"
                                )
                    ctx_2(body)
                    if ctx_2.function_state.should_halt:
                        ret_e = ctx_2.function_state.return_value
                        if not fisinstance(ret_e, ret_t):
                            raise TypeError(
                                f"Return value {ret_e} is not of type {ret_t} "
                                f"for function '{func_n}'."
                            )
                        return ret_e
                    raise ValueError(
                        f"Function '{func_n}' did not return a value, "
                        f"but expected type {ret_t}."
                    )

                self.bindings[func_n] = AssemblyInterpreterKernel(my_func, func_type)
                self.types[func_n] = func_type
                return None
            case asm.Return(value):
                assert self.function_state is not None
                self.function_state.return_value = self(value)
                self.function_state.should_halt = True
                return None
            case asm.Break():
                assert self.loop_state is not None
                self.loop_state.should_halt = True
                return None
            case asm.Module(funcs):
                ctx_2 = self.scope()
                for func in funcs:
                    ctx_2(func)
                kernels = {}
                for func in funcs:
                    match func:
                        case asm.Function(
                            asm.Variable(
                                func_n,
                                ret_t,
                            ),
                            args,
                            _,
                        ):
                            kernels[func_n] = ctx_2.bindings[func_n]
                        case _:
                            raise NotImplementedError(
                                f"Unrecognized function definition: {func}"
                            )
                return AssemblyInterpreterLibrary(ctx_2, kernels)
            case _:
                raise NotImplementedError(
                    f"Unrecognized assembly node type: {type(prgm)}"
                )

__call__(prgm, *, mode=None)

__call__(
    prgm: asm.Module, *, mode: CompilerMode | None = None
) -> AssemblyLibrary

__call__(prgm: asm.AssemblyNode) -> Any

Run the program.

Source code in src/finch/finch_assembly/interpreter.py

def __call__(self, prgm, *, mode=None):
    """
    Run the program.
    """
    if isinstance(prgm, asm.Module):
        return super().__call__(prgm, mode=mode)
    return self._dispatch(prgm)

scope(bindings=None, slots=None, types=None, loop_state=None, function_state=None, stdout=None, mode=None)

Create a new scope for the interpreter. This allows for nested scopes and variable shadowing.

Source code in src/finch/finch_assembly/interpreter.py

def scope(
    self,
    bindings=None,
    slots=None,
    types=None,
    loop_state=None,
    function_state=None,
    stdout=None,
    mode=None,
):
    """
    Create a new scope for the interpreter.
    This allows for nested scopes and variable shadowing.
    """
    if bindings is None:
        bindings = self.bindings.scope()
    if slots is None:
        slots = self.slots.scope()
    if types is None:
        types = self.types.scope()
    if loop_state is None:
        loop_state = self.loop_state
    if function_state is None:
        function_state = self.function_state
    if stdout is None:
        stdout = self.stdout
    return AssemblyInterpreter(
        bindings=bindings,
        slots=slots,
        types=types,
        loop_state=loop_state,
        function_state=function_state,
        stdout=stdout,
        mode=self.mode if mode is None else mode,
    )

should_halt()

Check if the interpreter should halt execution. This is used to stop execution in loops or when a return statement is encountered.

Source code in src/finch/finch_assembly/interpreter.py

def should_halt(self):
    """
    Check if the interpreter should halt execution.
    This is used to stop execution in loops or when a return
    statement is encountered.
    """
    return (
        self.loop_state
        and self.loop_state.should_halt
        or self.function_state
        and self.function_state.should_halt
    )

AssemblyInterpreterKernel

Bases: AssemblyKernel

A kernel for interpreting FinchAssembly code. This is a simple interpreter that executes the assembly code.

Source code in src/finch/finch_assembly/interpreter.py

class AssemblyInterpreterKernel(AssemblyKernel):
    """
    A kernel for interpreting FinchAssembly code.
    This is a simple interpreter that executes the assembly code.
    """

    def __init__(self, func, type_):
        super().__init__(type_)
        self.func = func

    def __call__(self, *args):
        return self.func(*args)

AssemblyInterpreterLibrary

Bases: AssemblyLibrary

A class to represent an interpreted module of FinchAssembly.

Source code in src/finch/finch_assembly/interpreter.py

class AssemblyInterpreterLibrary(AssemblyLibrary):
    """
    A class to represent an interpreted module of FinchAssembly.
    """

    def __init__(self, ctx, kernels):
        self.ctx = ctx
        self.kernels = kernels

    def __getattr__(self, name):
        # Allow attribute access to kernels by name
        if name in self.kernels:
            return self.kernels[name]
        raise AttributeError(
            f"{type(self).__name__!r} object has no attribute {name!r}"
        )

HaltState dataclass

A class to represent the halt state of an assembly program. This is used to indicate whether we should break or return, and what the return value is if applicable.

Source code in src/finch/finch_assembly/interpreter.py

@dataclass(eq=True)
class HaltState:
    """
    A class to represent the halt state of an assembly program.
    This is used to indicate whether we should break or return, and
    what the return value is if applicable.
    """

    should_halt: bool = False
    return_value: Any = None