finch_assembly

AssemblyCFGBuilder

Incrementally builds control-flow graph for Finch Assembly IR.

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

class AssemblyCFGBuilder:
    """Incrementally builds control-flow graph for Finch Assembly IR."""

    def __init__(self, namespace: Namespace):
        self.cfg: ControlFlowGraph = ControlFlowGraph()
        self.current_block: BasicBlock = self.cfg.entry_block
        self.namespace = namespace

    def emit(self, stmt) -> None:
        """Add a statement to the current block."""
        self.current_block.add_statement(stmt)

    def build(self, node: AssemblyNode) -> ControlFlowGraph:
        return self(node)

    def __call__(
        self,
        node: AssemblyNode,
        break_block: BasicBlock | None = None,
        return_block: BasicBlock | None = None,
    ) -> ControlFlowGraph:
        match node:
            case NumberedStatement(stmt, _):
                match stmt:
                    case Return(_):
                        self.emit(node)
                        assert return_block
                        self.current_block.add_successor(return_block)
                        unreachable_block = self.cfg.new_block()
                        self.current_block = unreachable_block
                    case Break():
                        self.emit(node)
                        assert break_block
                        self.current_block.add_successor(break_block)
                        unreachable_block = self.cfg.new_block()
                        self.current_block = unreachable_block
                    case _:
                        self.emit(node)
            case Block(bodies):
                for body in bodies:
                    self(body, break_block, return_block)
            case IfElse(_, body, else_body):
                before_block = self.current_block

                # create blocks for if, else, and after
                if_block = self.cfg.new_block()
                else_block = self.cfg.new_block()
                after_block = self.cfg.new_block()

                # connect before block to if and else blocks
                before_block.add_successor(if_block)
                before_block.add_successor(else_block)

                # fill in the if block
                self.current_block = if_block
                self(body, break_block, return_block)
                self.current_block.add_successor(after_block)

                # fill in the else block
                self.current_block = else_block
                self(else_body, break_block, return_block)
                self.current_block.add_successor(after_block)

                # continue building after the if-else
                self.current_block = after_block
            case WhileLoop(_, body):
                before_block = self.current_block

                # create blocks for the loop body and the code after the loop
                body_block = self.cfg.new_block()
                after_block = self.cfg.new_block()

                # connect before block to the loop body and the after block
                before_block.add_successor(body_block)
                before_block.add_successor(after_block)

                # fill in the loop body
                self.current_block = body_block
                self(body, after_block, return_block)

                # connect the end of loop body back to the beginning to form the loop
                self.current_block.add_successor(body_block)
                self.current_block.add_successor(after_block)
                self.current_block = after_block
            case Function(_, _, body):
                # Function argument definitions are inserted by `assembly_desugar`.
                self(body, break_block, return_block)
            case Module(funcs):
                for func in funcs:
                    if not isinstance(func, Function):
                        raise NotImplementedError(
                            f"Unrecognized function type: {type(func)}"
                        )

                    if not isinstance(func.name, Variable):
                        raise NotImplementedError(
                            f"Unrecognized function name type: {type(func.name)}"
                        )
                    func_name = func.name.name

                    # set block names to the function name
                    self.cfg.block_name = func_name

                    # create entry/exit block for the function
                    func_entry_block = self.cfg.new_block()
                    func_exit_block = self.cfg.new_block()

                    # connect CFG entry block to the function's entry block
                    self.cfg.entry_block.add_successor(func_entry_block)

                    # dive into the body of the function
                    self.current_block = func_entry_block
                    self(func, break_block, func_exit_block)

                    # connect last block in the function to the
                    # exit block of the function
                    self.current_block.add_successor(func_exit_block)

                    # connect function exit block to the exit block of the CFG
                    func_exit_block.add_successor(self.cfg.exit_block)
            case Assign(_, _):
                # unnumbered temporary variables for loops
                self.emit(node)
            case node:
                raise NotImplementedError(node, "AssemblyCFGBuilder")

        return self.cfg

emit(stmt)

Add a statement to the current block.

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

def emit(self, stmt) -> None:
    """Add a statement to the current block."""
    self.current_block.add_statement(stmt)

AssemblyCopyPropagation

Bases: AbstractAssemblyDataflow

Copy propagation for FinchAssembly.

Lattice: - defs: mapping { var_name: sid | None } describing a unique reaching definition id for each variable (None means "not uniquely defined"). - copies: mapping { dst_var: (src_var, src_def_id) } describing a direct copy dst_var = src_var that is valid only if src_var still has the same unique reaching definition src_def_id.

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

class AssemblyCopyPropagation(AbstractAssemblyDataflow):
    """Copy propagation for FinchAssembly.

    Lattice:
    - defs: mapping ``{ var_name: sid | None }`` describing a unique reaching
        definition id for each variable (None means "not uniquely defined").
    - copies: mapping ``{ dst_var: (src_var, src_def_id) }`` describing a direct
        copy ``dst_var = src_var`` that is valid only if ``src_var`` still has the
        same unique reaching definition ``src_def_id``.
    """

    def direction(self) -> str:
        """Copy propagation is a forward analysis."""
        return "forward"

    def collect_copy_replacements(self) -> dict[tuple[int, str], str]:
        """Collect per-statement copy replacements.

        Returns:
            dict: Mapping ``(stmt_id, old_var_name) -> new_var_name`` indicating
                where ``old_var_name`` can be replaced by ``new_var_name`` at
                statement ``stmt_id`` based on copy-propagation facts.
        """
        replacements: dict[tuple[int, str], str] = {}

        for block in self.cfg.blocks.values():
            input_state = self.input_states.get(block.id, {})
            state = copy.deepcopy(input_state)

            for stmt in block.statements:
                sid = getattr(stmt, "sid", None)
                if sid is not None:
                    for name, value in self.get_lattice_value(state, stmt):
                        if isinstance(value, tuple) and value:
                            replacements[(sid, name)] = value[0]

                state = self.transfer([stmt], state)

        return replacements

    def get_lattice_value(self, state, stmt) -> list[tuple[str, object]]:
        """Collect lattice annotations for variables used in a stmt or expr."""
        annotated: list[tuple[str, object]] = []
        target = stmt

        if isinstance(target, NumberedStatement):
            target = target.stmt

        match target:
            case Assign(_, rhs):
                target = rhs

        copies = state.get("copies", {}) if isinstance(state, dict) else {}

        for node in PostOrderDFS(target):
            match node:
                case Variable(name, _):
                    if name in copies:
                        annotated.append((name, copies[name]))
                case _:
                    continue
        return annotated

    def transfer(self, stmts: Iterable[Any], state: dict) -> dict:
        """Transfer function over a sequence of statements.

        Applies copy-propagation effects of each statement in order, returning
        the updated lattice mapping. Only copies of variables are recorded; any
        existing mappings that point to a variable being reassigned are
        invalidated first.

        Args:
            stmts: Iterable of Assembly statements in a basic block.
            state: Incoming lattice mapping.

        Returns:
            dict: The outgoing lattice mapping after processing ``stmts``.
        """
        state = self._normalize_state(state)
        new_state = {"defs": state["defs"].copy(), "copies": state["copies"].copy()}

        defs: dict[str, int | None] = new_state["defs"]
        copies: dict[str, tuple[str, int | None]] = new_state["copies"]

        for wrapped in stmts:
            sid, stmt = self._unpack_stmt(wrapped)
            match stmt:
                case Assign(Variable(lhs_name, _), rhs):
                    # Any assignment kills previous copy info involving lhs.
                    copies.pop(lhs_name, None)

                    # If some other variable was known to be a copy of lhs, kill it
                    # because lhs's value changed.
                    to_remove = [
                        dst for dst, (src, _) in copies.items() if src == lhs_name
                    ]
                    for dst in to_remove:
                        copies.pop(dst, None)

                    # Update reaching definition for lhs.
                    defs[lhs_name] = sid

                    # If rhs is a variable with a unique reaching def, record a copy.
                    if isinstance(rhs, Variable):
                        rhs_name = rhs.name
                        rhs_def = defs.get(rhs_name)
                        if rhs_def is not None:
                            copies[lhs_name] = (rhs_name, rhs_def)

                    # Ensure all recorded copies remain consistent with current defs.
                    new_state["copies"] = self._prune_inconsistent_copies(defs, copies)
                    copies = new_state["copies"]
                case _:
                    continue

        return new_state

    def join(self, state_1: dict, state_2: dict) -> dict:
        """Meet operator for must copy-propagation.

        - defs join: keep a def id only if both agree, else None.
        - copies join: keep a copy only if both agree exactly, and it remains
          consistent with the joined defs.
        """

        s1 = self._normalize_state(state_1)
        s2 = self._normalize_state(state_2)

        defs_1: dict[str, int | None] = s1["defs"]
        defs_2: dict[str, int | None] = s2["defs"]
        copies_1: dict[str, tuple[str, int | None]] = s1["copies"]
        copies_2: dict[str, tuple[str, int | None]] = s2["copies"]

        joined_defs: dict[str, int | None] = {}
        for name in set(defs_1) | set(defs_2):
            v1 = defs_1.get(name)
            v2 = defs_2.get(name)
            joined_defs[name] = v1 if v1 == v2 else None

        joined_copies: dict[str, tuple[str, int | None]] = {
            dst: val
            for dst, val in copies_1.items()
            if dst in copies_2 and copies_2[dst] == val
        }

        joined_copies = self._prune_inconsistent_copies(joined_defs, joined_copies)

        return {"defs": joined_defs, "copies": joined_copies}

    def _normalize_state(self, state: dict) -> dict:
        if not state:
            return {"defs": {}, "copies": {}}

        if "defs" not in state or "copies" not in state:
            # allow old/empty shapes; upgrade in place
            return {"defs": state.get("defs", {}), "copies": state.get("copies", {})}

        return state

    def _unpack_stmt(self, stmt):
        if isinstance(stmt, NumberedStatement):
            return stmt.sid, stmt.stmt
        return None, stmt

    def _prune_inconsistent_copies(self, defs: dict, copies: dict) -> dict:
        pruned: dict[str, tuple[str, int | None]] = {}
        for dst, (src, src_def) in copies.items():
            if src_def is None:
                continue
            if defs.get(src) != src_def:
                continue
            pruned[dst] = (src, src_def)
        return pruned

