import os import platform import numpy as np from tqdm import trange import math import subprocess as sp import string import random from functools import reduce import re import modules.scripts as scripts import gradio as gr from modules import processing, shared, sd_samplers, images from modules.processing import Processed from modules.sd_samplers import samplers from modules.shared import opts, cmd_opts, state import subprocess wave_completed_regex = r'@wave_completed\(([\-]?[0-9]*\.?[0-9]+), ?([\-]?[0-9]*\.?[0-9]+)\)' wave_remaining_regex = r'@wave_remaining\(([\-]?[0-9]*\.?[0-9]+), ?([\-]?[0-9]*\.?[0-9]+)\)' def run_cmd(cmd): cmd = list(map(lambda arg: str(arg), cmd)) print("Executing %s" % " ".join(cmd)) popen_params = {"stdout": sp.DEVNULL, "stderr": sp.PIPE, "stdin": sp.DEVNULL} if os.name == "nt": popen_params["creationflags"] = 0x08000000 proc = sp.Popen(cmd, **popen_params) out, err = proc.communicate() # proc.wait() proc.stderr.close() if proc.returncode: raise IOError(err.decode("utf8")) del proc def encode_video(input_pattern, starting_number, output_dir, fps, quality, encoding, create_segments, segment_duration, ffmpeg_path): two_pass = (encoding == "VP9 (webm)") alpha_channel = ("webm" in encoding) suffix = "webm" if "webm" in encoding else "mp4" output_location = output_dir + f".{suffix}" encoding_lib = { "VP9 (webm)": "libvpx-vp9", "VP8 (webm)": "libvpx", "H.264 (mp4)": "libx264", "H.265 (mp4)": "libx265", }[encoding] args = [ "-framerate", fps, "-start_number", int(starting_number), "-i", input_pattern, "-c:v", encoding_lib, "-b:v","0", "-crf", quality, ] if encoding_lib == "libvpx-vp9": args += ["-pix_fmt", "yuva420p"] if(ffmpeg_path == ""): ffmpeg_path = "ffmpeg" if(platform.system == "Windows"): ffmpeg_path += ".exe" print("\n\n") if two_pass: first_pass_args = args + [ "-pass", "1", "-an", "-f", "null", os.devnull ] second_pass_args = args + [ "-pass", "2", output_location ] print("Running first pass ffmpeg encoding") run_cmd([ffmpeg_path] + first_pass_args) print("Running second pass ffmpeg encoding. This could take awhile...") run_cmd([ffmpeg_path] + second_pass_args) else: print("Running ffmpeg encoding. This could take awhile...") run_cmd([ffmpeg_path] + args + [output_location]) if(create_segments): print("Segmenting video") run_cmd([ffmpeg_path] + [ "-i", output_location, "-f", "segment", "-segment_time", segment_duration, "-vcodec", "copy", "-acodec", "copy", f"{output_dir}.%d.{suffix}" ]) def set_weights(match_obj, wave_progress): weight_0 = 0 weight_1 = 0 if match_obj.group(1) is not None: weight_0 = float(match_obj.group(1)) if match_obj.group(2) is not None: weight_1 = float(match_obj.group(2)) max_weight = max(weight_0, weight_1) min_weight = min(weight_0, weight_1) weight_range = max_weight - min_weight weight = min_weight + weight_range * wave_progress return str(weight) class Script(scripts.Script): def title(self): return "Loopback Wave V1.4.1" def show(self, is_img2img): return is_img2img def ui(self, is_img2img): frames = gr.Slider(minimum=1, maximum=2048, step=1, label='Frames', value=100) frames_per_wave = gr.Slider(minimum=0, maximum=120, step=1, label='Frames Per Wave', value=20) denoising_strength_change_amplitude = gr.Slider(minimum=0, maximum=1, step=0.01, label='Max additional denoise', value=0.6) denoising_strength_change_offset = gr.Number(minimum=0, maximum=180, step=1, label='Wave offset (ignore this if you don\'t know what it means)', value=0) initial_image_number = gr.Number(minimum=0, label='Initial generated image number', value=0) save_prompts = gr.Checkbox(label='Save prompts as text file', value=True) prompts = gr.Textbox(label="Prompt Changes", lines=5, value="") save_video = gr.Checkbox(label='Save results as video', value=True) output_dir = gr.Textbox(label="Video Name", lines=1, value="") video_fps = gr.Slider(minimum=1, maximum=120, step=1, label='Frames per second', value=10) video_quality = gr.Slider(minimum=0, maximum=60, step=1, label='Video Quality (crf)', value=40) video_encoding = gr.Dropdown(label='Video encoding', value="VP9 (webm)", choices=["VP9 (webm)", "VP8 (webm)", "H.265 (mp4)", "H.264 (mp4)"]) ffmpeg_path = gr.Textbox(label="ffmpeg binary. Only set this if it fails otherwise.", lines=1, value="") segment_video = gr.Checkbox(label='Cut video in to segments', value=True) video_segment_duration = gr.Slider(minimum=10, maximum=60, step=1, label='Video Segment Duration (seconds)', value=20) return [frames, denoising_strength_change_amplitude, frames_per_wave, denoising_strength_change_offset,initial_image_number, prompts, save_prompts, save_video, output_dir, video_fps, video_quality, video_encoding, ffmpeg_path, segment_video, video_segment_duration] def run(self, p, frames, denoising_strength_change_amplitude, frames_per_wave, denoising_strength_change_offset, initial_image_number, prompts: str,save_prompts, save_video, output_dir, video_fps, video_quality, video_encoding, ffmpeg_path, segment_video, video_segment_duration): processing.fix_seed(p) batch_count = p.n_iter p.extra_generation_params = { "Max Additional Denoise": denoising_strength_change_amplitude, "Frames per wave": frames_per_wave, "Wave Offset": denoising_strength_change_offset, } # We save them ourselves for the sake of ffmpeg p.do_not_save_samples = True changes_dict = {} p.batch_size = 1 p.n_iter = 1 output_images, info = None, None initial_seed = None initial_info = None grids = [] all_images = [] original_init_image = p.init_images state.job_count = frames * batch_count initial_color_corrections = [processing.setup_color_correction(p.init_images[0])] initial_denoising_strength = p.denoising_strength if(output_dir==""): output_dir = str(p.seed) else: output_dir = output_dir + "-" + str(p.seed) loopback_wave_path = os.path.join(p.outpath_samples, "loopback-wave") loopback_wave_images_path = os.path.join(loopback_wave_path, output_dir) os.makedirs(loopback_wave_images_path, exist_ok=True) p.outpath_samples = loopback_wave_images_path prompts = prompts.strip() if save_prompts: with open(loopback_wave_images_path + "-prompts.txt", "w") as f: generation_settings = [ "Generation Settings", f"Total Frames: {frames}", f"Frames Per Wave: {frames_per_wave}", f"Wave Offset: {denoising_strength_change_offset}", f"Base Denoising Strength: {initial_denoising_strength}", f"Max Additional Denoise: {denoising_strength_change_amplitude}", f"Initial Image Number: {initial_image_number}", "", "Video Encoding Settings", f"Save Video: {save_video}" ] if save_video: generation_settings = generation_settings + [ f"Framerate: {video_fps}", f"Quality: {video_quality}", f"Encoding: {video_encoding}", f"Create Segmented Video: {segment_video}" ] if segment_video: generation_settings = generation_settings + [f"Segment Duration: {video_segment_duration}"] generation_settings = generation_settings + [ "", "Prompt Details", "Initial Prompt:" + p.prompt, "", "Negative Prompt:" + p.negative_prompt, "", "Frame change prompts:", prompts ] f.write('\n'.join(generation_settings)) if prompts: lines = prompts.split("\n") for prompt_line in lines: params = prompt_line.split("::") if len(params) == 2: changes_dict[params[0]] = { "prompt": params[1] } elif len(params) == 3: changes_dict[params[0]] = { "seed": params[1], "prompt": params[2] } else: raise IOError(f"Invalid input in prompt line: {prompt_line}") raw_prompt = p.prompt for n in range(batch_count): history = [] # Reset to original init image at the start of each batch p.init_images = original_init_image seed_state = "adding" current_seed = p.seed for i in range(frames): current_seed = p.seed state.job = "" if str(i) in changes_dict: raw_prompt = changes_dict[str(i)]["prompt"] state.job = "New prompt: %s\n" % raw_prompt if "seed" in changes_dict[str(i)]: current_seed = changes_dict[str(i)]["seed"] if current_seed.startswith("+"): seed_state = "adding" current_seed = current_seed.strip("+") elif current_seed.startswith("-"): seed_state = "subtracting" current_seed = current_seed.strip("-") else: seed_state = "constant" current_seed = int(current_seed) p.seed = current_seed p.n_iter = 1 p.batch_size = 1 p.do_not_save_grid = True if opts.img2img_color_correction: p.color_corrections = initial_color_corrections wave_progress = float(1)/(float(frames_per_wave - 1))*float(((float(i)%float(frames_per_wave)) + ((float(1)/float(180))*denoising_strength_change_offset))) print(wave_progress) new_prompt = re.sub(wave_completed_regex, lambda x: set_weights(x, wave_progress), raw_prompt) new_prompt = re.sub(wave_remaining_regex, lambda x: set_weights(x, 1 - wave_progress), new_prompt) p.prompt = new_prompt print(new_prompt) denoising_strength_change_rate = 180/frames_per_wave cos = abs(math.cos(math.radians(i*denoising_strength_change_rate + denoising_strength_change_offset))) p.denoising_strength = initial_denoising_strength + denoising_strength_change_amplitude - (cos * denoising_strength_change_amplitude) state.job += f"Iteration {i + 1}/{frames}, batch {n + 1}/{batch_count}. Denoising Strength: {p.denoising_strength}" processed = processing.process_images(p) if initial_seed is None: initial_seed = processed.seed initial_info = processed.info init_img = processed.images[0] p.init_images = [init_img] if seed_state == "adding": p.seed = processed.seed + 1 elif seed_state == "subtracting": p.seed = processed.seed - 1 image_number = int(initial_image_number + i) images.save_image(init_img, p.outpath_samples, "", processed.seed, processed.prompt, forced_filename=str(image_number)) history.append(init_img) grid = images.image_grid(history, rows=1) if opts.grid_save: images.save_image(grid, p.outpath_grids, "grid", initial_seed, p.prompt, opts.grid_format, info=info, short_filename=not opts.grid_extended_filename, grid=True, p=p) grids.append(grid) all_images += history if opts.return_grid: all_images = grids + all_images if save_video: input_pattern = os.path.join(loopback_wave_images_path, "%d.png") encode_video(input_pattern, initial_image_number, loopback_wave_images_path, video_fps, video_quality, video_encoding, segment_video, video_segment_duration, ffmpeg_path) processed = Processed(p, all_images, initial_seed, initial_info) return processed