# demo/app.py import gradio as gr import torch import time import logging import random import numpy as np from diffusers import StableDiffusionXLPipeline, EulerAncestralDiscreteScheduler import threading import queue import spaces # Import the component and ALL its utilities from gradio_livelog import LiveLog from gradio_livelog.utils import ProgressTracker, capture_logs # --- 1. LOGIC FOR THE "LIVELOG FEATURE DEMO" TAB --- # Configure logging for the demo logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') def run_process(disable_console: bool, run_error_case: bool): """ The generator function for the interactive feature demo. """ with capture_logs(disable_console=disable_console) as get_logs: total_steps = 100 tracker = ProgressTracker(total=total_steps, description="Simulating a process...") logging.info(f"Starting simulated process with {total_steps} steps...") for record in get_logs(): yield {"type": "log", "level": record.levelname, "content": record.getMessage()} for i in range(total_steps): time.sleep(0.03) current_step = i + 1 if current_step == 10: logging.warning(f"Low disk space warning at step {current_step}.") elif current_step == 30: logging.log(logging.INFO + 5, f"Asset pack loaded successfully at step {current_step}.") elif current_step == 75: logging.critical(f"Checksum mismatch! Data may be corrupt at step {current_step}.") if run_error_case and current_step == 50: logging.error("A fatal simulation error occurred! Aborting.") for record in get_logs(): yield {"type": "log", "level": record.levelname, "content": record.getMessage()} yield tracker.update(advance=0, status="error") return yield tracker.update() for record in get_logs(): level_name = "SUCCESS" if record.levelno == logging.INFO + 5 else record.levelname yield {"type": "log", "level": level_name, "content": record.getMessage()} logging.log(logging.INFO + 5, "Process completed successfully!") for record in get_logs(): level_name = "SUCCESS" if record.levelno == logging.INFO + 5 else record.levelname yield {"type": "log", "level": level_name, "content": record.getMessage()} yield tracker.update(advance=0, status="success") def update_livelog_properties(mode, color, lines, scroll): return gr.update(display_mode=mode, background_color=color, line_numbers=lines, autoscroll=scroll) def clear_output(): return None def run_success_case(disable_console: bool): yield from run_process(disable_console=disable_console, run_error_case=False) def run_error_case(disable_console: bool): yield from run_process(disable_console=disable_console, run_error_case=True) # --- 2. LOGIC FOR THE "Diffusion Pipeline Integration" TAB --- MODEL_ID = "SG161222/RealVisXL_V5.0" MAX_SEED = np.iinfo(np.int32).max print("Loading Stable Diffusion model... This may take a moment.") pipe = StableDiffusionXLPipeline.from_pretrained( MODEL_ID, torch_dtype=torch.float16, use_safetensors=True, add_watermarker=False ) pipe.enable_vae_tiling() pipe.enable_model_cpu_offload() pipe.scheduler = EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config) print("Model loaded successfully.") pipe.set_progress_bar_config(disable=True) @spaces.GPU(duration=60, enable_queue=True) def run_diffusion_in_thread(prompt: str, update_queue: queue.Queue): """This function contains the blocking pipe() call and runs in a worker thread.""" tracker = None try: seed = random.randint(0, MAX_SEED) generator = torch.Generator(device="cuda").manual_seed(seed) prompt_style = f"hyper-realistic 8K image of {prompt}. ultra-detailed, lifelike, high-resolution, sharp, vibrant colors, photorealistic" negative_prompt_style = "cartoonish, low resolution, blurry, simplistic, abstract, deformed, ugly" num_inference_steps = 20 update_queue.put((None, {"type": "log", "level": "INFO", "content": f"Using seed: {seed}"})) update_queue.put((None, {"type": "log", "level": "INFO", "content": "Starting diffusion process..."})) tracker = ProgressTracker(total=num_inference_steps, description="Diffusion Steps") def progress_callback(pipe_instance, step, timestep, callback_kwargs): update_dict = tracker.update() update_queue.put((None, update_dict)) return callback_kwargs images = pipe( prompt=prompt_style, negative_prompt=negative_prompt_style, width=1024, height=1024, guidance_scale=3.0, num_inference_steps=num_inference_steps, generator=generator, callback_on_step_end=progress_callback ).images update_queue.put((images, {"type": "log", "level": "SUCCESS", "content": "Image generated successfully!"})) final_update = tracker.update(advance=0, status="success") update_queue.put((images, final_update)) except Exception as e: logging.error(f"Error in diffusion thread: {e}", exc_info=True) if tracker: error_update = tracker.update(advance=0, status="error") update_queue.put((None, error_update)) update_queue.put((None, {"type": "log", "level": "ERROR", "content": f"An error occurred: {e}"})) finally: update_queue.put(None) def generate(prompt): """This function starts the worker thread and yields updates from the queue.""" yield None, {"type": "log", "level": "INFO", "content": "Preparing generation..."} update_queue = queue.Queue() diffusion_thread = threading.Thread(target=run_diffusion_in_thread, args=(prompt, update_queue)) diffusion_thread.start() while True: update = update_queue.get() if update is None: break yield update # --- 3. THE COMBINED GRADIO UI with TABS --- with gr.Blocks(theme=gr.themes.Ocean()) as demo: gr.HTML("