collect_copy_replacements()

Collect per-statement copy replacements.

Returns:

Name	Type	Description
dict	dict[tuple[int, str], str]	Mapping (stmt_id, old_var_name) -> new_var_name indicating where old_var_name can be replaced by new_var_name at statement stmt_id based on copy-propagation facts.

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

def collect_copy_replacements(self) -> dict[tuple[int, str], str]:
    """Collect per-statement copy replacements.

    Returns:
        dict: Mapping ``(stmt_id, old_var_name) -> new_var_name`` indicating
            where ``old_var_name`` can be replaced by ``new_var_name`` at
            statement ``stmt_id`` based on copy-propagation facts.
    """
    replacements: dict[tuple[int, str], str] = {}

    for block in self.cfg.blocks.values():
        input_state = self.input_states.get(block.id, {})
        state = copy.deepcopy(input_state)

        for stmt in block.statements:
            sid = getattr(stmt, "sid", None)
            if sid is not None:
                for name, value in self.get_lattice_value(state, stmt):
                    if isinstance(value, tuple) and value:
                        replacements[(sid, name)] = value[0]

            state = self.transfer([stmt], state)

    return replacements

direction()

Copy propagation is a forward analysis.

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

def direction(self) -> str:
    """Copy propagation is a forward analysis."""
    return "forward"

get_lattice_value(state, stmt)

Collect lattice annotations for variables used in a stmt or expr.

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

def get_lattice_value(self, state, stmt) -> list[tuple[str, object]]:
    """Collect lattice annotations for variables used in a stmt or expr."""
    annotated: list[tuple[str, object]] = []
    target = stmt

    if isinstance(target, NumberedStatement):
        target = target.stmt

    match target:
        case Assign(_, rhs):
            target = rhs

    copies = state.get("copies", {}) if isinstance(state, dict) else {}

    for node in PostOrderDFS(target):
        match node:
            case Variable(name, _):
                if name in copies:
                    annotated.append((name, copies[name]))
            case _:
                continue
    return annotated

join(state_1, state_2)

Meet operator for must copy-propagation.

• defs join: keep a def id only if both agree, else None.
• copies join: keep a copy only if both agree exactly, and it remains consistent with the joined defs.

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

def join(self, state_1: dict, state_2: dict) -> dict:
    """Meet operator for must copy-propagation.

    - defs join: keep a def id only if both agree, else None.
    - copies join: keep a copy only if both agree exactly, and it remains
      consistent with the joined defs.
    """

    s1 = self._normalize_state(state_1)
    s2 = self._normalize_state(state_2)

    defs_1: dict[str, int | None] = s1["defs"]
    defs_2: dict[str, int | None] = s2["defs"]
    copies_1: dict[str, tuple[str, int | None]] = s1["copies"]
    copies_2: dict[str, tuple[str, int | None]] = s2["copies"]

    joined_defs: dict[str, int | None] = {}
    for name in set(defs_1) | set(defs_2):
        v1 = defs_1.get(name)
        v2 = defs_2.get(name)
        joined_defs[name] = v1 if v1 == v2 else None

    joined_copies: dict[str, tuple[str, int | None]] = {
        dst: val
        for dst, val in copies_1.items()
        if dst in copies_2 and copies_2[dst] == val
    }

    joined_copies = self._prune_inconsistent_copies(joined_defs, joined_copies)

    return {"defs": joined_defs, "copies": joined_copies}

transfer(stmts, state)

Transfer function over a sequence of statements.

Applies copy-propagation effects of each statement in order, returning the updated lattice mapping. Only copies of variables are recorded; any existing mappings that point to a variable being reassigned are invalidated first.

Parameters:

Name	Type	Description	Default
stmts	Iterable[Any]	Iterable of Assembly statements in a basic block.	required
state	dict	Incoming lattice mapping.	required

Returns:

Name	Type	Description
dict	dict	The outgoing lattice mapping after processing stmts.

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

def transfer(self, stmts: Iterable[Any], state: dict) -> dict:
    """Transfer function over a sequence of statements.

    Applies copy-propagation effects of each statement in order, returning
    the updated lattice mapping. Only copies of variables are recorded; any
    existing mappings that point to a variable being reassigned are
    invalidated first.

    Args:
        stmts: Iterable of Assembly statements in a basic block.
        state: Incoming lattice mapping.

    Returns:
        dict: The outgoing lattice mapping after processing ``stmts``.
    """
    state = self._normalize_state(state)
    new_state = {"defs": state["defs"].copy(), "copies": state["copies"].copy()}

    defs: dict[str, int | None] = new_state["defs"]
    copies: dict[str, tuple[str, int | None]] = new_state["copies"]

    for wrapped in stmts:
        sid, stmt = self._unpack_stmt(wrapped)
        match stmt:
            case Assign(Variable(lhs_name, _), rhs):
                # Any assignment kills previous copy info involving lhs.
                copies.pop(lhs_name, None)

                # If some other variable was known to be a copy of lhs, kill it
                # because lhs's value changed.
                to_remove = [
                    dst for dst, (src, _) in copies.items() if src == lhs_name
                ]
                for dst in to_remove:
                    copies.pop(dst, None)

                # Update reaching definition for lhs.
                defs[lhs_name] = sid

                # If rhs is a variable with a unique reaching def, record a copy.
                if isinstance(rhs, Variable):
                    rhs_name = rhs.name
                    rhs_def = defs.get(rhs_name)
                    if rhs_def is not None:
                        copies[lhs_name] = (rhs_name, rhs_def)

                # Ensure all recorded copies remain consistent with current defs.
                new_state["copies"] = self._prune_inconsistent_copies(defs, copies)
                copies = new_state["copies"]
            case _:
                continue

    return new_state

