123 lines
7.0 KiB
Python
123 lines
7.0 KiB
Python
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
import torch
|
|
|
|
from test_model_loading import load_nodes
|
|
from feature_analysis import CHART_TYPES, analyze_features, prepare_features
|
|
|
|
|
|
class FeatureAnalysisTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
torch.set_num_threads(2)
|
|
cls.vectors=torch.tensor([[1.,0.,0.],[.9,.1,0.],[0.,1.,0.],[0.,.9,.1],[0.,0.,1.],[.1,0.,.9]])
|
|
cls.nodes=load_nodes()
|
|
|
|
def test_every_chart_returns_finite_image_and_original_distance_matrix(self):
|
|
expected=1-torch.nn.functional.normalize(self.vectors,dim=1)@torch.nn.functional.normalize(self.vectors,dim=1).T
|
|
expected.fill_diagonal_(0)
|
|
for chart in CHART_TYPES:
|
|
with self.subTest(chart=chart):
|
|
features=self.vectors
|
|
options={}
|
|
if chart=='patch_energy':
|
|
features=self.vectors[:,None,:].repeat(1,4,1)
|
|
options['tensor_layout']='tokens'
|
|
image,text,distances=analyze_features(features,chart_type=chart,width=800,height=600,**options)
|
|
report=json.loads(text)
|
|
self.assertEqual(tuple(image.shape),(1,600,800,3))
|
|
self.assertTrue(torch.isfinite(image).all())
|
|
self.assertGreater(float(image.max()-image.min()),0.5)
|
|
torch.testing.assert_close(distances,expected,atol=2e-7,rtol=1e-6)
|
|
self.assertEqual(report['chart_type'],chart)
|
|
self.assertEqual(len(report['nearest_neighbors']),6)
|
|
for index,row in enumerate(report['nearest_neighbors']):
|
|
self.assertNotIn(index,[item['index'] for item in row])
|
|
|
|
def test_graph_edges_are_knn_union_and_use_original_distance(self):
|
|
_,text,distances=analyze_features(self.vectors,top_k=1,width=800,height=600)
|
|
report=json.loads(text)
|
|
expected={tuple(sorted((i,row[0]['index']))) for i,row in enumerate(report['nearest_neighbors'])}
|
|
actual={(edge['source'],edge['target']) for edge in report['edges']}
|
|
self.assertEqual(actual,expected)
|
|
for edge in report['edges']:
|
|
self.assertAlmostEqual(edge['distance'],float(distances[edge['source'],edge['target']]),places=6)
|
|
|
|
def test_distances_and_normalization_match_known_values(self):
|
|
features=torch.tensor([[3.,4.],[0.,5.]])
|
|
for metric,value in (('euclidean',np.sqrt(10)),('manhattan',4.0),('cosine',0.2)):
|
|
_,_,dist=analyze_features(features,chart_type='distance_heatmap',metric=metric,normalize=False,width=800,height=600)
|
|
self.assertAlmostEqual(float(dist[0,1]),value,places=6)
|
|
_,text,dist=analyze_features(features,chart_type='distance_heatmap',metric='euclidean',normalize=True,width=800,height=600)
|
|
self.assertAlmostEqual(float(dist[0,1]),np.sqrt(.4),places=6)
|
|
self.assertEqual(json.loads(text)['normalization'],'row L2')
|
|
|
|
def test_layout_reductions_preserve_image_batch(self):
|
|
tokens=torch.arange(2*3*4,dtype=torch.float32).reshape(2,3,4)
|
|
vector,patches,_,_=prepare_features(tokens,'tokens')
|
|
np.testing.assert_allclose(vector,tokens.mean(1).numpy())
|
|
self.assertEqual(patches.shape,(2,3,4))
|
|
spatial=torch.arange(2*4*2*3,dtype=torch.float32).reshape(2,4,2,3)
|
|
vector,patches,grid,_=prepare_features(spatial,'spatial')
|
|
np.testing.assert_allclose(vector,spatial.mean((2,3)).numpy())
|
|
self.assertEqual(grid,(2,3))
|
|
layers=torch.stack((tokens,tokens+10),dim=1)
|
|
vector,_,_,_=prepare_features(layers,'layer_tokens',layer_index=-1)
|
|
np.testing.assert_allclose(vector,(tokens+10).mean(1).numpy())
|
|
vector,_,_,_=prepare_features(layers,'layer_tokens',layer_pooling='mean')
|
|
np.testing.assert_allclose(vector,(tokens+5).mean(1).numpy())
|
|
vector,_,_,_=prepare_features(layers,'flatten')
|
|
self.assertEqual(vector.shape,(2,24))
|
|
with self.assertRaisesRegex(ValueError,'ambiguous'):
|
|
prepare_features(layers)
|
|
|
|
def test_bad_inputs_and_undefined_cosine_fail_explicitly(self):
|
|
for features in (torch.tensor([[0.,0.],[1.,0.]]),torch.tensor([[float('nan'),1.]])):
|
|
with self.assertRaises(ValueError):
|
|
analyze_features(features,width=800,height=600)
|
|
with self.assertRaisesRegex(ValueError,'labels'):
|
|
analyze_features(self.vectors,labels='one',width=800,height=600)
|
|
with self.assertRaisesRegex(ValueError,'matching features'):
|
|
analyze_features(self.vectors,images=torch.ones(1,32,32,3),width=800,height=600)
|
|
with self.assertRaisesRegex(ValueError,'patch_energy requires'):
|
|
analyze_features(self.vectors,chart_type='patch_energy',width=800,height=600)
|
|
|
|
def test_degenerate_inputs_do_not_fabricate_silhouette_or_pca(self):
|
|
identical=torch.ones(4,3)
|
|
_,text,dist=analyze_features(identical,chart_type='silhouette',width=800,height=600)
|
|
report=json.loads(text)
|
|
self.assertIsNone(report['silhouette_scores'])
|
|
self.assertEqual(report['pca_explained_variance_ratio'],[0.,0.,0.])
|
|
self.assertTrue(report['warnings'])
|
|
torch.testing.assert_close(dist,torch.zeros(4,4))
|
|
_,text,_=analyze_features(torch.zeros(3,2),metric='euclidean',chart_type='dimension_correlation',width=800,height=600)
|
|
self.assertEqual(json.loads(text)['dimension_correlations'],[[None,None],[None,None]])
|
|
|
|
def test_feature_node_never_invokes_model_and_image_node_invokes_extraction_once(self):
|
|
node=self.nodes.KaloscopeFeatureAnalysisNode()
|
|
self.assertNotIn('model',node.INPUT_TYPES()['required'])
|
|
with patch.object(self.nodes.KaloscopeExtractFeaturesNode,'extract',side_effect=AssertionError('unexpected inference')):
|
|
image,_,matrix=node.analyze(self.vectors,chart_type='relationship_graph',width=800,height=600)
|
|
self.assertEqual(matrix.shape,(6,6))
|
|
with patch.object(self.nodes.KaloscopeExtractFeaturesNode,'extract',return_value=(self.vectors,)) as extraction:
|
|
outputs=self.nodes.KaloscopeImageAnalysisNode().analyze(torch.ones(6,32,32,3),{},width=800,height=600)
|
|
extraction.assert_called_once()
|
|
torch.testing.assert_close(outputs[2],self.vectors)
|
|
self.assertEqual(tuple(outputs[0].shape),(1,600,800,3))
|
|
|
|
def test_clustering_options_and_seed_are_reproducible(self):
|
|
for method in ('kmeans','agglomerative','dbscan','none'):
|
|
_,text,_=analyze_features(self.vectors,chart_type='cluster_sizes',cluster_method=method,width=800,height=600)
|
|
self.assertEqual(len(json.loads(text)['cluster_labels']),6)
|
|
_,first,_=analyze_features(self.vectors,chart_type='pca_scatter',seed=7,width=800,height=600)
|
|
_,second,_=analyze_features(self.vectors,chart_type='pca_scatter',seed=7,width=800,height=600)
|
|
self.assertEqual(json.loads(first)['cluster_labels'],json.loads(second)['cluster_labels'])
|
|
self.assertEqual(json.loads(first)['projection'],json.loads(second)['projection'])
|
|
|
|
|
|
if __name__=='__main__':
|
|
unittest.main()
|