File size: 1,917 Bytes
bedfeec | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | import math
from dataclasses import dataclass
import torch
@dataclass
class SchedulerOutput:
prev_sample: torch.FloatTensor
class FlowMatchEulerScheduler:
def __init__(self, num_train_timesteps: int = 1000):
self.num_train_timesteps = num_train_timesteps
self.sigmas = None
self.timesteps = None
self._step_index = None
def set_timesteps(self, sigmas, device):
if isinstance(sigmas, (list, tuple)):
sigmas = torch.tensor(sigmas, dtype=torch.float32)
elif not isinstance(sigmas, torch.Tensor):
sigmas = torch.from_numpy(sigmas).to(dtype=torch.float32)
sigmas = sigmas.to(device=device)
self.timesteps = sigmas * self.num_train_timesteps
self.sigmas = torch.cat([sigmas, torch.zeros(1, device=device)])
self._step_index = None
def step(
self,
model_output: torch.FloatTensor,
timestep: torch.FloatTensor,
sample: torch.FloatTensor,
) -> SchedulerOutput:
if self._step_index is None:
self._step_index = (self.timesteps == timestep).nonzero()
self._step_index = 0 if self._step_index.numel() == 0 else self._step_index[0].item()
sample = sample.to(torch.float32)
sigma = self.sigmas[self._step_index]
sigma_next = self.sigmas[self._step_index + 1]
prev_sample = sample + (sigma_next - sigma) * model_output
prev_sample = prev_sample.to(model_output.dtype)
self._step_index += 1
return SchedulerOutput(prev_sample=prev_sample)
def compute_sway_sigmas(num_steps: int, sway_sampling_coef: float = -1.0):
t = torch.linspace(0, 1, num_steps + 1)
t = t + sway_sampling_coef * (torch.cos(math.pi / 2.0 * t) - 1.0 + t)
sigmas = 1.0 - t
return sigmas
def compute_linear_sigmas(num_steps: int):
return torch.linspace(1.0, 1.0 / num_steps, num_steps)
|