AssemblyExpression

Bases: AssemblyNode, ExpressionTerm

Assembly AST expression base class.

An assembly expression is a program node which evaluates to a value.

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

class AssemblyExpression(AssemblyNode, ExpressionTerm):
    """
    Assembly AST expression base class.

    An assembly expression is a program node which evaluates to a value.
    """

    @property
    @abstractmethod
    def result_type(self) -> BufferFType:
        """Returns the type of the expression."""
        ...

result_type abstractmethod property

Returns the type of the expression.

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)

AssemblyKernel

Bases: FTyped, ABC

A callable implementation of an AssemblyKernelFType.

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

class AssemblyKernel(FTyped, ABC):
    """A callable implementation of an AssemblyKernelFType."""

    def __init__(self, type_: AssemblyKernelFType):
        self._ftype = type_

    @property
    def ftype(self) -> AssemblyKernelFType:
        return self._ftype

    @abstractmethod
    def __call__(self, *args) -> Any: ...

AssemblyKernelFType dataclass

Bases: CallableFType

The identity and signature of a function defined in a Finch module.

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

@dataclass(eq=False, frozen=True)
class AssemblyKernelFType(CallableFType):
    """The identity and signature of a function defined in a Finch module."""

    name: str
    arg_types: tuple[FType, ...]
    result_type: FType
    definition: str = field(default_factory=lambda: uuid4().hex)

    def __eq__(self, other):
        match other:
            case AssemblyKernelFType():
                return (
                    self.name,
                    self.arg_types,
                    self.result_type,
                    self.definition,
                ) == (other.name, other.arg_types, other.result_type, other.definition)
        return NotImplemented

    def __hash__(self):
        return hash((self.name, self.arg_types, self.result_type, self.definition))

    def __call__(self, value):
        if isinstance(value, AssemblyKernel) and value.ftype == self:
            return value
        raise TypeError(f"Expected a kernel of type {self}")

    def return_type(self, *args: FType) -> FType:
        if args != self.arg_types:
            raise TypeError(f"{self.name} expects {self.arg_types}, got {args}")
        return self.result_type

AssemblyLibrary

Bases: ABC

Represents a module containing assembly kernels.

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

class AssemblyLibrary(ABC):
    """
    Represents a module containing assembly kernels.
    """

    @abstractmethod
    def __getattr__(self, name: str) -> AssemblyKernel:
        """
        Get the assembly Kernel corresponding to the given name.
        """
        ...

__getattr__(name) abstractmethod

Get the assembly Kernel corresponding to the given name.

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

@abstractmethod
def __getattr__(self, name: str) -> AssemblyKernel:
    """
    Get the assembly Kernel corresponding to the given name.
    """
    ...

AssemblyLoader

Bases: Stage

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

class AssemblyLoader(Stage):
    @abstractmethod
    def lower(
        self, term: asm.Module, *, mode: CompilerMode | None = None
    ) -> AssemblyLibrary:
        """
        Load the given assembly program into a runnable module.
        """

lower(term, *, mode=None) abstractmethod

Load the given assembly program into a runnable module.

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

@abstractmethod
def lower(
    self, term: asm.Module, *, mode: CompilerMode | None = None
) -> AssemblyLibrary:
    """
    Load the given assembly program into a runnable module.
    """

AssemblyNode

Bases: Term

AssemblyNode

Represents a FinchAssembly IR node. FinchAssembly is the final intermediate representation before code generation (translation to the output language). It is a low-level imperative description of the program, with control flow, linear memory regions called "buffers", and explicit memory management.

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

class AssemblyNode(Term):
    """
    AssemblyNode

    Represents a FinchAssembly IR node. FinchAssembly is the final intermediate
    representation before code generation (translation to the output language).
    It is a low-level imperative description of the program, with control flow,
    linear memory regions called "buffers", and explicit memory management.
    """

    @classmethod
    def head(cls):
        """Returns the head of the node."""
        return cls

    @classmethod
    def make_term(cls, head, *args):
        """Creates a term with the given head and arguments."""
        return head.from_children(*args)

    @classmethod
    def from_children(cls, *children):
        """
        Creates a term from the given children. This is used to create terms
        from the children of a node.
        """
        return cls(*children)

    def __str__(self):
        """Returns a string representation of the node."""
        ctx = AssemblyPrinterContext()
        res = ctx(self)
        return res if res is not None else ctx.emit()

__str__()

Returns a string representation of the node.

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

def __str__(self):
    """Returns a string representation of the node."""
    ctx = AssemblyPrinterContext()
    res = ctx(self)
    return res if res is not None else ctx.emit()

from_children(*children) classmethod

Creates a term from the given children. This is used to create terms from the children of a node.

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

@classmethod
def from_children(cls, *children):
    """
    Creates a term from the given children. This is used to create terms
    from the children of a node.
    """
    return cls(*children)

head() classmethod

Returns the head of the node.

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

@classmethod
def head(cls):
    """Returns the head of the node."""
    return cls

make_term(head, *args) classmethod

Creates a term with the given head and arguments.

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

@classmethod
def make_term(cls, head, *args):
    """Creates a term with the given head and arguments."""
    return head.from_children(*args)

AssemblyTransform

Bases: Stage

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

class AssemblyTransform(Stage):
    @abstractmethod
    def lower(self, term: asm.Module) -> asm.Module:
        """
        Transform the given assembly term into another assembly term.
        """

lower(term) abstractmethod

Transform the given assembly term into another assembly term.

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

@abstractmethod
def lower(self, term: asm.Module) -> asm.Module:
    """
    Transform the given assembly term into another assembly term.
    """

AssemblyTypeChecker

A type checker for FinchAssembly

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

