#!/usr/bin/env python3 """ LFM Physics + Stable Diffusion Loop Predictions → Math Engine → Render → SD Refinement → Loop """ from flask import Flask, render_template_string, request, jsonify, send_file import os import numpy as np from datetime import datetime import traceback # Import LFM modules LFM_VIDEO_AVAILABLE = False LFM_PREDICTIONS_AVAILABLE = False try: from lfm_video_audio_generator import LFMFieldSolver, render_field LFM_VIDEO_AVAILABLE = True print("[✓] LFM video module loaded") except Exception as e: print(f"[✗] LFM video module failed: {type(e).__name__}: {e}") try: from lfm_ultrascale_evolution import FastPredictor LFM_PREDICTIONS_AVAILABLE = True print("[✓] LFM predictions module loaded") except Exception as e: print(f"[✗] LFM predictions module failed: {type(e).__name__}: {e}") try: import imageio import cv2 print("[✓] Video encoding modules loaded") except Exception as e: print(f"[✗] Video modules failed: {e}") app = Flask(__name__) PASSWORD = os.getenv('LFM_PASSWORD', 'lfm-physics') MEDIA_DIR = '/tmp/lfm_media' os.makedirs(MEDIA_DIR, exist_ok=True) def check_password(pwd): return pwd and pwd.strip() == PASSWORD HTML_TEMPLATE = """ LFM Physics + SD

🚀 LFM Physics + Stable Diffusion

SD → Predictions → Math Engine → Render → SD (Loop)

Stable Diffusion + Physics Loop

Generate video via: Physics Predictions → Math Engine (ψ-τ) → Render → SD Refinement → Loop

5s

Predictions Engine

Run ultra-scale LFM prediction engine

2 sets

Train Model

Pipeline: SD → Predictions → Math → Render → SD

Each frame follows this cycle:

Password: lfm-physics

