Chaitanya-02 commited on
Commit
3ff558d
·
verified ·
1 Parent(s): 35670be

Update sourcecode.py

Browse files
Files changed (1) hide show
  1. sourcecode.py +10 -334
sourcecode.py CHANGED
@@ -1,29 +1,5 @@
1
- """
2
- Helper scripts for generating synthetic images using diffusion model.
3
- Functions:
4
- - get_top_misclassified
5
- - get_class_list
6
- - generateClassPairs
7
- - outputDirectory
8
- - pipe_img
9
- - createPrompts
10
- - interpolatePrompts
11
- - slerp
12
- - get_middle_elements
13
- - remove_middle
14
- - genClassImg
15
- - getMetadata
16
- - groupbyInterpolation
17
- - ungroupInterpolation
18
- - groupAllbyInterpolation
19
- - getPairIndices
20
- - generateImagesFromDataset
21
- - generateTrace
22
- """
23
-
24
  import json
25
  import os
26
-
27
  import numpy as np
28
  import pandas as pd
29
  import torch
@@ -36,16 +12,7 @@ from torch import nn
36
  from torchmetrics.functional.image import structural_similarity_index_measure as ssim
37
  from torchvision import transforms
38
 
39
-
40
  def get_top_misclassified(val_classifier_json):
41
- """
42
- Retrieves the top misclassified classes from a validation classifier JSON file.
43
- Args:
44
- val_classifier_json (str): The path to the validation classifier JSON file.
45
- Returns:
46
- dict: A dictionary containing the top misclassified classes, where the keys are the class names
47
- and the values are the number of misclassifications.
48
- """
49
  with open(val_classifier_json) as f:
50
  val_output = json.load(f)