class AssemblyTypeChecker:
    """
    A type checker for FinchAssembly
    """

    def __init__(
        self,
        ctxt=None,
        loop_state=None,
        function_state=None,
    ):
        if ctxt is None:
            ctxt = ScopedDict()
        self.ctxt = ctxt
        self.loop_state = loop_state
        self.function_state = function_state

    def scope(
        self,
        ctxt=None,
        loop_state=None,
        function_state=None,
    ):
        if ctxt is None:
            ctxt = self.ctxt.scope()
        if loop_state is None:
            loop_state = self.loop_state
        if function_state is None:
            function_state = self.function_state
        return AssemblyTypeChecker(
            ctxt=ctxt,
            loop_state=loop_state,
            function_state=function_state,
        )

    def check_return_type(self, return_type):
        if self.function_state:
            check_type_match(
                self.function_state.return_type,
                return_type,
            )
        else:
            raise AssemblyTypeError("Cannot return outside of function.")

    def check_in_ctxt(self, var_n, var_t):
        try:
            check_type_match(self.ctxt[var_n], var_t)
        except KeyError:
            raise AssemblyTypeError(
                f"The variable '{var_n}' is not defined in the current context."
            ) from KeyError

    def check_buffer(self, buffer):
        buffer_type = self.check_expr(buffer)
        if isinstance(buffer_type, BufferFType):
            return buffer_type
        raise AssemblyTypeError(f"Expected buffer, got {buffer_type}.")

    def check_struct(self, struct):
        struct_type = self.check_expr(struct)
        if isinstance(struct_type, (StructFType)):
            return struct_type
        raise AssemblyTypeError(f"Expected struct, got {struct_type}.")

    def check_cond(self, cond):
        cond_type = self.check_expr(cond)
        cond_ftype = ftype(cond_type)
        if isinstance(cond_ftype, (FDTypeNumeric, FDTypeBoolean)):
            return
        raise AssemblyTypeError("Conditional must be number or boolean.")

    def check_expr(self, expr: asm.AssemblyExpression):
        match expr:
            case asm.Literal(value):
                return ftype(value)
            case asm.Variable(var_n, var_t) | asm.Slot(var_n, var_t):
                check_type(var_t)
                self.check_in_ctxt(var_n, var_t)
                return var_t
            case asm.GetAttr(obj, asm.Literal(attr)):
                obj_type = self.check_struct(obj)
                return check_attrtype(obj_type, attr)
            case asm.Call(op, args):
                op_type = self.check_expr(op)
                if not isinstance(op_type, algebra.CallableFType):
                    raise AssemblyTypeError(f"Expected a callable type, got {op_type}")
                arg_types = tuple(self.check_expr(arg) for arg in args)
                try:
                    return op_type.return_type(*arg_types)
                except (TypeError, ValueError, AssertionError) as exc:
                    raise AssemblyTypeError(
                        f"Cannot call {op_type} with {arg_types}"
                    ) from exc
            case asm.Load(buffer, index):
                buffer_type = self.check_buffer(buffer)
                index_type = self.check_expr(index)
                check_type_match(buffer_type.length_type, index_type)
                return buffer_type.element_type
            case asm.Length(buffer):
                buffer_type = self.check_buffer(buffer)
                return buffer_type.length_type
            case _:
                raise ValueError(f"Ill-formed AssemblyExpression:  {type(expr)}.")

    def check_stmt(self, stmt: asm.AssemblyStatement):
        if isinstance(stmt, asm.AssemblyExpression):
            self.check_expr(stmt)
            return None
        match stmt:
            case asm.Unpack(asm.Slot(var_n, var_t), rhs):
                check_type(var_t)
                rhs_type = self(rhs)
                check_type_match(var_t, rhs_type)
                if var_n in self.ctxt:
                    raise AssemblyTypeError(
                        f"Slot {var_n} is already defined in the current "
                        f"context, cannot overwrite with slot."
                    )
                self.ctxt[var_n] = var_t
                return None
            case asm.Repack(asm.Slot(var_n, var_t)):
                check_type(var_t)
                self.check_in_ctxt(var_n, var_t)
                return None
            case asm.Assign(asm.Variable(var_n, var_t), rhs):
                check_type(var_t)
                rhs_type = self.check_expr(rhs)
                check_type_match(var_t, rhs_type)
                if var_n in self.ctxt:
                    check_type_match(self.ctxt[var_n], var_t)
                else:
                    self.ctxt[var_n] = var_t
                return None
            case asm.SetAttr(obj, asm.Literal(attr), value):
                obj_type = self.check_struct(obj)
                attrtype = check_attrtype(obj_type, attr)
                value_type = self.check_expr(value)
                check_type_match(attrtype, value_type)
                return None
            case asm.Store(buffer, index, value):
                buffer_type = self.check_buffer(buffer)
                index_type = self.check_expr(index)
                check_type_match(buffer_type.length_type, index_type)
                value_type = self.check_expr(value)
                check_type_match(buffer_type.element_type, value_type)
                return None
            case asm.Resize(buffer, new_size):
                buffer_type = self.check_buffer(buffer)
                new_size_type = self.check_expr(new_size)
                check_type_match(buffer_type.length_type, new_size_type)
                return None
            case asm.ForLoop(asm.Variable(var_n, var_t), start, end, body):
                check_type(var_t)
                check_is_index_type(var_t)
                start_type = self.check_expr(start)
                end_type = self.check_expr(end)
                check_type_match(var_t, start_type)
                check_type_match(var_t, end_type)
                loop = self.scope(loop_state=LoopState())
                loop.ctxt[var_n] = var_t
                loop.check_stmt(body)
                return None
            case asm.BufferLoop(buffer, asm.Variable(var_n, var_t), body):
                check_type(var_t)
                buffer_type = self.check_buffer(buffer)
                check_type_match(buffer_type.element_type, var_t)
                loop = self.scope(loop_state=LoopState())
                loop.ctxt[var_n] = var_t
                loop.check_stmt(body)
                return None
            case asm.WhileLoop(cond, body):
                self.check_cond(cond)
                self.loop_state = LoopState()
                self.check_stmt(body)
                self.loop_state = None
                return None
            case asm.If(cond, body):
                self.check_cond(cond)
                self.check_stmt(body)
                return None
            case asm.IfElse(cond, body, else_body):
                self.check_cond(cond)
                body_type = self.check_stmt(body)
                else_body_type = self.check_stmt(else_body)
                if body_type is None or else_body_type is None:
                    return None
                return body_type
            case asm.Return(arg):
                return_type = self.check_expr(arg)
                self.check_return_type(return_type)
                return return_type
            case asm.Assert(arg):
                t = self.check_expr(arg)
                check_type_match(ftype(np.bool_), t)
                return None
            case asm.Break():
                if self.loop_state is None:
                    raise AssemblyTypeError("Cannot break outside of loop.")
                return None
            case asm.Block(bodies):
                block = self.scope()
                return_type = None
                left_already = False
                for body in bodies:
                    if left_already:
                        raise AssemblyTypeError("Unreachable statements in block.")
                    return_type = block.check_stmt(body)
                    left_already = return_type is not None or body is asm.Break
                return return_type
            case _:
                raise ValueError(f"Ill-formed statement:  {type(stmt)}.")

    def check_function(self, func: asm.AssemblyNode):
        match func:
            case asm.Function(
                asm.Variable(
                    func_name,
                    asm.AssemblyKernelFType(result_type=return_type) as func_type,
                ),
                args,
                body,
            ):
                check_type(return_type)
                if tuple(arg.result_type for arg in args) != func_type.arg_types:
                    raise AssemblyTypeError(
                        f"Function '{func_name}' arguments do not match {func_type}."
                    )
                if self.function_state:
                    raise AssemblyTypeError(
                        f"Cannot nest function definitions:  '{func_name}'."
                    )
                body_scope = self.scope(function_state=FunctionState(return_type))
                body_scope.ctxt[func_name] = func_type
                for arg in args:
                    check_type(arg.type)
                    body_scope.ctxt[arg.name] = arg.type
                body_type = body_scope.check_stmt(body)
                if body_type is None:
                    raise AssemblyTypeError(
                        f"Function '{func_name}' is not guaranteed to return."
                    )
                check_type_match(return_type, body_type)
                return
            case _:
                raise ValueError(f"Ill-formed function:  {type(func)}.")

    def check_module(self, mod: asm.AssemblyNode):
        defined_funcs = []
        match mod:
            case asm.Module(funcs):
                for func in funcs:
                    func_name = check_is_func(func)  # can be removed later
                    if func_name in defined_funcs:
                        raise AssemblyTypeError(
                            f"Two functions defined with the name '{func_name}'."
                        )
                    self.ctxt[func_name] = func.name.result_type
                    defined_funcs.append(func_name)
                for func in funcs:
                    self.check_function(func)
                return
            case _:
                raise AssemblyTypeError(f"Expected module, got {type(mod)}.")

    def __call__(self, prgm: asm.AssemblyNode):
        if isinstance(prgm, asm.Module):
            self.check_module(prgm)
            return None
        if isinstance(prgm, asm.Function):
            self.check_function(prgm)
            return None
        if isinstance(prgm, asm.AssemblyStatement):
            self.check_stmt(prgm)
            return None
        if isinstance(prgm, asm.AssemblyExpression):
            return self.check_expr(prgm)
        raise ValueError(f"Ill-formed program:  {type(prgm)}.")

Assert dataclass

Bases: AssemblyTree, AssemblyStatement

