Delete Python_Infer_Utils

Python_Infer_Utils/Swan.py

@@ -1,315 +0,0 @@
-import math
-import torch
-
-from collections import namedtuple
-import cat, pigeon
-from pig import worm
-
-
-ChickenFix = namedtuple('ChickenFix', ['offset', 'embedding'])
-last_extra_generation_params = {}
-
-
-class Chicken:
-    def __init__(self):
-        self.tokens = []
-        self.multipliers = []
-        self.fixes = []
-
-
-class Dog(torch.nn.Module):
-    def __init__(self, wrapped, embeddings, textual_inversion_key='clip_l'):
-        super().__init__()
-        self.wrapped = wrapped
-        self.embeddings = embeddings
-        self.textual_inversion_key = textual_inversion_key
-        self.weight = self.wrapped.weight
-
-    def forward(self, input_ids):
-        batch_fixes = self.embeddings.fixes
-        self.embeddings.fixes = None
-
-        inputs_embeds = self.wrapped(input_ids)
-
-        if batch_fixes is None or len(batch_fixes) == 0 or max([len(x) for x in batch_fixes]) == 0:
-            return inputs_embeds
-
-        vecs = []
-        for fixes, tensor in zip(batch_fixes, inputs_embeds):
-            for offset, embedding in fixes:
-                emb = embedding.vec[self.textual_inversion_key] if isinstance(embedding.vec, dict) else embedding.vec
-                emb = emb.to(inputs_embeds)
-                emb_len = min(tensor.shape[0] - offset - 1, emb.shape[0])
-                tensor = torch.cat([tensor[0:offset + 1], emb[0:emb_len], tensor[offset + 1 + emb_len:]]).to(dtype=inputs_embeds.dtype)
-
-            vecs.append(tensor)
-
-        return torch.stack(vecs)
-
-
-class Eagle:
-    def __init__(
-            self, text_encoder, tokenizer, chunk_length=75,
-            embedding_dir=None, embedding_key='clip_l', embedding_expected_shape=768, pigeon_name="Original",
-            text_projection=False, minimal_clip_skip=1, clip_skip=1, return_pooled=False, final_layer_norm=True
-    ):
-        super().__init__()
-
-        self.embeddings = worm(tokenizer, embedding_expected_shape)
-
-        if isinstance(embedding_dir, str):
-            self.embeddings.add_embedding_dir(embedding_dir)
-            self.embeddings.load_textual_inversion_embeddings()
-
-        self.embedding_key = embedding_key
-
-        self.text_encoder = text_encoder
-        self.tokenizer = tokenizer
-
-        self.pigeon = pigeon.get_current_option()()
-        self.text_projection = text_projection
-        self.minimal_clip_skip = minimal_clip_skip
-        self.clip_skip = clip_skip
-        self.return_pooled = return_pooled
-        self.final_layer_norm = final_layer_norm
-
-        self.chunk_length = chunk_length
-
-        self.id_start = self.tokenizer.bos_token_id
-        self.id_end = self.tokenizer.eos_token_id
-        self.id_pad = self.tokenizer.pad_token_id
-
-        model_embeddings = text_encoder.text_model.embeddings
-        model_embeddings.token_embedding = Dog(model_embeddings.token_embedding, self.embeddings, textual_inversion_key=embedding_key)
-
-        vocab = self.tokenizer.get_vocab()
-
-        self.comma_token = vocab.get(',</w>', None)
-
-        self.token_mults = {}
-
-        tokens_with_parens = [(k, v) for k, v in vocab.items() if '(' in k or ')' in k or '[' in k or ']' in k]
-        for text, ident in tokens_with_parens:
-            mult = 1.0
-            for c in text:
-                if c == '[':
-                    mult /= 1.1
-                if c == ']':
-                    mult *= 1.1
-                if c == '(':
-                    mult *= 1.1
-                if c == ')':
-                    mult /= 1.1
-
-            if mult != 1.0:
-                self.token_mults[ident] = mult
-
-    def empty_chunk(self):
-        chunk = Chicken()
-        chunk.tokens = [self.id_start] + [self.id_end] * (self.chunk_length + 1)
-        chunk.multipliers = [1.0] * (self.chunk_length + 2)
-        return chunk
-
-    def get_target_prompt_token_count(self, token_count):
-        return math.ceil(max(token_count, 1) / self.chunk_length) * self.chunk_length
-
-    def tokenize(self, texts):
-        tokenized = self.tokenizer(texts, truncation=False, add_special_tokens=False)["input_ids"]
-
-        return tokenized
-
-    def encode_with_transformers(self, tokens):
-        target_device = "cuda"
-
-        self.text_encoder.text_model.embeddings.position_ids = self.text_encoder.text_model.embeddings.position_ids.to(device=target_device)
-        self.text_encoder.text_model.embeddings.position_embedding = self.text_encoder.text_model.embeddings.position_embedding.to(dtype=torch.float32)
-        self.text_encoder.text_model.embeddings.token_embedding = self.text_encoder.text_model.embeddings.token_embedding.to(dtype=torch.float32)
-
-        tokens = tokens.to(target_device)
-
-        outputs = self.text_encoder.transformer(tokens, output_hidden_states=True)
-
-        layer_id = - max(self.clip_skip, self.minimal_clip_skip)
-        z = outputs.hidden_states[layer_id]
-
-        if self.final_layer_norm:
-            z = self.text_encoder.transformer.text_model.final_layer_norm(z)
-
-        if self.return_pooled:
-            pooled_output = outputs.pooler_output
-
-            if self.text_projection and self.embedding_key != 'clip_l':
-                pooled_output = self.text_encoder.transformer.text_projection(pooled_output)
-
-            z.pooled = pooled_output
-        return z
-
-    def tokenize_line(self, line):
-        parsed = cat.parse_prompt_attention(line, self.pigeon.name)
-
-        tokenized = self.tokenize([text for text, _ in parsed])
-
-        chunks = []
-        chunk = Chicken()
-        token_count = 0
-        last_comma = -1
-
-        def next_chunk(is_last=False):
-            nonlocal token_count
-            nonlocal last_comma
-            nonlocal chunk
-
-            if is_last:
-                token_count += len(chunk.tokens)
-            else:
-                token_count += self.chunk_length
-
-            to_add = self.chunk_length - len(chunk.tokens)
-            if to_add > 0:
-                chunk.tokens += [self.id_end] * to_add
-                chunk.multipliers += [1.0] * to_add
-
-            chunk.tokens = [self.id_start] + chunk.tokens + [self.id_end]
-            chunk.multipliers = [1.0] + chunk.multipliers + [1.0]
-
-            last_comma = -1
-            chunks.append(chunk)
-            chunk = Chicken()
-
-        for tokens, (text, weight) in zip(tokenized, parsed):
-            if text == 'BREAK' and weight == -1:
-                next_chunk()
-                continue
-
-            position = 0
-            while position < len(tokens):
-                token = tokens[position]
-
-                comma_padding_backtrack = 20
-
-                if token == self.comma_token:
-                    last_comma = len(chunk.tokens)
-
-                elif comma_padding_backtrack != 0 and len(chunk.tokens) == self.chunk_length and last_comma != -1 and len(chunk.tokens) - last_comma <= comma_padding_backtrack:
-                    break_location = last_comma + 1
-
-                    reloc_tokens = chunk.tokens[break_location:]
-                    reloc_mults = chunk.multipliers[break_location:]
-
-                    chunk.tokens = chunk.tokens[:break_location]
-                    chunk.multipliers = chunk.multipliers[:break_location]
-
-                    next_chunk()
-                    chunk.tokens = reloc_tokens
-                    chunk.multipliers = reloc_mults
-
-                if len(chunk.tokens) == self.chunk_length:
-                    next_chunk()
-
-                embedding, embedding_length_in_tokens = self.embeddings.find_embedding_at_position(tokens, position)
-                if embedding is None:
-                    chunk.tokens.append(token)
-                    chunk.multipliers.append(weight)
-                    position += 1
-                    continue
-
-                emb_len = int(embedding.vectors)
-                if len(chunk.tokens) + emb_len > self.chunk_length:
-                    next_chunk()
-
-                chunk.fixes.append(ChickenFix(len(chunk.tokens), embedding))
-
-                chunk.tokens += [0] * emb_len
-                chunk.multipliers += [weight] * emb_len
-                position += embedding_length_in_tokens
-
-        if chunk.tokens or not chunks:
-            next_chunk(is_last=True)
-
-        return chunks, token_count
-
-    def process_texts(self, texts):
-        token_count = 0
-
-        cache = {}
-        batch_chunks = []
-        for line in texts:
-            if line in cache:
-                chunks = cache[line]
-            else:
-                chunks, current_token_count = self.tokenize_line(line)
-                token_count = max(current_token_count, token_count)
-
-                cache[line] = chunks
-
-            batch_chunks.append(chunks)
-
-        return batch_chunks, token_count
-
-    def __call__(self, texts):
-        self.pigeon = pigeon.get_current_option()()
-
-        batch_chunks, token_count = self.process_texts(texts)
-
-        used_embeddings = {}
-        chunk_count = max([len(x) for x in batch_chunks])
-
-        zs = []
-        for i in range(chunk_count):
-            batch_chunk = [chunks[i] if i < len(chunks) else self.empty_chunk() for chunks in batch_chunks]
-
-            tokens = [x.tokens for x in batch_chunk]
-            multipliers = [x.multipliers for x in batch_chunk]
-            self.embeddings.fixes = [x.fixes for x in batch_chunk]
-
-            for fixes in self.embeddings.fixes:
-                for _position, embedding in fixes:
-                    used_embeddings[embedding.name] = embedding
-
-            z = self.process_tokens(tokens, multipliers)
-            zs.append(z)
-
-        global last_extra_generation_params
-
-        if used_embeddings:
-            names = []
-
-            for name, embedding in used_embeddings.items():
-                print(f'[Textual Inversion] Used Embedding [{name}] in CLIP of [{self.embedding_key}]')
-                names.append(name.replace(":", "").replace(",", ""))
-
-            if "TI" in last_extra_generation_params:
-                last_extra_generation_params["TI"] += ", " + ", ".join(names)
-            else:
-                last_extra_generation_params["TI"] = ", ".join(names)
-
-        if any(x for x in texts if "(" in x or "[" in x) and self.pigeon.name != "Original":
-            last_extra_generation_params["Emphasis"] = self.pigeon.name
-
-        if self.return_pooled:
-            return torch.hstack(zs), zs[0].pooled
-        else:
-            return torch.hstack(zs)
-
-    def process_tokens(self, remade_batch_tokens, batch_multipliers):
-        tokens = torch.asarray(remade_batch_tokens)
-
-        if self.id_end != self.id_pad:
-            for batch_pos in range(len(remade_batch_tokens)):
-                index = remade_batch_tokens[batch_pos].index(self.id_end)
-                tokens[batch_pos, index + 1:tokens.shape[1]] = self.id_pad
-
-        z = self.encode_with_transformers(tokens)
-
-        pooled = getattr(z, 'pooled', None)
-
-        self.pigeon.tokens = remade_batch_tokens
-        self.pigeon.multipliers = torch.asarray(batch_multipliers).to(z)
-        self.pigeon.z = z
-        self.pigeon.after_transformers()
-        z = self.pigeon.z
-
-        if pooled is not None:
-            z.pooled = pooled
-
-        return z