51
  val_metrics_df = pd.DataFrame.from_dict(
@@ -58,26 +25,11 @@ def get_top_misclassified(val_classifier_json):
58
 
59
 
60
  def get_class_list(val_classifier_json):
61
- """
62
- Retrieves the list of classes from the given validation classifier JSON file.
63
- Args:
64
- val_classifier_json (str): The path to the validation classifier JSON file.
65
- Returns:
66
- list: A sorted list of class names extracted from the JSON file.
67
- """
68
  with open(val_classifier_json, "r") as f:
69
  data = json.load(f)
70
  return sorted(list(data["val_metrics_details"].keys()))
71
 
72
-
73
  def generateClassPairs(val_classifier_json):
74
- """
75
- Generate pairs of misclassified classes from the given validation classifier JSON.
76
- Args:
77
- val_classifier_json (str): The path to the validation classifier JSON file.
78
- Returns:
79
- list: A sorted list of pairs of misclassified classes.
80
- """
81
  pairs = set()
82
  misclassified_classes = get_top_misclassified(val_classifier_json)
83
  for key, value in misclassified_classes.items():
@@ -85,17 +37,7 @@ def generateClassPairs(val_classifier_json):
85
  pairs.add(tuple(sorted([key, v])))
86
  return sorted(list(pairs))
87
 
88
-
89
  def outputDirectory(class_pairs, synth_path, metadata_path):
90
- """
91
- Creates the output directory structure for the synthesized data.
92
- Args:
93
- class_pairs (list): A list of class pairs.
94
- synth_path (str): The path to the directory where the synthesized data will be stored.
95
- metadata_path (str): The path to the directory where the metadata will be stored.
96
- Returns:
97
- None
98
- """
99
  for id in class_pairs:
100
  class_folder = f"{synth_path}/{id}"
101
  if not (os.path.exists(class_folder)):
@@ -104,7 +46,6 @@ def outputDirectory(class_pairs, synth_path, metadata_path):
104
  os.makedirs(metadata_path)
105
  print("Info: Output directory ready.")
106
 
107
-
108
  def pipe_img(
109
  model_path,
110
  device="cuda",
@@ -115,24 +56,6 @@ def pipe_img(
115
  cpu_offload=False,
116
  scheduler=None,
117
  ):
118
- """
119
- Creates and returns an image-to-image pipeline for stable diffusion.
120
- Args:
121
- model_path (str): The path to the pretrained model.
122
- device (str, optional): The device to use for computation. Defaults to "cuda".
123
- apply_optimization (bool, optional): Whether to apply optimization techniques. Defaults to True.
124
- use_torchcompile (bool, optional): Whether to use torchcompile for model compilation. Defaults to False.
125
- ci_cb (tuple, optional): A tuple containing the cache interval and cache branch ID. Defaults to (5, 1).
126
- use_safetensors (bool, optional): Whether to use safetensors. Defaults to None.
127
- cpu_offload (bool, optional): Whether to enable CPU offloading. Defaults to False.
128
- scheduler (LMSDiscreteScheduler, optional): The scheduler for the pipeline. Defaults to None.
129
- Returns:
130
- StableDiffusionImg2ImgPipeline: The image-to-image pipeline for stable diffusion.
131
- """
132
- ###############################
133
- # Reference:
134
- # Akimov, R. (2024) Images Interpolation with Stable Diffusion - Hugging Face Open-Source AI Cookbook. Available at: https://huggingface.co/learn/cookbook/en/stable_diffusion_interpolation (Accessed: 4 June 2024).
135
- ###############################
136
  if scheduler is None:
137
  scheduler = LMSDiscreteScheduler(
138
  beta_start=0.00085,
@@ -150,16 +73,14 @@ def pipe_img(
150
  if cpu_offload:
151
  pipe.enable_model_cpu_offload()
152
  if apply_optimization:
153
- # tomesd.apply_patch(pipe, ratio=0.5)
154
  helper = DeepCacheSDHelper(pipe=pipe)
155
  cache_interval, cache_branch_id = ci_cb
156
  helper.set_params(
157
  cache_interval=cache_interval, cache_branch_id=cache_branch_id
158
- ) # lower is faster but lower quality
159
  helper.enable()
160
- # if torch.cuda.is_available():
161
- # pipe.to("cuda")
162
- # pipe.enable_xformers_memory_efficient_attention()
163
  if use_torchcompile:
164
  pipe.unet = torch.compile(pipe.unet, mode="reduce-overhead", fullgraph=True)
165
  return pipe
@@ -171,18 +92,7 @@ def createPrompts(
171
  use_default_negative_prompt=False,
172
  negative_prompt=None,
173
  ):
174
- """
175
- Create prompts for image generation.
176
- Args:
177
- class_name_pairs (list): A list of two class names.
178
- prompt_structure (str, optional): The structure of the prompt. Defaults to "a photo of a <class_name>".
179
- use_default_negative_prompt (bool, optional): Whether to use the default negative prompt. Defaults to False.
180
- negative_prompt (str, optional): The negative prompt to steer the generation away from certain features.
181
- Returns:
182
- tuple: A tuple containing two lists - prompts and negative_prompts.
183
- prompts (list): Text prompts that describe the desired output image.
184
- negative_prompts (list): Negative prompts that can be used to steer the generation away from certain features.
185
- """
186
  if prompt_structure is None:
187
  prompt_structure = "a photo of a <class_name>"
188
  elif "<class_name>" not in prompt_structure:
@@ -204,11 +114,10 @@ def createPrompts(
204
  print("Info: Negative prompt not provided, returning as None.")
205
  return prompts, None
206
  else:
207
- # Negative prompts that can be used to steer the generation away from certain features.
208
  negative_prompts = [negative_prompt] * len(prompts)
209
  return prompts, negative_prompts
210
 
211
-
212
  def interpolatePrompts(
213
  prompts,
214
  pipeline,
@@ -217,51 +126,10 @@ def interpolatePrompts(
217
  remove_n_middle=0,
218
  device="cuda",
219
  ):
220
- """
221
- Interpolates prompts by generating intermediate embeddings between pairs of prompts.
222
- Args:
223
- prompts (List[str]): A list of prompts to be interpolated.
224
- pipeline: The pipeline object containing the tokenizer and text encoder.
225
- num_interpolation_steps (int): The number of interpolation steps between each pair of prompts.
226
- sample_mid_interpolation (int): The number of intermediate embeddings to sample from the middle of the interpolated prompts.
227
- remove_n_middle (int, optional): The number of middle embeddings to remove from the interpolated prompts. Defaults to 0.
228
- device (str, optional): The device to run the interpolation on. Defaults to "cuda".
229
- Returns:
230
- interpolated_prompt_embeds (torch.Tensor): The interpolated prompt embeddings.
231
- prompt_metadata (dict): Metadata about the interpolation process, including similarity scores and nearest class information.
232
- e.g. if num_interpolation_steps = 10, sample_mid_interpolation = 6, remove_n_middle = 2
233
- Interpolated: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
234
- Sampled: [2, 3, 4, 5, 6, 7]
235
- Removed: x x
236
- Returns: [2, 3, 6, 7]
237
- """
238
-
239
- ###############################
240
- # Reference:
241
- # Akimov, R. (2024) Images Interpolation with Stable Diffusion - Hugging Face Open-Source AI Cookbook. Available at: https://huggingface.co/learn/cookbook/en/stable_diffusion_interpolation (Accessed: 4 June 2024).
242
- ###############################
243
-
244
  def slerp(v0, v1, num, t0=0, t1=1):
245
- """
246
- Performs spherical linear interpolation between two vectors.
247
- Args:
248
- v0 (torch.Tensor): The starting vector.
249
- v1 (torch.Tensor): The ending vector.
250
- num (int): The number of interpolation points.
251
- t0 (float, optional): The starting time. Defaults to 0.
252
- t1 (float, optional): The ending time. Defaults to 1.
253
- Returns:
254
- torch.Tensor: The interpolated vectors.
255
- """
256
- ###############################
257
- # Reference:
258
- # Karpathy, A. (2022) hacky stablediffusion code for generating videos, Gist. Available at: https://gist.github.com/karpathy/00103b0037c5aaea32fe1da1af553355 (Accessed: 4 June 2024).
259
- ###############################
260
  v0 = v0.detach().cpu().numpy()
261
  v1 = v1.detach().cpu().numpy()
262
-
263
  def interpolation(t, v0, v1, DOT_THRESHOLD=0.9995):
264
- """helper function to spherically interpolate two arrays v1 v2"""
265
  dot = np.sum(v0 * v1 / (np.linalg.norm(v0) * np.linalg.norm(v1)))
266
  if np.abs(dot) > DOT_THRESHOLD:
267
  v2 = (1 - t) * v0 + t * v1
@@ -274,28 +142,11 @@ def interpolatePrompts(
274
  s1 = sin_theta_t / sin_theta_0
275
  v2 = s0 * v0 + s1 * v1
276
  return v2
277
-
278
  t = np.linspace(t0, t1, num)
279
-
280
  v3 = torch.tensor(np.array([interpolation(t[i], v0, v1) for i in range(num)]))
281
-
282
  return v3
283
 
284
  def get_middle_elements(lst, n):
285
- """
286
- Returns a tuple containing a sublist of the middle elements of the given list `lst` and a range of indices of those elements.
287
- Args:
288
- lst (list): The list from which to extract the middle elements.
289
- n (int): The number of middle elements to extract.
290
- Returns:
291
- tuple: A tuple containing the sublist of middle elements and a range of indices.
292
- Raises:
293
- None
294
- Examples:
295
- lst = [1, 2, 3, 4, 5]
296
- get_middle_elements(lst, 3)
297
- ([2, 3, 4], range(2, 5))
298
- """
299
  if n % 2 == 0: # Even number of elements
300
  middle_index = len(lst) // 2 - 1
301
  start = middle_index - n // 2 + 1
@@ -308,35 +159,18 @@ def interpolatePrompts(
308
  return lst[start:end], range(start, end)
309
 
310
  def remove_middle(data, n):
311
- """
312
- Remove the middle n elements from a list.
313
- Args:
314
- data (list): The input list.
315
- n (int): The number of elements to remove from the middle of the list.
316
- Returns:
317
- list: The modified list with the middle n elements removed.
318
- Raises:
319
- ValueError: If n is negative or greater than the length of the list.
320
- """
321
  if n < 0 or n > len(data):
322
  raise ValueError(
323
  "Invalid value for n. It should be non-negative and less than half the list length"
324
  )
325
-
326
- # Find the middle index
327
  middle = len(data) // 2
328
-
329
- # Create slices to exclude the middle n elements
330
  if n == 1:
331
  return data[:middle] + data[middle + 1 :]
332
  elif n % 2 == 0:
333
  return data[: middle - n // 2] + data[middle + n // 2 :]
334
  else:
335
  return data[: middle - n // 2] + data[middle + n // 2 + 1 :]
336
-
337
  batch_size = len(prompts)
338
-
339
- # Tokenizing and encoding prompts into embeddings.
340
  prompts_tokens = pipeline.tokenizer(
341
  prompts,
342
  padding="max_length",
@@ -345,25 +179,19 @@ def interpolatePrompts(
345
  return_tensors="pt",
346
  )
347
  prompts_embeds = pipeline.text_encoder(prompts_tokens.input_ids.to(device))[0]
348
-
349
- # Interpolating between embeddings pairs for the given number of interpolation steps.
350
  interpolated_prompt_embeds = []
351
-
352
  for i in range(batch_size - 1):
353
  interpolated_prompt_embeds.append(
354
  slerp(prompts_embeds[i], prompts_embeds[i + 1], num_interpolation_steps)
355
  )
356
-
357
  full_interpolated_prompt_embeds = interpolated_prompt_embeds[:]
358
  interpolated_prompt_embeds[0], sample_range = get_middle_elements(
359
  interpolated_prompt_embeds[0], sample_mid_interpolation
360
  )
361
-
362
  if remove_n_middle > 0:
363
  interpolated_prompt_embeds[0] = remove_middle(
364
  interpolated_prompt_embeds[0], remove_n_middle
365
  )
366
-
367
  prompt_metadata = dict()
368
  similarity = nn.CosineSimilarity(dim=-1, eps=1e-6)
369
  for i in range(num_interpolation_steps):
@@ -384,7 +212,6 @@ def interpolatePrompts(
384
  .item()
385
  )
386
  relative_distance = class1_sim / (class1_sim + class2_sim)
387
-
388
  prompt_metadata[i] = {
389
  "selected": i in sample_range,
390
  "similarity": {
@@ -398,8 +225,7 @@ def interpolatePrompts(
398
 
399
  interpolated_prompt_embeds = torch.cat(interpolated_prompt_embeds, dim=0).to(device)
400
  return interpolated_prompt_embeds, prompt_metadata
401
-
402
-
403
  def genClassImg(
404
  pipeline,
405
  pos_embed,
@@ -413,24 +239,6 @@ def genClassImg(
413
  num_inference_steps=25,
414
  guidance_scale=7.5,
415
  ):
416
- """
417
- Generate class image using the given inputs.
418
- Args:
419
- pipeline: The pipeline object used for image generation.
420
- pos_embed: The positive embedding for the class.
421
- neg_embed: The negative embedding for the class (optional).
422
- input_image: The input image for guidance (optional).
423
- generator: The generator model used for image generation.
424
- latents: The latent vectors used for image generation.
425
- num_imgs: The number of images to generate (default is 1).
426
- height: The height of the generated images (default is 512).
427
- width: The width of the generated images (default is 512).
428
- num_inference_steps: The number of inference steps for image generation (default is 25).
429
- guidance_scale: The scale factor for guidance (default is 7.5).
430
- Returns:
431
- The generated class image.
432
- """
433
-
434
  if neg_embed is not None:
435
  npe = neg_embed[None, ...]
436
  else:
@@ -449,7 +257,6 @@ def genClassImg(
449
  image=input_image,
450
  ).images[0]
451
 
452
-
453
  def getMetadata(
454
  class_pairs,
455
  path,
@@ -469,32 +276,7 @@ def getMetadata(
469
  save_json=True,
470
  save_path=".",
471
  ):
472
- """
473
- Generate metadata for the given parameters.
474
- Args:
475
- class_pairs (list): List of class pairs.
476
- path (str): Path to the data.
477
- seed (int): Seed value for randomization.
478
- guidance_scale (float): Scale factor for guidance.
479
- num_inference_steps (int): Number of inference steps.
480
- num_interpolation_steps (int): Number of interpolation steps.
481
- sample_mid_interpolation (bool): Flag to sample mid-interpolation.
482
- height (int): Height of the image.
483
- width (int): Width of the image.
484
- prompts (list): List of prompts.
485
- negative_prompts (list): List of negative prompts.
486
- pipeline (object): Pipeline object.
487
- prompt_metadata (dict): Metadata for prompts.
488
- negative_prompt_metadata (dict): Metadata for negative prompts.
489
- ssim_metadata (dict, optional): SSIM scores metadata. Defaults to None.
490
- save_json (bool, optional): Flag to save metadata as JSON. Defaults to True.
491
- save_path (str, optional): Path to save the JSON file. Defaults to ".".
492
- Returns:
493
- dict: Generated metadata.
494
- """
495
-
496
  metadata = dict()
497
-
498
  metadata["class_pairs"] = class_pairs
499
  metadata["path"] = path
500
  metadata["seed"] = seed
@@ -524,36 +306,18 @@ def getMetadata(
524
  json.dump(metadata, f, indent=4)
525
  return metadata
526
 
527
-
528
  def groupbyInterpolation(dir_to_classfolder):
529
- """
530
- Group files in a directory by interpolation step.
531
- Args:
532
- dir_to_classfolder (str): The path to the directory containing the files.
533
- Returns:
534
- None
535
- """
536
  files = [
537
  (f.split(sep="_")[1].split(sep=".")[0], os.path.join(dir_to_classfolder, f))
538
  for f in os.listdir(dir_to_classfolder)
539
  ]
540
- # create a subfolder for each step of the interpolation
541
  for interpolation_step, file_path in files:
542
  new_dir = os.path.join(dir_to_classfolder, interpolation_step)
543
  if not os.path.exists(new_dir):
544
  os.makedirs(new_dir)
545
  os.rename(file_path, os.path.join(new_dir, os.path.basename(file_path)))
546
 
547
-
548
  def ungroupInterpolation(dir_to_classfolder):
549
- """
550
- Moves all files from subdirectories within `dir_to_classfolder` to `dir_to_classfolder` itself,
551
- and then removes the subdirectories.
552
- Args:
553
- dir_to_classfolder (str): The path to the directory containing the subdirectories.
554
- Returns:
555
- None
556
- """
557
  for interpolation_step in os.listdir(dir_to_classfolder):
558
  if os.path.isdir(os.path.join(dir_to_classfolder, interpolation_step)):
559
  for f in os.listdir(os.path.join(dir_to_classfolder, interpolation_step)):
@@ -563,21 +327,13 @@ def ungroupInterpolation(dir_to_classfolder):
563
  )
564
  os.rmdir(os.path.join(dir_to_classfolder, interpolation_step))
565
 
566
-
567
  def groupAllbyInterpolation(
568
  data_path,
569
  group=True,
570
  fn_group=groupbyInterpolation,
571
  fn_ungroup=ungroupInterpolation,
572
  ):
573
- """
574
- Group or ungroup all data classes by interpolation.
575
- Args:
576
- data_path (str): The path to the data.
577
- group (bool, optional): Whether to group the data. Defaults to True.
578
- fn_group (function, optional): The function to use for grouping. Defaults to groupbyInterpolation.
579
- fn_ungroup (function, optional): The function to use for ungrouping. Defaults to ungroupInterpolation.
580
- """
581
  data_classes = sorted(os.listdir(data_path))
582
  if group:
583
  fn = fn_group
@@ -589,17 +345,7 @@ def groupAllbyInterpolation(
589
  fn(c_path)
590
  print(f"Processed {c}")
591
 
592
-
593
  def getPairIndices(subset_len, total_pair_count=1, seed=None):
594
- """
595
- Generate pairs of indices for a given subset length.
596
- Args:
597
- subset_len (int): The length of the subset.
598
- total_pair_count (int, optional): The total number of pairs to generate. Defaults to 1.
599
- seed (int, optional): The seed value for the random number generator. Defaults to None.
600
- Returns:
601
- list: A list of pairs of indices.
602
- """
603
  rng = np.random.default_rng(seed)
604
  group_size = (subset_len + total_pair_count - 1) // total_pair_count
605
  numbers = list(range(subset_len))
@@ -611,7 +357,6 @@ def getPairIndices(subset_len, total_pair_count=1, seed=None):
611
  groups = numbers[: group_size * total_pair_count].reshape(-1, group_size)
612
  return groups.tolist()
613
 
614
-
615
  def generateImagesFromDataset(
616
  img_subsets,
617
  class_iterables,
@@ -631,31 +376,6 @@ def generateImagesFromDataset(
631
  device="cuda",
632
  return_images=False,
633
  ):
634
- """
635
- Generates images from a dataset using the given parameters.
636
- Args:
637
- img_subsets (dict): A dictionary containing image subsets for each class.
638
- class_iterables (dict): A dictionary containing iterable objects for each class.
639
- pipeline (object): The pipeline object used for image generation.
640
- interpolated_prompt_embeds (list): A list of interpolated prompt embeddings.
641
- interpolated_negative_prompts_embeds (list): A list of interpolated negative prompt embeddings.
642
- num_inference_steps (int): The number of inference steps for image generation.
643
- guidance_scale (float): The scale factor for guidance loss during image generation.
644
- height (int, optional): The height of the generated images. Defaults to 512.
645
- width (int, optional): The width of the generated images. Defaults to 512.
646
- seed (int, optional): The seed value for random number generation. Defaults to None.
647
- save_path (str, optional): The path to save the generated images. Defaults to ".".
648
- class_pairs (tuple, optional): A tuple containing pairs of class identifiers. Defaults to ("0", "1").
649
- save_image (bool, optional): Whether to save the generated images. Defaults to True.
650
- image_type (str, optional): The file format of the saved images. Defaults to "jpg".
651
- interpolate_range (str, optional): The range of interpolation for prompt embeddings.
652
- Possible values are "full", "nearest", or "furthest". Defaults to "full".
653
- device (str, optional): The device to use for image generation. Defaults to "cuda".
654
- return_images (bool, optional): Whether to return the generated images. Defaults to False.
655
- Returns:
656
- dict or tuple: If return_images is True, returns a dictionary containing the generated images for each class and a dictionary containing the SSIM scores for each class and interpolation step.
657
- If return_images is False, returns a dictionary containing the SSIM scores for each class and interpolation step.
658
- """
659
  if interpolate_range == "nearest":
660
  nearest_half = True
661
  furthest_half = False
@@ -675,14 +395,12 @@ def generateImagesFromDataset(
675
  (1, pipeline.unet.config.in_channels, height // 8, width // 8),
676
  generator=generator,
677
  ).to(device)
678
-
679
  embed_len = len(interpolated_prompt_embeds)
680
  embed_pairs = zip(interpolated_prompt_embeds, interpolated_negative_prompts_embeds)
681
  embed_pairs_list = list(embed_pairs)
682
  if return_images:
683
  class_images = dict()
684
  class_ssim = dict()
685
-
686
  if nearest_half or furthest_half:
687
  if nearest_half:
688
  steps_range = (range(0, embed_len // 2), range(embed_len // 2, embed_len))
@@ -694,7 +412,6 @@ def generateImagesFromDataset(
694
  else:
695
  steps_range = (range(embed_len), range(embed_len))
696
  mutiplier = 1
697
-
698
  for class_iter, class_id in enumerate(class_pairs):
699
  if return_images:
700
  class_images[class_id] = list()
@@ -702,20 +419,14 @@ def generateImagesFromDataset(
702
  i: {"ssim_sum": 0, "ssim_count": 0, "ssim_avg": 0} for i in range(embed_len)
703
  }
704
  subset_len = len(img_subsets[class_id])
705
- # to efficiently randomize the steps to interpolate for each image in the class, group_map is used
706
- # group_map: index is the image id, element is the group id
707
- # steps_range[class_iter] determines the range of steps to interpolate for the class,
708
- # so the first half of the steps are for the first class and so on. range(0,7) and range(8,15) for 16 steps
709
- # then the rest is to multiply the steps to cover the whole subset + remainder
710
  group_map = (
711
  list(steps_range[class_iter]) * mutiplier * (subset_len // embed_len + 1)
712
  )
713
  rng.shuffle(
714
  group_map
715
- ) # shuffle the steps to interpolate for each image, position in the group_map is mapped to the image id
716
-
717
  iter_indices = class_iterables[class_id].pop()
718
- # generate images for each image in the class, randomly selecting an interpolated step
719
  for image_id in iter_indices:
720
  img, trg = img_subsets[class_id][image_id]
721
  input_image = img.unsqueeze(0)
@@ -756,21 +467,17 @@ def generateImagesFromDataset(
756
  generated_image.save(
757
  f"{save_path}/{class_id}/{seed}-{image_id}_{interpolate_step}.{image_type}"
758
  )
759
-
760
- # calculate ssim avg for the class
761
  for i_step in range(embed_len):
762
  if class_ssim[class_id][i_step]["ssim_count"] > 0:
763
  class_ssim[class_id][i_step]["ssim_avg"] = (
764
  class_ssim[class_id][i_step]["ssim_sum"]
765
  / class_ssim[class_id][i_step]["ssim_count"]
766
  )
767
-
768
  if return_images:
769
  return class_images, class_ssim
770
  else:
771
  return class_ssim
772
 
773
-
774
  def generateTrace(
775
  prompts,
776
  img_subsets,
@@ -785,24 +492,6 @@ def generateTrace(
785
  interpolate_range="full",
786
  save_prompt_embeds=False,
787
  ):
788
- """
789
- Generate a trace dictionary containing information about the generated images.
790
- Args:
791
- prompts (list): List of prompt texts.
792
- img_subsets (dict): Dictionary containing image subsets for each class.
793
- class_iterables (dict): Dictionary containing iterable objects for each class.
794
- interpolated_prompt_embeds (torch.Tensor): Tensor containing interpolated prompt embeddings.
795
- interpolated_negative_prompts_embeds (torch.Tensor): Tensor containing interpolated negative prompt embeddings.
796
- subset_indices (dict): Dictionary containing indices of subsets for each class.
797
- seed (int, optional): Seed value for random number generation. Defaults to None.
798
- save_path (str, optional): Path to save the generated images. Defaults to ".".
799
- class_pairs (tuple, optional): Tuple containing class pairs. Defaults to ("0", "1").
800
- image_type (str, optional): Type of the generated images. Defaults to "jpg".
801
- interpolate_range (str, optional): Range of interpolation. Defaults to "full".
802
- save_prompt_embeds (bool, optional): Flag to save prompt embeddings. Defaults to False.
803
- Returns:
804
- dict: Trace dictionary containing information about the generated images.
805
- """
806
  trace_dict = {
807
  "class_pairs": list(),
808
  "class_id": list(),
@@ -815,7 +504,6 @@ def generateTrace(
815
  "output_file_path": list(),
816
  "input_prompts_embed": list(),
817
  }
818
-
819
  if interpolate_range == "nearest":
820
  nearest_half = True
821
  furthest_half = False
@@ -825,7 +513,6 @@ def generateTrace(
825
  else:
826
  nearest_half = False
827
  furthest_half = False
828
-
829
  if seed is None:
830
  seed = torch.Generator().seed()
831
  rng = np.random.default_rng(seed)
@@ -836,7 +523,6 @@ def generateTrace(
836
  interpolated_negative_prompts_embeds.cpu().numpy(),
837
  )
838
  embed_pairs_list = list(embed_pairs)
839
-
840
  if nearest_half or furthest_half:
841
  if nearest_half:
842
  steps_range = (range(0, embed_len // 2), range(embed_len // 2, embed_len))
@@ -848,24 +534,15 @@ def generateTrace(
848
  else:
849
  steps_range = (range(embed_len), range(embed_len))
850
  mutiplier = 1
851
-
852
  for class_iter, class_id in enumerate(class_pairs):
853
-
854
  subset_len = len(img_subsets[class_id])
855
- # to efficiently randomize the steps to interpolate for each image in the class, group_map is used
856
- # group_map: index is the image id, element is the group id
857
- # steps_range[class_iter] determines the range of steps to interpolate for the class,
858
- # so the first half of the steps are for the first class and so on. range(0,7) and range(8,15) for 16 steps
859
- # then the rest is to multiply the steps to cover the whole subset + remainder
860
  group_map = (
861
  list(steps_range[class_iter]) * mutiplier * (subset_len // embed_len + 1)
862
  )
863
  rng.shuffle(
864
  group_map
865
- ) # shuffle the steps to interpolate for each image, position in the group_map is mapped to the image id
866
-
867
  iter_indices = class_iterables[class_id].pop()
868
- # generate images for each image in the class, randomly selecting an interpolated step
869
  for image_id in iter_indices:
870
  class_ds = img_subsets[class_id]
871
  interpolate_step = group_map[image_id]
@@ -878,7 +555,6 @@ def generateTrace(
878
  input_prompts_embed = embed_pairs_list[interpolate_step]
879
  else:
880
  input_prompts_embed = None
881
-
882
  trace_dict["class_pairs"].append(class_pairs)
883
  trace_dict["class_id"].append(class_id)
884
  trace_dict["image_id"].append(image_id)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
  import json
2
  import os
 
3
  import numpy as np
4
  import pandas as pd
5
  import torch
 
12
  from torchmetrics.functional.image import structural_similarity_index_measure as ssim
13
  from torchvision import transforms
14
 
 
15
  def get_top_misclassified(val_classifier_json):
 
 
 
 
 
 
 
 
16
  with open(val_classifier_json) as f:
17
  val_output = json.load(f)
18
  val_metrics_df = pd.DataFrame.from_dict(
 
25
 
26
 
27
  def get_class_list(val_classifier_json):
 
 
 
 
 
 
 
28
  with open(val_classifier_json, "r") as f:
29
  data = json.load(f)
30
  return sorted(list(data["val_metrics_details"].keys()))
31
 
 
32
  def generateClassPairs(val_classifier_json):
 
 
 
 
 
 
 
33
  pairs = set()
34
  misclassified_classes = get_top_misclassified(val_classifier_json)
35
  for key, value in misclassified_classes.items():
 
37
  pairs.add(tuple(sorted([key, v])))
38
  return sorted(list(pairs))
39
 
 
40
  def outputDirectory(class_pairs, synth_path, metadata_path):
 
 
 
 
 
 
 
 
 
41
  for id in class_pairs:
42
  class_folder = f"{synth_path}/{id}"
43
  if not (os.path.exists(class_folder)):
 
46
  os.makedirs(metadata_path)
47
  print("Info: Output directory ready.")
48
 
 
49
  def pipe_img(
50
  model_path,
51
  device="cuda",
 
56
  cpu_offload=False,
57
  scheduler=None,
58
  ):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
59
  if scheduler is None:
60
  scheduler = LMSDiscreteScheduler(
61
  beta_start=0.00085,
 
73
  if cpu_offload:
74
  pipe.enable_model_cpu_offload()
75
  if apply_optimization:
76
+
77
  helper = DeepCacheSDHelper(pipe=pipe)
78
  cache_interval, cache_branch_id = ci_cb
79
  helper.set_params(
80
  cache_interval=cache_interval, cache_branch_id=cache_branch_id
81
+ )
82
  helper.enable()
83
+
 
 
84
  if use_torchcompile:
85
  pipe.unet = torch.compile(pipe.unet, mode="reduce-overhead", fullgraph=True)
86
  return pipe
 
92
  use_default_negative_prompt=False,
93
  negative_prompt=None,
94
  ):
95
+
 
 
 
 
 
 
 
 
 
 
 
96
  if prompt_structure is None:
97
  prompt_structure = "a photo of a <class_name>"
98
  elif "<class_name>" not in prompt_structure:
 
114
  print("Info: Negative prompt not provided, returning as None.")
115
  return prompts, None
116
  else:
117
+
118
  negative_prompts = [negative_prompt] * len(prompts)
119
  return prompts, negative_prompts
120
 
 
121
  def interpolatePrompts(
122
  prompts,
123
  pipeline,
 
126
  remove_n_middle=0,
127
  device="cuda",
128
  ):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
129
  def slerp(v0, v1, num, t0=0, t1=1):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
130
  v0 = v0.detach().cpu().numpy()
131
  v1 = v1.detach().cpu().numpy()
 
132
  def interpolation(t, v0, v1, DOT_THRESHOLD=0.9995):
 
133
  dot = np.sum(v0 * v1 / (np.linalg.norm(v0) * np.linalg.norm(v1)))
134
  if np.abs(dot) > DOT_THRESHOLD:
135
  v2 = (1 - t) * v0 + t * v1
 
142
  s1 = sin_theta_t / sin_theta_0
143
  v2 = s0 * v0 + s1 * v1
144
  return v2
 
145
  t = np.linspace(t0, t1, num)
 
146
  v3 = torch.tensor(np.array([interpolation(t[i], v0, v1) for i in range(num)]))
 
147
  return v3
148
 
149
  def get_middle_elements(lst, n):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
150
  if n % 2 == 0: # Even number of elements
151
  middle_index = len(lst) // 2 - 1
152
  start = middle_index - n // 2 + 1
 
159
  return lst[start:end], range(start, end)
160
 
161
  def remove_middle(data, n):
 
 
 
 
 
 
 
 
 
 
162
  if n < 0 or n > len(data):
163
  raise ValueError(
164
  "Invalid value for n. It should be non-negative and less than half the list length"
165
  )
 
 
166
  middle = len(data) // 2
 
 
167
  if n == 1:
168
  return data[:middle] + data[middle + 1 :]
169
  elif n % 2 == 0:
170
  return data[: middle - n // 2] + data[middle + n // 2 :]
171
  else:
172
  return data[: middle - n // 2] + data[middle + n // 2 + 1 :]
 
173
  batch_size = len(prompts)
 
 
174
  prompts_tokens = pipeline.tokenizer(
175
  prompts,
176
  padding="max_length",
 
179
  return_tensors="pt",
180
  )
181
  prompts_embeds = pipeline.text_encoder(prompts_tokens.input_ids.to(device))[0]
 
 
182
  interpolated_prompt_embeds = []
 
183
  for i in range(batch_size - 1):
184
  interpolated_prompt_embeds.append(
185
  slerp(prompts_embeds[i], prompts_embeds[i + 1], num_interpolation_steps)
186
  )
 
187
  full_interpolated_prompt_embeds = interpolated_prompt_embeds[:]
188
  interpolated_prompt_embeds[0], sample_range = get_middle_elements(
189
  interpolated_prompt_embeds[0], sample_mid_interpolation
190
  )
 
191
  if remove_n_middle > 0:
192
  interpolated_prompt_embeds[0] = remove_middle(
193
  interpolated_prompt_embeds[0], remove_n_middle
194
  )
 
195
  prompt_metadata = dict()
196
  similarity = nn.CosineSimilarity(dim=-1, eps=1e-6)
197
  for i in range(num_interpolation_steps):
 
212
  .item()
213
  )
214
  relative_distance = class1_sim / (class1_sim + class2_sim)
 
215
  prompt_metadata[i] = {
216
  "selected": i in sample_range,
217
  "similarity": {
 
225
 
226
  interpolated_prompt_embeds = torch.cat(interpolated_prompt_embeds, dim=0).to(device)
227
  return interpolated_prompt_embeds, prompt_metadata
228
+
 
229
  def genClassImg(
230
  pipeline,
231
  pos_embed,
 
239
  num_inference_steps=25,
240
  guidance_scale=7.5,
241
  ):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
242
  if neg_embed is not None:
243
  npe = neg_embed[None, ...]
244
  else:
 
257
  image=input_image,
258
  ).images[0]
259
 
 
260
  def getMetadata(
261
  class_pairs,
262
  path,
 
276
  save_json=True,
277
  save_path=".",
278
  ):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
279
  metadata = dict()
 
280
  metadata["class_pairs"] = class_pairs
281
  metadata["path"] = path
282
  metadata["seed"] = seed
 
306
  json.dump(metadata, f, indent=4)
307
  return metadata
308
 
 
309
  def groupbyInterpolation(dir_to_classfolder):
 
 
 
 
 
 
 
310
  files = [
311
  (f.split(sep="_")[1].split(sep=".")[0], os.path.join(dir_to_classfolder, f))
312
  for f in os.listdir(dir_to_classfolder)
313
  ]
 
314
  for interpolation_step, file_path in files:
315
  new_dir = os.path.join(dir_to_classfolder, interpolation_step)
316
  if not os.path.exists(new_dir):
317
  os.makedirs(new_dir)
318
  os.rename(file_path, os.path.join(new_dir, os.path.basename(file_path)))
319
 
 
320
  def ungroupInterpolation(dir_to_classfolder):
 
 
 
 
 
 
 
 
321
  for interpolation_step in os.listdir(dir_to_classfolder):
322
  if os.path.isdir(os.path.join(dir_to_classfolder, interpolation_step)):
323
  for f in os.listdir(os.path.join(dir_to_classfolder, interpolation_step)):
 
327
  )
328
  os.rmdir(os.path.join(dir_to_classfolder, interpolation_step))
329
 
 
330
  def groupAllbyInterpolation(
331
  data_path,
332
  group=True,
333
  fn_group=groupbyInterpolation,
334
  fn_ungroup=ungroupInterpolation,
335
  ):
336
+
 
 
 
 
 
 
 
337
  data_classes = sorted(os.listdir(data_path))
338
  if group:
339
  fn = fn_group
 
345
  fn(c_path)
346
  print(f"Processed {c}")
347
 
 
348
  def getPairIndices(subset_len, total_pair_count=1, seed=None):
 
 
 
 
 
 
 
 
 
349
  rng = np.random.default_rng(seed)
350
  group_size = (subset_len + total_pair_count - 1) // total_pair_count
351
  numbers = list(range(subset_len))
 
357
  groups = numbers[: group_size * total_pair_count].reshape(-1, group_size)
358
  return groups.tolist()
359
 
 
360
  def generateImagesFromDataset(
361
  img_subsets,
362
  class_iterables,
 
376
  device="cuda",
377
  return_images=False,
378
  ):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
379
  if interpolate_range == "nearest":
380
  nearest_half = True
381
  furthest_half = False
 
395
  (1, pipeline.unet.config.in_channels, height // 8, width // 8),
396
  generator=generator,
397
  ).to(device)
 
398
  embed_len = len(interpolated_prompt_embeds)
399
  embed_pairs = zip(interpolated_prompt_embeds, interpolated_negative_prompts_embeds)
400
  embed_pairs_list = list(embed_pairs)
401
  if return_images:
402
  class_images = dict()
403
  class_ssim = dict()
 
404
  if nearest_half or furthest_half:
405
  if nearest_half:
406
  steps_range = (range(0, embed_len // 2), range(embed_len // 2, embed_len))
 
412
  else:
413
  steps_range = (range(embed_len), range(embed_len))
414
  mutiplier = 1
 
415
  for class_iter, class_id in enumerate(class_pairs):
416
  if return_images:
417
  class_images[class_id] = list()
 
419
  i: {"ssim_sum": 0, "ssim_count": 0, "ssim_avg": 0} for i in range(embed_len)
420
  }
421
  subset_len = len(img_subsets[class_id])
422
+
 
 
 
 
423
  group_map = (
424
  list(steps_range[class_iter]) * mutiplier * (subset_len // embed_len + 1)
425
  )
426
  rng.shuffle(
427
  group_map
428
+ )
 
429
  iter_indices = class_iterables[class_id].pop()
 
430
  for image_id in iter_indices:
431
  img, trg = img_subsets[class_id][image_id]
432
  input_image = img.unsqueeze(0)
 
467
  generated_image.save(
468
  f"{save_path}/{class_id}/{seed}-{image_id}_{interpolate_step}.{image_type}"
469
  )
 
 
470
  for i_step in range(embed_len):
471
  if class_ssim[class_id][i_step]["ssim_count"] > 0:
472
  class_ssim[class_id][i_step]["ssim_avg"] = (
473
  class_ssim[class_id][i_step]["ssim_sum"]
474
  / class_ssim[class_id][i_step]["ssim_count"]
475
  )
 
476
  if return_images:
477
  return class_images, class_ssim
478
  else:
479
  return class_ssim
480
 
 
481
  def generateTrace(
482
  prompts,
483
  img_subsets,
 
492
  interpolate_range="full",
493
  save_prompt_embeds=False,
494
  ):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
495
  trace_dict = {
496
  "class_pairs": list(),
497
  "class_id": list(),
 
504
  "output_file_path": list(),
505
  "input_prompts_embed": list(),
506
  }
 
507
  if interpolate_range == "nearest":
508
  nearest_half = True
509
  furthest_half = False
 
513
  else:
514
  nearest_half = False
515
  furthest_half = False
 
516
  if seed is None:
517
  seed = torch.Generator().seed()
518
  rng = np.random.default_rng(seed)
 
523
  interpolated_negative_prompts_embeds.cpu().numpy(),
524
  )
525
  embed_pairs_list = list(embed_pairs)
 
526
  if nearest_half or furthest_half:
527
  if nearest_half:
528
  steps_range = (range(0, embed_len // 2), range(embed_len // 2, embed_len))
 
534
  else:
535
  steps_range = (range(embed_len), range(embed_len))
536
  mutiplier = 1
 
537
  for class_iter, class_id in enumerate(class_pairs):
 
538
  subset_len = len(img_subsets[class_id])
 
 
 
 
 
539
  group_map = (
540
  list(steps_range[class_iter]) * mutiplier * (subset_len // embed_len + 1)
541
  )
542
  rng.shuffle(
543
  group_map
544
+ )
 
545
  iter_indices = class_iterables[class_id].pop()
 
546
  for image_id in iter_indices:
547
  class_ds = img_subsets[class_id]
548
  interpolate_step = group_map[image_id]
 
555
  input_prompts_embed = embed_pairs_list[interpolate_step]
556
  else:
557
  input_prompts_embed = None
 
558
  trace_dict["class_pairs"].append(class_pairs)
559
  trace_dict["class_id"].append(class_id)
560
  trace_dict["image_id"].append(image_id)