Checks that an expression is true at runtime. Also used in dataflow analysis to record conditions in conditionals and loops.

Attributes:

Name	Type	Description
exp	AssemblyExpression	Expression which is being asserted.

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

@dataclass(eq=True, frozen=True)
class Assert(AssemblyTree, AssemblyStatement):
    """
    Checks that an expression is true at runtime. Also used in dataflow analysis
    to record conditions in conditionals and loops.

    Attributes:
        exp: Expression which is being asserted.
    """

    exp: AssemblyExpression

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.exp]

children property

Returns the children of the node.

Assign dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a logical AST statement that evaluates rhs, binding the result to lhs.

Attributes:

Name	Type	Description
lhs	Variable	The left-hand side of the binding.
rhs	AssemblyExpression	The right-hand side to evaluate.

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

@dataclass(eq=True, frozen=True)
class Assign(AssemblyTree, AssemblyStatement):
    """
    Represents a logical AST statement that evaluates `rhs`, binding the result
    to `lhs`.

    Attributes:
        lhs: The left-hand side of the binding.
        rhs: The right-hand side to evaluate.
    """

    lhs: Variable
    rhs: AssemblyExpression

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.lhs, self.rhs]

children property

Returns the children of the node.

Block dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a statement that executes a sequence of statements bodies....

Attributes:

Name	Type	Description
bodies	tuple[AssemblyStatement | AssemblyExpression, ...]	The sequence of statements to execute.

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

@dataclass(eq=True, frozen=True)
class Block(AssemblyTree, AssemblyStatement):
    """
    Represents a statement that executes a sequence of statements `bodies...`.

    Attributes:
        bodies: The sequence of statements to execute.
    """

    bodies: tuple[AssemblyStatement | AssemblyExpression, ...] = ()

    @property
    def children(self):
        """Returns the children of the node."""
        return [*self.bodies]

    @classmethod
    def from_children(cls, *bodies):
        return cls(bodies)

children property

Returns the children of the node.

Break dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a break statement that exits the current loop.

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

@dataclass(eq=True, frozen=True)
class Break(AssemblyTree, AssemblyStatement):
    """
    Represents a break statement that exits the current loop.
    """

    @property
    def children(self):
        """Returns the children of the node."""
        return []

children property

Returns the children of the node.

Buffer

Bases: FTyped[BufferFType[FT]], ABC

Abstract base class for buffer-like data structures. Buffers support random access, and can be resized. They are used to store data in a way that allows for efficient reading and writing of elements.

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

class Buffer(FTyped[BufferFType[FT]], ABC):
    """
    Abstract base class for buffer-like data structures. Buffers support random access,
    and can be resized. They are used to store data in a way that allows for efficient
    reading and writing of elements.
    """

    @abstractmethod
    def __init__(self, length: int, dtype: type): ...

    @abstractmethod
    def length(self):
        """
        Return the length of the buffer.
        """
        ...

    @property
    def element_type(self) -> FT:
        """
        Return the type of elements stored in the buffer.
        This is typically the same as the dtype used to create the buffer.
        """
        return self.ftype.element_type

    @property
    def length_type(self) -> FDTypeInteger:
        """
        Return the type of indices used to access elements in the buffer.
        This is typically an integer type.
        """
        return self.ftype.length_type

    @abstractmethod
    def load(self, idx: int): ...

    @abstractmethod
    def store(self, idx: int, val): ...

    @abstractmethod
    def resize(self, len: int):
        """
        Resize the buffer to the new length.
        """
        ...

element_type property

Return the type of elements stored in the buffer. This is typically the same as the dtype used to create the buffer.

length_type property

Return the type of indices used to access elements in the buffer. This is typically an integer type.

length() abstractmethod

Return the length of the buffer.

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

@abstractmethod
def length(self):
    """
    Return the length of the buffer.
    """
    ...

resize(len) abstractmethod

Resize the buffer to the new length.

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

@abstractmethod
def resize(self, len: int):
    """
    Resize the buffer to the new length.
    """
    ...

BufferFType

Bases: FType, Generic[FT]

Abstract base class for the ftype of arguments. The ftype defines how the data structures store data, and can construct a data structure with the call method.

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

class BufferFType(FType, Generic[FT]):
    """
    Abstract base class for the ftype of arguments. The ftype defines how the
    data structures store data, and can construct a data structure with the call method.
    """

    @abstractmethod
    def __call__(self, *args, **kwargs):
        """
        Create an instance of an object in this ftype with the given arguments.
        """
        ...

    @property
    @abstractmethod
    def element_type(self) -> FT:
        """
        Return the type of elements stored in the buffer.
        This is typically the same as the dtype used to create the buffer.
        """
        ...

    @property
    def length_type(self) -> FDTypeInteger:
        """
        Returns the type used for the length of the buffer.
        """
        return intp

element_type abstractmethod property

Return the type of elements stored in the buffer. This is typically the same as the dtype used to create the buffer.

length_type property

Returns the type used for the length of the buffer.

__call__(*args, **kwargs) abstractmethod

Create an instance of an object in this ftype with the given arguments.

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

@abstractmethod
def __call__(self, *args, **kwargs):
    """
    Create an instance of an object in this ftype with the given arguments.
    """
    ...

BufferLoop dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a loop that iterates over the elements of a buffer.

Attributes:

Name	Type	Description
buffer	AssemblyExpression	The buffer to iterate over.
var	Variable	The loop variable for each element in the buffer.
body	AssemblyStatement	The body of the loop to execute for each element.

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

@dataclass(eq=True, frozen=True)
class BufferLoop(AssemblyTree, AssemblyStatement):
    """
    Represents a loop that iterates over the elements of a buffer.

    Attributes:
        buffer: The buffer to iterate over.
        var: The loop variable for each element in the buffer.
        body: The body of the loop to execute for each element.
    """

    buffer: AssemblyExpression
    var: Variable
    body: AssemblyStatement

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.buffer, self.var, self.body]

children property

Returns the children of the node.

Call dataclass

Bases: AssemblyExpression, AssemblyTree, CallTerm

Represents an expression for calling the function op on args....

Attributes:

Name	Type	Description
op	AssemblyExpression	The function to call.
args	tuple[AssemblyExpression, ...]	The arguments to call on the function.

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

@dataclass(eq=True, frozen=True)
class Call(AssemblyExpression, AssemblyTree, CallTerm):
    """
    Represents an expression for calling the function `op` on `args...`.

    Attributes:
        op: The function to call.
        args: The arguments to call on the function.
    """

    op: AssemblyExpression
    args: tuple[AssemblyExpression, ...]

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.op, *self.args]

    @classmethod
    def from_children(cls, op, *args):
        return cls(op, args)

    @property
    def result_type(self):
        """Returns the type of the expression."""
        arg_types = [arg.result_type for arg in self.args]
        op_type = self.op.result_type
        assert isinstance(op_type, CallableFType)
        return return_type(op_type, *arg_types)

children property

Returns the children of the node.

result_type property

Returns the type of the expression.

ForLoop dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a for loop that iterates over a range of values.

Attributes:

Name	Type	Description
var	Variable	The loop variable.
start	AssemblyExpression	The starting value of the range.
end	AssemblyExpression	The ending value of the range.
body	AssemblyStatement	The body of the loop to execute.

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

@dataclass(eq=True, frozen=True)
class ForLoop(AssemblyTree, AssemblyStatement):
    """
    Represents a for loop that iterates over a range of values.

    Attributes:
        var: The loop variable.
        start: The starting value of the range.
        end: The ending value of the range.
        body: The body of the loop to execute.
    """

    var: Variable
    start: AssemblyExpression
    end: AssemblyExpression
    body: AssemblyStatement

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.var, self.start, self.end, self.body]

children property

Returns the children of the node.

Function dataclass

Bases: AssemblyTree

