Upload 127 files

a1c8a3d verified 8 months ago

6.88 kB

"""
TD Lang Executor — Runs compiled .td scripts and tracks lineage.
Two modes:
    - compile: Parse .td -> generate .py file (no execution)
    - run: Parse .td -> generate .py -> execute it
All outputs go to td_lang_outputs/<script_name>_<timestamp>/
    - compiled.py   — The generated Python script
    - lineage.json  — What happened, in what order (artifact tracking)
Pipeline: .td file -> Parser -> AST -> Compiler -> Python string -> **Executor**
"""

import ast as python_ast
import hashlib
import json
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional

from .grammar import parse_td_file, parse_td_string
from .compiler import compile_program
from .ast_nodes import TDProgram
from .errors import TDCompileError, TDLangError


# ============================================================================
# EXECUTOR
# ============================================================================

class TDExecutor:
    """Execute td_lang programs — compile and optionally run.
    Usage:
        executor = TDExecutor()
        # Compile only (check + generate .py)
        py_path = executor.compile("demo.td")
        # Compile and run
        result = executor.run("demo.td")
        # Just check syntax
        executor.check("demo.td")
    """

    def __init__(self, output_dir: str = "td_lang_outputs"):
        self.output_dir = Path(output_dir)

    def check(self, td_path: str) -> TDProgram:
        """Parse and validate a .td file without compiling or running.
        Args:
            td_path: Path to the .td file.
        Returns:
            The parsed TDProgram.
        Raises:
            TDSyntaxError: If syntax is invalid.
            TDCompileError: If semantic validation fails.
        """
        print(f"[td_lang] Checking {td_path}...")
        program = parse_td_file(td_path)

        # Count what we found
        n_commands = len(program.commands)
        has_gates = program.gates is not None
        has_budget = program.budget is not None

        print(f"[td_lang] OK — {n_commands} commands", end="")
        if has_gates:
            print(f", gates: {program.gates.must_pass}", end="")
        if has_budget:
            print(f", budget set", end="")
        print()

        return program

    def compile(self, td_path: str) -> Path:
        """Parse, validate, and compile a .td file into Python.
        Args:
            td_path: Path to the .td file.
        Returns:
            Path to the generated .py file.
        Raises:
            TDSyntaxError: If syntax is invalid.
            TDCompileError: If compilation fails.
        """
        print(f"[td_lang] Compiling {td_path}...")

        # Parse
        program = parse_td_file(td_path)

        # Compile
        python_code = compile_program(program)

        # Validate the generated Python is valid syntax
        try:
            python_ast.parse(python_code)
        except SyntaxError as e:
            raise TDCompileError(
                f"Generated Python has a syntax error (this is a td_lang bug): {e}",
                hint="Please report this — the compiler generated bad code.",
            ) from e

        # Save to output directory
        out_dir = self._make_output_dir(td_path)
        py_path = out_dir / "compiled.py"
        py_path.write_text(python_code)

        # Save source hash for lineage
        source_text = Path(td_path).read_text()
        meta = {
            "source_file": str(td_path),
            "source_hash": hashlib.sha256(source_text.encode()).hexdigest(),
            "compiled_at": datetime.now().isoformat(),
            "td_lang_version": "0.2.0",
            "python_file": str(py_path),
            "n_commands": len(program.commands),
            "has_gates": program.gates is not None,
            "has_budget": program.budget is not None,
        }
        meta_path = out_dir / "compile_meta.json"
        meta_path.write_text(json.dumps(meta, indent=2))

        print(f"[td_lang] Compiled to: {py_path}")
        return py_path

    def run(self, td_path: str, dry_run: bool = False) -> dict:
        """Parse, compile, and execute a .td file.
        Args:
            td_path: Path to the .td file.
            dry_run: If True, compile but don't execute.
        Returns:
            Dict with execution results.
        """
        # Compile first
        py_path = self.compile(td_path)

        if dry_run:
            print("[td_lang] Dry run — compiled but not executed.")
            return {"status": "dry_run", "compiled": str(py_path)}

        # Execute the generated Python script
        print(f"[td_lang] Executing {py_path}...")
        print()

        try:
            result = subprocess.run(
                [sys.executable, str(py_path.resolve())],
                capture_output=False,       # Let output stream to console
                # Run from the original working directory, NOT the output dir.
                # The compiled code uses relative paths like "td_lang_outputs/..."
                # which must resolve from the project root.
            )

            if result.returncode == 0:
                print()
                print("[td_lang] Execution completed successfully.")
                return {"status": "success", "compiled": str(py_path)}
            else:
                print()
                print(f"[td_lang] Execution failed (exit code {result.returncode}).")
                return {
                    "status": "failed",
                    "compiled": str(py_path),
                    "exit_code": result.returncode,
                }

        except Exception as e:
            print(f"\n[td_lang] Execution error: {e}")
            return {"status": "error", "compiled": str(py_path), "error": str(e)}

    def _make_output_dir(self, td_path: str) -> Path:
        """Create a timestamped output directory for this run."""
        name = Path(td_path).stem
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        out_dir = self.output_dir / f"{name}_{timestamp}"
        out_dir.mkdir(parents=True, exist_ok=True)
        return out_dir


