Queue: 0 | Completed: 0
") with gr.Column(scale=0): refresh_stats_btn = gr.Button("⟳", elem_id="refresh-stats-btn") with gr.Tabs(): with gr.Tab("Generate", id="generate_tab"): with gr.Row(): with gr.Column(scale=2): model_type = gr.Radio( choices=["Original", "F1"], value="Original", label="Model", info="Select which model to use for generation" ) input_image = gr.Image( sources='upload', type="numpy", label="Image (optional)", height=420, elem_classes="contain-image", image_mode="RGB", show_download_button=False, show_label=True, container=True ) with gr.Accordion("Latent Image Options", open=False): latent_type = gr.Dropdown( ["Black", "White", "Noise", "Green Screen"], label="Latent Image", value="Black", info="Used as a starting point if no image is provided" ) prompt = gr.Textbox(label="Prompt", value=default_prompt) with gr.Accordion("Prompt Parameters", open=False): n_prompt = gr.Textbox(label="Negative Prompt", value="", visible=True) # Make visible for both models blend_sections = gr.Slider( minimum=0, maximum=10, value=4, step=1, label="Number of sections to blend between prompts" ) with gr.Accordion("Generation Parameters", open=True): with gr.Row(): steps = gr.Slider(label="Steps", minimum=1, maximum=100, value=25, step=1) total_second_length = gr.Slider(label="Video Length (Seconds)", minimum=1, maximum=120, value=6, step=0.1) with gr.Group(): with gr.Row("Resolution"): resolutionW = gr.Slider( label="Width", minimum=128, maximum=768, value=640, step=32, info="Nearest valid width will be used." ) resolutionH = gr.Slider( label="Height", minimum=128, maximum=768, value=640, step=32, info="Nearest valid height will be used." ) resolution_text = gr.Markdown(value="Settings saved successfully! Restart required for theme change.
" except Exception as e: return f"Error saving settings: {str(e)}
" save_btn.click( fn=save_settings, inputs=[save_metadata, gpu_memory_preservation, mp4_crf, clean_up_videos, output_dir, metadata_dir, lora_dir, gradio_temp_dir, auto_save, theme_dropdown], outputs=[status] ) def cleanup_temp_files(): """Clean up temporary files and folders in the Gradio temp directory""" temp_dir = settings.get("gradio_temp_dir") if not temp_dir or not os.path.exists(temp_dir): return "No temporary directory found or directory does not exist." try: # Get all items in the temp directory items = os.listdir(temp_dir) removed_count = 0 print(f"Finding items in {temp_dir}") for item in items: item_path = os.path.join(temp_dir, item) try: if os.path.isfile(item_path) or os.path.islink(item_path): print(f"Removing {item_path}") os.remove(item_path) removed_count += 1 elif os.path.isdir(item_path): print(f"Removing directory {item_path}") shutil.rmtree(item_path) removed_count += 1 except Exception as e: print(f"Error removing {item_path}: {e}") return f"Cleaned up {removed_count} temporary files/folders." except Exception as e: return f"Error cleaning up temporary files: {str(e)}" # --- Event Handlers and Connections (Now correctly indented) --- # Connect the main process function (wrapper for adding to queue) def process_with_queue_update(model_type, *args): # Extract all arguments (ensure order matches inputs lists) input_image, prompt_text, n_prompt, seed_value, total_second_length, latent_window_size, steps, cfg, gs, rs, gpu_memory_preservation, use_teacache, mp4_crf, randomize_seed_checked, save_metadata_checked, blend_sections, latent_type, clean_up_videos, selected_loras, resolutionW, resolutionH, *lora_args = args # DO NOT parse the prompt here. Parsing happens once in the worker. # Use the current seed value as is for this job # Call the process function with all arguments # Pass the model_type and the ORIGINAL prompt_text string to the backend process function result = process_fn(model_type, input_image, prompt_text, n_prompt, seed_value, total_second_length, # Pass original prompt_text string latent_window_size, steps, cfg, gs, rs, use_teacache, blend_sections, latent_type, clean_up_videos, selected_loras, resolutionW, resolutionH, *lora_args) # If randomize_seed is checked, generate a new random seed for the next job new_seed_value = None if randomize_seed_checked: new_seed_value = random.randint(0, 21474) print(f"Generated new seed for next job: {new_seed_value}") # If a job ID was created, automatically start monitoring it and update queue if result and result[1]: # Check if job_id exists in results job_id = result[1] queue_status_data = update_queue_status_fn() # Add the new seed value to the results if randomize is checked if new_seed_value is not None: return [result[0], job_id, result[2], result[3], result[4], result[5], result[6], queue_status_data, new_seed_value] else: return [result[0], job_id, result[2], result[3], result[4], result[5], result[6], queue_status_data, gr.update()] # If no job ID was created, still return the new seed if randomize is checked if new_seed_value is not None: return result + [update_queue_status_fn(), new_seed_value] else: return result + [update_queue_status_fn(), gr.update()] # Custom end process function that ensures the queue is updated def end_process_with_update(): queue_status_data = end_process_fn() # Make sure to return the queue status data return queue_status_data # --- Inputs Lists --- # --- Inputs for Original Model --- ips = [ input_image, prompt, n_prompt, seed, total_second_length, latent_window_size, steps, cfg, gs, rs, gpu_memory_preservation, use_teacache, mp4_crf, randomize_seed, save_metadata, blend_sections, latent_type, clean_up_videos, lora_selector, resolutionW, resolutionH, lora_names_states ] # Add LoRA sliders to the input list ips.extend([lora_sliders[lora] for lora in lora_names]) # --- Connect Buttons --- start_button.click( # Pass the selected model type from the radio buttons fn=lambda selected_model, *args: process_with_queue_update(selected_model, *args), inputs=[model_type] + ips, outputs=[result_video, current_job_id, preview_image, progress_desc, progress_bar, start_button, end_button, queue_status, seed] ) # Connect the end button to cancel the current job and update the queue end_button.click( fn=end_process_with_update, outputs=[queue_status] ) # --- Connect Monitoring --- # Auto-monitor the current job when job_id changes # Monitor original tab current_job_id.change( fn=monitor_fn, inputs=[current_job_id], outputs=[result_video, current_job_id, preview_image, progress_desc, progress_bar, start_button, end_button] ) cleanup_btn.click( fn=cleanup_temp_files, outputs=[cleanup_output] ) # --- Connect Queue Refresh --- refresh_stats_btn.click( fn=lambda: update_queue_status_fn(), # Use update_queue_status_fn passed in inputs=None, outputs=[queue_status] # Removed queue_stats_display from outputs ) # Set up auto-refresh for queue status (using a timer) refresh_timer = gr.Number(value=0, visible=False) def refresh_timer_fn(): """Updates the timer value periodically to trigger queue refresh""" return int(time.time()) # This timer seems unused, maybe intended for block.load()? Keeping definition for now. # refresh_timer.change( # fn=update_queue_status_fn, # Use the function passed in # outputs=[queue_status] # Update shared queue status display # ) # --- Connect LoRA UI --- # Function to update slider visibility based on selection def update_lora_sliders(selected_loras): updates = [] # Need to handle potential missing keys if lora_names changes dynamically # For now, assume lora_names passed to create_interface is static for lora in lora_names: updates.append(gr.update(visible=(lora in selected_loras))) # Ensure the output list matches the number of sliders defined num_sliders = len(lora_sliders) return updates[:num_sliders] # Return only updates for existing sliders # Connect the dropdown to the sliders lora_selector.change( fn=update_lora_sliders, inputs=[lora_selector], outputs=[lora_sliders[lora] for lora in lora_names] # Assumes lora_sliders keys match lora_names ) # --- Connect Metadata Loading --- # Function to load metadata from JSON file def load_metadata_from_json(json_path): if not json_path: # Return updates for all potentially affected components num_orig_sliders = len(lora_sliders) return [gr.update()] * (2 + num_orig_sliders) try: with open(json_path, 'r') as f: metadata = json.load(f) prompt_val = metadata.get('prompt') seed_val = metadata.get('seed') # Check for LoRA values in metadata lora_weights = metadata.get('loras', {}) # Changed key to 'loras' based on studio.py worker print(f"Loaded metadata from JSON: {json_path}") print(f"Prompt: {prompt_val}, Seed: {seed_val}") # Update the UI components updates = [ gr.update(value=prompt_val) if prompt_val else gr.update(), gr.update(value=seed_val) if seed_val is not None else gr.update() ] # Update LoRA sliders if they exist in metadata for lora in lora_names: if lora in lora_weights: updates.append(gr.update(value=lora_weights[lora])) else: updates.append(gr.update()) # No change if LoRA not in metadata # Ensure the number of updates matches the number of outputs num_orig_sliders = len(lora_sliders) return updates[:2 + num_orig_sliders] # Return updates for prompt, seed, and sliders except Exception as e: print(f"Error loading metadata: {e}") num_orig_sliders = len(lora_sliders) return [gr.update()] * (2 + num_orig_sliders) # Connect JSON metadata loader for Original tab json_upload.change( fn=load_metadata_from_json, inputs=[json_upload], outputs=[prompt, seed] + [lora_sliders[lora] for lora in lora_names] ) # --- Helper Functions (defined within create_interface scope if needed by handlers) --- # Function to get queue statistics def get_queue_stats(): try: # Get all jobs from the queue jobs = job_queue.get_all_jobs() # Count jobs by status status_counts = { "QUEUED": 0, "RUNNING": 0, "COMPLETED": 0, "FAILED": 0, "CANCELLED": 0 } for job in jobs: if hasattr(job, 'status'): status = str(job.status) # Use str() for safety if status in status_counts: status_counts[status] += 1 # Format the display text stats_text = f"Queue: {status_counts['QUEUED']} | Running: {status_counts['RUNNING']} | Completed: {status_counts['COMPLETED']} | Failed: {status_counts['FAILED']} | Cancelled: {status_counts['CANCELLED']}" return f"{stats_text}
" except Exception as e: print(f"Error getting queue stats: {e}") return "Error loading queue stats
" # Add footer with social links with gr.Row(elem_id="footer"): with gr.Column(scale=1): gr.HTML(""" """) # Add CSS for footer return block # --- Top-level Helper Functions (Used by Gradio callbacks, must be defined outside create_interface) --- def format_queue_status(jobs): """Format job data for display in the queue status table""" rows = [] for job in jobs: created = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(job.created_at)) if job.created_at else "" started = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(job.started_at)) if job.started_at else "" completed = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(job.completed_at)) if job.completed_at else "" # Calculate elapsed time elapsed_time = "" if job.started_at: if job.completed_at: start_datetime = datetime.datetime.fromtimestamp(job.started_at) complete_datetime = datetime.datetime.fromtimestamp(job.completed_at) elapsed_seconds = (complete_datetime - start_datetime).total_seconds() elapsed_time = f"{elapsed_seconds:.2f}s" else: # For running jobs, calculate elapsed time from now start_datetime = datetime.datetime.fromtimestamp(job.started_at) current_datetime = datetime.datetime.now() elapsed_seconds = (current_datetime - start_datetime).total_seconds() elapsed_time = f"{elapsed_seconds:.2f}s (running)" # Get generation type from job data generation_type = getattr(job, 'generation_type', 'Original') # Removed thumbnail processing rows.append([ job.id[:6] + '...', generation_type, job.status.value, created, started, completed, elapsed_time # Removed thumbnail from row data ]) return rows # Create the queue status update function (wrapper around format_queue_status) def update_queue_status_with_thumbnails(): # Function name is now slightly misleading, but keep for now to avoid breaking clicks # This function is likely called by the refresh button and potentially the timer # It needs access to the job_queue object # Assuming job_queue is accessible globally or passed appropriately # For now, let's assume it's globally accessible as defined in studio.py # If not, this needs adjustment based on how job_queue is managed. try: # Need access to the global job_queue instance from studio.py # This might require restructuring or passing job_queue differently. # For now, assuming it's accessible (this might fail if run standalone) from __main__ import job_queue # Attempt to import from main script scope jobs = job_queue.get_all_jobs() for job in jobs: if job.status == JobStatus.PENDING: job.queue_position = job_queue.get_queue_position(job.id) if job_queue.current_job: job_queue.current_job.status = JobStatus.RUNNING return format_queue_status(jobs) except ImportError: print("Error: Could not import job_queue. Queue status update might fail.") return [] # Return empty list on error except Exception as e: print(f"Error updating queue status: {e}") return []