A module-level function definition with arguments args....

Attributes:

Name	Type	Description
name	Variable	The function variable, annotated with its AssemblyKernelFType.
args	tuple[Variable, ...]	The arguments to the function.
body	AssemblyStatement	The body of the function. If it does not contain a return statement, the function returns the value of body.

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

@dataclass(eq=True, frozen=True)
class Function(AssemblyTree):
    """
    A module-level function definition with arguments `args...`.

    Attributes:
        name: The function variable, annotated with its AssemblyKernelFType.
        args: The arguments to the function.
        body: The body of the function. If it does not contain a return statement,
            the function returns the value of `body`.
    """

    name: Variable
    args: tuple[Variable, ...]
    body: AssemblyStatement

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.name, *self.args, self.body]

    @classmethod
    def from_children(cls, *children):
        """Creates a term with the given head and arguments."""
        name, *arg_nodes, body = children
        return cls(name, tuple(arg_nodes), body)

children property

Returns the children of the node.

from_children(*children) classmethod

Creates a term with the given head and arguments.

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

@classmethod
def from_children(cls, *children):
    """Creates a term with the given head and arguments."""
    name, *arg_nodes, body = children
    return cls(name, tuple(arg_nodes), body)

GetAttr dataclass

Bases: AssemblyExpression, AssemblyTree

Represents a getter for an attribute attr of an object obj. Attributes: obj: The object to get the attribute from. attr: The name of the attribute to get.

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

@dataclass(eq=True, frozen=True)
class GetAttr(AssemblyExpression, AssemblyTree):
    """
    Represents a getter for an attribute `attr` of an object `obj`.
    Attributes:
        obj: The object to get the attribute from.
        attr: The name of the attribute to get.
    """

    obj: AssemblyExpression
    attr: Literal

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.obj, self.attr]

    @property
    def result_type(self):
        """Returns the type of the expression."""
        assert isinstance(self.obj.result_type, StructFType)
        return dict(self.obj.result_type.struct_fields)[self.attr.val]

children property

Returns the children of the node.

result_type property

Returns the type of the expression.

If dataclass

Bases: AssemblyTree, AssemblyStatement

Represents an if statement that executes the body if the condition is true.

Attributes:

Name	Type	Description
condition	AssemblyExpression	The condition to evaluate for the if to execute the body.
body	AssemblyStatement	The body of the if statement to execute.

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

@dataclass(eq=True, frozen=True)
class If(AssemblyTree, AssemblyStatement):
    """
    Represents an if statement that executes the body if the condition is true.

    Attributes:
        condition: The condition to evaluate for the if to execute the body.
        body: The body of the if statement to execute.
    """

    condition: AssemblyExpression
    body: AssemblyStatement

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.condition, self.body]

children property

Returns the children of the node.

IfElse dataclass

Bases: AssemblyTree, AssemblyStatement

Represents an if-else statement that executes the body if the condition is true, otherwise executes else_body.

Attributes:

Name	Type	Description
condition	AssemblyExpression	The condition to evaluate for the if to execute the body.
body	AssemblyStatement	The body of the if statement to execute.
else_body	AssemblyStatement	An alternative body to execute if the condition is false.

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

@dataclass(eq=True, frozen=True)
class IfElse(AssemblyTree, AssemblyStatement):
    """
    Represents an if-else statement that executes the body if the condition
    is true, otherwise executes else_body.

    Attributes:
        condition: The condition to evaluate for the if to execute the body.
        body: The body of the if statement to execute.
        else_body: An alternative body to execute if the condition is false.
    """

    condition: AssemblyExpression
    body: AssemblyStatement
    else_body: AssemblyStatement

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.condition, self.body, self.else_body]

children property

Returns the children of the node.

Length dataclass

Bases: AssemblyExpression, AssemblyTree

Represents getting the length of a buffer.

Attributes:

Name	Type	Description
buffer	AssemblyExpression	The buffer whose length is queried.

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

@dataclass(eq=True, frozen=True)
class Length(AssemblyExpression, AssemblyTree):
    """
    Represents getting the length of a buffer.

    Attributes:
        buffer: The buffer whose length is queried.
    """

    buffer: AssemblyExpression

    @property
    def children(self):
        return [self.buffer]

    @property
    def result_type(self):
        """Returns the type of the expression."""
        return length_type(self.buffer.result_type)

result_type property

Returns the type of the expression.

Literal dataclass

Bases: AssemblyExpression, LiteralTerm[T]

Represents the literal value val.

Attributes:

Name	Type	Description
val	T	The literal value.

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

class Literal(AssemblyExpression, LiteralTerm[T]):
    """
    Represents the literal value `val`.

    Attributes:
        val: The literal value.
    """

    @property
    def result_type(self):
        """Returns the type of the expression."""
        return ftype(self.val)

    def __repr__(self) -> str:
        return literal_repr(type(self).__name__, {"val": self.val})

result_type property

Returns the type of the expression.

Load dataclass

Bases: AssemblyExpression, AssemblyTree

Represents loading a value from a buffer at a given index.

Attributes:

Name	Type	Description
buffer	AssemblyExpression	The buffer to load from.
index	AssemblyExpression	The index to load at.

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

@dataclass(eq=True, frozen=True)
class Load(AssemblyExpression, AssemblyTree):
    """
    Represents loading a value from a buffer at a given index.

    Attributes:
        buffer: The buffer to load from.
        index: The index to load at.
    """

    buffer: AssemblyExpression
    index: AssemblyExpression

    @property
    def children(self):
        return [self.buffer, self.index]

    @property
    def result_type(self) -> FType:
        """Returns the type of the expression."""
        assert isinstance(self.buffer.result_type, BufferFType)
        return self.buffer.result_type.element_type

result_type property

Returns the type of the expression.

Module dataclass

Bases: AssemblyTree

Represents a group of functions. This is the toplevel translation unit for FinchAssembly.

Attributes:

Name	Type	Description
funcs	tuple[Function, ...]	The functions defined in the module.

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

@dataclass(eq=True, frozen=True)
class Module(AssemblyTree):
    """
    Represents a group of functions. This is the toplevel translation unit for
    FinchAssembly.

    Attributes:
        funcs: The functions defined in the module.
    """

    funcs: tuple[Function, ...]

    @property
    def children(self):
        """Returns the children of the node."""
        return [*self.funcs]

    @classmethod
    def from_children(cls, *funcs):
        return cls(funcs)

children property

Returns the children of the node.

NumberedStatement dataclass

Bases: AssemblyStatement

Wrapper for AssemblyStatement that assigns a unique id to each statement for easier tracking in the CFG.

Attributes:

Name	Type	Description
stmt	AssemblyStatement	The original AssemblyStatement being wrapped.
sid	int	A unique integer identifier for the statement.

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

@dataclass(eq=True, frozen=True)
class NumberedStatement(AssemblyStatement):
    """
    Wrapper for AssemblyStatement that assigns a unique id to each statement
    for easier tracking in the CFG.

    Attributes:
        stmt: The original AssemblyStatement being wrapped.
        sid: A unique integer identifier for the statement.
    """

    stmt: AssemblyStatement
    sid: int

    def __str__(self) -> str:
        return f"[{self.sid}] {str(self.stmt)}"

    @property
    def children(self):
        return (self.stmt, self.sid)

Repack dataclass

Bases: AssemblyTree, AssemblyStatement

Registers updates from a symbolic object val with the original object. The original object may now be accessed and modified.

Attributes:

Name	Type	Description
slot		The symbolic object to read from.

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

@dataclass(eq=True, frozen=True)
class Repack(AssemblyTree, AssemblyStatement):
    """
    Registers updates from a symbolic object `val` with the original
    object. The original object may now be accessed and modified.

    Attributes:
        slot: The symbolic object to read from.
    """

    val: Slot

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.val]

children property

