import torchdata.datapipes as dp
import json
from PIL import Image
import functools
import numpy as np
import torch
import pickle
import os
import cv2
import random
from torchvision import transforms
from braceexpand import braceexpand
import hydra
from random import choice
import tarfile
from torchdata.datapipes.iter import TarArchiveLoader
from typing import cast, IO, Iterable, Iterator, Optional, Tuple, Dict
from torchdata.datapipes import functional_datapipe
from io import BufferedIOBase
from torchdata.datapipes.utils import StreamWrapper
from torchdata.datapipes.utils.common import validate_pathname_binary_tuple
import warnings
from torchdata.datapipes.iter import IterDataPipe
import pyrootutils
pyrootutils.setup_root(__file__, indicator='.project-root', pythonpath=True)
BOI_TOKEN = ''
EOI_TOKEN = ''
IMG_TOKEN = ''
gen_prompt = [
"Please show me a picture of ",
"Please design an image of ",
"Please produce a photo of ",
"Please generate an image of ",
"Please draw a painting of ",
"I'd like to see a drawing of ",
"I'd love to see an illustration of ",
"I'd like to view an image of ",
"I want to see a picture of ",
"I would like to see a photo of ",
"Show me a photo of ",
"Generate a picture of ",
"Show me a photograph of ",
"Generate an image of ",
"Generate an image: ",
"Generate a picture: ",
"Generate a painting: ",
"Generate a photograph: ",
"Show me a photograph: ",
"Draw a picture: ",
"Draw a painting: ",
"Draw an image: ",
"Can you make an image of ",
"Can you draw a painting of ",
"Can you produce a picture of ",
"Can you generate a photo of ",
"Can you depict a picture of ",
"Can you show me an illustration of ",
]
gen_prompt_response = [
"Here is a picture.",
"I have designed an image.",
"Here is a photo.",
"I have generated an image.",
"Here's a painting.",
"Here's a drawing.",
"Enjoy this illustration.",
"Take a look at this image.",
"Here is a picture.",
"I have created a photo.",
"Enjoy this photo.",
"I have generated a picture.",
"Here is a photograph.",
"Here's an image.",
"Certainly, here's an image.",
"Absolutely, here is a painting.",
"Sure, here is a picture.",
"Of course, here is a photo.",
"Certainly, please enjoy this picture.",
"Sure, please enjoy this illustration.",
"",
]
jdb_filter_vocab = ['watermark', 'watermark,', 'chaos 100', 'chaos 100,']
def filter_data_with_image_ids(item):
if ('images' not in item):
# print(item['__key__'])
# print('filtered because no images')
return False
elif 'input_ids' not in item:
return False
else:
return True
def calculate_new_dimensions(height, width, target_size):
if height < width:
new_height = target_size
new_width = int(width * (target_size / height))
else:
new_width = target_size
new_height = int(height * (target_size / width))
return new_height, new_width
def unwarp_data(item):
unwarpped = {}
for key, value in item.items():
if isinstance(value, dict):
unwarpped.update(value)
elif value is not None:
unwarpped[key] = value
if 'metadata' not in unwarpped:
unwarpped['metadata'] = '{}'
# if '__key__' in unwarpped:
# unwarpped['__key__'] = unwarpped['__key__'].split('/')[-1]
return unwarpped
# def filter_data_with_similarity(item, similarity_thr=0.2, min_resolution=180, min_aspect_ratio=0.666):
def filter_data_with_similarity(item, similarity_thr=0.2, assure_text=True):
if ('images' not in item):
# print(item['__key__'])
# print('filtered because no images')
return False
elif (not item.get('filter_flag', True)):
# print(item['__key__'])
# print('filtered because filter flag.')
return False
elif assure_text and ('text' not in item):
# print(item['__key__'])
# print('filtered because assure_text')
return False
else:
metadata = json.loads(item['metadata'])
if 'all_similarities' in metadata:
similarity = max(metadata['all_similarities'])
elif 'similarity' in metadata:
similarity = metadata['similarity']
elif 'score' in metadata:
similarity = metadata['score']
elif 'SCORE' in metadata:
similarity = metadata['SCORE']
else:
similarity = None
if similarity is not None:
if similarity < similarity_thr:
# print(item['__key__'])
# print('filtered because similarity')
return False
return True
def single_turn_edit_collate(batch):
results = {}
keys = batch[0].keys()
for key in keys:
cur = [batch[i][key] for i in range(len(batch)) if batch[i][key] is not None]
if len(cur) == 0:
results[key] = None
elif isinstance(cur[0], torch.Tensor):
if key in ['embeds_gen_mask', 'embeds_cmp_mask', 'images']:
results[key] = torch.cat(cur, dim=0)
else:
results[key] = torch.stack(cur, dim=0)
else:
results[key] = cur
return results
def decode_t2i_data(item,
image_dir,
tokenizer,
image_transform=None,
sd_image_transform=None,
max_length=128,
min_resolution=400,
instruction_prompt='[INST] {instruction} [/INST]\n',
turn_sep='\n',
system_message='',
min_aspect_ratio=0.666,
num_img_in_tokens=64,
num_img_out_tokens=64):
key, value = item
if 'image' not in value or 'caption' not in value:
return {}
image_path = os.path.join(image_dir, value["image"])
try:
image = Image.open(image_path).convert('RGB')
width, height = image.size
aspect_ratio = height / width
if height < min_resolution or width < min_resolution:
print(f'filtered because resolution: ({width},{height})')
return {}
if aspect_ratio < min_aspect_ratio or aspect_ratio > 1 / min_aspect_ratio:
print(f'filtered because aspect ratio: ({width},{height})')
return {}
### SD related
image_data = {}
if sd_image_transform is not None:
# image_data['original_sizes'] = torch.tensor([height, width])
sd_image_tensor = sd_image_transform(image)
target_size = sd_image_tensor.shape[-2]
target_width, target_height = calculate_new_dimensions(height=height, width=width, target_size=target_size)
y1 = max(0, int(round((target_height - target_size) / 2.0)))
x1 = max(0, int(round((target_width - target_size) / 2.0)))
# image_data['crop_top_lefts'] = torch.tensor([y1, x1])
image_data['time_ids'] = torch.tensor([height, width, y1, x1, target_size, target_size])
image_data['sd_images'] = sd_image_tensor
if image_transform is not None:
image = image_transform(image)
except Exception as e:
print('Error while decode image: ', e)
return {}
input_ids = []
labels = []
input_text = ''
if system_message != '':
if not system_message.endswith('\n'):
system_message += '\n'
input_text += system_message
item_ids = tokenizer.encode(system_message, add_special_tokens=False)
item_labels = [-100] * len(item_ids)
input_ids.extend(item_ids)
labels.extend(item_labels)
caption = value["caption"]
image_cmp_tokens = BOI_TOKEN + ''.join(
[IMG_TOKEN.format(int(item)) for item in range(num_img_in_tokens)]) + EOI_TOKEN
image_gen_tokens = BOI_TOKEN + ''.join(
[IMG_TOKEN.format(int(item)) for item in range(num_img_out_tokens)]) + EOI_TOKEN
instruction = instruction_prompt.format_map({'instruction': caption})
response = image_gen_tokens
images = torch.stack([image], dim=0)
# print(instruction)
item_ids = tokenizer.encode(instruction, add_special_tokens=False)
item_labels = [-100] * len(item_ids)
input_text += instruction
input_ids.extend(item_ids)
labels.extend(item_labels)
item_ids = tokenizer.encode(response, add_special_tokens=False)
item_labels = item_ids
input_text += response
input_ids.extend(item_ids)
labels.extend(item_labels)
input_ids = [tokenizer.bos_token_id] + input_ids + [tokenizer.eos_token_id]
attention_mask = [1] * len(input_ids)
labels = [-100] + labels + [tokenizer.eos_token_id]
boi_token_id = tokenizer.encode(BOI_TOKEN, add_special_tokens=False)[0]
eoi_token_id = tokenizer.encode(EOI_TOKEN, add_special_tokens=False)[0]
ids_cmp_mask = [False] * len(input_ids)
ids_gen_mask = [False] * len(input_ids)
embeds_cmp_mask = [False]
embeds_gen_mask = [True]
# print(len(input_ids))
if len(input_ids) >= max_length:
# input_ids = input_ids[:max_length]
# attention_mask = attention_mask[:max_length]
# labels = labels[:max_length]
# ids_cmp_mask = ids_cmp_mask[:max_length]
# ids_gen_mask = ids_gen_mask[:max_length]
# print('An edit sample has been removed because of max length. input_text: ', input_text)
return {}
else:
padding_length = max_length - len(input_ids)
input_ids = input_ids + [tokenizer.pad_token_id] * padding_length
attention_mask = attention_mask + [0] * padding_length
labels = labels + [-100] * padding_length
ids_cmp_mask = ids_cmp_mask + [False] * padding_length
ids_gen_mask = ids_gen_mask + [False] * padding_length
input_ids = torch.tensor(input_ids, dtype=torch.long)
attention_mask = torch.tensor(attention_mask, dtype=torch.long)
labels = torch.tensor(labels, dtype=torch.long)
ids_cmp_mask = torch.tensor(ids_cmp_mask, dtype=torch.bool)
ids_gen_mask = torch.tensor(ids_gen_mask, dtype=torch.bool)
embeds_cmp_mask = torch.tensor(embeds_cmp_mask) if embeds_cmp_mask is not None else None
embeds_gen_mask = torch.tensor(embeds_gen_mask) if embeds_gen_mask is not None else None
boi_idx = torch.where(input_ids == boi_token_id)[0].tolist()
eoi_idx = torch.where(input_ids == eoi_token_id)[0].tolist()
ids_gen_mask[boi_idx[0] + 1:eoi_idx[0]] = True
labels[boi_idx[0] + 1:eoi_idx[0] + 1] = -100
ret = {
'input_ids': input_ids,
'attention_mask': attention_mask,
'labels': labels,
'ids_gen_mask': ids_gen_mask,
'ids_cmp_mask': ids_cmp_mask,
'embeds_gen_mask': embeds_gen_mask,
'embeds_cmp_mask': embeds_cmp_mask,
'images': images,
'text': input_text,
}
ret.update(image_data)
return ret
def build_t2i_datapipe(data_dir,
image_dir,
tokenizer=None,
max_length=77,
batch_size=None,
min_resolution=180,
image_transform=None,
sd_image_transform=None,
instruction_prompt='[INST] {instruction} [INST]\n',
turn_sep='\n',
system_message='',
min_aspect_ratio=0.666,
num_img_in_tokens=64,
num_img_out_tokens=64,
cycle_count=None):
decode_partial = functools.partial(decode_t2i_data,
image_dir=image_dir,
tokenizer=tokenizer,
image_transform=image_transform,
sd_image_transform=sd_image_transform,
max_length=max_length,
instruction_prompt=instruction_prompt,
turn_sep=turn_sep,
system_message=system_message,
min_resolution=min_resolution,
min_aspect_ratio=min_aspect_ratio,
num_img_in_tokens=num_img_in_tokens,
num_img_out_tokens=num_img_out_tokens)
filter_partial = functools.partial(filter_data_with_image_ids)
if isinstance(data_dir, str):
data_dir = list(braceexpand(data_dir))
datapipe = dp.iter.FileLister(root=data_dir, masks='*.jsonl', recursive=True)
datapipe = datapipe.shuffle()
datapipe = datapipe.cycle(count=cycle_count)
datapipe = datapipe.shuffle()
# datapipe = dp.iter.FileLister(root=data_dir, masks='0000000.tar', recursive=True)
datapipe = datapipe.sharding_filter()
# datapipe = datapipe.sharding_round_robin_dispatch(SHARDING_PRIORITIES.MULTIPROCESSING)
datapipe = datapipe.open_files(mode='r')
datapipe = datapipe.parse_jsonl_files()
datapipe = datapipe.map(decode_partial)
datapipe = datapipe.filter(filter_partial)
# datapipe = datapipe.shuffle(buffer_size=1024)
if batch_size is not None:
datapipe = datapipe.batch(batch_size)
datapipe = datapipe.collate(single_turn_edit_collate)
return datapipe
def decode_long_story_data(item,
image_dir,
tokenizer,
story_len,
image_transform=None,
sd_image_transform=None,
max_length=128,
min_resolution=400,
instruction_prompt='{instruction}',
turn_sep='\n',
system_message='',
min_aspect_ratio=0.666,
num_img_in_tokens=64,
num_img_out_tokens=64, ):
key, value = item
if 'images' not in value or 'captions' not in value:
return {}
image_paths = [os.path.join(image_dir, image_path) for image_path in value["images"]]
# assert len(image_paths) == story_len
story_len = len(image_paths)
num_image_given = random.randint(0, story_len - 2)
try:
images = []
for image_path in image_paths:
image = Image.open(image_path).convert('RGB')
images.append(image)
width, height = image.size
aspect_ratio = height / width
if height < min_resolution or width < min_resolution:
print(f'filtered because resolution: ({width},{height})')
return {}
if aspect_ratio < min_aspect_ratio or aspect_ratio > 1 / min_aspect_ratio:
print(f'filtered because aspect ratio: ({width},{height})')
return {}
image_data = {}
sd_image = images[num_image_given + 1]
if sd_image_transform is not None:
# image_data['original_sizes'] = torch.tensor([height, width])
sd_image_tensor = sd_image_transform(sd_image)
target_size = sd_image_tensor.shape[-2]
target_width, target_height = calculate_new_dimensions(height=height, width=width, target_size=target_size)
y1 = max(0, int(round((target_height - target_size) / 2.0)))
x1 = max(0, int(round((target_width - target_size) / 2.0)))
# image_data['crop_top_lefts'] = torch.tensor([y1, x1])
image_data['time_ids'] = torch.tensor([height, width, y1, x1, target_size, target_size])
image_data['sd_images'] = sd_image_tensor
if image_transform is not None:
for i in range(len(images)):
images[i] = image_transform(images[i])
images = torch.stack(images, dim=0)
except Exception as e:
print('Error while decode image: ', e)
return {}
input_ids = []
labels = []
input_text = ''
if system_message != '':
if not system_message.endswith('\n'):
system_message += '\n'
input_text += system_message
item_ids = tokenizer.encode(system_message, add_special_tokens=False)
item_labels = [-100] * len(item_ids)
input_ids.extend(item_ids)
labels.extend(item_labels)
captions_all = []
for i in range(story_len):
caption = value["captions"][i]
captions_all.append(caption)
image_cmp_tokens = BOI_TOKEN + ''.join(
[IMG_TOKEN.format(int(item)) for item in range(num_img_in_tokens)]) + EOI_TOKEN
image_gen_tokens = BOI_TOKEN + ''.join(
[IMG_TOKEN.format(int(item)) for item in range(num_img_out_tokens)]) + EOI_TOKEN
instruction = instruction_prompt.format_map({'instruction': captions_all[0] + image_cmp_tokens})
for i in range(num_image_given):
instruction = instruction + "[INST]" + captions_all[i + 1] + image_cmp_tokens
response = "[INST]" + captions_all[num_image_given + 1] + image_gen_tokens
images = images[:num_image_given + 2]
# print(instruction)
item_ids = tokenizer.encode(instruction, add_special_tokens=False)
item_labels = [-100] * len(item_ids)
input_text += instruction
input_ids.extend(item_ids)
labels.extend(item_labels)
item_ids = tokenizer.encode(response, add_special_tokens=False)
item_labels = item_ids
input_text += response
input_ids.extend(item_ids)
labels.extend(item_labels)
input_ids = [tokenizer.bos_token_id] + input_ids + [tokenizer.eos_token_id]
attention_mask = [1] * len(input_ids)
labels = [-100] + labels + [tokenizer.eos_token_id]
boi_token_id = tokenizer.encode(BOI_TOKEN, add_special_tokens=False)[0]
eoi_token_id = tokenizer.encode(EOI_TOKEN, add_special_tokens=False)[0]
ids_cmp_mask = [False] * len(input_ids)
ids_gen_mask = [False] * len(input_ids)
embeds_cmp_mask = [True] + [True] * num_image_given + [False]
embeds_gen_mask = [False] + [False] * num_image_given + [True]
# print(len(input_ids))
if len(input_ids) >= max_length:
# input_ids = input_ids[:max_length]
# attention_mask = attention_mask[:max_length]
# labels = labels[:max_length]
# ids_cmp_mask = ids_cmp_mask[:max_length]
# ids_gen_mask = ids_gen_mask[:max_length]
# print('An edit sample has been removed because of max length. input_text: ', input_text)
return {}
else:
padding_length = max_length - len(input_ids)
input_ids = input_ids + [tokenizer.pad_token_id] * padding_length
attention_mask = attention_mask + [0] * padding_length
labels = labels + [-100] * padding_length
ids_cmp_mask = ids_cmp_mask + [False] * padding_length
ids_gen_mask = ids_gen_mask + [False] * padding_length
input_ids = torch.tensor(input_ids, dtype=torch.long)
attention_mask = torch.tensor(attention_mask, dtype=torch.long)
labels = torch.tensor(labels, dtype=torch.long)
ids_cmp_mask = torch.tensor(ids_cmp_mask, dtype=torch.bool)
ids_gen_mask = torch.tensor(ids_gen_mask, dtype=torch.bool)
embeds_cmp_mask = torch.tensor(embeds_cmp_mask) if embeds_cmp_mask is not None else None
embeds_gen_mask = torch.tensor(embeds_gen_mask) if embeds_gen_mask is not None else None
boi_idx = torch.where(input_ids == boi_token_id)[0].tolist()
eoi_idx = torch.where(input_ids == eoi_token_id)[0].tolist()
ids_cmp_mask[boi_idx[0] + 1:eoi_idx[0]] = True
for i in range(num_image_given):
ids_cmp_mask[boi_idx[i + 1] + 1:eoi_idx[i + 1]] = True
ids_gen_mask[boi_idx[-1] + 1:eoi_idx[-1]] = True
labels[boi_idx[-1] + 1:eoi_idx[-1] + 1] = -100
ret = {
'input_ids': input_ids,
'attention_mask': attention_mask,
'labels': labels,
'ids_gen_mask': ids_gen_mask,
'ids_cmp_mask': ids_cmp_mask,
'embeds_gen_mask': embeds_gen_mask,
'embeds_cmp_mask': embeds_cmp_mask,
'images': images,
'text': input_text,
}
ret.update(image_data)
return ret
def build_long_story_datapipe(data_dir,
image_dir,
tokenizer=None,
story_len=30,
max_length=77,
batch_size=None,
min_resolution=180,
image_transform=None,
sd_image_transform=None,
instruction_prompt='{instruction}',
turn_sep='\n',
system_message='',
min_aspect_ratio=0.666,
num_img_in_tokens=64,
num_img_out_tokens=64,
cycle_count=None):
decode_partial = functools.partial(decode_long_story_data,
image_dir=image_dir,
tokenizer=tokenizer,
story_len=story_len,
image_transform=image_transform,
sd_image_transform=sd_image_transform,
max_length=max_length,
instruction_prompt=instruction_prompt,
turn_sep=turn_sep,
system_message=system_message,
min_resolution=min_resolution,
min_aspect_ratio=min_aspect_ratio,
num_img_in_tokens=num_img_in_tokens,
num_img_out_tokens=num_img_out_tokens)
filter_partial = functools.partial(filter_data_with_image_ids)
if isinstance(data_dir, str):
data_dir = list(braceexpand(data_dir))
datapipe = dp.iter.FileLister(root=data_dir, masks='*.jsonl', recursive=True)
datapipe = datapipe.shuffle()
datapipe = datapipe.cycle(count=cycle_count)
datapipe = datapipe.shuffle()
# datapipe = dp.iter.FileLister(root=data_dir, masks='0000000.tar', recursive=True)
datapipe = datapipe.sharding_filter()
# datapipe = datapipe.sharding_round_robin_dispatch(SHARDING_PRIORITIES.MULTIPROCESSING)
datapipe = datapipe.open_files(mode='r')
datapipe = datapipe.parse_jsonl_files()
datapipe = datapipe.map(decode_partial)
datapipe = datapipe.filter(filter_partial)
# datapipe = datapipe.shuffle(buffer_size=1024)
if batch_size is not None:
datapipe = datapipe.batch(batch_size)
datapipe = datapipe.collate(single_turn_edit_collate)
return datapipe
def build_multi_datapipes(datapipes, tokenizer=None, image_transform=None, sd_image_transform=None,
sample_weights=None):
# assert concat_type in ['concat', 'mux_longest', 'sample']
if sample_weights is None:
sample_weights = [1] * len(datapipes)
else:
assert len(sample_weights) == len(datapipes)
datapipes = [
hydra.utils.instantiate(datapipe, tokenizer=tokenizer, image_transform=image_transform,
sd_image_transform=sd_image_transform) for datapipe in datapipes
]
datasets_to_weights_dict = {}
for dataset, sample_weight in zip(datapipes, sample_weights):
datasets_to_weights_dict[dataset] = sample_weight
datapipe = dp.iter.SampleMultiplexer(datasets_to_weights_dict)
return datapipe