Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 8 additions & 7 deletions core/data_processing/plotting.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
import numpy as np

from core.data_processing.measurement_set import MeasurementSet
from core.pipeline.fit_pipeline import FitResult
from core.pipeline.fit_pipeline import FitResult, resolve_fit_curve


def prepare_plot_data(
Expand Down Expand Up @@ -65,14 +65,15 @@ def prepare_plot_data(
fits: list[Dict[str, Any]] = []
if fit_results:
for fr in fit_results:
label = 'Median Fit'
x = fr.x_fit.magnitude
y = fr.y_fit.magnitude
# Re-derive the dense curve from the fitted parameters so a reloaded
# fit renders the same smooth line as the original fit, regardless of
# the resolution serialized in x_fit/y_fit.
x_fit, y_fit = resolve_fit_curve(fr)
fits.append(
{
'x': x,
'y': y,
'label': label,
'x': x_fit.magnitude,
'y': y_fit.magnitude,
'label': 'Median Fit',
'id': fr.id,
}
)
Expand Down
88 changes: 88 additions & 0 deletions core/pipeline/fit_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,13 @@
import numpy as np

from core.assays.base import BaseAssay
from core.assays.dba import DBAAssay
from core.assays.dye_alone import DyeAloneAssay
from core.assays.gda import GDAAssay
from core.assays.h2g import H2GAssay
from core.assays.hg2 import HG2Assay
from core.assays.ida import IDAAssay
from core.assays.registry import AssayType
from core.data_processing.measurement_set import MeasurementSet
from core.optimizer.filters import calculate_fit_metrics, filter_fits
from core.optimizer.multistart import multistart_minimize
Expand All @@ -45,6 +51,88 @@ def _dense_fit_curve(assay: BaseAssay, params: np.ndarray) -> tuple[Quantity, Qu
return Q_(x_dense, 'M'), assay.forward_model(params, x=x_dense)


# Concrete assay classes keyed by serialized AssayType, used to rebuild a
# minimal assay when re-deriving the smooth display curve from a stored fit.
# DYE_ALONE is intentionally absent: its curve is a straight line (sparse and
# dense are visually identical) and its forward model takes regression
# coefficients rather than equilibrium conditions.
_RECONSTRUCTABLE_ASSAYS: Dict[AssayType, Type[BaseAssay]] = {
AssayType.GDA: GDAAssay,
AssayType.IDA: IDAAssay,
AssayType.DBA_HtoD: DBAAssay,
AssayType.DBA_DtoH: DBAAssay,
AssayType.DBA_HG2: HG2Assay,
AssayType.DBA_H2G: H2GAssay,
}


def _conditions_as_quantities(conditions: Dict[str, Any]) -> Dict[str, Any]:
"""Re-wrap serialized condition magnitudes as Quantities.

Fit-result JSON stores conditions as bare magnitudes in base units (M for
concentrations, 1/M for association constants); fresh fits already carry
them as Quantities. Non-numeric values (e.g. the DBA ``mode`` string) pass
through untouched.
"""
out: Dict[str, Any] = {}
for key, value in conditions.items():
if isinstance(value, Quantity) or isinstance(value, bool):
out[key] = value
elif isinstance(value, (int, float)):
out[key] = Q_(float(value), '1/M' if key.startswith('Ka') else 'M')
else:
out[key] = value
return out


def resolve_fit_curve(result: 'FitResult') -> Tuple[Quantity, Quantity]:
"""Return the dense Median Fit curve to draw for *result*.

The smooth line is fully determined by the fitted parameters, the assay
conditions, the assay type, and the data x-range — so it is re-derived here
on demand rather than trusted from the serialized ``x_fit``/``y_fit``. This
keeps a reloaded fit visually identical to the original even when the
serialized curve was sampled only at the measured concentrations (older
exports, the linear/failed-fit paths).

Falls back to the stored ``result.x_fit``/``result.y_fit`` when the curve
cannot be reconstructed (unknown or linear assay type, no fitted
parameters, or missing/invalid conditions).
"""
stored = (result.x_fit, result.y_fit)
if not result.parameters:
return stored
try:
Comment on lines +102 to +105
assay_type = AssayType[result.assay_type]
except KeyError:
return stored
assay_cls = _RECONSTRUCTABLE_ASSAYS.get(assay_type)
if assay_cls is None:
return stored

conditions = _conditions_as_quantities(result.conditions)
# The assay type is authoritative for the DBA titration mode.
if assay_type is AssayType.DBA_HtoD:
conditions['mode'] = 'HtoD'
elif assay_type is AssayType.DBA_DtoH:
conditions['mode'] = 'DtoH'

try:
x_ref = result.x_fit.to('M')
assay = assay_cls(
x_data=x_ref,
y_data=Q_(np.zeros_like(x_ref.magnitude), 'au'),
**conditions,
)
params = np.array(
[float(result.parameters[k].magnitude) for k in assay.parameter_keys],
dtype=float,
)
return _dense_fit_curve(assay, params)
except (KeyError, ValueError, TypeError):
return stored


@dataclass
class FitResult:
"""Serializable container for fitting results.
Expand Down
15 changes: 8 additions & 7 deletions gui/fitting_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
from core.data_processing.measurement_set import MeasurementSet
from core.data_processing.plotting import prepare_plot_data
from core.io.formats.bmg_reader import BMG_PLACEHOLDER_KEY
from core.pipeline.fit_pipeline import FitConfig, FitResult
from core.pipeline.fit_pipeline import FitConfig, FitResult, resolve_fit_curve
from gui.app_state import SessionState
from gui.plotting.distribution_widget import DistributionWidget
from gui.plotting.fit_summary_widget import FitSummaryWidget
Expand Down Expand Up @@ -407,23 +407,24 @@ def import_results(self) -> None:
self._state.fit_results = results
self._refresh_plot()
else:
# Source file not available — show fit curves only
# Source file not available — show fit curves only. Re-derive
# the dense curve so it matches the original (see prepare_plot_data).
self._state.fit_results = results
meta = ASSAY_REGISTRY[self._state.assay_type]
last = results[-1]
curves = [resolve_fit_curve(r) for r in results]
plot_data = {
'concentrations': last.x_fit.magnitude,
'concentrations': curves[-1][0].magnitude,
'active_replicas': [],
'dropped_replicas': [],
'average': None,
'fits': [
{
'x': r.x_fit.magnitude,
'y': r.y_fit.magnitude,
'x': x_fit.magnitude,
'y': y_fit.magnitude,
'label': 'Median Fit',
'id': r.id,
}
for r in results
for r, (x_fit, y_fit) in zip(results, curves)
],
}
self._plot_widget.update_plot(
Expand Down
128 changes: 128 additions & 0 deletions tests/unit/test_fit_curve_reload.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
"""Regression: a reloaded Median Fit curve must match the originally drawn one.

The smooth fit line is fully determined by the fitted parameters, the assay
conditions, the assay type, and the data x-range. Older exports (and the
linear / failed-fit paths) serialised ``x_fit``/``y_fit`` only at the measured
concentrations, so on reload the line was drawn as coarse straight segments
that did not match the dense curve a fresh fit renders. The display path now
re-derives the dense curve from the stored parameters, so reload matches the
original regardless of the serialised resolution (issue #25).
"""

import json

import numpy as np

from core.data_processing.measurement_set import MeasurementSet
from core.data_processing.plotting import prepare_plot_data
from core.pipeline.fit_pipeline import _FIT_CURVE_POINTS, FitResult, _dense_fit_curve
from core.units import Q_
from tests.conftest import IDA_TRUE, _make_ida_data


def _ida_setup():
"""Known IDA assay + ground-truth parameter vector (no optimizer)."""
from core.assays.ida import IDAAssay

x, y = _make_ida_data(IDA_TRUE)
assay = IDAAssay(
x_data=Q_(x, 'M'),
y_data=Q_(y, 'au'),
Ka_dye=Q_(IDA_TRUE['Ka_dye'], '1/M'),
h0=Q_(IDA_TRUE['h0'], 'M'),
d0=Q_(IDA_TRUE['d0'], 'M'),
)
params = np.array(
[IDA_TRUE['Ka_guest'], IDA_TRUE['I0'], IDA_TRUE['I_dye_free'], IDA_TRUE['I_dye_bound']],
dtype=float,
)
return assay, x, y, params


def _ida_result(x, y, params, *, x_fit, y_fit) -> FitResult:
"""Build an IDA FitResult with caller-chosen serialised curve arrays."""
keys = ('Ka_guest', 'I0', 'I_dye_free', 'I_dye_bound')
units = ('1/M', 'au', 'au/M', 'au/M')
return FitResult(
parameters={k: Q_(float(v), u) for k, v, u in zip(keys, params, units)},
uncertainties={k: Q_(np.nan, u) for k, u in zip(keys, units)},
rmse=1.0,
r_squared=0.999,
n_passing=1,
n_total=1,
x_fit=x_fit,
y_fit=y_fit,
assay_type='IDA',
model_name='equilibrium_4param',
conditions={
'Ka_dye': Q_(IDA_TRUE['Ka_dye'], '1/M'),
'h0': Q_(IDA_TRUE['h0'], 'M'),
'd0': Q_(IDA_TRUE['d0'], 'M'),
},
)


def _reload(result: FitResult) -> FitResult:
"""Round-trip through the on-disk JSON form (faithful, not regenerated)."""
return FitResult.from_dict(json.loads(json.dumps(result.to_dict())))


def test_reloaded_curve_matches_original_smooth_curve():
"""A pre-smooth-curve (sparse) export reloads to the dense original line."""
assay, x, y, params = _ida_setup()

# The line a fresh fit draws: the dense forward-model curve.
x_dense, y_dense = _dense_fit_curve(assay, params)

# Simulate the old export: curve stored only at the measured points.
sparse = _ida_result(x, y, params, x_fit=Q_(x, 'M'), y_fit=assay.forward_model(params))
reloaded = _reload(sparse)
# The serialised arrays really are sparse (precondition of the bug).
assert len(reloaded.x_fit.magnitude) == len(x)

ms = MeasurementSet(concentrations=x, signals=np.vstack([y]), replica_ids=('r1',))
fit = prepare_plot_data(ms, [reloaded])['fits'][0]

assert len(fit['x']) == _FIT_CURVE_POINTS
np.testing.assert_allclose(fit['x'], x_dense.magnitude, rtol=1e-12, atol=0.0)
np.testing.assert_allclose(fit['y'], y_dense.magnitude, rtol=1e-9, atol=1e-9)


def test_dense_export_curve_unchanged_after_reload():
"""A current (dense) export still renders the identical curve after reload."""
assay, x, y, params = _ida_setup()
x_dense, y_dense = _dense_fit_curve(assay, params)

dense = _ida_result(x, y, params, x_fit=x_dense, y_fit=y_dense)
reloaded = _reload(dense)

ms = MeasurementSet(concentrations=x, signals=np.vstack([y]), replica_ids=('r1',))
fit = prepare_plot_data(ms, [reloaded])['fits'][0]

np.testing.assert_allclose(fit['x'], x_dense.magnitude, rtol=1e-12, atol=0.0)
np.testing.assert_allclose(fit['y'], y_dense.magnitude, rtol=1e-12, atol=0.0)


def test_linear_fit_curve_falls_back_to_stored_arrays():
"""DYE_ALONE (linear) has no equilibrium curve to rebuild — keep stored arrays."""
x = np.linspace(0.0, 20e-6, 12)
y = 5e10 * x + 100.0
result = FitResult(
parameters={'slope': Q_(5e10, 'au/M'), 'intercept': Q_(100.0, 'au')},
uncertainties={'slope': Q_(np.nan, 'au/M'), 'intercept': Q_(np.nan, 'au')},
rmse=0.1,
r_squared=1.0,
n_passing=1,
n_total=1,
x_fit=Q_(x, 'M'),
y_fit=Q_(y, 'au'),
assay_type='DYE_ALONE',
model_name='linear',
)
reloaded = _reload(result)

ms = MeasurementSet(concentrations=x, signals=np.vstack([y]), replica_ids=('r1',))
fit = prepare_plot_data(ms, [reloaded])['fits'][0]

np.testing.assert_array_equal(fit['x'], x)
np.testing.assert_array_equal(fit['y'], y)