Returns the children of the node.

Resize dataclass

Bases: AssemblyTree, AssemblyStatement

Represents resizing a buffer to a new size.

Attributes:

Name	Type	Description
buffer	AssemblyExpression	The buffer to resize.
new_size	AssemblyExpression	The new size for the buffer.

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

@dataclass(eq=True, frozen=True)
class Resize(AssemblyTree, AssemblyStatement):
    """
    Represents resizing a buffer to a new size.

    Attributes:
        buffer: The buffer to resize.
        new_size: The new size for the buffer.
    """

    buffer: AssemblyExpression
    new_size: AssemblyExpression

    @property
    def children(self):
        return [self.buffer, self.new_size]

Return dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a return statement that returns arg from the current function. Halts execution of the function body.

Attributes:

Name	Type	Description
arg	AssemblyExpression	The argument to return.

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

@dataclass(eq=True, frozen=True)
class Return(AssemblyTree, AssemblyStatement):
    """
    Represents a return statement that returns `arg` from the current function.
    Halts execution of the function body.

    Attributes:
        arg: The argument to return.
    """

    arg: AssemblyExpression

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.arg]

children property

Returns the children of the node.

SetAttr dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a setter for an attribute attr of an object obj. Attributes: obj: The object to set the attribute on. attr: The name of the attribute to set. value: The value to set the attribute to.

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

@dataclass(eq=True, frozen=True)
class SetAttr(AssemblyTree, AssemblyStatement):
    """
    Represents a setter for an attribute `attr` of an object `obj`.
    Attributes:
        obj: The object to set the attribute on.
        attr: The name of the attribute to set.
        value: The value to set the attribute to.
    """

    obj: AssemblyExpression
    attr: Literal
    value: AssemblyExpression

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.obj, self.attr, self.value]

children property

Returns the children of the node.

Slot dataclass

Bases: AssemblyExpression

Represents a register to a symbolic object. Using a register in an expression creates a copy of the object.

Attributes:

Name	Type	Description
name	str	The name of the symbolic object to register.
type	Any	The type of the symbolic object.

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

@dataclass(eq=True, frozen=True)
class Slot(AssemblyExpression):
    """
    Represents a register to a symbolic object. Using a register in an
    expression creates a copy of the object.

    Attributes:
        name: The name of the symbolic object to register.
        type: The type of the symbolic object.
    """

    name: str
    type: Any

    @property
    def result_type(self):
        """Returns the type of the expression."""
        return self.type

    def __repr__(self) -> str:
        return literal_repr(type(self).__name__, asdict(self))

    @property
    def symbol(self) -> str:
        return self.name

result_type property

Returns the type of the expression.

Store dataclass

Bases: AssemblyTree, AssemblyStatement

Represents storing a value into a buffer at a given index.

Attributes:

Name	Type	Description
buffer	AssemblyExpression	The buffer to store into.
index	AssemblyExpression	The index to store at.
value	AssemblyExpression	The value to store.

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

@dataclass(eq=True, frozen=True)
class Store(AssemblyTree, AssemblyStatement):
    """
    Represents storing a value into a buffer at a given index.

    Attributes:
        buffer: The buffer to store into.
        index: The index to store at.
        value: The value to store.
    """

    buffer: AssemblyExpression
    index: AssemblyExpression
    value: AssemblyExpression

    @property
    def children(self):
        return [self.buffer, self.index, self.value]

Unpack dataclass

Bases: AssemblyTree, AssemblyStatement

Attempts to convert rhs into a symbolic, which can be registerd with lhs. The original object must not be accessed or modified until the corresponding Repack node is reached.

Attributes:

Name	Type	Description
lhs	Slot	The symbolic object to write to.
rhs	AssemblyExpression	The original object to read from.

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

@dataclass(eq=True, frozen=True)
class Unpack(AssemblyTree, AssemblyStatement):
    """
    Attempts to convert `rhs` into a symbolic, which can be registerd with
    `lhs`. The original object must not be accessed or modified until the
    corresponding `Repack` node is reached.

    Attributes:
        lhs: The symbolic object to write to.
        rhs: The original object to read from.
    """

    lhs: Slot
    rhs: AssemblyExpression

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.lhs, self.rhs]

children property

Returns the children of the node.

Variable dataclass

Bases: AssemblyExpression, NamedTerm

Represents a logical AST expression for a variable named name, which will hold a value of type type.

Attributes:

Name	Type	Description
name	str	The name of the variable.
type	FType	The type of the variable.

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

@dataclass(eq=True, frozen=True)
class Variable(AssemblyExpression, NamedTerm):
    """
    Represents a logical AST expression for a variable named `name`, which
    will hold a value of type `type`.

    Attributes:
        name: The name of the variable.
        type: The type of the variable.
    """

    name: str
    type: FType

    @property
    def result_type(self) -> FType:
        """Returns the type of the expression."""
        return self.type

    def __repr__(self) -> str:
        return literal_repr(type(self).__name__, {"name": self.name, "type": self.type})

    @property
    def symbol(self) -> str:
        return self.name

result_type property

Returns the type of the expression.

WhileLoop dataclass

Bases: AssemblyTree, AssemblyStatement

Represents a while loop that executes as long as the condition is true.

Attributes:

Name	Type	Description
condition	AssemblyExpression	The condition to evaluate for the loop to continue.
body	AssemblyStatement	The body of the loop to execute.

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

@dataclass(eq=True, frozen=True)
class WhileLoop(AssemblyTree, AssemblyStatement):
    """
    Represents a while loop that executes as long as the condition is true.

    Attributes:
        condition: The condition to evaluate for the loop to continue.
        body: The body of the loop to execute.
    """

    condition: AssemblyExpression
    body: AssemblyStatement

    @property
    def children(self):
        """Returns the children of the node."""
        return [self.condition, self.body]

children property

Returns the children of the node.

assembly_copy_propagation(node)

Apply copy-propagation to a FinchAssembly node. Args: node: Root FinchAssembly node to optimize. Returns: AssemblyNode: The optimized FinchAssembly node.

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

def assembly_copy_propagation(node: AssemblyNode) -> AssemblyNode:
    """
    Apply copy-propagation to a FinchAssembly node.
    Args:
        node: Root FinchAssembly node to optimize.
    Returns:
        AssemblyNode: The optimized FinchAssembly node.
    """

    def apply(pre_node: AssemblyNode, ctx: AssemblyCopyPropagation) -> AssemblyNode:
        lattice: dict[tuple[int, str], str] = ctx.collect_copy_replacements()

        # Replace variables in a statement according to the collected replacements.
        def replace_vars(target: AssemblyNode, sid: int):
            def rw_var(n: AssemblyNode):
                match n:
                    case Variable(name, vtype):
                        key = (sid, name)
                        if key in lattice:
                            return Variable(lattice[key], vtype)

                return None

            return Rewrite(PostWalk(rw_var))(target)

        # Rewrite each numbered statement to replace variables.
        def rw(x: AssemblyNode):
            match x:
                case NumberedStatement(stmt, sid):
                    match stmt:
                        # if Assign, replace vars only on rhs to avoid replacing lhs
                        case Assign(lhs, rhs):
                            rhs = replace_vars(rhs, sid)
                            new_stmt = Assign(lhs, rhs)
                            return NumberedStatement(new_stmt, sid)
                        case _:
                            new_stmt = replace_vars(stmt, sid)
                            return NumberedStatement(new_stmt, sid)

            return None

        return Rewrite(PostWalk(rw))(pre_node)

    return assembly_dataflow_run(node, AssemblyCopyPropagation, apply)

assembly_dataflow_analyze(node, analysis_cls)

Run preprocessing + CFG build + analysis for a dataflow pass.

Returns:

Type	Description
tuple[AnalysisT, AssemblyNode]	(analysis_ctx, preprocessed_node)

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