# ============================================================================
# PUBLIC API
# ============================================================================

def check_td_file(td_path: str) -> TDProgram:
    """Quick syntax check on a .td file."""
    return TDExecutor().check(td_path)


def compile_td_file(td_path: str, output_dir: str = "td_lang_outputs") -> Path:
    """Compile a .td file to Python."""
    return TDExecutor(output_dir=output_dir).compile(td_path)


def run_td_file(td_path: str, output_dir: str = "td_lang_outputs", dry_run: bool = False) -> dict:
    """Compile and run a .td file."""
    return TDExecutor(output_dir=output_dir).run(td_path, dry_run=dry_run)

	"""
	TD Lang Executor — Runs compiled .td scripts and tracks lineage.

	Two modes:
	- compile: Parse .td -> generate .py file (no execution)
	- run: Parse .td -> generate .py -> execute it

	All outputs go to td_lang_outputs/<script_name>_<timestamp>/
	- compiled.py — The generated Python script
	- lineage.json — What happened, in what order (artifact tracking)

	Pipeline: .td file -> Parser -> AST -> Compiler -> Python string -> Executor
	"""

	import ast as python_ast
	import hashlib
	import json
	import subprocess
	import sys
	from datetime import datetime
	from pathlib import Path
	from typing import Optional

	from .grammar import parse_td_file, parse_td_string
	from .compiler import compile_program
	from .ast_nodes import TDProgram
	from .errors import TDCompileError, TDLangError


	# ============================================================================
	# EXECUTOR
	# ============================================================================

	class TDExecutor:
	"""Execute td_lang programs — compile and optionally run.

	Usage:
	executor = TDExecutor()

	# Compile only (check + generate .py)
	py_path = executor.compile("demo.td")

	# Compile and run
	result = executor.run("demo.td")

	# Just check syntax
	executor.check("demo.td")
	"""

	def __init__(self, output_dir: str = "td_lang_outputs"):
	self.output_dir = Path(output_dir)

	def check(self, td_path: str) -> TDProgram:
	"""Parse and validate a .td file without compiling or running.

	Args:
	td_path: Path to the .td file.

	Returns:
	The parsed TDProgram.

	Raises:
	TDSyntaxError: If syntax is invalid.
	TDCompileError: If semantic validation fails.
	"""
	print(f"[td_lang] Checking {td_path}...")
	program = parse_td_file(td_path)

	# Count what we found
	n_commands = len(program.commands)
	has_gates = program.gates is not None
	has_budget = program.budget is not None

	print(f"[td_lang] OK — {n_commands} commands", end="")
	if has_gates:
	print(f", gates: {program.gates.must_pass}", end="")
	if has_budget:
	print(f", budget set", end="")
	print()

	return program

	def compile(self, td_path: str) -> Path:
	"""Parse, validate, and compile a .td file into Python.

	Args:
	td_path: Path to the .td file.

	Returns:
	Path to the generated .py file.

	Raises:
	TDSyntaxError: If syntax is invalid.
	TDCompileError: If compilation fails.
	"""
	print(f"[td_lang] Compiling {td_path}...")

	# Parse
	program = parse_td_file(td_path)

	# Compile
	python_code = compile_program(program)

	# Validate the generated Python is valid syntax
	try:
	python_ast.parse(python_code)
	except SyntaxError as e:
	raise TDCompileError(
	f"Generated Python has a syntax error (this is a td_lang bug): {e}",
	hint="Please report this — the compiler generated bad code.",
	) from e

	# Save to output directory
	out_dir = self._make_output_dir(td_path)
	py_path = out_dir / "compiled.py"
	py_path.write_text(python_code)

	# Save source hash for lineage
	source_text = Path(td_path).read_text()
	meta = {
	"source_file": str(td_path),
	"source_hash": hashlib.sha256(source_text.encode()).hexdigest(),
	"compiled_at": datetime.now().isoformat(),
	"td_lang_version": "0.2.0",
	"python_file": str(py_path),
	"n_commands": len(program.commands),
	"has_gates": program.gates is not None,
	"has_budget": program.budget is not None,
	}
	meta_path = out_dir / "compile_meta.json"
	meta_path.write_text(json.dumps(meta, indent=2))

	print(f"[td_lang] Compiled to: {py_path}")
	return py_path

	def run(self, td_path: str, dry_run: bool = False) -> dict:
	"""Parse, compile, and execute a .td file.

	Args:
	td_path: Path to the .td file.
	dry_run: If True, compile but don't execute.

	Returns:
	Dict with execution results.
	"""
	# Compile first
	py_path = self.compile(td_path)

	if dry_run:
	print("[td_lang] Dry run — compiled but not executed.")
	return {"status": "dry_run", "compiled": str(py_path)}

	# Execute the generated Python script
	print(f"[td_lang] Executing {py_path}...")
	print()

	try:
	result = subprocess.run(
	[sys.executable, str(py_path.resolve())],
	capture_output=False, # Let output stream to console
	# Run from the original working directory, NOT the output dir.
	# The compiled code uses relative paths like "td_lang_outputs/..."
	# which must resolve from the project root.
	)

	if result.returncode == 0:
	print()
	print("[td_lang] Execution completed successfully.")
	return {"status": "success", "compiled": str(py_path)}
	else:
	print()
	print(f"[td_lang] Execution failed (exit code {result.returncode}).")
	return {
	"status": "failed",
	"compiled": str(py_path),
	"exit_code": result.returncode,
	}

	except Exception as e:
	print(f"\n[td_lang] Execution error: {e}")
	return {"status": "error", "compiled": str(py_path), "error": str(e)}

	def _make_output_dir(self, td_path: str) -> Path:
	"""Create a timestamped output directory for this run."""
	name = Path(td_path).stem
	timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
	out_dir = self.output_dir / f"{name}_{timestamp}"
	out_dir.mkdir(parents=True, exist_ok=True)
	return out_dir


	# ============================================================================
	# PUBLIC API
	# ============================================================================

	def check_td_file(td_path: str) -> TDProgram:
	"""Quick syntax check on a .td file."""
	return TDExecutor().check(td_path)


	def compile_td_file(td_path: str, output_dir: str = "td_lang_outputs") -> Path:
	"""Compile a .td file to Python."""
	return TDExecutor(output_dir=output_dir).compile(td_path)


	def run_td_file(td_path: str, output_dir: str = "td_lang_outputs", dry_run: bool = False) -> dict:
	"""Compile and run a .td file."""
	return TDExecutor(output_dir=output_dir).run(td_path, dry_run=dry_run)