import streamlit as st import subprocess import os import logging import threading import time from huggingface_hub import snapshot_download # Configure logging logging.basicConfig( filename='supercoder_launcher.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) # Directories REPO_DIR = os.path.join(os.getcwd(), "SuperCoder") # Session state initialization if 'terminal_output' not in st.session_state: st.session_state.terminal_output = [] if 'setup_complete' not in st.session_state: st.session_state.setup_complete = False if 'running' not in st.session_state: st.session_state.running = False # ---------- Utility functions ---------- def log_terminal(line: str): st.session_state.terminal_output.append(line) logging.info(line) def clear_terminal(): st.session_state.terminal_output = [] def get_command_output(command: str, cwd: str = None) -> str: try: result = subprocess.run( command, shell=True, cwd=cwd, capture_output=True, text=True, timeout=60 ) return result.stdout + "\n" + result.stderr except Exception as e: return f"Error executing command: {str(e)}" def run_command_background(command: str, cwd: str = None): log_terminal(f"> {command}") def target(): try: process = subprocess.Popen( command, shell=True, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, universal_newlines=True ) for line in process: if line: log_terminal(line.rstrip()) returncode = process.wait() log_terminal(f"Process finished with return code: {returncode}") except Exception as e: log_terminal(f"Exception in background command: {str(e)}") threading.Thread(target=target, daemon=True).start() st.success("Command started in background. Refresh terminal to see progress.") # Hugging Face handling (enhanced) def handle_huggingface(link: str, instruction: str, hf_token: str = None): log_terminal(f"Processing Hugging Face link: {link}") log_terminal(f"Instruction: {instruction}") if "spaces" in link: parts = link.split("huggingface.co/spaces/")[-1].strip("/") clone_url = f"https://huggingface.co/spaces/{parts}" if hf_token: clone_url = f"https:{hf_token}@huggingface.co/spaces/{parts}" dir_name = parts.split("/")[-1] cmd = f"git clone {clone_url} workspaces/{dir_name}" run_command_background(cmd, cwd=REPO_DIR) elif "datasets" in link or "models" in link: repo_id = link.split("huggingface.co/")[-1].strip("/") local_dir = os.path.join(REPO_DIR, "models", repo_id.split("/")[-1]) try: path = snapshot_download( repo_id=repo_id, local_dir=local_dir, token=hf_token or None, ignore_patterns=["*.bin"] if "small" in instruction.lower() else None ) log_terminal(f"Downloaded {repo_id} → {path}") except Exception as e: log_terminal(f"HF download error: {str(e)}") else: log_terminal("Unsupported Hugging Face link type.") # ---------- Main UI ---------- st.set_page_config(page_title="SuperCoder Advanced Launcher", page_icon="🚀", layout="wide") st.title("🚀 SuperCoder Advanced Launcher") st.markdown("Full-featured launcher for [TransformerOptimus/SuperCoder](https://github.com/TransformerOptimus/SuperCoder) with Docker Compose, .env config, live logs, preview, and Hugging Face/GitHub integrations.") tabs = st.tabs(["Setup", "Config", "Run", "Logs & Status", "Preview", "Integrations", "Terminal"]) with tabs[0]: # Setup st.subheader("Repository Management") repo_git = os.path.join(REPO_DIR, ".git") cloned = os.path.isdir(repo_git) if not cloned: if st.button("Clone SuperCoder Repository", type="primary"): with st.spinner("Cloning repository..."): cmd = f"git clone https://github.com/TransformerOptimus/SuperCoder {REPO_DIR}" success = subprocess.run(cmd, shell=True, capture_output=True, text=True).returncode == 0 if success: st.session_state.setup_complete = True st.success("Repository cloned successfully.") st.rerun() else: st.error("Clone failed. Check logs.") else: st.success("✅ Repository cloned") st.session_state.setup_complete = True if st.button("Update Repository (git pull)"): with st.spinner("Pulling latest changes..."): run_command_background("git pull", cwd=REPO_DIR) with tabs[1]: # Config st.subheader(".env Configuration (API keys, etc.)") env_path = os.path.join(REPO_DIR, ".env") default_env = """OPENAI_API_KEY=sk-... ANTHROPIC_API_KEY=... HF_TOKEN=hf_... # Add other keys as needed """ if os.path.exists(env_path): with open(env_path) as f: env_content = f.read() else: env_content = default_env new_env = st.text_area(".env content", value=env_content, height=400) if st.button("Save .env"): try: with open(env_path, "w") as f: f.write(new_env) st.success(".env saved successfully.") except Exception as e: st.error(f"Save failed: {e}") with tabs[2]: # Run st.subheader("Docker Compose Controls") col1, col2, col3 = st.columns(3) with col1: if st.button("Start / Rebuild", type="primary", disabled=not st.session_state.setup_complete): run_command_background("docker compose up --build -d", cwd=REPO_DIR) st.session_state.running = True with col2: if st.button("Stop"): run_command_background("docker compose down", cwd=REPO_DIR) st.session_state.running = False with col3: if st.button("Restart"): run_command_background("docker compose restart", cwd=REPO_DIR) if st.button("Full Rebuild (down + up --build)"): run_command_background("docker compose down && docker compose up --build -d", cwd=REPO_DIR) with tabs[3]: # Logs & Status st.subheader("Service Status & Logs") auto = st.checkbox("Auto-refresh every 8 seconds") if auto: st.autorefresh(interval=8000, key="logsrefresh") c1, c2 = st.columns(2) with c1: st.markdown("**docker compose ps**") status = get_command_output("docker compose ps", cwd=REPO_DIR) st.code(status) with c2: st.markdown("**Recent logs**") logs = get_command_output("docker compose logs --tail=1000", cwd=REPO_DIR) st.code(logs, language="log") with tabs[4]: # Preview st.subheader("Live Application Preview") if st.session_state.running or "Up" in get_command_output("docker compose ps -q", cwd=REPO_DIR): st.components.v1.iframe("http://localhost:3000", height=900, scrolling=True) else: st.warning("Application not running. Start it in the Run tab first.") with tabs[5]: # Integrations st.subheader("Hugging Face & GitHub Integrations") hf_token = st.text_input("Hugging Face Token (for private repos)", type="password") link = st.text_input("Hugging Face or GitHub Link") instruction = st.text_area("Instructions (e.g., 'small model only', path hints)") if st.button("Process Link"): if not link: st.error("Provide a link.") else: if "huggingface.co" in link: handle_huggingface(link, instruction, hf_token) elif "github.com" in link: dir_name = link.split("/")[-1].replace(".git", "") cmd = f"git clone {link} workspaces/{dir_name}" run_command_background(cmd, cwd=REPO_DIR) st.success("Processing started.") st.markdown("---") st.subheader("Manual HF Model/Dataset Download") repo_id = st.text_input("Repo ID (e.g. meta-llama/Llama-3-8B)") target_dir = st.text_input("Target subdirectory (relative to SuperCoder)", default="models/custom") if st.button("Download Now"): full_path = os.path.join(REPO_DIR, target_dir) os.makedirs(full_path, exist_ok=True) try: snapshot_download(repo_id=repo_id, local_dir=full_path, token=hf_token or None) st.success(f"Downloaded to {full_path}") except Exception as e: st.error(str(e)) with tabs[6]: # Terminal st.subheader("Command Output Terminal") st.code("\n".join(st.session_state.terminal_output), language="bash") col1, col2 = st.columns(2) with col1: if st.button("Refresh Terminal"): st.rerun() with col2: if st.button("Clear Terminal"): clear_terminal() st.rerun() st.sidebar.markdown("### Quick Actions") if st.sidebar.button("Full Reset (stop + remove repo)"): if st.sidebar.checkbox("Confirm dangerous action"): run_command_background("docker compose down -v", cwd=REPO_DIR) if os.path.exists(REPO_DIR): import shutil shutil.rmtree(REPO_DIR) st.session_state.setup_complete = False st.session_state.running = False clear_terminal() st.success("Reset complete.") st.sidebar.markdown("---") st.sidebar.caption(f"Launcher running from: {os.getcwd()}") st.sidebar.caption("SuperCoder directory: ./SuperCoder")