def assembly_dataflow_analyze(
    node: AssemblyNode, analysis_cls: type[AnalysisT]
) -> tuple[AnalysisT, AssemblyNode]:
    """
    Run preprocessing + CFG build + analysis for a dataflow pass.

    Returns:
        (analysis_ctx, preprocessed_node)
    """
    namespace = Namespace(node)
    pre_node, sid = assembly_dataflow_preprocess(node)
    ctx = analysis_cls(assembly_build_cfg(pre_node, sid=sid, namespace=namespace))
    ctx.analyze()
    return ctx, pre_node

assembly_dataflow_run(node, analysis_cls, apply)

Run a full dataflow pass (preprocess -> analyze -> apply -> postprocess).

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

def assembly_dataflow_run(
    node: AssemblyNode,
    analysis_cls: type[AnalysisT],
    apply: Callable[[AssemblyNode, AnalysisT], AssemblyNode],
) -> AssemblyNode:
    """
    Run a full dataflow pass (preprocess -> analyze -> apply -> postprocess).
    """
    ctx, pre_node = assembly_dataflow_analyze(node, analysis_cls)
    updated = apply(pre_node, ctx)
    return assembly_dataflow_postprocess(updated)

length_type(arg)

The length type of the given argument. The length type is the type of the value returned by len(arg).

Parameters:

Name	Type	Description	Default
arg	FTyped | FType	The object to determine the length type for.	required

Returns:

Type	Description
FDTypeInteger	The length type of the given object.

Raises:

Type	Description
AttributeError	If the length type is not implemented for the given type.

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

def length_type(arg: FTyped | FType) -> FDTypeInteger:
    """The length type of the given argument. The length type is the type of
    the value returned by len(arg).

    Args:
        arg: The object to determine the length type for.

    Returns:
        The length type of the given object.

    Raises:
        AttributeError: If the length type is not implemented for the given type.
    """
    length_type: FDTypeInteger | None = getattr(arg, "length_type", None)
    if length_type is None:
        raise AttributeError(f"{type(arg).__name__} has no length_type")
    return length_type

parse_assembly(code, vars, position_type=np.intp)

Parse Finch Assembly code and convert it to assembly node objects.

Takes a string containing Finch Assembly code and transforms it into a structured representation using assembly nodes. The parser supports assignments, increments, for/while loops, if/if-else statements, buffer and field accesses, and calls.

Parameters:

Name	Type	Description	Default
code	str	The Finch Assembly code to parse. Should start with "finch" or "finch-asm" followed by assembly statements. Comments (C/C++ style) are supported.	required
vars	dict[str, AssemblyExpression]	Dictionary mapping names to Assembly expressions. Used to resolve variable references in the assembly code.	required
position_type	type	NumPy integer type to use for integer literals. Affects the dtype of parsed integer constants. (default: np.intp)	intp

Returns:

Type	Description
Block	A Finch Assembly Block representing the parsed code.

Raises:

Type	Description
Exception	If the parser encounters unrecognized syntax or tree nodes.

Example

from finch.finch_assembly import nodes as asm vars = {"i": asm.Variable("i", int), "arr": asm.Variable("arr", np.ndarray)} code = '''finch ... arr[i] = 42 ... ''' stmt = parse_assembly(code, vars)

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

def parse_assembly(
    code: str, vars: dict[str, asm.AssemblyExpression], position_type: type = np.intp
) -> asm.Block:
    """
    Parse Finch Assembly code and convert it to assembly node objects.

    Takes a string containing Finch Assembly code and transforms it into a structured
    representation using assembly nodes. The parser supports assignments, increments,
    for/while loops, if/if-else statements, buffer and field accesses, and calls.

    Args:
        code: The Finch Assembly code to parse. Should start with "finch" or "finch-asm"
            followed by assembly statements. Comments (C/C++ style) are supported.
        vars: Dictionary mapping names to Assembly expressions.
            Used to resolve variable references in the assembly code.
        position_type: NumPy integer type to use for integer literals. Affects the dtype
            of parsed integer constants. (default: np.intp)

    Returns:
        A Finch Assembly Block representing the parsed code.

    Raises:
        Exception: If the parser encounters unrecognized syntax or tree nodes.

    Example:
        >>> from finch.finch_assembly import nodes as asm
        >>> vars = {"i": asm.Variable("i", int), "arr": asm.Variable("arr", np.ndarray)}
        >>> code = '''finch
        ... arr[i] = 42
        ... '''
        >>> stmt = parse_assembly(code, vars)
    """
    tree = assembly_parser.parse(code.strip())

    def ctx(tree: Tree):
        match tree:
            case Token("CNAME", "true"):
                return asm.Literal(True)
            case Token("CNAME", "false"):
                return asm.Literal(False)
            case Token("CNAME", val):
                return vars[val]
            case Token("OP" | "INFIX_OP", val):
                return _OPS[val]
            case Token("INT", val):
                return asm.Literal(position_type(val))
            case Token("DECIMAL", val):
                return asm.Literal(float(val))
            case Tree("start", [Tree("block", bodies)]):
                return asm.Block(tuple(ctx(b) for b in bodies))
            case Tree("for_loop", [i, start, stop, Tree("block", bodies)]):
                return asm.ForLoop(
                    ctx(i),
                    ctx(start),
                    ctx(stop),
                    asm.Block(tuple(ctx(b) for b in bodies)),
                )
            case Tree("if", [cond, Tree("block", bodies)]):
                return asm.If(ctx(cond), asm.Block(tuple(ctx(b) for b in bodies)))
            case Tree("while_loop", [cond, Tree("block", bodies)]):
                return asm.WhileLoop(
                    ctx(cond), asm.Block(tuple(ctx(b) for b in bodies))
                )
            case Tree(
                "if_else", [cond, Tree("block", bodies), Tree("block", else_bodies)]
            ):
                return asm.IfElse(
                    ctx(cond),
                    asm.Block(tuple(ctx(b) for b in bodies)),
                    asm.Block(tuple(ctx(b) for b in else_bodies)),
                )
            case Tree("resize", [arr, size]):
                return asm.Resize(ctx(arr), ctx(size))
            case Tree("attribute", [obj, Token("CNAME", name)]):
                return asm.GetAttr(ctx(obj), asm.Literal(name))
            case Tree("call", [op, *args]):
                return asm.Call(ctx(op), tuple(ctx(arg) for arg in args))
            case Tree("scansearch", [arr, x, lo, hi]):
                return asm.Call(
                    asm.Literal(ffuncs.scansearch), (ctx(arr), ctx(x), ctx(lo), ctx(hi))
                )
            case Tree("bin_op", [op, expr1, expr2]):
                return asm.Call(asm.Literal(ctx(op)), (ctx(expr1), ctx(expr2)))
            case Tree("assign", [lhs, expr]):
                return assign(ctx(lhs), ctx(expr))
            case Tree("access", [tns, access_expr]):
                return asm.Load(ctx(tns), ctx(access_expr))
            case Tree("increment", [lhs, op, expr]):
                lhs_e = ctx(lhs)
                return assign(lhs_e, asm.Call(asm.Literal(ctx(op)), (lhs_e, ctx(expr))))
            case Tree("expr" | "access_expr", [expr1, op, expr2]):
                return asm.Call(asm.Literal(ctx(op)), (ctx(expr1), ctx(expr2)))
            case other:
                raise Exception(f"{other} not recognized.")

    def assign(lhs, rhs):
        match lhs:
            case asm.Variable():
                return asm.Assign(lhs, rhs)
            case asm.Load(buffer, index):
                return asm.Store(buffer, index, rhs)
            case asm.GetAttr(obj, attr):
                return asm.SetAttr(obj, attr, rhs)
            case _:
                raise TypeError(f"Invalid assembly assignment target: {lhs}")

    return ctx(tree)