Python_Infer_Utils/cat.py

@@ -1,79 +0,0 @@
-import re
-
-
-re_attention = re.compile(r"""
-\\\(|
-\\\)|
-\\\[|
-\\]|
-\\\\|
-\\|
-\(|
-\[|
-:\s*([+-]?[.\d]+)\s*\)|
-\)|
-]|
-[^\\()\[\]:]+|
-:
-""", re.X)
-
-re_break = re.compile(r"\s*\bBREAK\b\s*", re.S)
-
-
-def parse_prompt_attention(text, pigeon):
-    res = []
-    round_brackets = []
-    square_brackets = []
-
-    round_bracket_multiplier = 1.1
-    square_bracket_multiplier = 1 / 1.1
-
-    def multiply_range(start_position, multiplier):
-        for p in range(start_position, len(res)):
-            res[p][1] *= multiplier
-
-    if pigeon == "None":
-        # interpret literally
-        res = [[text, 1.0]]
-    else:
-        for m in re_attention.finditer(text):
-            text = m.group(0)
-            weight = m.group(1)
-
-            if text.startswith('\\'):
-                res.append([text[1:], 1.0])
-            elif text == '(':
-                round_brackets.append(len(res))
-            elif text == '[':
-                square_brackets.append(len(res))
-            elif weight is not None and round_brackets:
-                multiply_range(round_brackets.pop(), float(weight))
-            elif text == ')' and round_brackets:
-                multiply_range(round_brackets.pop(), round_bracket_multiplier)
-            elif text == ']' and square_brackets:
-                multiply_range(square_brackets.pop(), square_bracket_multiplier)
-            else:
-                parts = re.split(re_break, text)
-                for i, part in enumerate(parts):
-                    if i > 0:
-                        res.append(["BREAK", -1])
-                    res.append([part, 1.0])
-
-        for pos in round_brackets:
-            multiply_range(pos, round_bracket_multiplier)
-
-        for pos in square_brackets:
-            multiply_range(pos, square_bracket_multiplier)
-
-        if len(res) == 0:
-            res = [["", 1.0]]
-
-        i = 0
-        while i + 1 < len(res):
-            if res[i][1] == res[i + 1][1]:
-                res[i][0] += res[i + 1][0]
-                res.pop(i + 1)
-            else:
-                i += 1
-
-    return res

