mgtotaro commited on
Commit
92ea1b5
·
1 Parent(s): 74fa148

add fasta support; add unittest suite

Browse files
Files changed (8) hide show
  1. .gitignore +1 -0
  2. data.py +19 -14
  3. instructions.md +1 -1
  4. model.py +2 -2
  5. test/__init__.py +0 -0
  6. test/test_app.py +144 -0
  7. test/test_data.py +297 -0
  8. test/test_model.py +109 -0
.gitignore CHANGED
@@ -2,3 +2,4 @@ Dockerfile
2
  *.ipynb
3
  out.*
4
  */
 
 
2
  *.ipynb
3
  out.*
4
  */
5
+ __pycache__/
data.py CHANGED
@@ -15,8 +15,13 @@ class Data:
15
  AA = "ACDEFGHIKLMNPQRSTVWY"
16
 
17
  def parse_seq(self, src: str):
18
- "Parse input sequence"
19
- self.seq = src.strip().upper().replace('\n', '')
 
 
 
 
 
20
  if not all(x in self.model.alphabet for x in self.seq):
21
  raise RuntimeError(f"Unsupported characters in sequence: {''.join(x for x in self.seq if x not in self.model.alphabet)}")
22
 
@@ -35,7 +40,7 @@ class Data:
35
  self.sub.append(f"{src_c}{resi}{trg_c}")
36
  self.resi.append(resi)
37
  elif all(match(r'\d+', x) for x in self.trg):
38
- self.mode = 'DMS'
39
  trh = []
40
  for resi in map(int, self.trg):
41
  if resi < 1 or resi > len(self.seq):
@@ -59,7 +64,7 @@ class Data:
59
  raise RuntimeError(f"Unrecognised input substitution: {self.seq[int(''.join(resi_str))]}{int(''.join(resi_str))} /= {s}{int(''.join(resi_str))}")
60
  self.trg = trh
61
  else:
62
- self.mode = 'TMS'
63
  self.trg = []
64
  for resi, src_c in enumerate(self.seq, 1):
65
  for trg_c in self.AA.replace(src_c, ''):
@@ -84,11 +89,11 @@ class Data:
84
 
85
  def parse_output(self) -> None:
86
  "Format output data for visualisation"
87
- if self.mode == "TMS":
88
- self._render_tms()
89
  self.out.to_csv(self.out_csv, float_format='%.2f')
90
- elif self.mode == "DMS":
91
- self._sort_dms()
92
  self._style_and_save()
93
  elif self.mode == "MUT":
94
  self.out = self.out.sort_values(self.model_name, ascending=False)
@@ -103,16 +108,16 @@ class Data:
103
  .background_gradient(cmap="RdYlGn", vmax=8, vmin=-8).hide(axis=0).hide(axis=1))
104
  self.out.to_csv(self.out_csv, float_format='%.2f', index=False, header=False)
105
 
106
- def _sort_dms(self):
107
- "Sort DMS output by residue then score, top 19 per position, reshaped to columns"
108
  self.out = (self.out.assign(resi=self.out['0'].str.extract(r'(\d+)', expand=False).astype(int))
109
  .sort_values(["resi", self.model_name], ascending=[True,False])
110
  .groupby(["resi"]).head(19).drop(["resi"], axis=1))