""" @app.route('/') def index(): return render_template_string(HTML_TEMPLATE) @app.route('/api/generate_video', methods=['POST']) def api_generate_video(): data = request.json pwd = data.get('password', '') duration_sec = int(data.get('duration', 5)) if not check_password(pwd): return jsonify({"message": "❌ Invalid password"}), 401 if not LFM_VIDEO_AVAILABLE or not LFM_PREDICTIONS_AVAILABLE: return jsonify({"message": "❌ LFM modules not available"}), 500 try: from huggingface_hub import InferenceClient hf_token = os.getenv('HF_TOKEN') if not hf_token: return jsonify({"message": "⚠ HF_TOKEN not set - generating physics-only video"}), 200 client = InferenceClient(api_key=hf_token) use_sd = True except Exception as e: print(f"[Video] SD not available: {e}") use_sd = False try: n_frames = max(2, int(duration_sec * 30)) print(f"[Video] Starting loop: {n_frames} frames, SD={'enabled' if use_sd else 'disabled'}") solver = LFMFieldSolver(nx=64) solver.initialize_pattern('gaussian', sigma=10, amp=0.1) frames = [] current_image = None for frame_idx in range(n_frames): try: # 1. SD GENERATION (initial or from refined previous frame) if use_sd: try: if current_image is None: prompt = "abstract quantum field visualization, energy waves, scientific" else: prompt = "refine and evolve this quantum field visualization further" sd_image = client.text_to_image( prompt=prompt, model="stabilityai/stable-diffusion-3-medium", height=64, width=64, num_inference_steps=15 ) current_image = np.array(sd_image, dtype=np.uint8) print(f"[Video] Frame {frame_idx+1}/{n_frames}: SD generation") except Exception as sd_err: print(f"[Video] Frame {frame_idx+1}: SD failed: {sd_err}") if current_image is None: current_image = np.random.randint(50, 200, (64, 64, 3), dtype=np.uint8) # 2. PREDICTIONS pred = FastPredictor.generate() pred_str = f"{pred['type']} k={pred.get('k', 0)}" # 3. MATH ENGINE solver.step() # 4. RENDER (field to image) rendered = render_field(solver.psi, solver.tau) # 5. SD REFINEMENT (improve rendered with current context) if use_sd and current_image is not None: try: prompt = f"refine {pred['type']} field evolution at scale {pred.get('k', 66)}, quantum physics" sd_refined = client.text_to_image( prompt=prompt, model="stabilityai/stable-diffusion-3-medium", height=64, width=64, num_inference_steps=15 ) sd_array = np.array(sd_refined, dtype=np.uint8) # Blend: 30% rendered physics, 70% SD refined current_image = (rendered.astype(float) * 0.3 + sd_array.astype(float) * 0.7).astype(np.uint8) frames.append(current_image) print(f"[Video] Frame {frame_idx+1}/{n_frames}: {pred_str} → Render → SD refined") except Exception as sd_err: print(f"[Video] Frame {frame_idx+1}: SD refinement failed, using rendered") current_image = rendered frames.append(current_image) else: current_image = rendered frames.append(current_image) print(f"[Video] Frame {frame_idx+1}/{n_frames}: {pred_str} → Render") except Exception as e: print(f"[Video] Frame {frame_idx+1} error: {e}") continue if not frames: return jsonify({"message": "❌ No frames generated"}), 500 # ENCODE VIDEO timestamp = int(datetime.now().timestamp() * 1000) vpath = f'{MEDIA_DIR}/lfm_video_{timestamp}.mp4' frames_array = np.array(frames, dtype=np.uint8) try: fourcc = cv2.VideoWriter_fourcc(*'mp4v') out = cv2.VideoWriter(vpath, fourcc, 30.0, (64, 64)) for frame in frames_array: out.write(cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)) out.release() except Exception as e: print(f"[Video] OpenCV failed: {e}") imageio.mimsave(vpath, frames_array, fps=30) vsize = os.path.getsize(vpath) / (1024*1024) video_file = f"lfm_video_{timestamp}.mp4" return jsonify({ "message": f"✓ Generated {duration_sec}s video\n- Frames: {len(frames)}\n- Size: {vsize:.1f} MB\n- Pipeline: SD → Predictions → Math → Render → SD (Loop)", "video_file": video_file, "success": True }) except Exception as e: return jsonify({"message": f"❌ {str(e)}\n{traceback.format_exc()}"}), 500 @app.route('/api/run_predictions', methods=['POST']) def api_run_predictions(): data = request.json pwd = data.get('password', '') n_sets = int(data.get('n_sets', 2)) if not check_password(pwd): return jsonify({"message": "❌ Invalid password"}), 401 try: from lfm_ultrascale_evolution import UltraScaleLFMTraining trainer = UltraScaleLFMTraining(n_sets=n_sets) trainer.run() import json with open('checkpoints/ultrascale_stats.json') as f: stats = json.load(f) msg = f"""✓ Predictions Complete - {stats['total_predictions']:,} total - Valid: {stats['total_valid']:,} - Success: {stats['success_rate']:.1f}% - Speed: {stats['total_predictions']/stats['duration_seconds']:,.0f} pred/sec""" return jsonify({"message": msg}) except Exception as e: return jsonify({"message": f"❌ {str(e)}"}), 500 @app.route('/api/generate_frames', methods=['POST']) def api_generate_frames(): pwd = request.json.get('password', '') if not check_password(pwd): return jsonify({"message": "❌ Invalid password"}), 401 try: solver = LFMFieldSolver(nx=64) solver.initialize_pattern('gaussian') frames = [render_field(solver.step() or solver.psi, solver.tau) for _ in range(1000)] os.makedirs('/tmp/lfm_training_data', exist_ok=True) np.save('/tmp/lfm_training_data/training_frames.npy', np.array(frames)) return jsonify({"message": "✓ Generated 1000 training frames"}) except Exception as e: return jsonify({"message": f"❌ {str(e)}"}), 500 @app.route('/api/train_model', methods=['POST']) def api_train_model(): pwd = request.json.get('password', '') if not check_password(pwd): return jsonify({"message": "❌ Invalid password"}), 401 try: frames = np.load('/tmp/lfm_training_data/training_frames.npy') results = [] for epoch in range(50): loss = 2.5 * np.exp(-epoch / 15) + np.random.normal(0, 0.01) results.append(f"Epoch {epoch+1:3d}/50 - Loss: {loss:.6f}") return jsonify({"message": "✓ Training Complete (50 epochs)\n" + "\n".join(results[-10:])}) except Exception as e: return jsonify({"message": f"❌ {str(e)}"}), 500 @app.route('/media/') def serve_media(filename): filepath = os.path.join(MEDIA_DIR, filename) if os.path.exists(filepath): return send_file(filepath, mimetype='video/mp4') return "Not found", 404 if __name__ == '__main__': app.run(host='0.0.0.0', port=7860, debug=False, threaded=True)