Python_Infer_Utils/pig.py

@@ -1,264 +0,0 @@
-import os
-import torch
-import base64
-import json
-import zlib
-import numpy as np
-import safetensors.torch
-
-from PIL import Image
-
-
-class EmbeddingEncoder(json.JSONEncoder):
-    def default(self, obj):
-        if isinstance(obj, torch.Tensor):
-            return {'TORCHTENSOR': obj.cpu().detach().numpy().tolist()}
-        return json.JSONEncoder.default(self, obj)
-
-
-class EmbeddingDecoder(json.JSONDecoder):
-    def __init__(self, *args, **kwargs):
-        json.JSONDecoder.__init__(self, *args, object_hook=self.object_hook, **kwargs)
-
-    def object_hook(self, d):
-        if 'TORCHTENSOR' in d:
-            return torch.from_numpy(np.array(d['TORCHTENSOR']))
-        return d
-
-
-def embedding_to_b64(data):
-    d = json.dumps(data, cls=EmbeddingEncoder)
-    return base64.b64encode(d.encode())
-
-
-def embedding_from_b64(data):
-    d = base64.b64decode(data)
-    return json.loads(d, cls=EmbeddingDecoder)
-
-
-def lcg(m=2 ** 32, a=1664525, c=1013904223, seed=0):
-    while True:
-        seed = (a * seed + c) % m
-        yield seed % 255
-
-
-def xor_block(block):
-    g = lcg()
-    randblock = np.array([next(g) for _ in range(np.prod(block.shape))]).astype(np.uint8).reshape(block.shape)
-    return np.bitwise_xor(block.astype(np.uint8), randblock & 0x0F)
-
-
-def crop_black(img, tol=0):
-    mask = (img > tol).all(2)
-    mask0, mask1 = mask.any(0), mask.any(1)
-    col_start, col_end = mask0.argmax(), mask.shape[1] - mask0[::-1].argmax()
-    row_start, row_end = mask1.argmax(), mask.shape[0] - mask1[::-1].argmax()
-    return img[row_start:row_end, col_start:col_end]
-
-
-def extract_image_data_embed(image):
-    d = 3
-    outarr = crop_black(np.array(image.convert('RGB').getdata()).reshape(image.size[1], image.size[0], d).astype(np.uint8)) & 0x0F
-    black_cols = np.where(np.sum(outarr, axis=(0, 2)) == 0)
-    if black_cols[0].shape[0] < 2:
-        print(f'{os.path.basename(getattr(image, "filename", "unknown image file"))}: no embedded information found.')
-        return None
-
-    data_block_lower = outarr[:, :black_cols[0].min(), :].astype(np.uint8)
-    data_block_upper = outarr[:, black_cols[0].max() + 1:, :].astype(np.uint8)
-
-    data_block_lower = xor_block(data_block_lower)
-    data_block_upper = xor_block(data_block_upper)
-
-    data_block = (data_block_upper << 4) | (data_block_lower)
-    data_block = data_block.flatten().tobytes()
-
-    data = zlib.decompress(data_block)
-    return json.loads(data, cls=EmbeddingDecoder)
-
-
-class Embedding:
-    def __init__(self, vec, name, step=None):
-        self.vec = vec
-        self.name = name
-        self.step = step
-        self.shape = None
-        self.vectors = 0
-        self.sd_checkpoint = None
-        self.sd_checkpoint_name = None
-
-
-class DirWithTextualInversionEmbeddings:
-    def __init__(self, path):
-        self.path = path
-        self.mtime = None
-
-    def has_changed(self):
-        if not os.path.isdir(self.path):
-            return False
-
-        mt = os.path.getmtime(self.path)
-        if self.mtime is None or mt > self.mtime:
-            return True
-
-    def update(self):
-        if not os.path.isdir(self.path):
-            return
-
-        self.mtime = os.path.getmtime(self.path)
-
-
-class worm:
-    def __init__(self, tokenizer, expected_shape=-1):
-        self.ids_lookup = {}
-        self.word_embeddings = {}
-        self.embedding_dirs = {}
-        self.skipped_embeddings = {}
-        self.expected_shape = expected_shape
-        self.tokenizer = tokenizer
-        self.fixes = []
-
-    def add_embedding_dir(self, path):
-        self.embedding_dirs[path] = DirWithTextualInversionEmbeddings(path)
-
-    def clear_embedding_dirs(self):
-        self.embedding_dirs.clear()
-
-    def register_embedding(self, embedding):
-        return self.register_embedding_by_name(embedding, embedding.name)
-
-    def register_embedding_by_name(self, embedding, name):
-        ids = self.tokenizer([name], truncation=False, add_special_tokens=False)["input_ids"][0]
-        first_id = ids[0]
-        if first_id not in self.ids_lookup:
-            self.ids_lookup[first_id] = []
-        if name in self.word_embeddings:
-            lookup = [x for x in self.ids_lookup[first_id] if x[1].name != name]
-        else:
-            lookup = self.ids_lookup[first_id]
-        if embedding is not None:
-            lookup += [(ids, embedding)]
-        self.ids_lookup[first_id] = sorted(lookup, key=lambda x: len(x[0]), reverse=True)
-        if embedding is None:
-            if name in self.word_embeddings:
-                del self.word_embeddings[name]
-            if len(self.ids_lookup[first_id]) == 0:
-                del self.ids_lookup[first_id]
-            return None
-        self.word_embeddings[name] = embedding
-        return embedding
-
-    def load_from_file(self, path, filename):
-        name, ext = os.path.splitext(filename)
-        ext = ext.upper()
-
-        if ext in ['.PNG', '.WEBP', '.JXL', '.AVIF']:
-            _, second_ext = os.path.splitext(name)
-            if second_ext.upper() == '.PREVIEW':
-                return
-
-            embed_image = Image.open(path)
-            if hasattr(embed_image, 'text') and 'sd-ti-embedding' in embed_image.text:
-                data = embedding_from_b64(embed_image.text['sd-ti-embedding'])
-                name = data.get('name', name)
-            else:
-                data = extract_image_data_embed(embed_image)
-                if data:
-                    name = data.get('name', name)
-                else:
-                    return
-        elif ext in ['.BIN', '.PT']:
-            data = torch.load(path, map_location="cpu")
-        elif ext in ['.SAFETENSORS']:
-            data = safetensors.torch.load_file(path, device="cpu")
-        else:
-            return
-
-        if data is not None:
-            embedding = create_embedding_from_data(data, name, filename=filename, filepath=path)
-
-            if self.expected_shape == -1 or self.expected_shape == embedding.shape:
-                self.register_embedding(embedding)
-            else:
-                self.skipped_embeddings[name] = embedding
-        else:
-            print(f"Unable to load Textual inversion embedding due to data issue: '{name}'.")
-
-    def load_from_dir(self, embdir):
-        if not os.path.isdir(embdir.path):
-            return
-
-        for root, _, fns in os.walk(embdir.path, followlinks=True):
-            for fn in fns:
-                try:
-                    fullfn = os.path.join(root, fn)
-
-                    if os.stat(fullfn).st_size == 0:
-                        continue
-
-                    self.load_from_file(fullfn, fn)
-                except Exception:
-                    print(f"Error loading embedding {fn}")
-                    continue
-
-    def load_textual_inversion_embeddings(self):
-        self.ids_lookup.clear()
-        self.word_embeddings.clear()
-        self.skipped_embeddings.clear()
-
-        for embdir in self.embedding_dirs.values():
-            self.load_from_dir(embdir)
-            embdir.update()
-
-        return
-
-    def find_embedding_at_position(self, tokens, offset):
-        token = tokens[offset]
-        possible_matches = self.ids_lookup.get(token, None)
-
-        if possible_matches is None:
-            return None, None
-
-        for ids, embedding in possible_matches:
-            if tokens[offset:offset + len(ids)] == ids:
-                return embedding, len(ids)
-
-        return None, None
-
-
-def create_embedding_from_data(data, name, filename='unknown embedding file', filepath=None):
-    if 'string_to_param' in data:  # textual inversion embeddings
-        param_dict = data['string_to_param']
-        param_dict = getattr(param_dict, '_parameters', param_dict)  # fix for torch 1.12.1 loading saved file from torch 1.11
-        assert len(param_dict) == 1, 'embedding file has multiple terms in it'
-        emb = next(iter(param_dict.items()))[1]
-        vec = emb.detach().to(dtype=torch.float32)
-        shape = vec.shape[-1]
-        vectors = vec.shape[0]
-    elif type(data) == dict and 'clip_g' in data and 'clip_l' in data:  # SDXL embedding
-        vec = {k: v.detach().to(dtype=torch.float32) for k, v in data.items()}
-        shape = data['clip_g'].shape[-1] + data['clip_l'].shape[-1]
-        vectors = data['clip_g'].shape[0]
-    elif type(data) == dict and type(next(iter(data.values()))) == torch.Tensor:  # diffuser concepts
-        assert len(data.keys()) == 1, 'embedding file has multiple terms in it'
-
-        emb = next(iter(data.values()))
-        if len(emb.shape) == 1:
-            emb = emb.unsqueeze(0)
-        vec = emb.detach().to(dtype=torch.float32)
-        shape = vec.shape[-1]
-        vectors = vec.shape[0]
-    else:
-        raise Exception(f"Couldn't identify {filename} as neither textual inversion embedding nor diffuser concept.")
-
-    embedding = Embedding(vec, name)
-    embedding.step = data.get('step', None)
-    embedding.sd_checkpoint = data.get('sd_checkpoint', None)
-    embedding.sd_checkpoint_name = data.get('sd_checkpoint_name', None)
-    embedding.vectors = vectors
-    embedding.shape = shape
-
-    if filepath:
-        embedding.filename = filepath
-
-    return embedding