111
  self.out = pd.concat([self.out.iloc[19*x:19*(x+1)].reset_index(drop=True) for x in range(self.out.shape[0]//19)]
112
  , axis=1).set_axis(range(self.out.shape[0]//19*2), axis="columns")
113
 
114
- def _render_tms(self):
115
- "Build TMS heatmap from scored mutations"
116
  # Group by residue, keep top 19 (all alternatives)
117
  grouped = (self.out.assign(resi=self.resi_cycle())
118
  .groupby("resi").head(19))
@@ -147,7 +152,7 @@ class Data:
147
  return (self.resi * (len(self.out)//len(self.resi) + 1))[:len(self.out)]
148
 
149
  def _plot_heatmap(self, ncols, nrows):
150
- "Render TMS heatmap to PNG"
151
  kw = dict(cmap="RdBu", cbar=False, square=True, xticklabels=1, yticklabels=1
152
  , center=0, fmt='s', annot_kws={"size": "xx-large"})
153
  annotate = lambda df: df.map(lambda x: ' ' if x != 0 else '\u00b7')
@@ -191,7 +196,7 @@ class Data:
191
 
192
  @property
193
  def image(self):
194
- "return PNG path (TMS) or Styler object (DMS/MUT)"
195
  if self.out_table is not None:
196
  return self.out_table
197
  return self.out_img_path
 
15
  AA = "ACDEFGHIKLMNPQRSTVWY"
16
 
17
  def parse_seq(self, src: str):
18
+ "Parse input sequence (plain or FASTA)"
19
+ lines = src.strip().splitlines()
20
+ if lines and lines[0].startswith('>'):
21
+ lines = lines[1:]
22
+ self.seq = ''.join(lines).upper().replace('\n', '').replace(' ', '')
23
+ if not self.seq:
24
+ raise RuntimeError("Sequence is empty")
25
  if not all(x in self.model.alphabet for x in self.seq):
26
  raise RuntimeError(f"Unsupported characters in sequence: {''.join(x for x in self.seq if x not in self.model.alphabet)}")
27
 
 
40
  self.sub.append(f"{src_c}{resi}{trg_c}")
41
  self.resi.append(resi)
42
  elif all(match(r'\d+', x) for x in self.trg):
43
+ self.mode = 'SMS'
44
  trh = []
45
  for resi in map(int, self.trg):
46
  if resi < 1 or resi > len(self.seq):
 
64
  raise RuntimeError(f"Unrecognised input substitution: {self.seq[int(''.join(resi_str))]}{int(''.join(resi_str))} /= {s}{int(''.join(resi_str))}")
65
  self.trg = trh
66
  else:
67
+ self.mode = 'DMS'
68
  self.trg = []
69
  for resi, src_c in enumerate(self.seq, 1):
70
  for trg_c in self.AA.replace(src_c, ''):
 
89
 
90
  def parse_output(self) -> None:
91
  "Format output data for visualisation"
92
+ if self.mode == "DMS":
93
+ self._render_dms()
94
  self.out.to_csv(self.out_csv, float_format='%.2f')
95
+ elif self.mode == "SMS":
96
+ self._sort_sms()
97
  self._style_and_save()
98
  elif self.mode == "MUT":
99
  self.out = self.out.sort_values(self.model_name, ascending=False)
 
108
  .background_gradient(cmap="RdYlGn", vmax=8, vmin=-8).hide(axis=0).hide(axis=1))
109
  self.out.to_csv(self.out_csv, float_format='%.2f', index=False, header=False)
110
 
111
+ def _sort_sms(self):
112
+ "Sort SMS output by residue then score, top 19 per position, reshaped to columns"
113
  self.out = (self.out.assign(resi=self.out['0'].str.extract(r'(\d+)', expand=False).astype(int))
114
  .sort_values(["resi", self.model_name], ascending=[True,False])
115
  .groupby(["resi"]).head(19).drop(["resi"], axis=1))
116
  self.out = pd.concat([self.out.iloc[19*x:19*(x+1)].reset_index(drop=True) for x in range(self.out.shape[0]//19)]
117
  , axis=1).set_axis(range(self.out.shape[0]//19*2), axis="columns")
118
 
119
+ def _render_dms(self):
120
+ "Build DMS heatmap from scored mutations"
121
  # Group by residue, keep top 19 (all alternatives)
122
  grouped = (self.out.assign(resi=self.resi_cycle())
123
  .groupby("resi").head(19))
 
152
  return (self.resi * (len(self.out)//len(self.resi) + 1))[:len(self.out)]
153
 
154
  def _plot_heatmap(self, ncols, nrows):
155
+ "Render DMS heatmap to PNG"
156
  kw = dict(cmap="RdBu", cbar=False, square=True, xticklabels=1, yticklabels=1
157
  , center=0, fmt='s', annot_kws={"size": "xx-large"})
158
  annotate = lambda df: df.map(lambda x: ' ' if x != 0 else '\u00b7')
 
196
 
197
  @property
198
  def image(self):
199
+ "return PNG path (DMS) or Styler object (SMS/MUT)"
200
  if self.out_table is not None:
201
  return self.out_table
202
  return self.out_img_path
instructions.md CHANGED
@@ -6,7 +6,7 @@ If the server remains idle for a period, it will enter standby mode. Running a c
6
 
7
  ## Input
8
 
9
- **Sequence**: Enter the full amino acid sequence to be analyzed in the **Sequence** text box.
10
  Note: While jolly characters (e.g., `-X.B`) can be included, they currently cannot be visualized.
11
 
12
  **Substitutions**: Specify the substitutions you wish to test in the **Substitutions** box. The tool supports three running modes based on your input:
 
6
 
7
  ## Input
8
 
9
+ **Sequence**: Enter the full amino acid sequence to be analyzed in the **Sequence** text box. Plain sequences or FASTA format (with `>` header line) are accepted.
10
  Note: While jolly characters (e.g., `-X.B`) can be included, they currently cannot be visualized.
11
 
12
  **Substitutions**: Specify the substitutions you wish to test in the **Substitutions** box. The tool supports three running modes based on your input:
model.py CHANGED
@@ -64,8 +64,8 @@ class ESMModel:
64
 
65
  # Vectorized scoring: extract WT chars, indices, mutant chars from mutation strings
66
  muts = data.sub['0'].values
67
- wt_ids = torch.tensor([self[c] for c in muts], device=self.device)
68
- idxs = torch.tensor([[int(d) for d in m if d.isdigit()] for m in muts], device=self.device).squeeze(-1) - 1
69
  mt_ids = torch.tensor([self[m[-1]] for m in muts], device=self.device)
70
  scores = token_probs.to(self.device)[0, 1 + idxs, mt_ids] - token_probs.to(self.device)[0, 1 + idxs, wt_ids]
71
  data.out[self.model_name] = scores.cpu().numpy()
 
64
 
65
  # Vectorized scoring: extract WT chars, indices, mutant chars from mutation strings
66
  muts = data.sub['0'].values
67
+ wt_ids = torch.tensor([self[m[0]] for m in muts], device=self.device)
68
+ idxs = torch.tensor([int(''.join(c for c in m if c.isdigit())) for m in muts], device=self.device) - 1
69
  mt_ids = torch.tensor([self[m[-1]] for m in muts], device=self.device)
70
  scores = token_probs.to(self.device)[0, 1 + idxs, mt_ids] - token_probs.to(self.device)[0, 1 + idxs, wt_ids]
71
  data.out[self.model_name] = scores.cpu().numpy()
test/__init__.py ADDED
File without changes
test/test_app.py ADDED
@@ -0,0 +1,144 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import os
2
+ import sys
3
+ import tempfile
4
+ import unittest
5
+ from unittest.mock import MagicMock, patch
6
+
7
+ sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
8
+
9
+
10
+ class FakeModel:
11
+ "Minimal model stub for testing without GPU"
12
+ def __init__(self):
13
+ self.alphabet = {c: i + 3 for i, c in enumerate("ACDEFGHIKLMNPQRSTVWY")}
14
+ self.alphabet.update({'<pad>': 1, '<eos>': 2, '<cls>': 0})
15
+
16
+
17
+ class TestAppValidation(unittest.TestCase):
18
+ "app() input guards raise gradio.Error on bad input"
19
+
20
+ @patch('data.ModelFactory', return_value=FakeModel())
21
+ def test_empty_sequence_raises(self, _mock_factory):
22
+ from app import app
23
+ from gradio import Error
24
+ with self.assertRaises(Error) as ctx:
25
+ app("", "scan", "test/model", True)
26
+ self.assertIn("empty", str(ctx.exception).lower())
27
+
28
+ @patch('data.ModelFactory', return_value=FakeModel())
29
+ def test_empty_substitutions_raises(self, _mock_factory):
30
+ from app import app
31
+ from gradio import Error
32
+ with self.assertRaises(Error) as ctx:
33
+ app("MVEQYLLEAI", "", "test/model", True)
34
+ self.assertIn("substitution", str(ctx.exception).lower())
35
+
36
+
37
+ class TestDmsLayout(unittest.TestCase):
38
+ "ncols/nrows layout calculation logic from _render_dms"
39
+
40
+ def _calc_layout(self, num_cols):
41
+ """Replicate the ncols selection logic from data.py"""
42
+ from math import ceil
43
+ ncols = min([d for d in range(1, num_cols + 1) if num_cols % d == 0 and 30 <= d <= 60] or [60],
44
+ key=lambda x: abs(x - 60))
45
+ nrows = ceil(num_cols / ncols)
46
+ while num_cols / ncols < nrows and ncols > 45 and ncols * nrows >= num_cols:
47
+ ncols -= 1
48
+ ncols += 1
49
+ return ncols, nrows
50
+
51
+ def test_short_sequence(self):
52
+ # 7-col sequence (like examples in app)
53
+ ncols, nrows = self._calc_layout(7)
54
+ self.assertEqual(nrows, 1)
55
+ self.assertGreater(ncols, 0)
56
+
57
+ def test_medium_sequence(self):
58
+ ncols, nrows = self._calc_layout(100)
59
+ self.assertGreaterEqual(ncols, 30)
60
+ self.assertTrue(ncols * nrows >= 100) # grid must cover all columns
61
+
62
+ def test_long_sequence_fits_grid(self):
63
+ ncols, nrows = self._calc_layout(300)
64
+ self.assertTrue(ncols * nrows >= 300)
65
+
66
+ def test_very_long_sequence(self):
67
+ ncols, nrows = self._calc_layout(600)
68
+ self.assertTrue(ncols * nrows >= 600)
69
+
70
+
71
+ class TestDmsRender(unittest.TestCase):
72
+ "Full _render_dms pipeline produces PNG file"
73
+
74
+ @patch('data.ModelFactory', return_value=FakeModel())
75
+ def test_render_creates_png(self, _mock_factory):
76
+ from data import Data
77
+ import pandas as pd
78
+ with tempfile.TemporaryDirectory() as tmpdir:
79
+ out_path = os.path.join(tmpdir, 'test_heatmap.png')
80
+ csv_path = os.path.join(tmpdir, 'test_out.csv')
81
+ d = object.__new__(Data)
82
+ d.model_name = 'test'
83
+ d.seq = "MVEQYLL"
84
+ d.mode = 'DMS'
85
+ d.resi = list(range(1, len(d.seq) + 1))
86
+ # Build a minimal scored output matching DMS expectations
87
+ AA = "ACDEFGHIKLMNPQRSTVWY"
88
+ rows = []
89
+ for i, src_c in enumerate(d.seq, 1):
90
+ for trg_c in AA.replace(src_c, ''):
91
+ rows.append(f"{src_c}{i}{trg_c}")
92
+ scores = [float(i % 5 - 2) for i in range(len(rows))]
93
+ d.out = pd.DataFrame({'0': rows, 'test': scores})
94
+ d.out_img_path = out_path
95
+ d.out_csv = csv_path
96
+ d._render_dms()
97
+ self.assertTrue(os.path.exists(out_path), f"PNG not created at {out_path}")
98
+ self.assertTrue(os.path.getsize(out_path) > 0, "PNG is empty")
99
+
100
+
101
+ class TestParseOutputDispatch(unittest.TestCase):
102
+ "parse_output routes to correct handler per mode"
103
+
104
+ @patch('data.ModelFactory', return_value=FakeModel())
105
+ def test_dms_dispatch_calls_render(self, _mock_factory):
106
+ from data import Data
107
+ import pandas as pd
108
+ with tempfile.TemporaryDirectory() as tmpdir:
109
+ d = object.__new__(Data)
110
+ d.model_name = 'test'
111
+ d.mode = 'DMS'
112
+ d.seq = "MV"
113
+ d.resi = [1, 2]
114
+ AA = "ACDEFGHIKLMNPQRSTVWY"
115
+ rows = []
116
+ for src_c in "MV":
117
+ pos = {"M": 1, "V": 2}[src_c]
118
+ for trg_c in AA.replace(src_c, ''):
119
+ rows.append(f"{src_c}{pos}{trg_c}")
120
+ d.out = pd.DataFrame({'0': rows, 'test': list(range(len(rows)))})
121
+ d.out_img_path = os.path.join(tmpdir, 't.png')
122
+ d.out_csv = os.path.join(tmpdir, 't.csv')
123
+ d.parse_output()
124
+ self.assertTrue(os.path.exists(d.out_img_path))
125
+
126
+ @patch('data.ModelFactory', return_value=FakeModel())
127
+ def test_mut_dispatch_sorts_and_styles(self, _mock_factory):
128
+ from data import Data
129
+ import pandas as pd
130
+ with tempfile.TemporaryDirectory() as tmpdir:
131
+ d = object.__new__(Data)
132
+ d.model_name = 'test'
133
+ d.mode = 'MUT'
134
+ d.out = pd.DataFrame({'0': ['V2A', 'E5K'], 'test': [3.0, -1.0]})
135
+ d.out_csv = os.path.join(tmpdir, 'm.csv')
136
+ d.out_img_path = os.path.join(tmpdir, 'm.png')
137
+ d.parse_output()
138
+ # MUT sorts descending by score; V2A (3.0) should be first
139
+ self.assertEqual(d.out.iloc[0]['0'], 'V2A')
140
+ self.assertIsNotNone(d.out_table)
141
+
142
+
143
+ if __name__ == '__main__':
144
+ unittest.main()
test/test_data.py ADDED
@@ -0,0 +1,297 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import os
2
+ import sys
3
+ import tempfile
4
+ import unittest
5
+ from unittest.mock import MagicMock, patch
6
+
7
+ # Ensure project root is on path
8
+ sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
9
+
10
+
11
+ class FakeModel:
12
+ "Minimal model stub for testing without GPU"
13
+ def __init__(self):
14
+ self.alphabet = {c: i + 3 for i, c in enumerate("ACDEFGHIKLMNPQRSTVWY")}
15
+ self.alphabet.update({'<pad>': 1, '<eos>': 2, '<cls>': 0})
16
+
17
+
18
+ class TestParseSeq(unittest.TestCase):
19
+ "Sequence parsing (plain and FASTA)"
20
+
21
+ @patch('data.ModelFactory', return_value=FakeModel())
22
+ def test_plain_sequence(self, _mock_factory):
23
+ from data import Data
24
+ d = object.__new__(Data)
25
+ d.model = FakeModel()
26
+ d.parse_seq("MVEQYLLEAI")
27
+ self.assertEqual(d.seq, "MVEQYLLEAI")
28
+
29
+ @patch('data.ModelFactory', return_value=FakeModel())
30
+ def test_fasta_single_line(self, _mock_factory):
31
+ from data import Data
32
+ d = object.__new__(Data)
33
+ d.model = FakeModel()
34
+ d.parse_seq(">my protein\nMVEQYLLEAI")
35
+ self.assertEqual(d.seq, "MVEQYLLEAI")
36
+ self.assertFalse(d.seq.startswith('>'))
37
+
38
+ @patch('data.ModelFactory', return_value=FakeModel())
39
+ def test_fasta_multi_line(self, _mock_factory):
40
+ from data import Data
41
+ d = object.__new__(Data)
42
+ d.model = FakeModel()
43
+ d.parse_seq(">seq1\nMV EQ YL LE AI")
44
+ self.assertEqual(d.seq, "MVEQYLLEAI")
45
+
46
+ @patch('data.ModelFactory', return_value=FakeModel())
47
+ def test_lowercase_converted_to_upper(self, _mock_factory):
48
+ from data import Data
49
+ d = object.__new__(Data)
50
+ d.model = FakeModel()
51
+ d.parse_seq("mveqylleai")
52
+ self.assertEqual(d.seq, "MVEQYLLEAI")
53
+
54
+ @patch('data.ModelFactory', return_value=FakeModel())
55
+ def test_whitespace_stripped(self, _mock_factory):
56
+ from data import Data
57
+ d = object.__new__(Data)
58
+ d.model = FakeModel()
59
+ d.parse_seq(" MV EQ YL \n LE AI ")
60
+ self.assertEqual(d.seq, "MVEQYLLEAI")
61
+
62
+ @patch('data.ModelFactory', return_value=FakeModel())
63
+ def test_invalid_characters_raises(self, _mock_factory):
64
+ from data import Data
65
+ d = object.__new__(Data)
66
+ d.model = FakeModel()
67
+ with self.assertRaises(RuntimeError):
68
+ d.parse_seq("MVXYZ")
69
+
70
+ @patch('data.ModelFactory', return_value=FakeModel())
71
+ def test_empty_sequence_raises(self, _mock_factory):
72
+ from data import Data
73
+ d = object.__new__(Data)
74
+ d.model = FakeModel()
75
+ with self.assertRaises(RuntimeError):
76
+ d.parse_seq("")
77
+
78
+ @patch('data.ModelFactory', return_value=FakeModel())
79
+ def test_whitespace_only_raises(self, _mock_factory):
80
+ from data import Data
81
+ d = object.__new__(Data)
82
+ d.model = FakeModel()
83
+ with self.assertRaises(RuntimeError):
84
+ d.parse_seq(" \n ")
85
+
86
+ @patch('data.ModelFactory', return_value=FakeModel())
87
+ def test_fasta_header_only_raises(self, _mock_factory):
88
+ from data import Data
89
+ d = object.__new__(Data)
90
+ d.model = FakeModel()
91
+ with self.assertRaises(RuntimeError):
92
+ d.parse_seq(">just a header")
93
+
94
+
95
+ class TestParseSub(unittest.TestCase):
96
+ "Substitution parsing and mode detection"
97
+
98
+ def setUp(self):
99
+ self.seq = "MVEQYLL"
100
+
101
+ @patch('data.ModelFactory', return_value=FakeModel())
102
+ def test_dms_mode(self, _mock_factory):
103
+ from data import Data
104
+ d = object.__new__(Data)
105
+ d.model = FakeModel()
106
+ d.seq = self.seq
107
+ d.parse_sub("2 5")
108
+ self.assertEqual(d.mode, 'SMS')
109
+ self.assertEqual(len(d.resi), 2)
110
+ self.assertIn(2, d.resi)
111
+ self.assertIn(5, d.resi)
112
+ # Each position has 19 alternatives (20 AA minus WT)
113
+ self.assertEqual(len(d.sub), 38)
114
+
115
+ @patch('data.ModelFactory', return_value=FakeModel())
116
+ def test_mut_explicit_mode(self, _mock_factory):
117
+ from data import Data
118
+ d = object.__new__(Data)
119
+ d.model = FakeModel()
120
+ d.seq = self.seq
121
+ d.parse_sub("V2A E3K")
122
+ self.assertEqual(d.mode, 'MUT')
123
+ self.assertEqual(len(d.sub), 2)
124
+ self.assertEqual(list(d.sub['0']), ['V2A', 'E3K'])
125
+
126
+ @patch('data.ModelFactory', return_value=FakeModel())
127
+ def test_mut_seq_vs_seq_mode(self, _mock_factory):
128
+ from data import Data
129
+ d = object.__new__(Data)
130
+ d.model = FakeModel()
131
+ d.seq = self.seq
132
+ d.parse_sub("MVEQYAL") # same length, differs at pos 6
133
+ self.assertEqual(d.mode, 'MUT')
134
+ self.assertTrue(any('L6A' in str(s) for s in d.sub['0']))
135
+
136
+ @patch('data.ModelFactory', return_value=FakeModel())
137
+ def test_tms_fallback_mode(self, _mock_factory):
138
+ from data import Data
139
+ d = object.__new__(Data)
140
+ d.model = FakeModel()
141
+ d.seq = self.seq
142
+ d.parse_sub("deep mutational scanning")
143
+ self.assertEqual(d.mode, 'DMS')
144
+ # All positions x all alternatives
145
+ self.assertEqual(len(d.resi), len(self.seq))
146
+
147
+ @patch('data.ModelFactory', return_value=FakeModel())
148
+ def test_dms_position_out_of_range(self, _mock_factory):
149
+ from data import Data
150
+ d = object.__new__(Data)
151
+ d.model = FakeModel()
152
+ d.seq = self.seq
153
+ with self.assertRaises(RuntimeError):
154
+ d.parse_sub("999")
155
+
156
+ @patch('data.ModelFactory', return_value=FakeModel())
157
+ def test_mut_wrong_wt_raises(self, _mock_factory):
158
+ from data import Data
159
+ d = object.__new__(Data)
160
+ d.model = FakeModel()
161
+ d.seq = self.seq
162
+ # V2A but position 2 is V — correct; try A2K where pos 2 is V not A
163
+ with self.assertRaises(RuntimeError):
164
+ d.parse_sub("A2K")
165
+
166
+
167
+ class TestSmsSort(unittest.TestCase):
168
+ "SMS output sorting and reshaping"
169
+
170
+ @patch('data.ModelFactory', return_value=FakeModel())
171
+ def test_sort_preserves_top_19_per_position(self, mock_factory):
172
+ from data import Data
173
+ import pandas as pd
174
+ d = object.__new__(Data)
175
+ d.model_name = 'test'
176
+ # _sort_sms requires exactly 19 rows per position; build minimal valid input
177
+ AA = "ACDEFGHIKLMNPQRSTVWY"
178
+ subs_pos2 = [f'V2{a}' for a in AA.replace('V', '')] # 19 alternatives (skip WT V)
179
+ subs_pos5 = [f'L5{a}' for a in AA.replace('L', '')] # 19 alternatives (skip WT L)
180
+ scores_p2 = list(range(19))
181
+ scores_p5 = list(range(18, -1, -1))
182
+ d.out = pd.DataFrame({
183
+ '0': subs_pos2 + subs_pos5,
184
+ 'test': scores_p2[:19] + scores_p5
185
+ })
186
+ d.resi = [2, 5]
187
+ d._sort_sms()
188
+ self.assertEqual(d.out.shape[0], 19) # 19 rows after reshape
189
+
190
+
191
+ class TestResidueCycle(unittest.TestCase):
192
+ "resi_cycle helper"
193
+
194
+ @patch('data.ModelFactory', return_value=FakeModel())
195
+ def test_resi_cycles_correctly(self, _mock_factory):
196
+ from data import Data
197
+ import pandas as pd
198
+ d = object.__new__(Data)
199
+ d.model = FakeModel()
200
+ d.resi = [1, 3, 5]
201
+ d.out = pd.DataFrame({'x': range(9)})
202
+ cycle = d.resi_cycle()
203
+ self.assertEqual(cycle, [1, 3, 5, 1, 3, 5, 1, 3, 5])
204
+
205
+ @patch('data.ModelFactory', return_value=FakeModel())
206
+ def test_resi_truncates_to_length(self, _mock_factory):
207
+ from data import Data
208
+ import pandas as pd
209
+ d = object.__new__(Data)
210
+ d.model = FakeModel()
211
+ d.resi = [1, 2]
212
+ d.out = pd.DataFrame({'x': range(5)})
213
+ cycle = d.resi_cycle()
214
+ self.assertEqual(len(cycle), 5)
215
+ self.assertEqual(cycle[:4], [1, 2, 1, 2])
216
+
217
+
218
+ class TestStyleAndSave(unittest.TestCase):
219
+ "Table styling and CSV output"
220
+
221
+ @patch('data.ModelFactory', return_value=FakeModel())
222
+ def test_style_creates_styler_and_csv(self, _mock_factory):
223
+ from data import Data
224
+ import pandas as pd
225
+ with tempfile.TemporaryDirectory() as tmpdir:
226
+ d = object.__new__(Data)
227
+ d.model_name = 'test'
228
+ d.out = pd.DataFrame({'0': ['V2A'], 'test': [1.5]})
229
+ d.out_csv = os.path.join(tmpdir, 'out.csv')
230
+ d._style_and_save()
231
+ self.assertIsNotNone(d.out_table)
232
+ self.assertTrue(os.path.exists(d.out_csv))
233
+
234
+
235
+ class TestParseSubEdgeCases(unittest.TestCase):
236
+ "parse_sub edge cases"
237
+
238
+ @patch('data.ModelFactory', return_value=FakeModel())
239
+ def test_dms_duplicate_positions_allowed(self, _mock_factory):
240
+ from data import Data
241
+ d = object.__new__(Data)
242
+ d.model = FakeModel()
243
+ d.seq = "MVEQYLL"
244
+ d.parse_sub("3 3")
245
+ # Duplicate positions produce duplicate mutation sets
246
+ self.assertEqual(len(d.resi), 2)
247
+ self.assertEqual(len(d.sub), 38)
248
+
249
+ @patch('data.ModelFactory', return_value=FakeModel())
250
+ def test_mut_identical_sequences_no_muts(self, _mock_factory):
251
+ from data import Data
252
+ d = object.__new__(Data)
253
+ d.model = FakeModel()
254
+ d.seq = "MV"
255
+ d.parse_sub("MV") # identical — no differences
256
+ self.assertEqual(d.mode, 'MUT')
257
+ self.assertEqual(len(d.sub), 0)
258
+
259
+ @patch('data.ModelFactory', return_value=FakeModel())
260
+ def test_single_residue_sequence_tms(self, _mock_factory):
261
+ from data import Data
262
+ d = object.__new__(Data)
263
+ d.model = FakeModel()
264
+ d.seq = "A"
265
+ d.parse_sub("scan all")
266
+ self.assertEqual(d.mode, 'DMS')
267
+ self.assertEqual(len(d.resi), 1)
268
+ self.assertEqual(len(d.sub), 19)
269
+
270
+
271
+ class TestAppReturns(unittest.TestCase):
272
+ "app.py callback return values per mode"
273
+
274
+ @patch('data.ModelFactory', return_value=FakeModel())
275
+ def test_image_property_returns_path_for_tms(self, _mock_factory):
276
+ from data import Data
277
+ d = object.__new__(Data)
278
+ d.out_table = None
279
+ d.out_img_path = 'out.png'
280
+ self.assertIsInstance(d.image, str)
281
+ self.assertEqual(d.image, 'out.png')
282
+
283
+ @patch('data.ModelFactory', return_value=FakeModel())
284
+ def test_image_property_returns_styler_for_dms(self, _mock_factory):
285
+ from data import Data
286
+ import pandas as pd
287
+ d = object.__new__(Data)
288
+ d.model_name = 'test'
289
+ d.out_csv = '/tmp/x.csv'
290
+ d.out = pd.DataFrame({'0': ['V2A'], 'test': [1.5]})
291
+ d._style_and_save()
292
+ # After styling, image returns Styler, not path
293
+ self.assertNotIsInstance(d.image, str)
294
+
295
+
296
+ if __name__ == '__main__':
297
+ unittest.main()
test/test_model.py ADDED
@@ -0,0 +1,109 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import json
2
+ import os
3
+ import sys
4
+ import tempfile
5
+ import unittest
6
+ from unittest.mock import MagicMock, patch
7
+
8
+ sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
9
+
10
+
11
+ class TestModelFactoryRegistry(unittest.TestCase):
12
+ "ModelFactory register() and models()"
13
+
14
+ def setUp(self):
15
+ # Isolate _models for each test
16
+ from model import ModelFactory
17
+ self._orig = dict(ModelFactory._models)
18
+
19
+ def tearDown(self):
20
+ from model import ModelFactory
21
+ ModelFactory._models.clear()
22
+ ModelFactory._models.update(self._orig)
23
+
24
+ @patch('model.HfApi')
25
+ def test_register_adds_new_model(self, _mock_api):
26
+ from model import ModelFactory
27
+ class StubModel:
28
+ pass
29
+ ModelFactory.register("stub/model", StubModel)
30
+ self.assertIn("stub/model", ModelFactory.models())
31
+ self.assertEqual(ModelFactory._models["stub/model"], StubModel)
32
+
33
+ @patch('model.HfApi')
34
+ def test_models_returns_list_of_keys(self, _mock_api):
35
+ from model import ModelFactory
36
+ keys = ModelFactory.models()
37
+ self.assertIsInstance(keys, list)
38
+ self.assertTrue(len(keys) > 0)
39
+
40
+ @patch('model.HfApi')
41
+ def test_register_overwrites_existing(self, _mock_api):
42
+ from model import ModelFactory
43
+ existing_key = ModelFactory.models()[0]
44
+ class Replacement:
45
+ pass
46
+ ModelFactory.register(existing_key, Replacement)
47
+ self.assertIs(ModelFactory._models[existing_key], Replacement)
48
+
49
+
50
+ class TestFetchCache(unittest.TestCase):
51
+ "_fetch_model_ids cache read/write/TTL"
52
+
53
+ @patch('model.time')
54
+ @patch('builtins.open', new_callable=unittest.mock.mock_open, read_data=json.dumps({"ts": 100, "ids": ["a/b"]}))
55
+ def test_cache_hit_within_ttl(self, mock_file, mock_time):
56
+ mock_time.return_value = 200 # 100s ago, within 3600 TTL
57
+ with patch('os.path.exists', return_value=True):
58
+ from model import _fetch_model_ids
59
+ result = _fetch_model_ids()
60
+ self.assertEqual(result, ["a/b"])
61
+
62
+ @patch('model.time')
63
+ @patch('builtins.open', new_callable=unittest.mock.mock_open, read_data=json.dumps({"ts": 100, "ids": ["a/b"]}))
64
+ def test_cache_miss_expired(self, mock_file, mock_time):
65
+ mock_time.return_value = 5000 # well past 3600 TTL
66
+ fallback_ids = ["fallback/model"]
67
+ mock_api = MagicMock()
68
+ mock_api.list_models.side_effect = [
69
+ [MagicMock(id="fallback/model")],
70
+ []
71
+ ]
72
+ with patch('os.path.exists', return_value=True), \
73
+ patch('model.HfApi', return_value=mock_api), \
74
+ tempfile.TemporaryDirectory() as tmpdir:
75
+ import model
76
+ orig_cache = model._MODEL_CACHE
77
+ model._MODEL_CACHE = os.path.join(tmpdir, "cache.json")
78
+ try:
79
+ result = model._fetch_model_ids()
80
+ self.assertIn("fallback/model", result)
81
+ finally:
82
+ model._MODEL_CACHE = orig_cache
83
+
84
+
85
+ class TestScoringIndexExtraction(unittest.TestCase):
86
+ "Verify digit extraction from mutation strings (used in vectorized scoring)"
87
+
88
+ def _extract_idx(self, m):
89
+ """Replicate the logic from ESMModel.run_model"""
90
+ return int(''.join(c for c in m if c.isdigit())) - 1
91
+
92
+ def test_simple_two_digit(self):
93
+ self.assertEqual(self._extract_idx("V2A"), 1)
94
+
95
+ def test_three_digit_position(self):
96
+ self.assertEqual(self._extract_idx("R218K"), 217)
97
+
98
+ def test_single_digit_position(self):
99
+ self.assertEqual(self._extract_idx("M1D"), 0)
100
+
101
+ def test_various_positions(self):
102
+ cases = [("A10F", 9), ("L100W", 99), ("G500S", 499)]
103
+ for mut, expected in cases:
104
+ with self.subTest(mut=mut):
105
+ self.assertEqual(self._extract_idx(mut), expected)
106
+
107
+
108
+ if __name__ == '__main__':
109
+ unittest.main()