Digital Drosophila Archive
205 lines
9.8 KiB
Python
205 lines
9.8 KiB
Python
"""Synthetic tests only; no saved scientific results are accessed."""
|
|
import copy
|
|
import json
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from structure_metrics import aggregate_structure, MODES, SEEDS, METRICS
|
|
|
|
|
|
def records():
|
|
rows = []
|
|
for mode in MODES:
|
|
for odor in range(6):
|
|
design = np.zeros(686, bool)
|
|
design[odor] = True
|
|
for seed in SEEDS:
|
|
count = np.zeros(686, np.int32)
|
|
count[odor] = 10
|
|
count[10] = seed - SEEDS[0]
|
|
count[11] = odor
|
|
rows.append(dict(mode=mode, odor_index=odor, seed=seed,
|
|
counts=count, design_mask=design.copy()))
|
|
return rows
|
|
|
|
|
|
class StructureMetricsTests(unittest.TestCase):
|
|
def test_fixed_complete_inventory_and_pair_specific_baseline(self):
|
|
result = aggregate_structure(records())
|
|
self.assertEqual(result['record_count'], 60)
|
|
self.assertEqual(result['design']['design_union_count'], 6)
|
|
self.assertEqual(result['design']['common_non_design_count'], 680)
|
|
self.assertEqual(result['design']['common_non_design_column_positions'], list(range(6, 686)))
|
|
for mode in MODES:
|
|
self.assertEqual(result['modes'][mode]['trials'], 30)
|
|
for view_name in ('fixed_686', 'common_non_design'):
|
|
view = result['modes'][mode][view_name]
|
|
self.assertEqual(len(view['within_comparisons']), 60)
|
|
self.assertEqual(len(view['between_comparisons']), 75)
|
|
self.assertEqual(len(view['pairs']), 15)
|
|
for pair in view['pairs']:
|
|
self.assertEqual(len(pair['within_comparisons']), 20)
|
|
self.assertEqual(len(pair['between_comparisons']), 5)
|
|
self.assertEqual({row['odor_index'] for row in pair['within_comparisons']},
|
|
{pair['odor_index_a'], pair['odor_index_b']})
|
|
self.assertEqual([row['seed'] for row in pair['between_comparisons']], list(SEEDS))
|
|
json.dumps(result, allow_nan=False)
|
|
|
|
def test_primary_retains_design_columns_and_common_view_is_fixed(self):
|
|
result = aggregate_structure(records())['modes'][MODES[0]]
|
|
pair = result['fixed_686']['pairs'][0]
|
|
common_pair = result['common_non_design']['pairs'][0]
|
|
# Odors zero and one differ by two ten-count design axes plus one
|
|
# shared-coordinate count; same-seed variability cancels between.
|
|
self.assertEqual(pair['between']['euclidean_distance']['mean'], np.sqrt(201.))
|
|
self.assertEqual(common_pair['between']['euclidean_distance']['mean'], 1.)
|
|
self.assertEqual(pair['within']['euclidean_distance']['mean'], 2.)
|
|
self.assertEqual(common_pair['within']['euclidean_distance']['mean'], 2.)
|
|
|
|
def test_input_order_does_not_change_result(self):
|
|
rows = records()
|
|
expected = aggregate_structure(rows)
|
|
actual = aggregate_structure(list(reversed(rows)))
|
|
self.assertEqual(expected, actual)
|
|
|
|
def test_inputs_unchanged_and_readonly_arrays_accepted(self):
|
|
rows = records()
|
|
before = copy.deepcopy(rows)
|
|
for row in rows:
|
|
row['counts'].setflags(write=False)
|
|
row['design_mask'].setflags(write=False)
|
|
aggregate_structure(rows)
|
|
for old, new in zip(before, rows):
|
|
np.testing.assert_array_equal(old['counts'], new['counts'])
|
|
np.testing.assert_array_equal(old['design_mask'], new['design_mask'])
|
|
self.assertFalse(new['counts'].flags.writeable)
|
|
|
|
def test_zero_vectors_do_not_become_similarity_evidence(self):
|
|
rows = records()
|
|
for row in rows:
|
|
row['counts'].fill(0)
|
|
result = aggregate_structure(rows)
|
|
for mode in MODES:
|
|
for name in ('fixed_686', 'common_non_design'):
|
|
view = result['modes'][mode][name]
|
|
for metric in ('cosine_distance', 'pearson_correlation', 'jaccard'):
|
|
self.assertIsNone(view['within'][metric]['mean'])
|
|
self.assertEqual(view['within'][metric]['valid_comparisons'], 0)
|
|
self.assertEqual(view['between'][metric]['undefined_comparisons'], 75)
|
|
self.assertEqual(view['between']['euclidean_distance']['mean'], 0.)
|
|
json.dumps(result, allow_nan=False)
|
|
|
|
def test_constant_nonzero_vectors_have_undefined_pearson(self):
|
|
rows = records()
|
|
for row in rows:
|
|
row['counts'].fill(1)
|
|
view = aggregate_structure(rows)['modes'][MODES[0]]['fixed_686']
|
|
self.assertIsNone(view['within']['pearson_correlation']['mean'])
|
|
self.assertEqual(view['within']['jaccard']['mean'], 1.)
|
|
self.assertAlmostEqual(view['within']['cosine_distance']['mean'], 0., places=14)
|
|
|
|
def test_empty_common_population_is_nonapplicable(self):
|
|
rows = records()
|
|
for row in rows:
|
|
row['design_mask'].fill(True)
|
|
result = aggregate_structure(rows)
|
|
self.assertEqual(result['design']['common_non_design_count'], 0)
|
|
for mode in MODES:
|
|
view = result['modes'][mode]['common_non_design']
|
|
self.assertFalse(view['applicable'])
|
|
self.assertEqual(view['dimensions'], 0)
|
|
for metric in METRICS:
|
|
self.assertIsNone(view['between'][metric]['mean'])
|
|
self.assertEqual(view['within'][metric]['undefined_comparisons'], 60)
|
|
|
|
def test_int32_extrema_are_promoted_before_subtraction_and_norm(self):
|
|
rows = records()
|
|
maximum = np.iinfo(np.int32).max
|
|
for row in rows:
|
|
row['counts'].fill(0)
|
|
row['counts'][row['odor_index']] = maximum
|
|
result = aggregate_structure(rows)['modes'][MODES[0]]['fixed_686']
|
|
distance = result['pairs'][0]['between']['euclidean_distance']['mean']
|
|
self.assertAlmostEqual(distance / float(maximum), np.sqrt(2.))
|
|
self.assertEqual(result['between']['cosine_distance']['mean'], 1.)
|
|
json.dumps(result, allow_nan=False)
|
|
|
|
def test_missing_record_or_extra_repeat_rejected(self):
|
|
rows = records()
|
|
for bad in (rows[:-1], rows + [copy.deepcopy(rows[0])]):
|
|
with self.assertRaisesRegex(ValueError, 'Exactly 60'):
|
|
aggregate_structure(bad)
|
|
|
|
def test_duplicate_identity_rejected_even_with_sixty_rows(self):
|
|
rows = records()
|
|
rows[-1] = copy.deepcopy(rows[0])
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate CAL identity'):
|
|
aggregate_structure(rows)
|
|
|
|
def test_mask_change_across_seed_rejected(self):
|
|
rows = records()
|
|
rows[1]['design_mask'][100] = True
|
|
with self.assertRaisesRegex(ValueError, 'mask changed across seed or mode'):
|
|
aggregate_structure(rows)
|
|
|
|
def test_mask_change_across_mode_rejected(self):
|
|
rows = records()
|
|
for row in rows:
|
|
if row['mode'] == MODES[1] and row['odor_index'] == 0:
|
|
row['design_mask'][100] = True
|
|
with self.assertRaisesRegex(ValueError, 'mask changed across seed or mode'):
|
|
aggregate_structure(rows)
|
|
|
|
def test_negative_counts_and_incorrect_dtype_or_shape_rejected(self):
|
|
changes = [('negative', lambda row: row['counts'].__setitem__(0, -1)),
|
|
('counts64', lambda row: row.update(counts=row['counts'].astype(np.int64))),
|
|
('counts_float', lambda row: row.update(counts=row['counts'].astype(np.float64))),
|
|
('shape', lambda row: row.update(counts=np.zeros(685, np.int32))),
|
|
('mask_dtype', lambda row: row.update(design_mask=row['design_mask'].astype(np.uint8))),
|
|
('mask_shape', lambda row: row.update(design_mask=np.zeros(685, bool)))]
|
|
for label, change in changes:
|
|
with self.subTest(label=label):
|
|
rows = records()
|
|
change(rows[0])
|
|
with self.assertRaises(ValueError):
|
|
aggregate_structure(rows)
|
|
|
|
def test_strict_record_schema_and_id_ranges(self):
|
|
changes = [lambda row: row.update(extra=1), lambda row: row.pop('counts'),
|
|
lambda row: row.update(mode='new_mode'), lambda row: row.update(odor_index=6),
|
|
lambda row: row.update(odor_index=True), lambda row: row.update(seed=51106),
|
|
lambda row: row.update(seed=float(SEEDS[0]))]
|
|
for change in changes:
|
|
rows = records()
|
|
change(rows[0])
|
|
with self.assertRaises(ValueError):
|
|
aggregate_structure(rows)
|
|
|
|
def test_partial_undefined_metrics_report_counts_without_imputation(self):
|
|
rows = records()
|
|
# One zero-count record per mode. Four of the 60 within comparisons
|
|
# and five of the 75 between comparisons then have undefined cosine.
|
|
for row in rows:
|
|
if row['odor_index'] == 0 and row['seed'] == SEEDS[0]:
|
|
row['counts'].fill(0)
|
|
view = aggregate_structure(rows)['modes'][MODES[0]]['fixed_686']
|
|
self.assertEqual(view['within']['cosine_distance']['undefined_comparisons'], 4)
|
|
self.assertEqual(view['between']['cosine_distance']['undefined_comparisons'], 5)
|
|
self.assertEqual(view['within']['cosine_distance']['valid_comparisons'], 56)
|
|
self.assertEqual(view['between']['cosine_distance']['valid_comparisons'], 70)
|
|
|
|
def test_one_common_channel_pearson_is_undefined(self):
|
|
rows = records()
|
|
for row in rows:
|
|
row['design_mask'][:685] = True
|
|
row['counts'][685] = row['odor_index'] + 1
|
|
view = aggregate_structure(rows)['modes'][MODES[0]]['common_non_design']
|
|
self.assertEqual(view['dimensions'], 1)
|
|
self.assertIsNone(view['between']['pearson_correlation']['mean'])
|
|
self.assertEqual(view['between']['cosine_distance']['mean'], 0.)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|