File size: 11,489 Bytes
4347fc3
 
a16f7a6
4347fc3
 
a16f7a6
4347fc3
 
625d699
70dfa79
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3f788ef
 
 
 
 
 
777653e
 
 
a426c90
 
 
 
 
 
 
 
 
 
 
 
 
 
f7614f0
a16f7a6
 
4347fc3
a16f7a6
 
 
 
 
 
2a71f28
a16f7a6
 
 
 
83d8189
 
70dfa79
a16f7a6
4347fc3
60af39a
70dfa79
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4347fc3
 
60af39a
edcca83
a559676
4347fc3
 
60af39a
4347fc3
 
 
266a83c
4347fc3
 
 
 
3f788ef
c3c13cc
 
 
60af39a
c3c13cc
 
 
3f788ef
60af39a
c3c13cc
13f53a4
60af39a
3f788ef
60af39a
 
4347fc3
 
60af39a
266a83c
3b15df8
60af39a
1efb52e
30b8c04
 
1efb52e
 
3f788ef
70dfa79
 
 
 
 
 
 
 
 
 
3b15df8
 
 
 
3f788ef
 
c3c13cc
3f788ef
 
 
 
 
3b15df8
a16f7a6
 
 
3f788ef
70dfa79
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a16f7a6
70dfa79
a16f7a6
 
3f788ef
a16f7a6
70dfa79
a16f7a6
 
e6dc87e
 
 
 
 
 
a16f7a6
e6dc87e
2a71f28
70dfa79
e6dc87e
3f788ef
60af39a
3f788ef
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
import numpy as np
from sklearn.svm import SVC
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
import torch
from umap import UMAP
import PIL

def get_separation_space(type_bin, annotations, df, samples=200, method='LR', C=0.1, latent_space='Z'):
    """
    The get_separation_space function takes in a type_bin, annotations, and df.
    It then samples 100 of the most representative abstracts for that type_bin and 100 of the least representative abstracts for that type_bin.
    It then trains an SVM or logistic regression model on these 200 samples to find a separation space between them. 
    The function returns this separation space as well as how many nodes are important in this separation space.
    
    :param type_bin: Select the type of abstracts to be used for training
    :param annotations: Access the z_vectors
    :param df: Get the abstracts that are used for training
    :param samples: Determine how many samples to take from the top and bottom of the distribution
    :param method: Specify the classifier to use
    :param C: Control the regularization strength
    :return: The weights of the linear classifier
    :doc-author: Trelent
    """
    
    if latent_space == 'Z':
        col = 'z_vectors'
    else:
        col = 'w_vectors'
    
    if len(type_bin) == 1:
        type_bin = type_bin[0]
    if type(type_bin) == str:
        abstracts = np.array([float(ann) for ann in df[type_bin]])
        abstract_idxs = list(np.argsort(abstracts))[:samples]
        repr_idxs = list(np.argsort(abstracts))[-samples:]
        X = np.array([annotations[col][i] for i in abstract_idxs+repr_idxs])
    elif len(type_bin) == 2:
        print('Using two concepts for separation space')
        first_concept = np.array([float(ann) for ann in df[type_bin[0]]])
        second_concept = np.array([float(ann) for ann in df[type_bin[1]]])
        first_idxs = list(np.argsort(first_concept))[:samples]
        second_idxs = list(np.argsort(second_concept))[:samples]
        X = np.array([annotations[col][i] for i in first_idxs+second_idxs])
    else:
        print('Error: type_bin must be either a string or a list of strings of len 2')
        return
    
    X = X.reshape((2*samples, 512))
    y = np.array([1]*samples + [0]*samples)
    x_train, x_val, y_train, y_val = train_test_split(X, y, test_size=0.2)
    if method == 'SVM':
        svc = SVC(gamma='auto', kernel='linear', random_state=0, C=C)
        svc.fit(x_train, y_train)
        print('Val performance SVM', svc.score(x_val, y_val))
        imp_features = (np.abs(svc.coef_) > 0.2).sum()
        imp_nodes = np.where(np.abs(svc.coef_) > 0.2)[1]
        return svc.coef_ / np.linalg.norm(clf.coef_), imp_features, imp_nodes, np.round(clf.score(x_val, y_val),2)
    elif method == 'LR':
        clf = LogisticRegression(random_state=0, C=C)
        clf.fit(x_train, y_train)
        print('Val performance logistic regression', clf.score(x_val, y_val))
        imp_features = (np.abs(clf.coef_) > 0.15).sum()
        imp_nodes = np.where(np.abs(clf.coef_) > 0.15)[1]
        return clf.coef_ / np.linalg.norm(clf.coef_), imp_features, imp_nodes, np.round(clf.score(x_val, y_val),2)


