-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
1474 lines (1310 loc) · 69 KB
/
Copy pathapp.py
File metadata and controls
1474 lines (1310 loc) · 69 KB
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
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Gradio GUI for MolDeTr — branded workbench with Detect + Simulate tabs.
Two tabs share the paper-branded theme (theme.py) and, on Detect, the interactive Plotly spectrum
(plotting.py):
- **Detect**: load a 1-D ¹H NMR window (.npz/.npy), get the multiplet assignment table + an
interactive annotated spectrum (drag to box-zoom, double-click resets), with CSV / JSON export.
- **Simulate**: build a known spin system on the model's grid, optionally add training-range
distortions, detect, and compare against ground truth. Both tabs render with Plotly
(``app_ui.plotting``); the ground-truth comparison uses ``comparison_figure``.
Run locally:
pip install -e ".[app]"
python app.py
Deploys unchanged as a Hugging Face Space (set the checkpoint via ``MOLDETR_CHECKPOINT`` or place it
at ``moldetr/model/``). Weights are on Zenodo (DOI 10.5281/zenodo.21217102). ``theme.py`` and
``plotting.py`` must sit next to ``app.py``.
MolDeTr is research code accompanying the paper: it handles congested, strongly-coupled ¹H NMR
spectra and is largely field-agnostic — it works in Hz, so it was tested across 80–600 MHz (and
simulated down to ~5 MHz). Results can deviate for inputs outside its trained regime — unusual
distortions, non-standard pulse sequences or processing, mixtures, or windows wider than 1200 Hz.
``max J`` is the dominant coupling per multiplet; the full set comes from the committed
``structured_output`` path. See docs/SCOPE.md.
"""
from __future__ import annotations
import json
import os
import sys
import tempfile
import warnings
from pathlib import Path
from typing import Any, Literal
import gradio as gr
import numpy as np
import pandas as pd
from numpy.typing import NDArray
from moldetr.distort import distort
from moldetr.inference import build_model, load_checkpoint, run
from moldetr.postprocess import decode_predictions, load_extrema
from moldetr.simulate import COUPLING_EPS_HZ, coupling_blocks, simulate_systems
from moldetr.validation import INPUT_LENGTH, POINTS_PER_HZ, validate_spectrum
from app_ui import grading
from app_ui.plotting import ( # BRAND: interactive Plotly plots
assignment_rows,
comparison_figure,
spectrum_figure,
)
from app_ui.theme import (
CUSTOM_CSS,
HEADER_HTML,
MOLDETR_THEME,
) # BRAND: palette / header / theme (at launch)
ROOT = Path(__file__).resolve().parent
CHECKPOINT = os.environ.get(
"MOLDETR_CHECKPOINT", str(ROOT / "moldetr" / "model" / "model_spin_system_ABCDEFG_exp2.pth")
)
EXTREMA = str(ROOT / "moldetr" / "assets" / "extrema.txt")
# The simulate->predict round-trip (phenotypes, coupling-matrix helper, GT matching) lives in the
# scripts/ deliverable; add it to the path so the "Simulate" tab reuses it rather than duplicating.
sys.path.insert(0, str(ROOT / "scripts"))
import simulate_and_predict as sp # noqa: E402 (scripts/ was just placed on sys.path above)
AUTO, MANUAL, NONE = "Auto (from file)", "Manual (window ppm)", "None (report in Hz)"
_MODEL = None
def _get_model():
global _MODEL
if _MODEL is None:
_MODEL = load_checkpoint(build_model(), CHECKPOINT)
return _MODEL
def _checkpoint_error_message(exc: Exception) -> str:
"""Render a checkpoint-load refusal for the status box.
`load_checkpoint` raises `RuntimeError` when the trust gate refuses a file, and that message is
the only place `MOLDETR_ALLOW_UNTRUSTED_CHECKPOINT` — the documented way to run weights you
trained yourself — is named. Unwrapped, it surfaced as a bare Gradio "Error" toast, so the one
piece of information needed to resolve the situation was the piece that never arrived.
Fenced rather than interpolated into the sentence: the gate's message is multi-line and carries
both MD5s, and markdown collapses those lines into an unreadable run precisely where the user
has to compare two hex digests.
"""
return f"⚠ Could not load the checkpoint:\n\n```\n{exc}\n```"
EXAMPLES_DIR = ROOT / "examples"
def _is_bundled_example(path: Path) -> bool:
"""True only for a file that ships inside this repo's ``examples/`` directory.
Relative paths resolve against ``ROOT``, **not** the process CWD. ``build_ui()`` wires the
examples as relative paths (``"examples/roi_S10_example.npz"``), so a CWD-relative resolution
silently disarms the gate whenever the app is launched from another directory — invisible
today only because no bundled example needs pickle.
This cannot widen the gate: a relative path can only resolve inside ``examples/`` if such a
file actually ships there, and uploads always arrive as absolute temp paths.
"""
try:
candidate = path if path.is_absolute() else ROOT / path
return candidate.resolve().is_relative_to(EXAMPLES_DIR)
except (OSError, ValueError): # unresolvable path (broken link, bad drive) → not ours
return False
def _load(path: str, *, trusted: bool = False):
"""Load a spectrum (+ ppm calibration if present) from .npz/.npy. Array is returned as-is
(possibly complex) so the caller can surface the dtype; validation takes the real part.
``trusted`` is what enables pickle, and only files we ship get it. Unpickling executes code
carried in the archive, so an uploaded ``.npz`` must never take that path. The gate costs
nothing: the one branch that touches an object array is the ``metadata`` fallback below, which
is reached only when ``ppm_axis_padded`` is absent — and every bundled example has that axis.
"""
p = Path(path)
cal: dict = {}
if p.suffix == ".npz":
# allow_pickle is gated on provenance, never on the caller's convenience.
data = np.load(p, allow_pickle=trusted)
# Prefer the per-point ppm axis (correct for the ROI); metadata left/right_ppm span the full
# spectrum and would mis-place peaks, so only fall back to them if the axis is absent.
if "ppm_axis_padded" in data:
axis = np.asarray(data["ppm_axis_padded"], dtype=float)
cal = {"ppm_left": float(axis[0]), "ppm_right": float(axis[-1])}
elif "metadata" in data:
md = data["metadata"].item()
cal = {"ppm_left": md.get("left_ppm"), "ppm_right": md.get("right_ppm")}
for key in ("spectrum_padded", "spec"):
if key in data:
return np.asarray(data[key]), cal
return np.asarray(data[list(data.keys())[0]]), cal
return np.asarray(np.load(p)), cal
def _resolve_points_per_hz(points_per_hz) -> float:
"""Digital resolution in points/Hz, or ``ValueError`` naming what is wrong with it.
A blank field means "unset" and falls back to the default. Zero and negatives are *stated*
values that cannot be right, and the old ``float(x) if x else DEFAULT`` silently replaced 0
with 5.12 — so clearing the box produced confident, wrongly-scaled results. A negative value
is truthy and sailed through entirely, mirroring the axis and yielding negative line widths.
"""
if points_per_hz is None or points_per_hz == "":
return POINTS_PER_HZ
pph = float(points_per_hz)
if pph <= 0:
raise ValueError("digital resolution must be positive (points/Hz)")
return pph
def _spec_report(file, points_per_hz) -> str:
"""Post-upload input check — same logic, glyphs instead of emoji.""" # BRAND
if file is None:
return ""
path = file if isinstance(file, str) else file.name
try:
raw, cal = _load(path, trusted=_is_bundled_example(Path(path)))
except Exception as exc: # noqa: BLE001
return f"⚠ Could not read the file: {exc}"
try:
pph = _resolve_points_per_hz(points_per_hz)
except ValueError as exc:
return f"⚠ Invalid input: {exc}"
arr = np.asarray(raw).ravel()
n = arr.shape[0]
window = INPUT_LENGTH / pph
ok_len = "✓" if n == INPUT_LENGTH else f"✗ needs exactly {INPUT_LENGTH}"
ok_res = (
"✓"
if abs(pph - POINTS_PER_HZ) <= 0.01
else "⚠ not 1200 Hz, so predictions may be unreliable"
)
dtype = "complex; the real (absorption) part is used" if np.iscomplexobj(arr) else "real ✓"
finite = "✓" if np.all(np.isfinite(np.real(arr))) else "✗ contains NaN/Inf"
axis = "yes ✓ (Auto works)" if cal.get("ppm_left") is not None else "no; use Manual or None"
return (
"**Input check**\n"
f"- Length: **{n}** points {ok_len}\n"
f"- Resolution: **{pph:g}** points/Hz → **{window:.0f} Hz** window {ok_res}\n"
f"- Data type: {dtype}\n"
f"- Finite values: {finite}\n"
f"- ppm axis in file: {axis}"
)
def predict(file, threshold, ppm_mode, manual_left, manual_right, points_per_hz):
"""Run detection and return (assignment table, annotated Plotly plot, status message)."""
if file is None:
return None, None, "Load a `.npz`/`.npy` spectrum, or pick an example below."
if not Path(CHECKPOINT).exists():
return (
None,
None,
(
f"Checkpoint not found at `{CHECKPOINT}`. "
"Download it from Zenodo (10.5281/zenodo.21217102) into `moldetr/model/`."
),
)
path = file if isinstance(file, str) else file.name
# `_spec_report` has always guarded this; `predict` did not, so the same bad file that produced
# a tidy "⚠ Could not read" above the button rendered a Python traceback below it.
try:
raw, cal = _load(path, trusted=_is_bundled_example(Path(path)))
except Exception as exc: # noqa: BLE001 - any unreadable file must surface as a message
return None, None, f"⚠ Could not read the file: {exc}"
try:
pph = _resolve_points_per_hz(points_per_hz)
except ValueError as exc:
return None, None, f"Invalid input: {exc}"
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
try:
amplitudes = validate_spectrum(raw, points_per_hz=pph)
except ValueError as exc:
return None, None, f"Invalid spectrum: {exc}"
warn_msg = " ".join(str(w.message) for w in caught)
if ppm_mode == MANUAL and manual_left is not None and manual_right is not None:
ppm_left, ppm_right = float(manual_left), float(manual_right)
elif ppm_mode == AUTO:
ppm_left, ppm_right = cal.get("ppm_left"), cal.get("ppm_right")
else: # NONE, or MANUAL without both bounds -> report shift in Hz
ppm_left = ppm_right = None
try:
model = _get_model()
except RuntimeError as exc: # the checkpoint trust gate; its text names the only remedy
return None, None, _checkpoint_error_message(exc)
preds = decode_predictions(
run(model, amplitudes),
load_extrema(EXTREMA),
pph,
ppm_left=ppm_left,
ppm_right=ppm_right,
threshold=threshold,
)
fig = spectrum_figure(
amplitudes, preds, ppm_left=ppm_left, ppm_right=ppm_right, points_per_hz=pph
) # BRAND: Plotly; Hz axis when no ppm calibration
rows = assignment_rows(preds, ppm_left is not None and ppm_right is not None)
table = pd.DataFrame(rows) if rows else pd.DataFrame()
if preds:
msg = f"Detected **{len(preds)}** multiplet(s). Numbers on the plot match the table rows."
else:
msg = "No multiplets passed the detection threshold. Try lowering it."
if warn_msg:
msg += f"\n\n⚠ {warn_msg}"
return table, fig, msg
_EXPORT_DIR: str | None = None
def _export_dir() -> str:
"""One temp directory per process, reused by every Detect click.
`mkdtemp` *per click* leaked a directory on every detection — unbounded on a process that
stays up for weeks. The two export files are overwritten in place instead.
Reusing one path looks like it should let a later detection hand an earlier user someone
else's numbers, but it cannot: ``gr.DownloadButton`` copies the file into Gradio's own
**content-addressed** cache when the event returns, so the link a user holds points at a hash
of the bytes they were shown, not at this file. Verified in
``test_download_links_are_content_addressed_so_the_shared_dir_is_safe`` — which exists
precisely so a future Gradio that served this path directly would fail loudly here.
"""
global _EXPORT_DIR
if _EXPORT_DIR is None or not os.path.isdir(_EXPORT_DIR):
_EXPORT_DIR = tempfile.mkdtemp(prefix="moldetr_")
return _EXPORT_DIR
def predict_ui(file, threshold, ppm_mode, manual_left, manual_right, points_per_hz):
"""predict() + CSV/JSON export files for the download buttons.""" # NEW
table, fig, msg = predict(file, threshold, ppm_mode, manual_left, manual_right, points_per_hz)
csv_path = json_path = None
if table is not None and not table.empty:
out = _export_dir()
csv_path = os.path.join(out, "moldetr_prediction.csv")
json_path = os.path.join(out, "moldetr_prediction.json")
table.to_csv(csv_path, index=False)
with open(json_path, "w", encoding="utf-8") as fh:
json.dump(table.to_dict(orient="records"), fh, ensure_ascii=False, indent=2)
return (
table,
fig,
msg,
gr.DownloadButton(value=csv_path, interactive=csv_path is not None),
gr.DownloadButton(value=json_path, interactive=json_path is not None),
)
# --- "Simulate" tab: reuse the scripts/ round-trip against the same model + decode + plot ---------
SIMULATE_INTRO = (
"Build a spin system on the model's grid (80 MHz, 15→0 ppm, 6144 pts), distort it within the "
"range the model was trained on, then run the detector on it and compare against ground truth "
"you defined. Start from a known system or edit the matrix directly: shifts on the diagonal, "
"couplings above it. A second, independent system can be switched on below the first and is "
"summed with it at matching per-proton integrals. "
"Once a spectrum is simulated, the distortion sliders re-distort it live — the spin dynamics "
"are not solved again."
)
PHENOTYPE_CHOICES = sorted(sp.PHENOTYPES)
MATRIX_HINT = (
"Each row is one spin. The **diagonal** holds its shift δ in ppm; cells **above** the diagonal "
"hold the coupling J in Hz between that pair. Leave a pair at 0 and the two spins belong to "
"separate spin systems, which are simulated independently and summed. Cells below the diagonal "
"are ignored. For a second molecule, either leave its couplings to this one at 0 here, or use "
"the **Second spin system** panel below — they describe the same computation."
)
#: Largest spin count the matrix editor offers, **per editor**. `simulate` pays 2**n per coupled
#: block, and `MAX_BLOCK_SPINS` caps a single block at 10, so offering more rows than that would only
#: ever produce systems the simulator refuses. With a second editor the combined spectrum may hold
#: more spins than this — legitimately, since the Hamiltonian is built per block and never on the
#: joined matrix.
MAX_MATRIX_SPINS = 8
#: What the optional second panel starts from. Any of `PHENOTYPE_CHOICES` works; this one is small,
#: strongly coupled, and pairs with the presets people reach for first.
SECOND_SYSTEM_PRESET = "AB"
#: Pople letters name the spins the way the spin-system literature and the paper's table do.
_POPLE = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
#: The cell types `gr.Dataframe` accepts, spelled out so the grid's column list type-checks.
DataframeCell = Literal["str", "number", "bool", "date", "markdown", "html"]
def _matrix_datatype(n_spins: int) -> list[DataframeCell]:
"""Column types for the spin grid: a text label column, then one numeric column per spin."""
columns: list[DataframeCell] = ["str"]
columns.extend("number" for _ in range(n_spins))
return columns
def _spin_label(index: int) -> str:
"""A, B, C ... then A', B', ... once the alphabet runs out (it will not, at 8 spins)."""
letter = _POPLE[index % len(_POPLE)]
return letter + "'" * (index // len(_POPLE))
def _cell_value(cell: object, row: int, col: int) -> float:
"""Read one grid cell as a float, treating blanks as zero and naming the cell on a typo.
Gradio hands back ``""`` or ``None`` for a cleared cell and the raw string for anything it could
not parse. Both must be resolved here: `float("")` raises, and a coupling that silently vanished
looks exactly like a spectrum the user meant to produce.
"""
if cell is None:
return 0.0
if isinstance(cell, str):
if not cell.strip():
return 0.0
try:
value = float(cell) # type: ignore[arg-type]
except (TypeError, ValueError):
raise ValueError(
f"row {row + 1}, column {col + 1} is not a number: {cell!r}. "
"Use a plain decimal, or clear the cell for zero."
) from None
# NaN and inf survive float(), then surface far away as "this spin system has no observable
# transition" — a description of the symptom rather than of the cell the user typed into.
if not np.isfinite(value):
raise ValueError(
f"row {row + 1}, column {col + 1} is not a finite number: {cell!r}. "
"Use a plain decimal, or clear the cell for zero."
)
return value
def _matrix_to_system(rows: list[list[object]]) -> tuple[list[float], NDArray[np.float64]]:
"""Split the editor grid into per-spin shifts and a coupling matrix.
The grid is ``[label, v0, v1, ...]`` per row, so every value is offset by one column. The
**diagonal** carries chemical shifts in ppm and the **upper triangle** the couplings in Hz. Only
the upper triangle is read, matching :func:`moldetr.simulate.simulate` and
:func:`moldetr.simulate.coupling_blocks`, so a stale value below the diagonal cannot couple two
spins the plotted spectrum treats as independent.
Both grids let a user add or remove rows and columns directly, so the shape is checked rather
than assumed. Indexing a ragged row unguarded raised ``IndexError``, which ``_simulate_stage``
does not catch — it escaped the error-string channel and took the tab down with a Gradio toast
instead of naming the row. A surplus column is a mismatch too, not something to drop silently.
"""
n = len(rows)
shifts: list[float] = []
couplings = np.zeros((n, n), dtype=float)
for i, row in enumerate(rows):
if len(row) != n + 1:
raise ValueError(
f"row {i + 1} has {len(row) - 1} value(s) but the matrix has {n} spin(s). "
"Use the spin-count slider to resize rather than editing rows directly."
)
shifts.append(_cell_value(row[i + 1], i, i + 1))
for k in range(i + 1, n):
couplings[i, k] = _cell_value(row[k + 1], i, k + 1)
return shifts, couplings
def _resize_matrix(rows: list[list[object]], n_spins: int) -> list[list[object]]:
"""Rebuild the grid at ``n_spins`` rows, keeping every value that still has a home.
Rebuilding from scratch on each slider step is simpler and throws away a half-entered system on
a mis-click, so the overlap is copied across instead. New spins arrive at 0 ppm and uncoupled.
"""
grid: list[list[object]] = []
for i in range(n_spins):
row: list[object] = [_spin_label(i)]
for k in range(n_spins):
keep = i < len(rows) and (k + 1) < len(rows[i])
row.append(rows[i][k + 1] if keep else 0.0)
grid.append(row)
return grid
def _phenotype_grid(name: str) -> tuple[list[list[object]], list[list[object]]]:
"""Populate the matrix and the per-group width table from a named phenotype.
The dropdown pre-fills the grid rather than feeding the simulator, so the matrix stays the one
description of what is being simulated and a preset is just a starting point the user can edit.
"""
pheno = sp.PHENOTYPES[name]
shifts = [float(s) for s in pheno["shifts_ppm"]]
matrix = sp.build_coupling_matrix(len(shifts), pheno["couplings"])
rows: list[list[object]] = []
for i, shift in enumerate(shifts):
row: list[object] = [_spin_label(i)]
for k in range(len(shifts)):
row.append(shift if k == i else float(matrix[i, k]) if k > i else 0.0)
rows.append(row)
return rows, _width_rows(shifts, matrix, [float(w) for w in pheno["widths_hz"]])
def _width_rows(
shifts: list[float], couplings: NDArray[np.float64], widths: list[float] | None = None
) -> list[list[object]]:
"""One editable line width per **spin system**: ``system | n H | FWHM (Hz)``.
Per coupling block, not per ground-truth group, because that is the finest grain the simulator
can honour. ``simulate`` collapses widths to a single mean *within* a block, so two groups of one
coupled system are physically incapable of carrying different line shapes — for ethyl, widths of
(1, 1, 1, 3, 3) produce a spectrum bit-identical to a uniform 1.8. Offering a row per group would
put two controls on screen that silently average into one. Splitting blocks apart is what made
per-system widths real in the first place.
"""
rows: list[list[object]] = []
for block in coupling_blocks(couplings):
default = float(widths[block[0]]) if widths is not None and block[0] < len(widths) else 1.0
rows.append([_block_label(block), _block_shifts(shifts, block), len(block), default])
return rows
def _block_label(block: list[int]) -> str:
"""Name a spin system by its member spins — "A, B" — which is the key widths are matched on.
Deliberately the spin letters and not the shifts. The label has to survive a shift edit, or
retyping δ silently resets that system's line width; and it has to *change* when a coupling
merges two systems, because the width then belongs to a system that no longer exists. Membership
does exactly that; a shift list does neither.
"""
return ", ".join(_spin_label(i) for i in block)
def _block_shifts(shifts: list[float], block: list[int]) -> str:
"""The shifts a system covers, down-field first — shown so the row is readable, never keyed on."""
members = sorted({round(float(shifts[i]), 4) for i in block}, reverse=True)
return ", ".join(f"{s:g}" for s in members)
def _widths_per_spin(
shifts: list[float], couplings: NDArray[np.float64], width_rows: list[list[object]]
) -> list[float]:
"""Expand the per-system width table back to the one-entry-per-spin list `simulate` wants.
Rows are matched to systems by **label**, not by position. Typing a coupling into the matrix
merges two blocks without rebuilding the table, so a positional match hands row 2's width to
whatever block now sits second — a different set of spins, with the label on screen contradicting
the value applied. An unmatched system falls back to 1.0, which is visibly wrong rather than
quietly wrong.
"""
by_label: dict[str, float] = {}
for n, row in enumerate(width_rows):
if len(row) < 4:
raise ValueError(
f"line-width row {n + 1} is missing its FWHM value. "
"Use the spin-count slider to rebuild the table."
)
by_label[str(row[0])] = _cell_value(row[3], n, 3)
per_spin = [1.0] * len(shifts)
for block in coupling_blocks(couplings):
width = by_label.get(_block_label(block), 1.0)
for spin in block:
per_spin[spin] = width
return per_spin
def _join_systems(
first: tuple[list[float], NDArray[np.float64], list[float]],
second: tuple[list[float], NDArray[np.float64], list[float]],
) -> tuple[list[float], NDArray[np.float64], list[float]]:
"""Lay two independent spin systems out on one block-diagonal coupling matrix.
The cross terms stay zero, which is exactly what `coupling_blocks` reads as "separate systems",
so the joined matrix goes through the unchanged `simulate_systems` path: each block simulated on
a per-proton scale and summed under a single global peak rescale. That equivalence is what makes
two editors and one matrix the same computation (`test_simulate_additivity`), and it only holds
while the rescale happens **once, at the end** — adding two separately peak-normalised spectra
would silently flatten the relative integrals.
Widths are expanded to per-spin **before** joining, so the result never depends on the order
`coupling_blocks` happens to return the combined blocks in. Matching a concatenated width table
positionally against those blocks is the one way this could hand a system the wrong line shape.
An empty second system needs no special case and does not have one: `joined[n:, n:]` is then a
0×0 slice and the concatenations are no-ops, so the result *is* the first system. An earlier
version guarded that explicitly; deleting the guard changed no test, which is how it was found
to be dead. `test_two_spin_systems` pins the behaviour rather than the branch.
"""
(shifts, couplings, widths), (shifts2, couplings2, widths2) = first, second
n, n2 = len(shifts), len(shifts2)
joined = np.zeros((n + n2, n + n2))
joined[:n, :n] = couplings
joined[n:, n:] = couplings2
return shifts + shifts2, joined, widths + widths2
def _build_gt_groups(
shifts: list[float], couplings_hz: NDArray[np.float64] | list[list[float]]
) -> list[dict]:
"""Group equivalent spins into ground-truth multiplets (shift, proton count, max J).
``max_j_hz`` is read **per group from the coupling matrix**: the largest coupling from any spin
in the group to any spin outside it. Couplings *within* a group are excluded deliberately —
equivalent protons may carry a mutual J, but it produces no observable splitting, so reporting it
would claim a multiplet the spectrum does not show.
Spins are grouped by **shift alone**, deliberately, and not by coupling block. Ground truth here
describes what the spectrum shows, because it is compared against a detector that sees only the
spectrum: protons sharing a shift produce one peak carrying their combined area whether or not a
coupling connects them. Grouping by block would split a methoxy — three equivalent uncoupled
protons — into three 1H multiplets where the spectrum has a single 3H line. See
``tests/test_gt_groups_from_matrix.py`` for the measured case and the one known limitation.
Groups are ordered down-field first, matching how the spectrum is read and how the comparison
table is numbered.
"""
given = np.asarray(couplings_hz, dtype=float)
# Mirror `coupling_blocks`: the upper triangle is the contract, so both read the same couplings.
upper = np.triu(given, 1)
j = upper + upper.T
groups: dict[float, list[int]] = {}
for idx, shift in enumerate(shifts):
groups.setdefault(round(float(shift), 4), []).append(idx)
gt = []
for shift_val, idxs in sorted(groups.items(), reverse=True):
members = set(idxs)
outside = [
abs(float(j[i, k]))
for i in idxs
for k in range(j.shape[0])
# Same tolerance the block decomposition uses, so a stray 1e-15 cannot be "absent" to
# one and a reported coupling to the other.
if k not in members and abs(float(j[i, k])) > COUPLING_EPS_HZ
]
gt.append(
{
"shift_ppm": shift_val,
"proton_count": len(idxs),
"max_j_hz": max(outside) if outside else None,
}
)
return gt
def _predicted_max_j(pred: dict) -> float | None:
"""The model's single predicted coupling, or ``None`` when it emitted none.
``coupling_constants_hz`` holds **0 or 1** entries by construction -- ``PARAM_NAMES[3:7]`` are
``[sum, min, max, std]`` of the multiset, not four separate J values -- so there is one number to
show, not a set. Mirrors ``sp._comparison_row``, which has had this column all along while the
GUI table showed ``GT J`` with no predicted counterpart to compare it against.
"""
js = pred.get("coupling_constants_hz") or []
return float(js[0]) if len(js) else None
def _comparison_dataframe(gt_groups: list[dict], preds: list[dict]) -> pd.DataFrame:
"""GT-vs-detected table with a graded status and explicit error columns.
Each GT group is paired with its nearest-δ prediction (``match_to_gt``); predictions matched to
no GT are appended as spurious rows.
``status`` grades the **chemical shift alone** -- ``✓ excellent`` · ``✓ good`` · ``✓ ok`` ·
``~ fair`` · ``✗ off``, see :mod:`app_ui.grading`. It used to be conjunctive
(``dd_hz <= 2.0 and dh == 0``), which forced ``~ off`` on a proton-count mismatch *at zero shift
error*. Proton count now travels in ``ΔH`` and the coupling in ``ΔJ (Hz)``, each reported on its
own terms; ``✗ missed`` and ``+ extra`` are unchanged, being the absence of a pairing rather than
a grade. These mirror the connector / marker colours in :func:`plotting.comparison_figure`.
"""
matched = sp.match_to_gt(gt_groups, preds)
matched_ids = {id(p) for _g, p in matched if p is not None}
rows: list[dict] = []
for i, (gt, pred) in enumerate(matched, 1):
gt_j = "–" if gt["max_j_hz"] is None else f"{gt['max_j_hz']:.1f}"
row = {
"#": i,
"status": "",
"GT δ (ppm)": f"{gt['shift_ppm']:.2f}",
"GT H": gt["proton_count"],
"GT J (Hz)": gt_j,
}
if pred is None:
row.update(
{
"status": grading.MISSED,
"pred δ (ppm)": "–",
"pred H": "–",
"pred J (Hz)": "–",
"Δδ (Hz)": "–",
"ΔH": "–",
"ΔJ (Hz)": "–",
"conf": "–",
}
)
else:
dd_hz = abs(float(pred["chemical_shift_ppm"]) - gt["shift_ppm"]) * sp.BASE_FREQ_MHZ
dh = int(pred["proton_count"]) - gt["proton_count"]
pred_j = _predicted_max_j(pred)
dj = (
None if (pred_j is None or gt["max_j_hz"] is None) else abs(pred_j - gt["max_j_hz"])
)
row.update(
{
"status": grading.grade_shift(dd_hz),
"pred δ (ppm)": f"{float(pred['chemical_shift_ppm']):.3f}",
"pred H": int(pred["proton_count"]),
"pred J (Hz)": "–" if pred_j is None else f"{pred_j:.1f}",
"Δδ (Hz)": f"{dd_hz:.2f}",
"ΔH": f"{dh:+d}",
"ΔJ (Hz)": "–" if dj is None else f"{dj:.2f}",
"conf": f"{float(pred['confidence']):.2f}",
}
)
rows.append(row)
for k, pred in enumerate((p for p in preds if id(p) not in matched_ids), len(rows) + 1):
pred_j = _predicted_max_j(pred)
rows.append(
{
"#": k,
"status": grading.EXTRA,
"GT δ (ppm)": "–",
"GT H": "–",
"GT J (Hz)": "–",
"pred δ (ppm)": f"{float(pred['chemical_shift_ppm']):.3f}",
"pred H": int(pred["proton_count"]),
"pred J (Hz)": "–" if pred_j is None else f"{pred_j:.1f}",
"Δδ (Hz)": "–",
"ΔH": "–",
"ΔJ (Hz)": "–",
"conf": f"{float(pred['confidence']):.2f}",
}
)
return pd.DataFrame(rows)
def _simulate_distort_kwargs(
add_noise: bool,
snr: float,
phase0: float,
broaden: float,
baseline: float,
satellites: bool,
sat_j: float,
) -> dict[str, float]:
"""Assemble ``distort`` kwargs from the sliders (neutral / identity values are skipped).
``satellites`` defaults **on** because training applied ¹³C satellites unconditionally --
``augment_distortions`` calls ``add_13C_satellites_with_variability`` on every sample, with no
coin toss and no custom-values path. Leaving them off produced Simulate spectra systematically
cleaner than anything the model was trained on, and the tab offered no control to fix it.
``distort`` treats them as opt-in (they apply only when a parameter is supplied), which is the
inverse of training semantics, so parity has to be asserted here by the caller.
"""
dk: dict[str, float] = {}
if add_noise:
dk["noise_snr_log10"] = float(snr)
if satellites:
dk["sat_j_hz"] = float(sat_j)
dk["sat_intensity"] = 0.01 # midpoint of the trained 0.005-0.015 range
if float(phase0) != 0.0:
dk["phase0_deg"] = float(phase0)
if float(broaden) > 0.0:
dk["broaden_hz"] = float(broaden)
if float(baseline) > 0.0:
dk["baseline"] = float(baseline)
return dk
#: What ``_simulate_stage`` hands to ``_detect_stage``: the clean **complex** spectrum plus the ppm
#: axis, phenotype label and ground-truth groups — everything needed to distort, detect and compare
#: without paying the 2**n eigendecomposition again.
SimCache = dict[str, Any]
def _simulate_stage(
matrix_rows: list[list[object]],
width_rows: list[list[object]],
second_enabled: bool = False,
matrix_rows2: list[list[object]] | None = None,
width_rows2: list[list[object]] | None = None,
) -> SimCache | str:
"""Run the spin dynamics once and return everything the cheap stage needs.
The spin system comes entirely from the matrix grid — shifts on the diagonal, couplings in the
upper triangle — so there is one description of what is being simulated rather than a phenotype
and an edited copy of it that can disagree. The preset dropdown fills the grid and then has no
further say.
A second, independent system can be supplied from its own grid. It is **off unless asked for**:
the panel's controls always hold real values, so gating on `second_enabled` rather than on the
grid being empty is what keeps every single-system caller — including the positional ones — bit
identical to before this existed.
The parameters here carry defaults because this helper is wired to no Gradio event; only
`simulate_to_state` is, and its arity is checked against the wired input list.
Returns a cache dict, or a **string** carrying the user-facing error. The string channel keeps
the two stages composable into the original single-return-shape callback.
The cached spectrum is the *clean* one, exactly as ``simulate_systems`` produced it, because
re-distorting an already distorted spectrum would compound the effects as a slider is dragged.
It is **real**: ``simulate_systems`` sums Lorentzian absorption lines and never forms an
analytic signal, so there is no dispersion component to preserve or to strip. ``distort``
documents a complex input and this path has always handed it a real one, on ``main`` as here, so
the phase controls rotate a spectrum with no imaginary part rather than the analytic signal
training used. That is a pre-existing limitation of the tab, noted for the distortion pass, not
something this cache introduces.
"""
if not Path(CHECKPOINT).exists():
return (
f"Checkpoint not found at `{CHECKPOINT}`. "
"Download it from Zenodo (10.5281/zenodo.21217102) into `moldetr/model/`."
)
try:
shifts, couplings = _matrix_to_system(matrix_rows)
widths = _widths_per_spin(shifts, couplings, width_rows)
except ValueError as exc:
return f"Invalid spin matrix: {exc}"
if second_enabled:
try:
shifts2, couplings2 = _matrix_to_system(matrix_rows2 or [])
widths2 = _widths_per_spin(shifts2, couplings2, width_rows2 or [])
except ValueError as exc:
# Named, because two grids are on screen and "row 1" alone would be ambiguous.
return f"Invalid second spin matrix: {exc}"
shifts, couplings, widths = _join_systems(
(shifts, couplings, widths), (shifts2, couplings2, widths2)
)
if not shifts:
return "Add at least one spin to the matrix."
try:
# simulate_systems, not simulate: it splits the coupling matrix into independent blocks,
# simulates each on a per-proton scale and sums them, so several spin systems in one window
# keep the right relative integrals (one proton = one unit of area, everywhere). The default
# "peak" rescale then restores max = 1 before distortion, which is what the distortion
# magnitudes are calibrated against.
spectrum, ppm_axis = simulate_systems(
shifts,
couplings,
widths,
sp.BASE_FREQ_MHZ,
sp.LEFT_PPM,
sp.RIGHT_PPM,
sp.N_POINTS,
)
except ValueError as exc:
return f"Invalid parameters: {exc}"
blocks = coupling_blocks(couplings)
return {
"label": f"{len(shifts)} spin(s) in {len(blocks)} system(s)",
"spectrum": spectrum,
"ppm_axis": ppm_axis,
"gt_groups": _build_gt_groups(shifts, couplings),
}
def _distorted_amplitudes(
cache: SimCache,
add_noise: bool,
snr: float,
phase0: float,
broaden: float,
baseline: float,
satellites: bool,
sat_j: float,
) -> NDArray[np.float64]:
"""Apply the distortions to the cached clean spectrum and return real amplitudes.
Always returns a fresh array, so ``cache`` survives any number of slider moves. ``distort``
copies its input, but with every distortion at its neutral value it is skipped entirely and
``np.asarray(np.real(...))`` is a no-op on a real float64 array — it would hand back the cached
array itself, and one in-place edit downstream would corrupt every later re-distortion.
"""
spectrum = cache["spectrum"]
dk = _simulate_distort_kwargs(add_noise, snr, phase0, broaden, baseline, satellites, sat_j)
if dk:
spectrum = distort(spectrum, cache["ppm_axis"], **dk)
return np.array(np.real(spectrum), dtype=float, copy=True)
def _detect_stage(
cache: SimCache | str,
add_noise: bool,
snr: float,
phase0: float,
broaden: float,
baseline: float,
threshold: float,
satellites: bool,
sat_j: float,
) -> tuple[pd.DataFrame | None, object | None, str]:
"""Distort, detect and compare — everything that must re-run when a slider moves, and no more.
Accepts the error string ``_simulate_stage`` returns instead of a cache and passes it straight
through, so a stage driven directly from a stored cache (rather than through
``simulate_and_detect``) still shows the message rather than indexing a ``str``.
"""
if isinstance(cache, str):
return None, None, cache
try:
amplitudes = _distorted_amplitudes(
cache, add_noise, snr, phase0, broaden, baseline, satellites, sat_j
)
except ValueError as exc:
return None, None, f"Invalid parameters: {exc}"
label = cache["label"]
gt_groups = cache["gt_groups"]
try:
model = _get_model()
except RuntimeError as exc: # the checkpoint trust gate; its text names the only remedy
return None, None, _checkpoint_error_message(exc)
preds = decode_predictions(
# Skip the in-model noise floor when the user has already added calibrated noise here.
# The floor (0.005 * max) exists to drag a perfectly clean FFT-resampled spectrum back
# in-distribution; once "Add noise" is on, that job is done and the floor only masks the
# slider. The two are directly comparable -- distort's std is max/(2*SNR) -- and they are
# equal only at the slider's 2.0 minimum, so at the 3.0 default the requested noise sits
# 10x under the floor and at 5.0 it is 1000x under. That is why the slider felt inert.
# Detect (and predict.py) deliberately keep the floor: they feed the frozen decode.
run(model, amplitudes, noise_frac=0.0 if add_noise else 0.005),
load_extrema(EXTREMA),
sp.POINTS_PER_HZ,
ppm_left=sp.LEFT_PPM,
ppm_right=sp.RIGHT_PPM,
threshold=threshold,
)
matched = sp.match_to_gt(gt_groups, preds)
matched_ids = {id(p) for _g, p in matched if p is not None}
spurious = [p for p in preds if id(p) not in matched_ids]
fig = comparison_figure(
amplitudes,
matched,
spurious,
ppm_left=sp.LEFT_PPM,
ppm_right=sp.RIGHT_PPM,
base_freq_mhz=sp.BASE_FREQ_MHZ,
)
table = _comparison_dataframe(gt_groups, preds)
n_match = sum(1 for _g, p in matched if p is not None)
msg = (
f"**Simulated {label}**: {len(gt_groups)} ground-truth multiplet(s); the model "
f"**detected** {len(preds)} ({n_match} matched, {len(spurious)} spurious). "
"Teal ▽ = ground truth · clay ● = model detection; a connector turns **green** within "
"tolerance and **amber** when off. Missed GT and spurious peaks are outlined in red."
)
return table, fig, msg
def simulate_and_detect(
matrix_rows: list[list[object]],
width_rows: list[list[object]],
add_noise: bool,
snr: float,
phase0: float,
broaden: float,
baseline: float,
threshold: float,
satellites: bool,
sat_j: float,
second_enabled: bool = False,
matrix_rows2: list[list[object]] | None = None,
width_rows2: list[list[object]] | None = None,
) -> tuple[pd.DataFrame | None, object | None, str]:
"""Simulate the matrix, optionally distort, detect, and compare to ground truth.
Kept as a thin composition of the two stages, so the one-shot and cached paths cannot drift.
It is the **direct-call** entry point — for tests and library users — and is wired to no event.
The button runs :func:`simulate_to_state`, which merely carries `api_name="simulate_and_detect"`;
that shared name is why an earlier version of this docstring claimed the Gradio event addressed
this function. It does not, and the distinction is load-bearing: because `test_ui_graph`'s arity
check only inspects *wired* callbacks, the second-system arguments can carry defaults here while
`simulate_to_state`'s must not — which is what leaves every existing positional caller binding
exactly as before.
"""
cache = _simulate_stage(matrix_rows, width_rows, second_enabled, matrix_rows2, width_rows2)
return _detect_stage(
cache, add_noise, snr, phase0, broaden, baseline, threshold, satellites, sat_j
)
def preset_grid(name: str) -> tuple[list[list[object]], list[list[object]], int]:
"""Fill the matrix and width table from a preset, and resize the spin slider to match."""
rows, widths = _phenotype_grid(name)
return rows, widths, len(rows)
def resize_spin_matrix(
matrix_rows: list[list[object]], width_rows: list[list[object]], n_spins: float
) -> tuple[list[list[object]], list[list[object]]]:
"""Grow or shrink the matrix, keeping what was typed in **both** grids.
The width table is re-derived rather than resized alongside the matrix, because its rows are
spin *systems*: adding a spin coupled to an existing one enlarges a system instead of adding a
row. Widths whose system survives the resize are carried across by label, so the count slider is
not a way to lose them.
"""
rows = _resize_matrix(matrix_rows, int(n_spins))
try:
shifts, couplings = _matrix_to_system(rows)
rebuilt = _width_rows(shifts, couplings)
carried = _widths_per_spin(shifts, couplings, width_rows)
except (ValueError, IndexError):
# A bad cell is reported when Simulate is pressed. Leave the width table untouched rather
# than replacing it with an empty one, which silently reset every width to the default.
return rows, width_rows
for row, block in zip(rebuilt, coupling_blocks(couplings)):
row[3] = carried[block[0]]
return rows, rebuilt
def invalidate_cache() -> None:
"""Drop the cached spectrum without touching the grids.
Used by the width table, whose edits change the spectrum but not which spin systems exist, so
there is nothing to re-derive — unlike a matrix edit, which can merge two systems into one.
"""
return None
def matrix_edited(
matrix_rows: list[list[object]], width_rows: list[list[object]]
) -> tuple[None, list[list[object]]]:
"""React to a matrix edit: drop the cached spectrum and re-derive the width table.
Two things go stale the moment a cell changes. The **cache** must go, or a slider re-distorts
whatever was last simulated — edit five spins down to two, drag the phase slider, and the plot
re-renders the old five, still labelled as five, beside a matrix saying otherwise. Clearing beats
re-simulating on every keystroke, which is the ``2**n`` cost the cache exists to avoid.
The **width table** goes stale differently: typing a coupling merges two systems into one, so the
rows no longer describe the systems that exist. Rebuilding here keeps the screen honest, and
widths whose system survived the edit are carried across by label.
"""
try:
shifts, couplings = _matrix_to_system(matrix_rows)
rebuilt = _width_rows(shifts, couplings)
carried = _widths_per_spin(shifts, couplings, width_rows)
except (ValueError, IndexError):
# A half-typed cell is reported when Simulate is pressed. Leave the table alone rather than
# destroying the user's widths mid-edit.
return None, width_rows
for row, block in zip(rebuilt, coupling_blocks(couplings)):
row[3] = carried[block[0]]