LiveLog Component Showcase

") with gr.Tabs(): with gr.TabItem("LiveLog Feature Demo"): gr.Markdown("### Test all features of the LiveLog component interactively.") with gr.Row(): with gr.Column(scale=3): feature_logger = LiveLog( label="Process Output", line_numbers=True, height=550, background_color="#000000", display_mode="full" ) with gr.Column(scale=1): with gr.Group(): gr.Markdown("### Component Properties") display_mode_radio = gr.Radio(["full", "log", "progress"], label="Display Mode", value="full") bg_color_picker = gr.ColorPicker(label="Background Color", value="#000000") line_numbers_checkbox = gr.Checkbox(label="Show Line Numbers", value=True) autoscroll_checkbox = gr.Checkbox(label="Autoscroll", value=True) with gr.Group(): gr.Markdown("### Simulation Controls") disable_console_checkbox = gr.Checkbox(label="Disable Python Console Output", value=False) start_btn = gr.Button("Run Success Case", variant="primary") error_btn = gr.Button("Run Error Case") start_btn.click(fn=run_success_case, inputs=[disable_console_checkbox], outputs=feature_logger) error_btn.click(fn=run_error_case, inputs=[disable_console_checkbox], outputs=feature_logger) feature_logger.clear(fn=clear_output, inputs=None, outputs=[feature_logger]) controls = [display_mode_radio, bg_color_picker, line_numbers_checkbox, autoscroll_checkbox] for control in controls: control.change(fn=update_livelog_properties, inputs=controls, outputs=feature_logger) with gr.TabItem("Diffusion Pipeline Integration"): gr.Markdown("### Use `LiveLog` to monitor a real image generation process.") with gr.Row(): with gr.Column(scale=3): with gr.Group(): prompt = gr.Textbox( label="Enter your prompt", show_label=False, placeholder="A cinematic photo of a robot in a floral garden...", scale=8, container=False ) run_button = gr.Button("Generate", scale=1, variant="primary") livelog_viewer = LiveLog( label="Process Monitor", height=250, display_mode="full", line_numbers=False, disable_console=True ) with gr.Column(scale=2): result_gallery = gr.Gallery( label="Result", columns=1, show_label=False, height=500, min_width=768 ) run_button.click(fn=generate, inputs=[prompt], outputs=[result_gallery, livelog_viewer]) prompt.submit(fn=generate, inputs=[prompt], outputs=[result_gallery, livelog_viewer]) if __name__ == "__main__": demo.queue(max_size=50).launch(debug=True)