def regenerate_images(model, z, decision_boundary, min_epsilon=-3, max_epsilon=3, count=5, latent_space='Z', layers=None):
    """
    The regenerate_images function takes a model, z, and decision_boundary as input.  It then
    constructs an inverse rotation/translation matrix and passes it to the generator.  The generator
    expects this matrix as an inverse to avoid potentially failing numerical operations in the network.
    The function then generates images using G(z_0, label) where z_0 is a linear combination of z and the decision boundary.
    
    :param model: Pass in the model to be used for image generation
    :param z: Generate the starting point of the line
    :param decision_boundary: Generate images along the direction of the decision boundary
    :param min_epsilon: Set the minimum value of lambda
    :param max_epsilon: Set the maximum distance from the original image to generate
    :param count: Determine the number of images that are generated
    :return: A list of images and a list of lambdas
    :doc-author: Trelent
    """
    device = torch.device('cpu')
    G = model.to(device) # type: ignore
    
    if False:
        decision_boundary = z - (np.dot(z, decision_boundary.T) / np.dot(decision_boundary, decision_boundary.T)) * decision_boundary
    # Labels.
    label = torch.zeros([1, G.c_dim], device=device)

    z = torch.from_numpy(z.copy()).to(device)
    decision_boundary = torch.from_numpy(decision_boundary.copy()).to(device)
        
    lambdas = np.linspace(min_epsilon, max_epsilon, count)
    images = []
    # Generate images.
    for _, lambda_ in enumerate(lambdas):
        z_0 = z + lambda_ * decision_boundary
        if latent_space == 'Z':
            W_0 = G.mapping(z_0, label, truncation_psi=1).to(torch.float32)
            W = G.mapping(z, label, truncation_psi=1).to(torch.float32)
            print(W.dtype)
        else:
            W_0 = z_0.expand((14, -1)).unsqueeze(0).to(torch.float32)
            W = z.expand((14, -1)).unsqueeze(0).to(torch.float32)
            print(W.dtype)
        
        if layers:
            W_f = torch.empty_like(W).copy_(W).to(torch.float32)
            W_f[:, layers, :] = W_0[:, layers, :]
            img = G.synthesis(W_f, noise_mode='const')
        else:
            img = G.synthesis(W_0, noise_mode='const')
                
        img = (img.permute(0, 2, 3, 1) * 127.5 + 128).clamp(0, 255).to(torch.uint8)
        images.append(PIL.Image.fromarray(img[0].cpu().numpy(), 'RGB'))
            
    return images, lambdas


def generate_joint_effect(model, z, decision_boundaries, min_epsilon=-3, max_epsilon=3, count=5, latent_space='Z'):
    decision_boundary_joint = np.sum(decision_boundaries, axis=0)
    print(decision_boundary_joint.shape)
    return regenerate_images(model, z, decision_boundary_joint, min_epsilon=min_epsilon, max_epsilon=max_epsilon, count=count, latent_space=latent_space)
    
def generate_original_image(z, model, latent_space='Z'):
    """
    The generate_original_image function takes in a latent vector and the model,
    and returns an image generated from that latent vector.
    
    
    :param z: Generate the image
    :param model: Generate the image
    :return: A pil image
    :doc-author: Trelent
    """
    device = torch.device('cpu')
    G = model.to(device) # type: ignore
    # Labels.
    label = torch.zeros([1, G.c_dim], device=device)
    if latent_space == 'Z':
        z = torch.from_numpy(z.copy()).to(device)
        img = G(z, label, truncation_psi=1, noise_mode='const')
    else:
        W = torch.from_numpy(np.repeat(z, 14, axis=0).reshape(1, 14, z.shape[1]).copy()).to(device)
        print(W.shape)
        img = G.synthesis(W, noise_mode='const')

    img = (img.permute(0, 2, 3, 1) * 127.5 + 128).clamp(0, 255).to(torch.uint8)
    return PIL.Image.fromarray(img[0].cpu().numpy(), 'RGB')


