ml-cpu / app.py
arudradey's picture
Update app.py
1a2d6eb verified
import gradio as gr
import os
import subprocess
import shutil
import numpy as np
from PIL import Image
import pywasm
# 1. Configuration & Persistent Paths
SAVE_DIR = "/data"
EMCC_DIR = os.path.join(SAVE_DIR, "emsdk")
if not os.path.exists(SAVE_DIR):
os.makedirs(SAVE_DIR, exist_ok=True)
def get_wasm_files():
"""Helper to list current WASM files in the bucket"""
if not os.path.exists(SAVE_DIR):
return []
files = [f for f in os.listdir(SAVE_DIR) if f.endswith('.wasm')]
return files if files else []
# 2. Compiler Logic (The Forge)
def ensure_compiler():
"""Checks for Emscripten in the bucket, installs if missing"""
emcc_path = os.path.join(EMCC_DIR, "upstream/emscripten/emcc")
if os.path.exists(emcc_path):
return emcc_path
try:
subprocess.run(["git", "clone", "https://github.com/emscripten-core/emsdk.git", EMCC_DIR],
check=True, capture_output=True)
subprocess.run([os.path.join(EMCC_DIR, "emsdk"), "install", "latest"],
check=True, capture_output=True)
subprocess.run([os.path.join(EMCC_DIR, "emsdk"), "activate", "latest"],
check=True, capture_output=True)
return emcc_path
except Exception as e:
return f"Error: {str(e)}"
def compile_c(c_code, filename):
if not filename:
return "❌ Error: Please provide a filename."
emcc_path = ensure_compiler()
if "Error" in emcc_path:
return emcc_path
# Clean up filename - remove extension if user included it
base = filename.split('.')[0]
if not base:
return "❌ Error: Invalid filename."
c_f = f"/tmp/{base}.c"
out_wasm = f"/tmp/{base}.wasm"
dest_wasm = os.path.join(SAVE_DIR, f"{base}.wasm")
with open(c_f, "w") as f:
f.write(c_code)
try:
# SIDE_MODULE=1 creates a clean standalone WASM for pywasm
res = subprocess.run([
emcc_path, c_f,
"-O3",
"-s", "WASM=1",
"-s", "SIDE_MODULE=1",
"-o", out_wasm
], capture_output=True, text=True, timeout=120)
if res.returncode != 0:
stderr = res.stderr[-2000:] if len(res.stderr) > 2000 else res.stderr
return f"❌ Compiler Error:\n{stderr}"
if os.path.exists(out_wasm):
if os.path.exists(dest_wasm):
os.remove(dest_wasm)
shutil.move(out_wasm, dest_wasm)
return f"βœ… Successfully Forged: {base}.wasm"
return "❌ Error: WASM file generation failed."
except subprocess.TimeoutExpired:
return "❌ Error: Compilation timed out (>120s)."
except Exception as e:
return f"❌ System Error: {str(e)}"
# 3. Execution Logic (The Runner) - FIXED
def run_wasm_func(wasm_file, func_name, args_str):
if not wasm_file:
return "❌ Select a WASM file."
if not func_name:
return "❌ Provide a function name."
path = os.path.join(SAVE_DIR, wasm_file)
if not os.path.exists(path):
return f"❌ WASM file not found: {wasm_file}"
try:
# Parse arguments
arg_list = []
if args_str and args_str.strip():
for x in args_str.split(","):
x = x.strip()
if x:
arg_list.append(int(x))
# Create runtime and load
runtime = pywasm.Runtime()
module_instance = runtime.instance_from_file(path)
# Try calling the function
# pywasm may need underscore prefix for exported C functions
errors = []
for name in [func_name, f"_{func_name}"]:
try:
result = runtime.invocate(module_instance, name, arg_list)
return f"βœ… Execution Successful!\nFunction: {name}\nResult: {result}"
except AssertionError as e:
# Usually wrong number of arguments
errors.append(f"'{name}': Wrong argument count or assertion failed")
except Exception as e:
errors.append(f"'{name}': {str(e)}")
return f"❌ Runner Error:\n" + "\n".join(errors)
except Exception as e:
return f"❌ Runner Error: {str(e)}"
# 4. Visualizer Logic (Normalized & Tiled)
def render_wasm(wasm_file):
if not wasm_file:
return None, "❌ Select a file."
path = os.path.join(SAVE_DIR, wasm_file)
if not os.path.exists(path):
return None, f"❌ File not found: {wasm_file}"
try:
with open(path, "rb") as f:
data = f.read()
if len(data) == 0:
return None, "❌ Empty WASM file."
arr = np.frombuffer(data, dtype=np.uint8).copy()
# Normalize bytes to visible pixel range
min_val = arr.min()
max_val = arr.max()
if max_val > min_val:
arr = ((arr - min_val) * (255.0 / (max_val - min_val))).astype(np.uint8)
# Tile to fill 1080p canvas
canvas_size = 1920 * 1080
if len(arr) > 0:
repeats = (canvas_size // len(arr)) + 1
tiled = np.tile(arr, repeats)
canvas = tiled[:canvas_size]
else:
canvas = np.zeros(canvas_size, dtype=np.uint8)
img = Image.fromarray(canvas.reshape((1080, 1920)), mode='L')
return img, f"βœ… Visualizing {len(data)} bytes β†’ 1920Γ—1080 grayscale"
except Exception as e:
return None, f"❌ Visualizer Error: {str(e)}"
# 5. Gradio UI
with gr.Blocks() as demo:
gr.Markdown("# πŸ› οΈ LCPU: Advanced Arithmetic Engine")
with gr.Tab("1. The Forge"):
wasm_name = gr.Textbox(label="Output Filename", value="logic_core",
placeholder="e.g. math_engine (no extension)")
c_code_input = gr.TextArea(
label="C Code Source",
value='int add(int a, int b) {\n return a + b;\n}\n\nint mul(int a, int b) {\n return a * b;\n}',
lines=10
)
compile_btn = gr.Button("πŸš€ Build Binary", variant="primary")
forge_status = gr.Textbox(label="Forge Output", interactive=False)
with gr.Tab("2. The Runner"):
with gr.Row():
run_select = gr.Dropdown(label="Select Binary", choices=get_wasm_files())
refresh_run = gr.Button("πŸ”„ Refresh", variant="secondary")
f_name = gr.Textbox(label="Function Name", value="add")
args_in = gr.Textbox(label="Arguments (comma separated)", placeholder="10, 20")
run_btn = gr.Button("⚑ Run Logic", variant="primary")
run_output = gr.Textbox(label="Output Result", interactive=False)
with gr.Tab("3. The Visualizer"):
with gr.Row():
vis_select = gr.Dropdown(label="Select Binary", choices=get_wasm_files())
refresh_vis = gr.Button("πŸ”„ Refresh", variant="secondary")
render_btn = gr.Button("🎨 Map to 1080p", variant="primary")
canvas_img = gr.Image(label="Logic Texture")
vis_status = gr.Textbox(label="Visualizer Status", interactive=False)
# Wiring - FIXED
compile_btn.click(compile_c, inputs=[c_code_input, wasm_name], outputs=forge_status)
# Refresh buttons - return new Dropdown with updated choices
refresh_run.click(lambda: gr.Dropdown(choices=get_wasm_files()), outputs=run_select)
refresh_vis.click(lambda: gr.Dropdown(choices=get_wasm_files()), outputs=vis_select)
run_btn.click(run_wasm_func, inputs=[run_select, f_name, args_in], outputs=run_output)
render_btn.click(render_wasm, inputs=vis_select, outputs=[canvas_img, vis_status])
if __name__ == "__main__":
demo.launch(theme=gr.themes.Monochrome())