Aaron Sinclair
e18baf3106
- Handle 'assign' type statements in _emit_handler_statement() - Evaluate value expression and store to signal global - Enables patterns like: setting_tab = 0 - All tests passing (30/30)
461 lines
19 KiB
Python
461 lines
19 KiB
Python
"""LLVM IR code generator with Signal support for Ritz components."""
|
|
|
|
from typing import Any
|
|
|
|
|
|
def generate_llvm_ir(ir: dict[str, Any] | list[dict[str, Any]]) -> str:
|
|
"""
|
|
Generate LLVM IR from Ritz IR with Signal support.
|
|
|
|
Args:
|
|
ir: Component IR dictionary or list of component dictionaries
|
|
|
|
Returns:
|
|
LLVM IR source code with signals stored in WASM memory
|
|
"""
|
|
generator = SignalLLVMIRGenerator(ir)
|
|
return generator.generate()
|
|
|
|
|
|
class SignalLLVMIRGenerator:
|
|
"""Generates LLVM IR with reactive Signal support."""
|
|
|
|
def __init__(self, ir: dict[str, Any] | list[dict[str, Any]]):
|
|
# Normalize to list for uniform processing
|
|
self.components = ir if isinstance(ir, list) else [ir]
|
|
self.lines: list[str] = []
|
|
self.strings: dict[str, str] = {} # value -> global name
|
|
self.signals: dict[str, dict[str, Any]] = {} # signal_name -> {initial, index}
|
|
self.event_handlers: list[dict[str, Any]] = [] # List of event handler functions
|
|
self.next_string_id = 0
|
|
self.next_var_id = 0
|
|
self.next_signal_id = 0
|
|
|
|
# Track signal-to-DOM bindings for updates
|
|
self.signal_dom_bindings: dict[str, list[str]] = {} # signal_name -> [dom_node_ids]
|
|
|
|
# Track current component for signal scoping
|
|
self.current_component = None
|
|
|
|
def _signal_global(self, signal_name: str) -> str:
|
|
"""Get the scoped global name for a signal."""
|
|
if self.current_component:
|
|
return f"@signal_{self.current_component['name']}_{signal_name}"
|
|
return f"@signal_{signal_name}"
|
|
|
|
def generate(self) -> str:
|
|
"""Generate complete LLVM IR module with signals."""
|
|
# Emit shared header and function declarations once
|
|
self._emit_header()
|
|
self._emit_function_declarations()
|
|
|
|
# Process each component
|
|
for component in self.components:
|
|
# Set current component for signal scoping
|
|
self.current_component = component
|
|
|
|
# Reset component-specific state
|
|
self.signals = {}
|
|
self.event_handlers = []
|
|
|
|
# Collect signals and handlers for this component
|
|
self._collect_signals_and_handlers(component)
|
|
|
|
# Generate component-specific code
|
|
self._emit_signal_globals(component)
|
|
self._emit_helper_functions()
|
|
self._emit_event_handlers()
|
|
self._emit_component(component)
|
|
|
|
# Prepend string constants
|
|
return self._finalize_with_strings()
|
|
|
|
def _collect_signals_and_handlers(self, component: dict[str, Any]) -> None:
|
|
"""Collect all signals and event handlers from component IR."""
|
|
for stmt in component["body"]:
|
|
if stmt["type"] == "signal":
|
|
signal_id = self.next_signal_id
|
|
self.next_signal_id += 1
|
|
self.signals[stmt["name"]] = {
|
|
"initial": stmt["initial"],
|
|
"id": signal_id,
|
|
}
|
|
elif stmt["type"] == "function":
|
|
self.event_handlers.append(stmt)
|
|
|
|
def _emit_header(self) -> None:
|
|
"""Emit module header."""
|
|
self.lines.append("; Generated by Ritz Framework with Signals")
|
|
component_names = [c["name"] for c in self.components]
|
|
self.lines.append(f'; Components: {", ".join(component_names)}')
|
|
self.lines.append("")
|
|
|
|
def _emit_signal_globals(self, component: dict[str, Any]) -> None:
|
|
"""Emit global variables for signals in this component."""
|
|
if not self.signals:
|
|
return
|
|
|
|
component_name = component["name"]
|
|
self.lines.append(f"; Signal storage for {component_name} (reactive state in WASM memory)")
|
|
for signal_name, signal_info in self.signals.items():
|
|
# Each signal is a global i32, scoped to component to avoid collisions
|
|
initial_value = signal_info["initial"].get("value", 0)
|
|
self.lines.append(f"@signal_{component_name}_{signal_name} = global i32 {initial_value}")
|
|
self.lines.append("")
|
|
|
|
def _emit_function_declarations(self) -> None:
|
|
"""Emit JS bridge function declarations."""
|
|
self.lines.append("; JavaScript bridge function declarations")
|
|
self.lines.append("declare i32 @js_create_element(i8*)")
|
|
self.lines.append("declare i32 @js_create_text(i8*)")
|
|
self.lines.append("declare void @js_append_child(i32, i32)")
|
|
self.lines.append("declare void @js_set_attribute(i32, i8*, i8*)")
|
|
self.lines.append("declare void @js_set_root(i32)")
|
|
self.lines.append(
|
|
"declare void @js_bind_event(i32, i8*, i8*)"
|
|
) # elem, event_name, handler_name
|
|
self.lines.append("declare void @js_update_text_content(i32, i8*)") # Update DOM node text
|
|
self.lines.append("")
|
|
|
|
def _emit_helper_functions(self) -> None:
|
|
"""Emit helper functions for signal operations."""
|
|
if not self.signals:
|
|
return
|
|
|
|
self.lines.append("; Signal helper functions")
|
|
|
|
# Emit getter for each signal (exported for JS to call)
|
|
for signal_name in self.signals:
|
|
self.lines.append(f"define i32 @get_{signal_name}() {{")
|
|
self.lines.append("entry:")
|
|
self.lines.append(f" %val = load i32, i32* {self._signal_global(signal_name)}")
|
|
self.lines.append(" ret i32 %val")
|
|
self.lines.append("}")
|
|
self.lines.append("")
|
|
|
|
# Emit setter for each signal (exported for JS to call)
|
|
for signal_name in self.signals:
|
|
self.lines.append(f"define void @set_{signal_name}(i32 %newval) {{")
|
|
self.lines.append("entry:")
|
|
self.lines.append(f" store i32 %newval, i32* {self._signal_global(signal_name)}")
|
|
self.lines.append(" ; TODO: Trigger DOM update callback")
|
|
self.lines.append(" ret void")
|
|
self.lines.append("}")
|
|
self.lines.append("")
|
|
|
|
def _emit_event_handlers(self) -> None:
|
|
"""Emit event handler functions (exported for JS to call)."""
|
|
for handler in self.event_handlers:
|
|
func_name = handler["name"]
|
|
self.lines.append(f"; Event handler: {func_name}")
|
|
self.lines.append(f"define void @{func_name}() {{")
|
|
self.lines.append("entry:")
|
|
|
|
# Emit handler body
|
|
for stmt in handler["body"]:
|
|
self._emit_handler_statement(stmt)
|
|
|
|
self.lines.append(" ret void")
|
|
self.lines.append("}")
|
|
self.lines.append("")
|
|
|
|
def _emit_handler_statement(self, stmt: dict[str, Any]) -> None:
|
|
"""Emit a statement inside an event handler."""
|
|
if stmt["type"] == "assign":
|
|
# Handle direct signal assignment (x = 5)
|
|
signal_name = stmt["target"]
|
|
if signal_name in self.signals:
|
|
# Evaluate the value expression
|
|
value_var = self._emit_expr(stmt["value"])
|
|
|
|
# Store the value
|
|
self.lines.append(f" store i32 {value_var}, i32* {self._signal_global(signal_name)}")
|
|
|
|
# TODO: Trigger DOM updates
|
|
self.lines.append(f" ; TODO: Update DOM nodes bound to {signal_name}")
|
|
elif stmt["type"] == "augassign":
|
|
# Handle signal updates (count += 1)
|
|
signal_name = stmt["target"]
|
|
if signal_name in self.signals:
|
|
# Load current value
|
|
load_var = self._new_var()
|
|
self.lines.append(f" %{load_var} = load i32, i32* {self._signal_global(signal_name)}")
|
|
|
|
# Perform operation
|
|
result_var = self._new_var()
|
|
if stmt["op"] == "Add":
|
|
value = stmt["value"]["value"]
|
|
self.lines.append(f" %{result_var} = add i32 %{load_var}, {value}")
|
|
|
|
# Store back
|
|
self.lines.append(f" store i32 %{result_var}, i32* {self._signal_global(signal_name)}")
|
|
|
|
# TODO: Trigger DOM updates for this signal
|
|
self.lines.append(f" ; TODO: Update DOM nodes bound to {signal_name}")
|
|
|
|
def _emit_component(self, ir: dict[str, Any]) -> None:
|
|
"""Emit component render function."""
|
|
component_name = ir["name"]
|
|
self.lines.append("; Component render function")
|
|
self.lines.append(f"define void @render_{component_name}() {{")
|
|
self.lines.append("entry:")
|
|
|
|
# Process component body (skip signals and functions, already handled)
|
|
root_var = None
|
|
for stmt in ir["body"]:
|
|
if stmt["type"] in ["signal", "function"]:
|
|
continue # Already processed
|
|
var = self._emit_statement(stmt)
|
|
if root_var is None and var:
|
|
root_var = var
|
|
|
|
# Set root element
|
|
if root_var:
|
|
self.lines.append(f" call void @js_set_root(i32 %{root_var})")
|
|
|
|
self.lines.append(" ret void")
|
|
self.lines.append("}")
|
|
self.lines.append("")
|
|
|
|
def _emit_statement(self, stmt: dict[str, Any]) -> str | None:
|
|
"""Emit a statement, return variable name if it creates an element."""
|
|
if stmt["type"] == "with":
|
|
return self._emit_with(stmt)
|
|
elif stmt["type"] == "yield":
|
|
return self._emit_yield(stmt)
|
|
elif stmt["type"] == "if":
|
|
return self._emit_if(stmt)
|
|
return None
|
|
|
|
def _emit_if(self, stmt: dict[str, Any]) -> str | None:
|
|
"""Emit an if statement."""
|
|
# Generate unique labels for this if statement
|
|
if_id = self._new_var()
|
|
then_label = f"if.then.{if_id}"
|
|
else_label = f"if.else.{if_id}"
|
|
end_label = f"if.end.{if_id}"
|
|
|
|
# Evaluate condition
|
|
test_var = self._emit_expr(stmt["test"])
|
|
|
|
# Convert to i1 (boolean)
|
|
cond_var = self._new_var()
|
|
self.lines.append(f" %{cond_var} = icmp ne i32 %{test_var}, 0")
|
|
|
|
# Branch based on whether there's an else
|
|
has_else = stmt.get("orelse") and len(stmt["orelse"]) > 0
|
|
if has_else:
|
|
self.lines.append(f" br i1 %{cond_var}, label %{then_label}, label %{else_label}")
|
|
else:
|
|
self.lines.append(f" br i1 %{cond_var}, label %{then_label}, label %{end_label}")
|
|
|
|
# Emit then block
|
|
self.lines.append(f"{then_label}:")
|
|
for then_stmt in stmt["body"]:
|
|
self._emit_statement(then_stmt)
|
|
self.lines.append(f" br label %{end_label}")
|
|
|
|
# Emit else block if present
|
|
if has_else:
|
|
self.lines.append(f"{else_label}:")
|
|
for else_stmt in stmt["orelse"]:
|
|
self._emit_statement(else_stmt)
|
|
self.lines.append(f" br label %{end_label}")
|
|
|
|
# Emit merge block
|
|
self.lines.append(f"{end_label}:")
|
|
|
|
return None
|
|
|
|
def _emit_expr(self, expr: dict[str, Any]) -> str:
|
|
"""Emit an expression and return the result variable name."""
|
|
if expr["type"] == "name":
|
|
# Load signal value
|
|
signal_name = expr["id"]
|
|
if signal_name in self.signals:
|
|
load_var = self._new_var()
|
|
self.lines.append(f" %{load_var} = load i32, i32* {self._signal_global(signal_name)}")
|
|
return load_var
|
|
else:
|
|
# Unknown variable
|
|
return "0"
|
|
elif expr["type"] == "constant":
|
|
# Return constant inline
|
|
return str(expr["value"])
|
|
elif expr["type"] == "compare":
|
|
# Handle comparison (e.g., count > 5)
|
|
left_var = self._emit_expr(expr["left"])
|
|
right_var = self._emit_expr(expr["comparators"][0])
|
|
|
|
# Map Python comparison ops to LLVM icmp predicates
|
|
op_map = {
|
|
"Gt": "sgt", # signed greater than
|
|
"Lt": "slt", # signed less than
|
|
"GtE": "sge", # signed greater or equal
|
|
"LtE": "sle", # signed less or equal
|
|
"Eq": "eq", # equal
|
|
"NotEq": "ne", # not equal
|
|
}
|
|
|
|
icmp_op = op_map.get(expr["op"], "eq")
|
|
result_var = self._new_var()
|
|
self.lines.append(f" %{result_var} = icmp {icmp_op} i32 %{left_var}, {right_var}")
|
|
|
|
# Convert i1 to i32 for compatibility
|
|
result_i32 = self._new_var()
|
|
self.lines.append(f" %{result_i32} = zext i1 %{result_var} to i32")
|
|
return result_i32
|
|
else:
|
|
# Unknown expression type
|
|
return "0"
|
|
|
|
def _emit_with(self, stmt: dict[str, Any]) -> str:
|
|
"""Emit a 'with Element()' block."""
|
|
element_info = stmt["element"]
|
|
tag = element_info["tag"].lower()
|
|
props = element_info["props"]
|
|
|
|
# Create element
|
|
elem_var = self._new_var()
|
|
tag_str = self._get_or_create_string(tag)
|
|
self.lines.append(
|
|
f" %{elem_var} = call i32 @js_create_element(i8* getelementptr inbounds ({self._string_type(tag)}, {self._string_type(tag)}* {tag_str}, i32 0, i32 0))"
|
|
)
|
|
|
|
# Handle onclick prop specially
|
|
onclick_handler = None
|
|
for prop_name, prop_value in props.items():
|
|
if prop_name == "onclick":
|
|
# Bind event handler
|
|
if prop_value["type"] == "name":
|
|
onclick_handler = prop_value["id"]
|
|
else:
|
|
# Regular attribute
|
|
attr_name = prop_name.rstrip("_")
|
|
if prop_value["type"] == "constant":
|
|
value = str(prop_value["value"])
|
|
name_str = self._get_or_create_string(attr_name)
|
|
value_str = self._get_or_create_string(value)
|
|
self.lines.append(
|
|
f" call void @js_set_attribute(i32 %{elem_var}, "
|
|
f"i8* getelementptr inbounds ({self._string_type(attr_name)}, {self._string_type(attr_name)}* {name_str}, i32 0, i32 0), "
|
|
f"i8* getelementptr inbounds ({self._string_type(value)}, {self._string_type(value)}* {value_str}, i32 0, i32 0))"
|
|
)
|
|
|
|
# Bind onclick if present
|
|
if onclick_handler:
|
|
event_str = self._get_or_create_string("click")
|
|
handler_str = self._get_or_create_string(onclick_handler)
|
|
self.lines.append(
|
|
f" call void @js_bind_event(i32 %{elem_var}, "
|
|
f"i8* getelementptr inbounds ({self._string_type('click')}, {self._string_type('click')}* {event_str}, i32 0, i32 0), "
|
|
f"i8* getelementptr inbounds ({self._string_type(onclick_handler)}, {self._string_type(onclick_handler)}* {handler_str}, i32 0, i32 0))"
|
|
)
|
|
|
|
# Process children
|
|
for child_stmt in stmt["body"]:
|
|
child_var = self._emit_statement(child_stmt)
|
|
if child_var:
|
|
self.lines.append(
|
|
f" call void @js_append_child(i32 %{elem_var}, i32 %{child_var})"
|
|
)
|
|
|
|
return elem_var
|
|
|
|
def _emit_yield(self, stmt: dict[str, Any]) -> str | None:
|
|
"""Emit a yield statement."""
|
|
value = stmt.get("value")
|
|
if not value:
|
|
return None
|
|
|
|
if value["type"] == "constant":
|
|
# Simple text node
|
|
text = str(value["value"])
|
|
text_str = self._get_or_create_string(text)
|
|
text_var = self._new_var()
|
|
self.lines.append(
|
|
f" %{text_var} = call i32 @js_create_text(i8* getelementptr inbounds ({self._string_type(text)}, {self._string_type(text)}* {text_str}, i32 0, i32 0))"
|
|
)
|
|
return text_var
|
|
|
|
elif value["type"] == "fstring":
|
|
# Handle f-strings with signal references
|
|
# For now, create text with signal value
|
|
# TODO: Implement proper string concatenation
|
|
parts = value["parts"]
|
|
|
|
# Find signal references
|
|
for part in parts:
|
|
if part["type"] == "formatted_value":
|
|
if part["value"]["type"] == "name":
|
|
signal_name = part["value"]["id"]
|
|
if signal_name in self.signals:
|
|
# Load signal value and create text node
|
|
# This is simplified - real impl would concat strings
|
|
text_var = self._new_var()
|
|
self.lines.append(
|
|
f" ; TODO: Format f-string with signal {signal_name}"
|
|
)
|
|
# For now, just create placeholder text
|
|
placeholder = self._get_or_create_string("0")
|
|
self.lines.append(
|
|
f" %{text_var} = call i32 @js_create_text(i8* getelementptr inbounds ([2 x i8], [2 x i8]* {placeholder}, i32 0, i32 0))"
|
|
)
|
|
return text_var
|
|
|
|
# Fallback: create empty text
|
|
empty_str = self._get_or_create_string("")
|
|
text_var = self._new_var()
|
|
self.lines.append(
|
|
f" %{text_var} = call i32 @js_create_text(i8* getelementptr inbounds ({self._string_type('')}, {self._string_type('')}* {empty_str}, i32 0, i32 0))"
|
|
)
|
|
return text_var
|
|
|
|
return None
|
|
|
|
def _get_or_create_string(self, value: str) -> str:
|
|
"""Get or create a string constant."""
|
|
if value in self.strings:
|
|
return self.strings[value]
|
|
global_name = f"@.str.{self.next_string_id}"
|
|
self.next_string_id += 1
|
|
self.strings[value] = global_name
|
|
return global_name
|
|
|
|
def _string_type(self, value: str) -> str:
|
|
"""Get LLVM type for a string (account for UTF-8 byte length)."""
|
|
# Get actual byte length of UTF-8 encoded string
|
|
byte_length = len(value.encode("utf-8")) + 1 # +1 for null terminator
|
|
return f"[{byte_length} x i8]"
|
|
|
|
def _new_var(self) -> str:
|
|
"""Generate a new SSA variable name."""
|
|
var_name = f"var{self.next_var_id}"
|
|
self.next_var_id += 1
|
|
return var_name
|
|
|
|
def _finalize_with_strings(self) -> str:
|
|
"""Prepend string constants and return final IR."""
|
|
string_lines = ["; String constants"]
|
|
for value, global_name in self.strings.items():
|
|
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
|
|
str_type = self._string_type(value)
|
|
string_lines.append(
|
|
f'{global_name} = private unnamed_addr constant {str_type} c"{escaped}\\00"'
|
|
)
|
|
string_lines.append("")
|
|
|
|
# Find insertion point (after header, before declarations)
|
|
result_lines = []
|
|
header_end = 0
|
|
for i, line in enumerate(self.lines):
|
|
if "JavaScript bridge" in line or "Signal storage" in line:
|
|
header_end = i
|
|
break
|
|
|
|
result_lines.extend(self.lines[:header_end])
|
|
result_lines.extend(string_lines)
|
|
result_lines.extend(self.lines[header_end:])
|
|
|
|
return "\n".join(result_lines)
|