"""Dynamic page compiler — generate executable Python scripts from µFrame IR. Takes a parsed IR tree containing dynamic nodes (source, if, for, on_submit, state) and generates a self-contained Python script that: 1. Sets shebang + cache header 2. Reads form data from environment variables 3. Executes source commands 4. Evaluates conditionals and loops 5. Renders the layout into a CharGrid 6. Emits Micron to stdout """ from __future__ import annotations from uframe.ir import ( IRNode, Page, Box, Spacer, Heading, Text, Label, Divider, Link, Gauge, Status, Form, Field, FormButton, Let, Source, IfBlock, ForLoop, CacheControl, OnSubmit, StateDecl, SourceType, ) def _indent(code: str, level: int = 1) -> str: """Indent a block of code.""" prefix = " " * level return "\n".join(prefix + line if line.strip() else "" for line in code.split("\n")) def _resolve_vars(text: str) -> str: """Convert $var references to Python f-string expressions. First escapes literal braces (e.g. @color{ff0}{text} → @color{{ff0}}{{text}}) so they survive f-string evaluation, then replaces $var → {var}. """ import re # Use a unique placeholder for $var refs, escape all braces, then restore _PH = "\x00VAR" counter = [0] placeholders: dict[str, str] = {} def stash_var(m: re.Match) -> str: var = m.group(1) if "." in var: parts = var.split(".") base = parts[0] chain = "".join(f"['{p}']" for p in parts[1:]) expr = "{" + base + chain + "}" else: expr = "{" + var + "}" key = f"{_PH}{counter[0]}\x00" counter[0] += 1 placeholders[key] = expr return key # 1. Stash $var references with placeholders result = re.sub(r'\$([a-zA-Z_][\w.]*)', stash_var, text) # 2. Escape all remaining braces for f-string safety result = result.replace("{", "{{").replace("}", "}}") # 3. Restore $var placeholders (unescaped) for key, expr in placeholders.items(): result = result.replace(key, expr) return result def _resolve_vars_code(text: str) -> str: """Convert $var references to bare Python identifiers. Simple vars: $name → name Dotted paths: $item.name → item['name'] (dict access) """ import re def replace_var(m: re.Match) -> str: var = m.group(1) if "." in var: parts = var.split(".") base = parts[0] chain = "".join(f"['{p}']" for p in parts[1:]) return base + chain return var return re.sub(r'\$([a-zA-Z_][\w.]*)', replace_var, text) def _emit_node(node: IRNode, indent_level: int = 0) -> list[str]: """Generate Python code lines for a single IR node.""" lines: list[str] = [] ind = " " * indent_level if isinstance(node, Let): val = node.var_value # Try to detect numeric values try: float(val) lines.append(f"{ind}{node.var_name} = {val}") except ValueError: if "," in val: # Comma-separated list lines.append(f"{ind}{node.var_name} = [{val}]") else: lines.append(f"{ind}{node.var_name} = {val!r}") elif isinstance(node, Source): var = node.var_name if node.source_type == SourceType.SHELL: lines.append(f"{ind}{var} = _shell({node.command!r}, timeout={node.timeout})") elif node.source_type == SourceType.FILE: lines.append(f"{ind}{var} = _read_file({node.command!r})") elif node.source_type == SourceType.JSON: lines.append(f"{ind}{var} = _read_json({node.command!r})") elif node.source_type == SourceType.PYTHON: lines.append(f"{ind}{var} = eval({node.command!r}, {{'datetime': _dt_cls, 'timedelta': timedelta, 'secrets': secrets, 'os': os, 'json': json}})") elif node.source_type == SourceType.PARAM: lines.append(f"{ind}{var} = _get_param({node.command!r})") elif node.source_type == SourceType.RNS: import shlex as _shlex safe_cmd = _shlex.quote(node.command) lines.append(f"{ind}{var} = _shell('rnstatus ' + shlex.quote({safe_cmd!r}), timeout={node.timeout})") elif node.source_type == SourceType.HTTP: method = node.http_method or "GET" url = _resolve_vars(node.command) hdrs = _resolve_vars(node.http_headers) if node.http_headers else "" body = _resolve_vars(node.http_body) if node.http_body else "" if body: lines.append(f"""{ind}{var} = _http(f'''{url}''', method={method!r}, body=f'''{body}''', headers=f'''{hdrs}''', timeout={node.timeout})""") elif hdrs: lines.append(f"""{ind}{var} = _http(f'''{url}''', method={method!r}, headers=f'''{hdrs}''', timeout={node.timeout})""") else: lines.append(f"{ind}{var} = _http({node.command!r}, method={method!r}, timeout={node.timeout})") elif node.source_type == SourceType.SQLITE: lines.append(f"{ind}{var} = _sqlite({node.command!r}, {node.query!r})") elif node.source_type == SourceType.ENV: lines.append(f"{ind}{var} = os.environ.get({node.command!r}, '')") elif isinstance(node, CacheControl): lines.append(f"{ind}_cache_seconds = {node.seconds}") elif isinstance(node, StateDecl): lines.append(f"{ind}{node.state_name} = _load_state({node.path!r})") elif isinstance(node, IfBlock): cond = _resolve_vars_code(node.condition) cond = cond.replace("&&", " and ").replace("||", " or ") lines.append(f"{ind}if {cond}:") if node.children: for child in node.children: lines.extend(_emit_node(child, indent_level + 1)) else: lines.append(f"{ind} pass") for elif_cond, elif_children in node.elif_branches: ec = _resolve_vars_code(elif_cond).replace("&&", " and ").replace("||", " or ") lines.append(f"{ind}elif {ec}:") if elif_children: for child in elif_children: lines.extend(_emit_node(child, indent_level + 1)) else: lines.append(f"{ind} pass") if node.else_children: lines.append(f"{ind}else:") for child in node.else_children: lines.extend(_emit_node(child, indent_level + 1)) elif isinstance(node, ForLoop): iterable = _resolve_vars_code(node.iterable) lines.append(f"{ind}for {node.var_name} in _iter({iterable}):") if node.children: for child in node.children: lines.extend(_emit_node(child, indent_level + 1)) else: lines.append(f"{ind} pass") elif isinstance(node, OnSubmit): lines.append(f"{ind}if _get_field({node.form_name!r}, ''):") lines.append(f"{ind} # Form '{node.form_name}' was submitted") for child in node.children: lines.extend(_emit_node(child, indent_level + 1)) elif isinstance(node, Page): lines.append(f"{ind}_page_title = {node.title!r}") lines.append(f"{ind}_page_width = {node.width}") lines.append(f"{ind}_uf_source_parts = []") for child in node.children: lines.extend(_emit_node(child, indent_level)) # Content nodes — emit as µFrame source that gets compiled elif isinstance(node, Heading): level = node.level.value text = _resolve_vars(node.text) lines.append(f"{ind}_uf_source_parts.append(f'heading {level} \"{text}\"')") elif isinstance(node, Text): text = _resolve_vars(node.content) lines.append(f"{ind}_uf_source_parts.append(f'text \"{text}\"')") elif isinstance(node, Label): key = _resolve_vars(node.key) val = _resolve_vars(node.value) lines.append(f"{ind}_uf_source_parts.append(f'label \"{key}\" \"{val}\"')") elif isinstance(node, Gauge): label = _resolve_vars(node.label) raw_val = getattr(node, "_raw_value", None) value = _resolve_vars(raw_val) if raw_val and "$" in raw_val else str(node.value) raw_max = getattr(node, "_raw_max_val", None) max_val = _resolve_vars(raw_max) if raw_max and "$" in raw_max else str(node.max_val) extra = "" if node.warn is not None: extra += f" warn={node.warn}" if node.crit is not None: extra += f" crit={node.crit}" lines.append(f"{ind}_uf_source_parts.append(f'gauge \"{label}\" {value} {max_val} {node.bar_width}{extra}')") elif isinstance(node, Status): label = _resolve_vars(node.label) state = _resolve_vars(node.state) lines.append(f"{ind}_uf_source_parts.append(f'status \"{label}\" {state}')") elif isinstance(node, Box): weight = node.weight.name.lower() title = _resolve_vars(node.title) lines.append(f"{ind}_uf_source_parts.append(f'box {weight} \"{title}\"')") for child in node.children: # Indent children for the box child_lines = _emit_node(child, indent_level) for cl in child_lines: if "_uf_source_parts.append" in cl: # Add 2-space indent to the µFrame source cl = cl.replace(".append(f'", ".append(f' ") cl = cl.replace(".append('", ".append(' ") lines.append(cl) elif isinstance(node, Divider): style = node.divider_style.name.lower() lines.append(f"{ind}_uf_source_parts.append('divider {style}')") elif isinstance(node, Spacer): lines.append(f"{ind}_uf_source_parts.append('spacer {node.lines}')") elif isinstance(node, Link): display = _resolve_vars(node.display) dest = _resolve_vars(node.dest) lines.append(f"{ind}_uf_source_parts.append(f'link \"{display}\" \"{dest}\"')") elif isinstance(node, Field): lines.append(f"{ind}_uf_source_parts.append('field \"{node.field_name}\" {node.field_width} \"{node.placeholder}\"')") elif isinstance(node, FormButton): lines.append(f"{ind}_uf_source_parts.append('button \"{node.button_label}\" \"{node.dest}\"')") elif isinstance(node, Form): lines.append(f"{ind}_uf_source_parts.append('form \"{node.form_name}\"')") for child in node.children: child_lines = _emit_node(child, indent_level) for cl in child_lines: if "_uf_source_parts.append" in cl: cl = cl.replace(".append(f'", ".append(f' ") cl = cl.replace(".append('", ".append(' ") lines.append(cl) else: # Generic: emit children for child in node.children: lines.extend(_emit_node(child, indent_level)) return lines # --------------------------------------------------------------------------- # Runtime template embedded in generated scripts # --------------------------------------------------------------------------- def _build_script(uframe_import: str, page_logic: str, page_title: str, page_width: int) -> str: """Build the dynamic script from parts (avoids str.format brace issues).""" lines = [ "#!/usr/bin/env python3", "# Auto-generated by uFrame", "# Do not edit — regenerate with: uframe compile .uf", "", "import os, sys, json, subprocess, datetime, secrets, shlex", "from datetime import datetime as _dt_cls, timedelta", "", "# ─── Runtime Helpers ─────────────────────────────────────────", "", 'def _shell(cmd, timeout=5):', ' """Execute shell command, return stdout."""', ' try:', ' return subprocess.check_output(cmd, shell=True, timeout=timeout).decode().strip()', ' except Exception:', ' return ""', "", 'def _read_file(path):', ' """Read file contents."""', ' try:', ' return open(path).read().strip()', ' except Exception:', ' return ""', "", 'def _read_json(path):', ' """Read and parse JSON file."""', ' try:', ' with open(path) as f:', ' return json.load(f)', ' except Exception:', ' return {}', "", 'def _get_field(name, default=""):', ' """Read submitted form field from environment."""', ' return os.environ.get(f"FIELD_{name}", default)', "", 'def _get_param(name, default=""):', ' """Read URL parameter."""', ' return os.environ.get(f"PARAM_{name}",', ' os.environ.get(f"var_{name}", default))', "", 'def _load_state(path):', ' """Load state from JSON file."""', ' try:', ' with open(path) as f:', ' return json.load(f)', ' except Exception:', ' return {}', "", 'def _save_state(path, data):', ' """Save state to JSON file."""', ' os.makedirs(os.path.dirname(path), exist_ok=True)', ' with open(path, "w") as f:', ' json.dump(data, f, indent=2)', "", 'def _iter(val):', ' """Make a value iterable for for-loops."""', ' if isinstance(val, (list, tuple)):', ' return val', ' if isinstance(val, dict):', ' return [val]', ' if isinstance(val, str):', ' return val.strip().splitlines()', ' return []', "", 'def _http(url, method="GET", body="", headers="", timeout=10):', ' """HTTP request, return response body (JSON parsed if possible)."""', ' import urllib.request, urllib.error', ' try:', ' data = body.encode("utf-8") if body else None', ' req = urllib.request.Request(url, data=data, method=method)', ' req.add_header("User-Agent", "uframe/1.0")', ' if body and not headers:', ' req.add_header("Content-Type", "application/json")', ' if headers:', ' for pair in headers.split(";"):', ' if ":" in pair:', ' k, v = pair.split(":", 1)', ' req.add_header(k.strip(), v.strip())', ' with urllib.request.urlopen(req, timeout=timeout) as resp:', ' raw = resp.read().decode("utf-8")', ' try:', ' return json.loads(raw)', ' except (json.JSONDecodeError, ValueError):', ' return raw.strip()', ' except Exception as e:', ' return {"error": str(e)}', "", 'def _sqlite(db_path, query):', ' """Run a SQLite query, return list of dicts."""', ' import sqlite3', ' try:', ' conn = sqlite3.connect(db_path)', ' conn.row_factory = sqlite3.Row', ' cur = conn.execute(query)', ' rows = [dict(r) for r in cur.fetchall()]', ' conn.close()', ' return rows if len(rows) != 1 else rows[0]', ' except Exception as e:', ' return {"error": str(e)}', "", f"# ─── µFrame Compile ──────────────────────────────────────────", "", uframe_import, "", "# ─── Page Logic ──────────────────────────────────────────────", "", "_cache_seconds = 0", "", page_logic, "", "# ─── Render & Output ─────────────────────────────────────────", "", f'_uf_source = f\'\'\'page "{page_title}" {page_width}', "''' + \"\\n\".join(_uf_source_parts)", "", f"result = uframe.compile(_uf_source, width={page_width})", "", "if _cache_seconds >= 0:", ' print(f"#!c={_cache_seconds}")', "print(result.micron)", ] return "\n".join(lines) def compile_dynamic(page: Page) -> str: """Generate a self-contained executable Python script from an IR tree. The generated script imports uframe at runtime and compiles the dynamically-built .uf source into Micron output. """ logic_lines = _emit_node(page, indent_level=0) page_logic = "\n".join(logic_lines) uframe_import = "import uframe" return _build_script( uframe_import=uframe_import, page_logic=page_logic, page_title=page.title, page_width=page.width, )