def get_concepts_vectors(concepts, annotations, df, samples=100, method='LR', C=0.1, latent_space='Z'):
    """
    The get_concepts_vectors function takes in a list of concepts, a dictionary of annotations, and the dataframe containing all the images.
    It returns two things:
        1) A numpy array with shape (len(concepts), 512) where each row is an embedding vector for one concept.
        2) A set containing all nodes that are important in this separation space.
    
    :param concepts: Specify the concepts to be used in the analysis
    :param annotations: Get the annotations for each concept
    :param df: Get the annotations for each concept
    :param samples: Determine the number of samples to use in training the logistic regression model
    :param method: Choose the method used to train the model
    :param C: Control the regularization of the logistic regression
    :return: The vectors of the concepts and the nodes that are in common for all concepts
    :doc-author: Trelent
    """
    important_nodes = []
    performances = []
    vectors = np.zeros((len(concepts), 512))
    for i, conc in enumerate(concepts):
        vec, _, imp_nodes, performance = get_separation_space(conc, annotations, df, samples=samples, method=method, C=C, latent_space=latent_space)
        vectors[i,:] = vec
        performances.append(performance)
        important_nodes.append(set(imp_nodes))
    
    # reducer = UMAP(n_neighbors=3, # default 15, The size of local neighborhood (in terms of number of neighboring sample points) used for manifold approximation.
    #                n_components=3, # default 2, The dimension of the space to embed into.
    #                min_dist=0.1, # default 0.1, The effective minimum distance between embedded points.
    #                spread=2.0, # default 1.0, The effective scale of embedded points. In combination with ``min_dist`` this determines how clustered/clumped the embedded points are.
    #                random_state=0, # default: None, If int, random_state is the seed used by the random number generator;
    #            )

    # projection = reducer.fit_transform(vectors)
    nodes_in_common = set.intersection(*important_nodes)
    return vectors, nodes_in_common, performances



def get_verification_score(concept, decision_boundary, model, annotations, samples=100, latent_space='Z'):
    import open_clip
    import os
    import random
    from tqdm import tqdm
    os.environ["CUDA_VISIBLE_DEVICES"] = ""
    
    
    model_clip, _, preprocess = open_clip.create_model_and_transforms('ViT-L-14', pretrained='laion2b_s32b_b82k')
    tokenizer = open_clip.get_tokenizer('ViT-L-14')

    # Prepare the text queries
    #@markdown _in the form pre_prompt {label}_:
    pre_prompt = "Artwork, " #@param {type:"string"}
    text_descriptions = [f"{pre_prompt}{label}" for label in [concept]]
    text_tokens = tokenizer(text_descriptions)


    listlen = len(annotations['fname'])
    items = random.sample(range(listlen), samples)
    changes = []
    for iterator in tqdm(items):
        chunk_imgs = []
        chunk_ids = []

        if latent_space == 'Z':
            z = annotations['z_vectors'][iterator]
        else:
            z = annotations['w_vectors'][iterator]
        images, lambdas = regenerate_images(model, z, decision_boundary, min_epsilon=0, max_epsilon=1, count=2, latent_space=latent_space)
        for im,l in zip(images, lambdas):
            
            chunk_imgs.append(preprocess(im.convert("RGB")))
            chunk_ids.append(l)
            
        image_input = torch.tensor(np.stack(chunk_imgs))

        with torch.no_grad(), torch.cuda.amp.autocast():
            text_features = model_clip.encode_text(text_tokens).float()
            image_features = model_clip.encode_image(image_input).float()

            # Rescale features
            image_features /= image_features.norm(dim=-1, keepdim=True)  
            text_features /= text_features.norm(dim=-1, keepdim=True)
                    
            # Analyze featues
            text_probs = (100.0 * image_features.cpu().numpy() @ text_features.cpu().numpy().T)#.softmax(dim=-1)
            
        change = max(text_probs[1][0].item() - text_probs[0][0].item(), 0)
        changes.append(change)
        
    return np.round(np.mean(np.array(changes)), 4)