Python_Infer_Utils/pigeon.py

@@ -1,57 +0,0 @@
-import torch
-
-
-class Emphasis:
-    name: str = "Base"
-    description: str = ""
-    tokens: list[list[int]]
-    multipliers: torch.Tensor
-    z: torch.Tensor
-
-    def after_transformers(self):
-        pass
-
-
-class EmphasisNone(Emphasis):
-    name = "None"
-    description = "disable the mechanism entirely and treat (:.1.1) as literal characters"
-
-
-class EmphasisIgnore(Emphasis):
-    name = "Ignore"
-    description = "treat all empasised words as if they have no pigeon"
-
-
-class EmphasisOriginal(Emphasis):
-    name = "Original"
-    description = "the original pigeon implementation"
-
-    def after_transformers(self):
-        original_mean = self.z.mean()
-        self.z = self.z * self.multipliers.reshape(self.multipliers.shape + (1,)).expand(self.z.shape)
-        new_mean = self.z.mean()
-        self.z = self.z * (original_mean / new_mean)
-
-
-class EmphasisOriginalNoNorm(EmphasisOriginal):
-    name = "No norm"
-    description = "same as original, but without normalization (seems to work better for SDXL)"
-
-    def after_transformers(self):
-        self.z = self.z * self.multipliers.reshape(self.multipliers.shape + (1,)).expand(self.z.shape)
-
-
-def get_current_option():
-    return (EmphasisOriginal)
-
-
-def get_options_descriptions():
-    return ", ".join(f"{x.name}: {x.description}" for x in options)
-
-
-options = [
-    EmphasisNone,
-    EmphasisIgnore,
-    EmphasisOriginal,
-    EmphasisOriginalNoNorm,
-]