From 3b6f4b519e59b07f3b1559858c301f8596c172e1 Mon Sep 17 00:00:00 2001 From: Joel Eliason Date: Sun, 26 Jul 2026 21:19:25 -0600 Subject: [PATCH 1/3] feat(cal): reject assumed/uncertainty inputs on calibration targets Ports SubmodelTarget.validate_no_assumed_or_uncertainty_inputs onto CalibrationTargetEstimates. Submodel targets have refused fabricated spread for a while; calibration targets never were, which is how a target reached the corpus carrying: assumed_cv_fraction = 1.0 'Neither standard deviation nor interquartile range was reported for any immune cell density measurement; CV assumed at 1.0 as a conservative estimate.' A fabricated CV is not conservative -- it sets the weight the target pulls with in the likelihood. Two rules. 'assumed' anywhere in an input name is rejected regardless of input_type. Uncertainty-smelling names (cv, sigma, uncertainty, fold_uncertainty, translation_sd) are rejected on non-measurement types (experimental_condition, inferred_estimate, derived_arithmetic) -- dispersion a paper actually reports is a measurement and belongs on direct_parameter / proxy_measurement with a snippet. Reported spread stays legal: sd_*, sem_*, *_q3 and friends are untouched, and a genuinely author-reported CV is fine when typed as a measurement. Measured against the pdac corpus before writing: 449 inputs across the live calibration targets, one violation (treg_fraction_baseline), zero false positives. Also hoists a late module-level import in the test file to satisfy ruff E402. --- .../calibration/calibration_target_models.py | 74 ++++++++++++++++ .../test_calibration_target_validators.py | 86 ++++++++++++++++++- 2 files changed, 158 insertions(+), 2 deletions(-) diff --git a/src/maple/core/calibration/calibration_target_models.py b/src/maple/core/calibration/calibration_target_models.py index f7a88db..b155227 100644 --- a/src/maple/core/calibration/calibration_target_models.py +++ b/src/maple/core/calibration/calibration_target_models.py @@ -10,6 +10,7 @@ """ import ast +import re from pathlib import Path from typing import Any, Dict, List, Literal, Optional @@ -396,6 +397,79 @@ def validate_scalar_output(self) -> "CalibrationTargetEstimates": return self + @model_validator(mode="after") + def validate_no_assumed_or_uncertainty_inputs(self) -> "CalibrationTargetEstimates": + """ + Reject inputs that encode a modeling assumption rather than paper data. + + Ported from ``SubmodelTarget.validate_no_assumed_or_uncertainty_inputs``. + Calibration targets were never held to this rule, which is how a target + reached the corpus carrying ``assumed_cv_fraction = 1.0`` with the snippet + "Neither standard deviation nor interquartile range was reported for any + immune cell density measurement; CV assumed at 1.0 as a conservative + estimate." A fabricated spread is not a conservative estimate — it sets the + weight the target pulls with in the likelihood. + + Two checks: + + 1. ``assumed`` anywhere in an input name, regardless of input_type. Naming + something ``assumed_*`` is an admission that it is a modeling choice. + 2. Uncertainty-smelling names on *non-measurement* input types. Dispersion + that a paper actually reports is a measurement and belongs on + ``direct_parameter`` / ``proxy_measurement`` with a snippet to prove it; + the same name on ``experimental_condition`` / ``inferred_estimate`` / + ``derived_arithmetic`` means it was invented. + + Reported spread stays legal — ``sd_*``, ``sem_*``, ``*_q1``, ``*_q3`` and + friends are untouched, because they are values the paper printed. + """ + UNCERTAINTY_PATTERNS = [ + "cv", + "sigma", + "uncertainty", + "fold_uncertainty", + "translation_sd", + ] + NON_MEASUREMENT_TYPES = { + InputType.EXPERIMENTAL_CONDITION, + InputType.INFERRED_ESTIMATE, + InputType.DERIVED_ARITHMETIC, + } + + errors = [] + for inp in self.inputs: + name_lower = inp.name.lower() + + if "assumed" in name_lower: + errors.append( + f"Input '{inp.name}' contains 'assumed' in its name, indicating a " + f"modeling choice rather than extracted data. If the paper reports " + f"no dispersion, that is a property of the source and must be " + f"handled in the error model downstream — not fabricated here as " + f"an input. Derive it in distribution_code if it is a genuine " + f"transformation of reported values, or drop it." + ) + continue + + if inp.input_type in NON_MEASUREMENT_TYPES: + for pattern in UNCERTAINTY_PATTERNS: + if re.search(rf"\b{pattern}\b|_{pattern}$|^{pattern}_", name_lower): + errors.append( + f"Input '{inp.name}' looks like an uncertainty factor " + f"(matches '{pattern}') but has " + f"input_type='{inp.input_type.value}'. Dispersion the paper " + f"actually reports should be typed as a measurement " + f"(direct_parameter / proxy_measurement) and carry a " + f"value_snippet; on a non-measurement type it is an " + f"invented uncertainty factor." + ) + break + + if errors: + raise ValueError("\n".join(errors)) + + return self + @model_validator(mode="after") def validate_derived_arithmetic_inputs(self) -> "CalibrationTargetEstimates": """ diff --git a/tests/unit/core/test_calibration_target_validators.py b/tests/unit/core/test_calibration_target_validators.py index 69ebf88..1a53d14 100644 --- a/tests/unit/core/test_calibration_target_validators.py +++ b/tests/unit/core/test_calibration_target_validators.py @@ -37,6 +37,7 @@ from pydantic import ValidationError from maple.core.calibration import CalibrationTarget, Observable +from maple.core.calibration.calibration_target_models import CalibrationTargetEstimates DEFAULT_CLINICAL_SOURCE_RELEVANCE = { @@ -1678,8 +1679,6 @@ def test_degenerate_samples_rejected( # observable's moments-form observed_distribution must use shape=logit_normal. # ============================================================================ -from maple.core.calibration.calibration_target_models import CalibrationTargetEstimates - def _bounded_cal_estimates(shape: str) -> dict: return { @@ -1727,3 +1726,86 @@ def test_percent_with_normal_shape_raises(self): def test_percent_with_logit_normal_passes(self): CalibrationTargetEstimates.model_validate(_bounded_cal_estimates("logit_normal")) + + +def _estimates_with_input( + name: str, input_type: str | None = None, dispersion_type: str | None = None +) -> dict: + """Minimal valid estimates payload carrying one named input.""" + inp = { + "name": name, + "value": 0.5, + "units": "dimensionless", + "description": "an input", + "source_ref": "smith_2020", + "value_location": "Table 1", + "value_snippet": "reported value 0.5", + } + if input_type is not None: + inp["input_type"] = input_type + if dispersion_type is not None: + inp["dispersion_type"] = dispersion_type + inp["dispersion_type_rationale"] = "Paper states this explicitly in the legend." + return { + "median": [0.5], + "ci95": [[0.3, 0.7]], + "units": "dimensionless", + "sample_size": 40, + "sample_size_rationale": "n=40 patients, Table 1", + "inputs": [inp], + "distribution_code": ( + "def derive_distribution(inputs, ureg):\n" + f" v = inputs['{name}']\n" + " return {'median_obs': v, 'ci95_lower': v * 0.6, 'ci95_upper': v * 1.4}" + ), + "population_spread": "center_only", + } + + +class TestCalNoAssumedOrUncertaintyInputs: + """CalibrationTargetEstimates: fabricated spread must not enter as an input. + + Ported from SubmodelTarget. The live catch is treg_fraction_baseline, which + carried assumed_cv_fraction = 1.0 because the source reported no dispersion. + """ + + def test_assumed_in_name_raises(self): + with pytest.raises(ValidationError, match="assumed"): + CalibrationTargetEstimates.model_validate( + _estimates_with_input("assumed_cv_fraction", "inferred_estimate") + ) + + def test_assumed_raises_regardless_of_input_type(self): + """'assumed' is rejected even when typed as a real measurement.""" + with pytest.raises(ValidationError, match="assumed"): + CalibrationTargetEstimates.model_validate( + _estimates_with_input("assumed_baseline_density", "direct_parameter") + ) + + def test_uncertainty_name_on_non_measurement_type_raises(self): + with pytest.raises(ValidationError, match="uncertainty factor"): + CalibrationTargetEstimates.model_validate( + _estimates_with_input("cv_translation", "derived_arithmetic") + ) + + def test_reported_dispersion_stays_legal(self): + """sd_/sem_/q3 are values the paper printed — must not be rejected. + + Dispersion-named inputs separately owe a dispersion_type (the existing + SEM-vs-SD guard in shared_models); that is orthogonal to this validator + and supplied here so the test isolates the rule under test. + """ + for nm, dt in ( + ("sd_nk_fraction_til", "sd"), + ("sem_frac_pd1_cd4_tumor", "se"), + ("treg_density_q3", None), + ): + CalibrationTargetEstimates.model_validate( + _estimates_with_input(nm, "direct_parameter", dispersion_type=dt) + ) + + def test_uncertainty_name_on_measurement_type_stays_legal(self): + """A paper may genuinely report a CV; typed as a measurement it is data.""" + CalibrationTargetEstimates.model_validate( + _estimates_with_input("cv_reported_by_authors", "direct_parameter") + ) From 0c8c5eb53f0215aee2c3fa4cdd5cad22fccd2a01 Mon Sep 17 00:00:00 2001 From: Joel Eliason Date: Mon, 27 Jul 2026 07:19:58 -0600 Subject: [PATCH 2/3] feat(cal): Tier-1 validator ports from SubmodelTarget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five validators the submodel schema has enforced for a while, brought over to CalibrationTarget. Each was measured against the 52 live pdac targets before landing, so the migration cost is known rather than assumed. - source relevance now walks primary AND secondary sources. Previously only the primary was inspected, so a secondary could be a wiki, a preprint or an unreviewed database and never be reported. 1 live hit (a UniProt molecular-weight lookup). Stays a warning; the message says whether the source supplies values used in the derivation. - validate_context_mismatch_justified: a declared pharmacological/genetic perturbation must carry perturbation_relevance, and tme_compatibility=low must carry notes. Both raise, matching the submodel side, and both cover secondary sources. Warnings do not survive a batch load. 0 live hits, so the escalation is free. - centre-channel check inside validate_derivation_code: an across_patient target must not return the population's own 2.5/97.5 percentiles as ci95. That encodes the spread twice — omega reads it from `samples`, and flat inference reads the same width as measurement noise, so the target is weighted as though one simulated patient could land anywhere in the cohort. 26 live hits, all via population.summarize() with no n. population.summarize() gains n=/rng=/n_boot=. With n it subsample- bootstraps the median, so ci95 shrinks with n while `samples` passes through untouched. Shape-agnostic, which matters for the skewed lognormal marginals the density targets use. Bare summarize() is still correct for a center_only target. Without this the error message had no actionable fix for 24 of the 26 (bootstrap_median needs per-patient values they do not have), and the only way to clear it would have deleted the omega channel. - hardcoded values in distribution_code, scoped to non-integer floats. The submodel allowlist ({0,1,2,1.96,1.645}) fails 50 of 50 targets because distribution_code is Monte-Carlo statistics — seeds, draw counts, percentile arguments, IQR->SD factors. A ban that flags everything gets switched off. Scoped to non-integer floats it flags 5, all real: a fraction range read out of a paper into np.log(), and a detection floor plus tolerance band belonging in assumptions. Integer-valued measurements are a known gap, documented on the validator. - validate_snippets_against_pdfs, context-guarded on papers_dir. validate_input_values_in_snippets checks the value against the snippet, but both are LLM-written and can be jointly invented; this checks the snippet against the paper. Zero cost until the extraction pipeline passes papers_dir. Two existing fixtures were themselves examples of what is now rejected: the population-sample fixture returned population percentiles as ci95, and the clipping fixture wrote its floor inline. Both updated. 408 passed. --- .../calibration/calibration_target_models.py | 419 ++++++++++++++-- src/maple/core/calibration/population.py | 55 +- .../test_calibration_target_validators.py | 474 +++++++++++++++++- 3 files changed, 898 insertions(+), 50 deletions(-) diff --git a/src/maple/core/calibration/calibration_target_models.py b/src/maple/core/calibration/calibration_target_models.py index b155227..e714c4e 100644 --- a/src/maple/core/calibration/calibration_target_models.py +++ b/src/maple/core/calibration/calibration_target_models.py @@ -1203,56 +1203,114 @@ def warn_observation_sd_unreasonable(self) -> "CalibrationTarget": return self + def _all_source_relevances(self) -> List[tuple]: + """Yield ``(source_tag, SourceRelevanceAssessment)`` for every declared source. + + Parity with ``SubmodelTarget._all_source_relevances``. The calibration-target + validators used to inspect ``primary_data_source`` alone, so a secondary + source could carry ``non_peer_reviewed`` quality, an unjustified + perturbation, or a low TME compatibility with no notes and never be seen. + """ + out: List[tuple] = [] + if self.primary_data_source is not None and self.primary_data_source.source_relevance: + out.append( + (self.primary_data_source.source_tag, self.primary_data_source.source_relevance) + ) + for src in self.secondary_data_sources or []: + if src.source_relevance: + out.append((src.source_tag, src.source_relevance)) + return out + @model_validator(mode="after") def validate_source_relevance_warnings(self) -> "CalibrationTarget": """Emit warnings for source-relevance issues (non-peer-reviewed, cross-indication, - cross-species, low TME compatibility, perturbation without justification).""" - if self.primary_data_source is None or self.primary_data_source.source_relevance is None: - return self + cross-species) across primary AND secondary sources. + + Ported from ``SubmodelTarget.validate_source_quality_peer_reviewed`` / + ``warn_cross_indication_extrapolation`` / ``warn_cross_species_extrapolation``, + which walk every source rather than the primary only. A non-peer-reviewed + secondary that supplies a value used in the derivation is the case worth + seeing — the submodel message says so explicitly, so this one does too. + The hard requirements (perturbation justification, low-TME notes) moved out + to ``validate_context_mismatch_justified``, where they raise. + """ import warnings - sr = self.primary_data_source.source_relevance + for tag, sr in self._all_source_relevances(): + if sr.source_quality == SourceQuality.NON_PEER_REVIEWED: + used_for_values = any(inp.source_ref == tag for inp in self.empirical_data.inputs) + severity = " AND supplies values used in the derivation" if used_for_values else "" + warnings.warn( + f"Source '{tag}' quality is 'non_peer_reviewed'{severity}. " + "This includes Wikipedia, preprints, and unreviewed databases. " + "Prefer peer-reviewed primary literature; if this source must be " + "used, document the rationale in key_assumptions or " + "key_study_limitations.", + UserWarning, + ) - if sr.source_quality == SourceQuality.NON_PEER_REVIEWED: - warnings.warn( - "source_relevance.source_quality is 'non_peer_reviewed'. " - "Prefer peer-reviewed primary literature for calibration targets. " - "If this source must be used, document the rationale in key_assumptions.", - UserWarning, - ) + if sr.indication_match in (IndicationMatch.PROXY, IndicationMatch.UNRELATED): + warnings.warn( + f"Cross-indication extrapolation in source '{tag}': " + f"indication_match='{sr.indication_match.value}'. " + "Translation sigma inflation will be applied automatically during " + "prior construction.", + UserWarning, + ) - if sr.indication_match in (IndicationMatch.PROXY, IndicationMatch.UNRELATED): - warnings.warn( - f"Cross-indication extrapolation: indication_match='{sr.indication_match.value}'. " - "Translation sigma inflation will be applied automatically during prior construction.", - UserWarning, - ) + if sr.species_source != sr.species_target: + warnings.warn( + f"Cross-species extrapolation in source '{tag}': " + f"{sr.species_source} → {sr.species_target}. " + "Translation sigma inflation will be applied automatically during " + "prior construction.", + UserWarning, + ) - if sr.species_source != sr.species_target: - warnings.warn( - f"Cross-species extrapolation: {sr.species_source} → {sr.species_target}. " - "Translation sigma inflation will be applied automatically during prior construction.", - UserWarning, - ) + return self - if ( - sr.perturbation_type in (PerturbationType.PHARMACOLOGICAL, PerturbationType.GENETIC) - and not sr.perturbation_relevance - ): - warnings.warn( - f"source_relevance.perturbation_type is '{sr.perturbation_type.value}' but " - "perturbation_relevance is not provided. Document how the perturbed " - "measurement relates to the physiological parameter being estimated.", - UserWarning, - ) + @model_validator(mode="after") + def validate_context_mismatch_justified(self) -> "CalibrationTarget": + """A declared context mismatch must carry its justification. + + Ported from ``SubmodelTarget.validate_pharmacological_perturbation_justification``, + ``validate_genetic_perturbation_justification`` and + ``validate_low_tme_compatibility_notes``, which raise where the calibration + side only warned. Warnings do not survive a batch load: nobody reads 50 + targets' worth of stderr, so an unjustified mismatch entered the likelihood + at full weight with the bias undocumented. Both checks now cover secondary + sources as well as the primary. + """ + from maple.core.calibration.exceptions import MissingFieldError - if sr.tme_compatibility == TMECompatibility.LOW and not sr.tme_compatibility_notes: - warnings.warn( - "source_relevance.tme_compatibility is 'low' but tme_compatibility_notes " - "is not provided. Document the TME differences and their expected impact.", - UserWarning, - ) + for tag, sr in self._all_source_relevances(): + if ( + sr.perturbation_type in (PerturbationType.PHARMACOLOGICAL, PerturbationType.GENETIC) + and not sr.perturbation_relevance + ): + raise MissingFieldError( + f"Source '{tag}': perturbation_type is " + f"'{sr.perturbation_type.value}' but perturbation_relevance is not " + "provided.\n\n" + "A perturbed measurement is not the physiological quantity. " + "Document:\n" + " - whether the value is an upper bound, a lower bound, or typical\n" + " - whether scaling or adjustment is needed\n" + " - for drugs, how supraphysiological exposure affects the reading\n" + " - for KO/knockdown/overexpression, compensatory mechanisms" + ) + + if sr.tme_compatibility == TMECompatibility.LOW and not sr.tme_compatibility_notes: + raise MissingFieldError( + f"Source '{tag}': tme_compatibility is 'low' but " + "tme_compatibility_notes is not provided.\n\n" + "Document the specific TME differences and their expected impact:\n" + " - stromal density differences\n" + " - immune infiltration patterns\n" + " - chemokine / cytokine milieu\n" + " - expected DIRECTION and MAGNITUDE of the bias" + ) return self @@ -1559,6 +1617,9 @@ def validate_derivation_code(self) -> "CalibrationTarget": f"median ({median_reported[0]:.4g}) within 10% — the declared " f"'samples' array must be the population draw the median/CI summarize" ) + self._check_center_channel_is_not_population( + finite, ci95_reported[0], self.empirical_data.sample_size + ) except CalibrationTargetValidationError: # Re-raise all our custom validation errors @@ -1601,6 +1662,76 @@ def validate_derivation_code(self) -> "CalibrationTarget": return self + @staticmethod + def _check_center_channel_is_not_population( + finite: "np.ndarray", ci95_pair: List[float], sample_size: int + ) -> None: + """The two channels must not both carry the population spread. + + Calibration-target analogue of + ``SubmodelTarget.validate_center_channel_sem_scale``. A target that declares + ``population_spread='across_patient'`` uses two channels: ``median`` + + ``ci95`` pin the CENTER (so the interval must shrink with n — a bootstrap or + SEM-scale interval on the median), and ``samples`` carries the POPULATION + spread that hierarchical inference reads as omega. Returning the population's + own 2.5th / 97.5th percentiles as ``ci95`` encodes the spread TWICE: omega + gets it from ``samples``, and the flat likelihood reads it as measurement + noise, so the target is weighted as though a single simulated patient were + allowed to land anywhere in the cohort. + + This is the ``notes/calibration`` position — biological variability is the + theta term, not the noise — made enforceable. + + The trap is ``population.summarize()``, whose ``ci95_lower`` /``ci95_upper`` + ARE ``np.percentile(samples, 2.5 / 97.5)``. It is the obvious helper to call + and it silently produces a population-scale center channel. Pair + ``population.empirical_population()`` with ``population.bootstrap_median()`` + instead, which bootstraps the median and so shrinks with n. + + Detection compares the reported interval against the sample's own 95% range. + Agreement within 15% on both edges means the center channel is the population + range. ``sample_size=1`` (a single subject) is exempt: there the two + coincide legitimately. + """ + if sample_size is not None and sample_size <= 1: + return + lo_rep, hi_rep = float(ci95_pair[0]), float(ci95_pair[1]) + lo_pop = float(np.percentile(finite, 2.5)) + hi_pop = float(np.percentile(finite, 97.5)) + if lo_rep == 0 or hi_rep == 0: + return + + def _agrees(a: float, b: float) -> bool: + scale = max(abs(a), abs(b)) + return scale > 0 and abs(a - b) <= 0.15 * scale + + if not (_agrees(lo_rep, lo_pop) and _agrees(hi_rep, hi_pop)): + return + + raise ScaleMismatchError( + f"population_spread='across_patient' but ci95 = [{lo_rep:.4g}, {hi_rep:.4g}] " + f"is the POPULATION range of 'samples' " + f"([{lo_pop:.4g}, {hi_pop:.4g}] at the 2.5th/97.5th percentiles), not the " + f"uncertainty on the center.\n\n" + "The spread is then encoded twice — once in 'samples' (the omega signal " + "hierarchical inference reads) and once in ci95 (which flat inference reads " + "as measurement noise). The target is weighted as if one simulated patient " + "could land anywhere in the cohort, so it constrains far less than its n " + f"(n={sample_size}) justifies.\n\n" + "FIX — keep 'samples' exactly as it is (that channel is correct) and give " + "the center its own interval that shrinks with n. Pass the study's real " + "sample size to summarize(); it subsample-bootstraps the median and leaves " + "'samples' untouched:\n\n" + " return pop.summarize(samples, n=int(inputs['sample_size'].magnitude))\n\n" + "For a target built from per-patient values, pop.bootstrap_median(values, " + "rng=rng) gives the same center interval directly.\n\n" + "DO NOT clear this error by setting population_spread='center_only' unless " + "the reported width was never real across-patient spread (a pooled-mean / " + "SEM interval, or an assumed CV). That switch DELETES the population " + "channel — the target stops contributing to omega. Here the spread looks " + "genuine, so the center channel is what needs fixing, not the spread one." + ) + @model_validator(mode="after") def validate_source_refs(self) -> "CalibrationTarget": """Validator: Check all source_refs in empirical_data.inputs point to defined sources.""" @@ -2046,6 +2177,113 @@ def validate_no_hardcoded_constants_in_observable_code(self) -> "CalibrationTarg return self + @model_validator(mode="after") + def validate_no_hardcoded_values_in_distribution_code(self) -> "CalibrationTarget": + """Reject measured values written as literals inside ``distribution_code``. + + Calibration-target analogue of + ``SubmodelTarget.validate_no_hardcoded_values_in_observation_code``. A number + that came from a paper must arrive through ``empirical_data.inputs``, where it + carries a ``value_snippet``, a ``source_ref`` and a unit — that is what makes + it auditable and what lets ``validate_input_values_in_snippets`` check it. A + literal in the body has none of that, and no validator can tell it from + arithmetic. + + The submodel rule allows only ``{0, 1, 2, 1.96, 1.645}``. Measured against the + live corpus that rule fails 50 of 50 targets, because ``distribution_code`` is + Monte-Carlo statistics: it is full of legitimate seeds (``42``), draw counts + (``10000``), percentile arguments (``2.5``, ``97.5``, ``25``, ``75``) and + reconstruction constants (``1.349`` for IQR→SD). A ban that flags every target + gets switched off, so this one is scoped to the class that actually hides data: + + * **Integer literals are exempt.** In this corpus every one is a seed, a draw + count, a ``range()`` bound, an index or a comparison threshold. *Known gap: + an integer-valued measurement (``n_cells = 500``) would pass. Tighten by + claiming literals by syntactic role if one ever appears.* + * **Non-integer floats are candidates**, minus the statistical constants below + and minus numerical guards under ``1e-5``. + + Scoped this way it flags two live targets, both real: a fraction range + (``0.02``, ``0.21``) read straight out of a paper into ``np.log()`` with no + input declared, and a detection floor plus tolerance band (``0.001``, + ``0.005``) that are modeling choices belonging in ``assumptions``. + """ + if type(self).__name__ == "IsolatedSystemTarget": + return self + + # Arithmetic, percentile arguments, normal quantiles, IQR->SD reconstruction. + ALLOWED_FLOATS = { + 0.0, + 0.5, + 1.0, + 1.5, + 2.0, + 3.0, + 10.0, + 100.0, # arithmetic / decimal scaling + 2.5, + 5.0, + 25.0, + 50.0, + 75.0, + 95.0, + 97.5, # percentile arguments + 1.96, + 1.959963984540054, # z(0.975) + 1.645, + 1.6448536269514722, # z(0.95) + 0.6744897501960817, + 0.67448975329236258, # z(0.75), IQR half-width + 1.349, + 1.35, # IQR -> SD + } + EPSILON_GUARD = 1e-5 # numerical floors (1e-6, 1e-9) are not measurements + + code = self.empirical_data.distribution_code or "" + try: + tree = ast.parse(code) + except SyntaxError: + return self # reported by validate_derivation_code + + src = code.splitlines() + offenders: Dict[float, str] = {} + for node in ast.walk(tree): + if not isinstance(node, ast.Constant): + continue + value = node.value + if isinstance(value, bool) or not isinstance(value, float): + continue + if value.is_integer(): + continue + if any(abs(value - a) < 1e-10 for a in ALLOWED_FLOATS): + continue + if abs(value) < EPSILON_GUARD: + continue + line = src[node.lineno - 1].strip() if 0 < node.lineno <= len(src) else "" + offenders.setdefault(value, line) + + if offenders: + listing = "\n".join(f" • {v!r} in: {line}" for v, line in sorted(offenders.items())) + raise HardcodedConstantError( + "distribution_code contains hardcoded non-integer values:\n" + f"{listing}\n\n" + "Every number taken from a paper must enter through " + "empirical_data.inputs, so it carries a value_snippet, a source_ref " + "and units — a literal in the body is untraceable and is not checked " + "against the source.\n\n" + " • measured in the paper -> empirical_data.inputs, " + "input_type='direct_measurement' with value_snippet\n" + " • a reference constant -> empirical_data.inputs, " + "input_type='reference_value' or 'unit_conversion'\n" + " • a modeling choice (detection floor, tolerance band) -> " + "empirical_data.assumptions, with a rationale\n\n" + "Seeds, draw counts, percentile arguments and normal/IQR constants are " + "already exempt; if a legitimate statistical constant is flagged here, " + "add it to ALLOWED_FLOATS rather than declaring it as data." + ) + + return self + @model_validator(mode="after") def validate_no_extreme_dimensionless_constants(self) -> "CalibrationTarget": """ @@ -2169,6 +2407,109 @@ def validate_species_completeness(self) -> "CalibrationTarget": return self + @model_validator(mode="after") + def validate_snippets_against_pdfs(self, info: ValidationInfo) -> "CalibrationTarget": + """Fuzzy-match each input's ``value_snippet`` against the source PDF text. + + Ported verbatim in intent from + ``SubmodelTarget.validate_snippets_against_pdfs``. ``validate_input_values_in_snippets`` + already checks that the VALUE appears in the SNIPPET — but the snippet itself + is LLM-written, so that pair can be internally consistent and jointly invented. + This is the check that closes the loop: the snippet must appear in the paper. + + Requires ``context={'papers_dir': ...}``; skipped silently otherwise, so + loading existing YAMLs outside the extraction pipeline costs nothing. + Mechanistic targets are exempt — they have no measured source text. + """ + if not info.context or "papers_dir" not in info.context: + return self + if self.epistemic_basis == "mechanistic": + return self + if self.primary_data_source is None: + return self + + from maple.core.calibration.snippet_validator import load_paper_texts + from maple.core.calibration.validators import fuzzy_find_snippet_in_text + + papers_dir = info.context["papers_dir"] + + source_tags = {self.primary_data_source.source_tag} + source_metadata = { + self.primary_data_source.source_tag: { + "doi": self.primary_data_source.doi, + "url": None, + } + } + for src in self.secondary_data_sources or []: + source_tags.add(src.source_tag) + doi_or_url = getattr(src, "doi_or_url", None) or "" + is_url = doi_or_url.startswith("http") + source_metadata[src.source_tag] = { + "doi": None if is_url else (doi_or_url or None), + "url": doi_or_url if is_url else None, + } + + paper_data = load_paper_texts(source_tags, source_metadata, papers_dir) + + # Same exemptions as the value-in-snippet check: these inputs are not + # verbatim paper text. (The submodel schema's InputType also has + # REFERENCE_VALUE / UNIT_CONVERSION; the calibration one does not.) + SKIP_TYPES = { + InputType.DERIVED_ARITHMETIC, + InputType.INFERRED_ESTIMATE, + } + + errors = [] + for inp in self.empirical_data.inputs: + if inp.input_type in SKIP_TYPES: + continue + if getattr(inp, "source_type", None) == SourceType.FIGURE: + continue + if not inp.value_snippet and not inp.table_excerpt: + continue + if inp.source_ref not in paper_data: + continue + + text, source_type = paper_data[inp.source_ref] + + if inp.value_snippet: + found, score, _ = fuzzy_find_snippet_in_text(inp.value_snippet, text, threshold=0.7) + if not found: + errors.append( + f"Input '{inp.name}': value_snippet not found in " + f"{inp.source_ref} [{source_type}] (best score: {score:.2f}). " + "The snippet may be hallucinated or paraphrased — use verbatim " + "text from the paper." + ) + + if inp.table_excerpt: + te = inp.table_excerpt + for field_name, threshold in [ + ("table_id", 0.7), + ("column", 0.7), + ("value", 0.7), + ("row", 0.6), + ]: + field_val = getattr(te, field_name, None) + if not field_val: + continue + found, score, _ = fuzzy_find_snippet_in_text( + str(field_val), text, threshold=threshold + ) + if not found: + errors.append( + f"Input '{inp.name}': table_excerpt.{field_name}=" + f"'{field_val}' not found in {inp.source_ref} " + f"[{source_type}] (best score: {score:.2f})." + ) + + if errors: + from maple.core.calibration.exceptions import SnippetNotInSourceError + + raise SnippetNotInSourceError.from_errors(errors) + + return self + @model_validator(mode="after") def validate_no_control_characters(self) -> "CalibrationTarget": """ diff --git a/src/maple/core/calibration/population.py b/src/maple/core/calibration/population.py index 79d7ff7..27854b1 100644 --- a/src/maple/core/calibration/population.py +++ b/src/maple/core/calibration/population.py @@ -303,17 +303,64 @@ def midpoint(low, high): return 0.5 * (low + high) -def summarize(samples): +def summarize(samples, *, n=None, rng=None, n_boot=2_000): """Reduce a population sample to the ``derive_distribution`` return dict. ``samples`` may be 1-D (scalar observable) or 2-D ``(n_patients, k)`` (joint); the median / ci95 reduce over the patient axis (axis 0). + + **Pass ``n`` (the study's real sample size) whenever the target declares + ``population_spread='across_patient'``.** The two channels mean different things: + ``samples`` carries the POPULATION spread that hierarchical inference reads as + omega, while ``median`` / ``ci95`` pin the CENTER, so the interval must shrink + with n. Without ``n`` this returns the population's own 2.5th / 97.5th + percentiles as ``ci95``, which encodes the spread twice — omega gets it from + ``samples``, and flat inference reads the same width as measurement noise, so the + target is weighted as though one simulated patient could land anywhere in the + cohort. ``CalibrationTarget`` rejects that (see + ``_check_center_channel_is_not_population``). + + With ``n``, ``ci95`` becomes a subsample-bootstrap interval on the median: draw + ``n`` patients from the population sample, take the median, repeat. This is + shape-agnostic (no normality assumption, correct for the skewed lognormal + marginals most density targets use) and it is the right width for a cohort of + that size. ``rng`` defaults to a fixed seed so the target stays reproducible. + + Bare ``summarize(samples)`` remains correct for a ``center_only`` target, where + there is no second channel to double-encode. """ axis = 0 if getattr(samples, "ndim", 1) > 1 else None + median_obs = np.median(samples, axis=axis) + + if n is None: + return { + "median_obs": median_obs, + "ci95_lower": np.percentile(samples, 2.5, axis=axis), + "ci95_upper": np.percentile(samples, 97.5, axis=axis), + "samples": samples, + } + + n = int(n) + if n < 1: + raise ValueError(f"summarize(n=...) needs a positive sample size, got {n}") + if rng is None: + rng = np.random.default_rng(_SHUFFLE_SEED) + + units = getattr(samples, "units", None) + mag = np.asarray(getattr(samples, "magnitude", samples), dtype=float) + flat = mag if mag.ndim == 1 else mag # index patients on axis 0 either way + n_pop = flat.shape[0] + idx = rng.integers(0, n_pop, size=(n_boot, n)) + boots = np.median(flat[idx], axis=1) # (n_boot,) or (n_boot, k) + lo = np.percentile(boots, 2.5, axis=0) + hi = np.percentile(boots, 97.5, axis=0) + if units is not None: + lo = lo * units + hi = hi * units return { - "median_obs": np.median(samples, axis=axis), - "ci95_lower": np.percentile(samples, 2.5, axis=axis), - "ci95_upper": np.percentile(samples, 97.5, axis=axis), + "median_obs": median_obs, + "ci95_lower": lo, + "ci95_upper": hi, "samples": samples, } diff --git a/tests/unit/core/test_calibration_target_validators.py b/tests/unit/core/test_calibration_target_validators.py index 1a53d14..3c0ce97 100644 --- a/tests/unit/core/test_calibration_target_validators.py +++ b/tests/unit/core/test_calibration_target_validators.py @@ -32,6 +32,7 @@ """ import copy +import warnings import pytest from unittest.mock import Mock, patch from pydantic import ValidationError @@ -703,6 +704,18 @@ def test_validate_clipping_suggests_lognormal( ): """Validator should warn when distribution_code uses clipping.""" data = copy.deepcopy(golden_calibration_target_data) + # The clip floor is a modeling choice, so it is declared as an assumption + # rather than written inline — validate_no_hardcoded_values_in_distribution_code + # rejects a bare float literal. Orthogonal to the rule under test. + data["empirical_data"]["assumptions"] = [ + { + "name": "clip_floor", + "value": 0.01, + "units": "dimensionless", + "description": "Clip floor for the lognormal draw.", + "rationale": "Lower bound applied to avoid non-positive draws.", + } + ] # Add clipping to distribution_code (use new input names) data["empirical_data"]["distribution_code"] = ( "def derive_distribution(inputs, ureg):\n" @@ -714,7 +727,8 @@ def test_validate_clipping_suggests_lognormal( " n = 10000\n" " mu_log = math.log(mean.magnitude)\n" " samples = np.random.lognormal(mu_log, sigma_log.magnitude, n)\n" - " samples = np.clip(samples, 0.01, None) * mean.units # Clipping!\n" + " floor = inputs['clip_floor'].magnitude\n" + " samples = np.clip(samples, floor, None) * mean.units # Clipping!\n" " median_obs = np.median(samples)\n" " ci95 = np.percentile(samples, [2.5, 97.5])\n" " ci95_lower = ci95[0]\n" @@ -1589,23 +1603,31 @@ class TestCalibrationTargetPopulationSample: """The optional declared 'samples' population draw + population_spread gate.""" # A lognormal population draw whose median matches the golden reported median (1.0). + # + # The centre channel is SEM-scale: summarize(n=42) subsample-bootstraps the + # median, so ci95 shrinks with n while `samples` stays the population draw. + # Returning np.percentile(samples, [2.5, 97.5]) as ci95 instead would put the + # SAME width in both channels, which _check_center_channel_is_not_population + # now rejects (see TestCalCenterChannelNotPopulation). _GOOD_CODE = ( "def derive_distribution(inputs, ureg):\n" " import numpy as np, math\n" + " from maple.core.calibration import population as pop\n" " np.random.seed(42)\n" " mean = inputs['cd8_ratio_mean']\n" " sigma_log = inputs['cd8_ratio_sigma_log']\n" " mu_log = math.log(mean.magnitude)\n" " samples = np.random.lognormal(mu_log, sigma_log.magnitude, 10000) * mean.units\n" - " ci95 = np.percentile(samples, [2.5, 97.5])\n" - " return {'median_obs': np.median(samples), 'ci95_lower': ci95[0],\n" - " 'ci95_upper': ci95[1], 'samples': samples}" + " return pop.summarize(samples, n=42)" ) + # The centre interval _GOOD_CODE produces, for fixtures that use it. + _GOOD_CI95 = [[0.829794, 1.196636]] def _with_code(self, golden, code=None, **ed_overrides): data = copy.deepcopy(golden) if code is not None: data["empirical_data"]["distribution_code"] = code + data["empirical_data"]["ci95"] = copy.deepcopy(self._GOOD_CI95) data["empirical_data"].update(ed_overrides) return data @@ -1652,7 +1674,15 @@ def test_samples_median_mismatch_rejected( self, model_structure, golden_calibration_target_data, mock_crossref_success ): # samples centered 5x off the reported/computed median -> rejected - code = self._GOOD_CODE.replace("'samples': samples}", "'samples': samples * 5.0}") + # Override only the samples key: median/ci95 stay tied to the good draw, + # so the mismatch this test targets is reached rather than the earlier + # computed-vs-reported median check. + code = self._GOOD_CODE.replace( + " return pop.summarize(samples, n=42)", + " out = pop.summarize(samples, n=42)\n" + " out['samples'] = samples * 5.0\n" + " return out", + ) data = self._with_code( golden_calibration_target_data, code, population_spread="across_patient" ) @@ -1664,8 +1694,10 @@ def test_degenerate_samples_rejected( ): # A flat (zero-variance) sample is not a usable population spread. code = self._GOOD_CODE.replace( - "'samples': samples}", - "'samples': np.ones(10000) * mean.magnitude * mean.units}", + " return pop.summarize(samples, n=42)", + " out = pop.summarize(samples, n=42)\n" + " out['samples'] = np.ones(10000) * mean.magnitude * mean.units\n" + " return out", ) data = self._with_code( golden_calibration_target_data, code, population_spread="across_patient" @@ -1809,3 +1841,431 @@ def test_uncertainty_name_on_measurement_type_stays_legal(self): CalibrationTargetEstimates.model_validate( _estimates_with_input("cv_reported_by_authors", "direct_parameter") ) + + +# ============================================================================ +# Tier-1 ports from SubmodelTarget (see +# notes/calibration/cal_target_schema_hardening_2026-07-26.md in pdac-build). +# Each block names the submodel validator it mirrors and the live defect it +# was measured against. +# ============================================================================ + + +def _secondary_source(**relevance_overrides) -> dict: + """A secondary source carrying its own source_relevance block.""" + sr = dict(DEFAULT_CLINICAL_SOURCE_RELEVANCE) + sr.update(relevance_overrides) + return { + "source_tag": "uniprot_ccl2", + "title": "UniProt entry P13500 (CCL2)", + "first_author": "UniProt", + "year": 2023, + "doi_or_url": "https://www.uniprot.org/uniprot/P13500", + "source_relevance": sr, + } + + +class TestCalSourceRelevanceCoversSecondarySources: + """Ported from SubmodelTarget.validate_source_quality_peer_reviewed. + + The calibration side inspected primary_data_source alone, so a secondary + source could be a wiki, a preprint or an unreviewed database and never be + reported. The live case is a UniProt molecular-weight lookup on + pdac_tumor_ccl2_mcp1_concentration_ghassemzadeh2017. + """ + + def test_non_peer_reviewed_secondary_now_warns( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = copy.deepcopy(golden_calibration_target_data) + data["secondary_data_sources"] = [_secondary_source(source_quality="non_peer_reviewed")] + with pytest.warns(UserWarning, match="uniprot_ccl2.*non_peer_reviewed"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_warning_flags_a_secondary_that_supplies_values( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """A non-peer-reviewed source used only for context is milder than one + whose numbers reach the derivation; the message must say which.""" + data = copy.deepcopy(golden_calibration_target_data) + data["secondary_data_sources"] = [_secondary_source(source_quality="non_peer_reviewed")] + data["empirical_data"]["inputs"][0]["source_ref"] = "uniprot_ccl2" + with pytest.warns(UserWarning, match="supplies values used in the derivation"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_cross_species_secondary_now_warns( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = copy.deepcopy(golden_calibration_target_data) + data["secondary_data_sources"] = [_secondary_source(species_source="mouse")] + with pytest.warns(UserWarning, match="Cross-species extrapolation in source"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_clean_sources_emit_no_relevance_warning( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = copy.deepcopy(golden_calibration_target_data) + data["secondary_data_sources"] = [_secondary_source()] + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + assert not [ + w for w in caught if "extrapolation" in str(w.message) or "quality is" in str(w.message) + ] + + +class TestCalContextMismatchJustified: + """Ported from SubmodelTarget.validate_pharmacological_perturbation_justification, + validate_genetic_perturbation_justification and + validate_low_tme_compatibility_notes — all of which raise where the + calibration side only warned. + + Measured against the live corpus first: zero targets violate either rule, so + escalating warning -> error costs no migration. + """ + + def test_pharmacological_without_relevance_raises( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = copy.deepcopy(golden_calibration_target_data) + sr = data["primary_data_source"]["source_relevance"] + sr["perturbation_type"] = "pharmacological" + sr["perturbation_relevance"] = "" + with pytest.raises(ValidationError, match="perturbation_relevance is not"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_genetic_without_relevance_raises( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = copy.deepcopy(golden_calibration_target_data) + sr = data["primary_data_source"]["source_relevance"] + sr["perturbation_type"] = "genetic_perturbation" + sr["perturbation_relevance"] = "" + with pytest.raises(ValidationError, match="perturbation_relevance is not"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_low_tme_without_notes_raises( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = copy.deepcopy(golden_calibration_target_data) + sr = data["primary_data_source"]["source_relevance"] + sr["tme_compatibility"] = "low" + sr["tme_compatibility_notes"] = "" + with pytest.raises(ValidationError, match="tme_compatibility_notes is not"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_secondary_source_mismatch_also_raises( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """The whole point of the port: secondary sources are checked too.""" + data = copy.deepcopy(golden_calibration_target_data) + data["secondary_data_sources"] = [ + _secondary_source(tme_compatibility="low", tme_compatibility_notes="") + ] + with pytest.raises(ValidationError, match="uniprot_ccl2.*tme_compatibility_notes"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_justified_mismatch_passes( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = copy.deepcopy(golden_calibration_target_data) + sr = data["primary_data_source"]["source_relevance"] + sr["perturbation_type"] = "pharmacological" + sr["perturbation_relevance"] = "Gemcitabine-exposed specimens; value is an upper bound." + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + +class TestCalCenterChannelNotPopulation: + """Ported from SubmodelTarget.validate_center_channel_sem_scale. + + An across_patient target has two channels: `samples` is the population + spread (omega), median/ci95 pin the centre and must shrink with n. Returning + the population percentiles as ci95 encodes the spread twice. Measured + against the live corpus: 26 targets do exactly this, all via + population.summarize() without an n. + """ + + # Population draw whose median matches the golden reported median (1.0). + _POP = ( + "def derive_distribution(inputs, ureg):\n" + " import numpy as np, math\n" + " rng = np.random.default_rng(42)\n" + " mean = inputs['cd8_ratio_mean']\n" + " sigma_log = inputs['cd8_ratio_sigma_log']\n" + " mu_log = math.log(mean.magnitude)\n" + " samples = rng.lognormal(mu_log, sigma_log.magnitude, 10000) * mean.units\n" + "{body}" + ) + + _POPULATION_CI = _POP.format( + body=( + " ci95 = np.percentile(samples, [2.5, 97.5])\n" + " return {'median_obs': np.median(samples), 'ci95_lower': ci95[0],\n" + " 'ci95_upper': ci95[1], 'samples': samples}" + ) + ) + + _SEM_CI = _POP.format( + body=( + " from maple.core.calibration import population as pop\n" + " return pop.summarize(samples, n=42, rng=rng)" + ) + ) + + def _data(self, golden, code, **ed): + data = copy.deepcopy(golden) + data["empirical_data"]["distribution_code"] = code + data["empirical_data"]["population_spread"] = "across_patient" + data["empirical_data"].update(ed) + return data + + def test_population_range_as_ci95_rejected( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = self._data(golden_calibration_target_data, self._POPULATION_CI) + with pytest.raises(ValidationError, match="is the POPULATION range"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_error_names_the_channel_that_needs_fixing( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """Clearing the error with population_spread='center_only' would delete + the omega channel. The message must steer to summarize(n=...) instead.""" + data = self._data(golden_calibration_target_data, self._POPULATION_CI) + with pytest.raises(ValidationError) as exc: + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + msg = str(exc.value) + assert "summarize(samples, n=" in msg + assert "DELETES the population channel" in msg + + def test_sem_scale_ci95_passes( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + # summarize(n=42) bootstraps the median, so ci95 is far narrower than the + # population range; median and samples are unchanged. + data = self._data(golden_calibration_target_data, self._SEM_CI, ci95=[[0.888, 1.124]]) + target = CalibrationTarget.model_validate( + data, context={"model_structure": model_structure} + ) + lo, hi = target.empirical_data.ci95[0] + assert 0.8 < lo < 1.0 and 1.0 < hi < 1.3 + + def test_single_subject_exempt( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """With n=1 the centre's uncertainty IS the population spread; no double + encoding is possible, so the check must not fire.""" + data = self._data( + golden_calibration_target_data, + self._POPULATION_CI, + sample_size=1, + sample_size_rationale="single reported subject", + ) + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_center_only_targets_unaffected( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """A center_only target has no second channel, so a wide ci95 is fine.""" + CalibrationTarget.model_validate( + golden_calibration_target_data, context={"model_structure": model_structure} + ) + + +class TestCalNoHardcodedValuesInDistributionCode: + """Ported from SubmodelTarget.validate_no_hardcoded_values_in_observation_code, + scoped to non-integer floats. + + The submodel allowlist ({0,1,2,1.96,1.645}) fails 50 of 50 live targets, + because distribution_code is Monte-Carlo statistics. Scoped to non-integer + floats it flags 5, all real: a fraction range read straight out of a paper + into np.log(), and a detection floor / tolerance band that belong in + assumptions. + """ + + def _with_code(self, golden, extra_lines): + """Inject extra statements into the golden derivation. + + The golden code already returns values matching the fixture's declared + median/ci95, so injecting keeps validate_derivation_code happy and + isolates the rule under test. The golden body is itself clean under this + validator (seed 42, n=10000, percentiles 2.5/97.5 are all exempt). + """ + data = copy.deepcopy(golden) + code = data["empirical_data"]["distribution_code"] + anchor = " import math\n" + assert anchor in code + data["empirical_data"]["distribution_code"] = code.replace(anchor, anchor + extra_lines, 1) + return data + + def test_paper_value_as_literal_rejected( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """The live catch: apcaf_fraction_of_caf hardcodes its 2%-21% range.""" + data = self._with_code( + golden_calibration_target_data, + " lo, hi = math.log(0.02), math.log(0.21)\n", + ) + with pytest.raises(ValidationError, match="hardcoded non-integer values"): + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_error_lists_the_offending_line( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + data = self._with_code( + golden_calibration_target_data, + " EPS_FLOOR = 0.001\n", + ) + with pytest.raises(ValidationError) as exc: + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + assert "EPS_FLOOR = 0.001" in str(exc.value) + + def test_statistical_constants_stay_legal( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """Seeds, draw counts, percentile arguments, z-values and the IQR->SD + factor are the vocabulary of the file — banning them switches the + validator off.""" + data = self._with_code( + golden_calibration_target_data, + " rng2 = np.random.default_rng(42)\n" + " draws = rng2.normal(1.0, 0.5, 10000)\n" + " q1, q3 = np.percentile(draws, [25, 75])\n" + " sd = (q3 - q1) / 1.349\n" + " half = 1.959963984540054 * sd\n" + " guard = max(sd, 1e-6)\n", + ) + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + def test_integers_are_exempt( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """Known gap, documented on the validator: every integer in the live + corpus is a count, index or bound, so integers are not candidates.""" + data = self._with_code( + golden_calibration_target_data, + " n_mc = 200000\n" " vals = [inputs['cd8_ratio_mean'] for _ in range(1, 11)]\n", + ) + CalibrationTarget.model_validate(data, context={"model_structure": model_structure}) + + +class TestCalSnippetsAgainstPdfs: + """Ported from SubmodelTarget.validate_snippets_against_pdfs. + + validate_input_values_in_snippets checks the VALUE against the SNIPPET, but + both are LLM-written and can be jointly invented. This closes the loop by + checking the snippet against the paper. + """ + + def test_skipped_without_papers_dir( + self, model_structure, golden_calibration_target_data, mock_crossref_success + ): + """No papers_dir in context -> no PDF work, no network, no failure.""" + with patch("maple.core.calibration.snippet_validator.load_paper_texts") as loader: + CalibrationTarget.model_validate( + golden_calibration_target_data, context={"model_structure": model_structure} + ) + loader.assert_not_called() + + def test_snippet_absent_from_paper_rejected( + self, model_structure, golden_calibration_target_data, mock_crossref_success, tmp_path + ): + with patch( + "maple.core.calibration.snippet_validator.load_paper_texts", + return_value={"smith_2020": ("The paper says nothing of the sort.", "pdf")}, + ): + with pytest.raises(ValidationError, match="value_snippet not found"): + CalibrationTarget.model_validate( + golden_calibration_target_data, + context={"model_structure": model_structure, "papers_dir": tmp_path}, + ) + + def test_snippet_present_in_paper_passes( + self, model_structure, golden_calibration_target_data, mock_crossref_success, tmp_path + ): + snippet = golden_calibration_target_data["empirical_data"]["inputs"][0]["value_snippet"] + with patch( + "maple.core.calibration.snippet_validator.load_paper_texts", + return_value={"smith_2020": (f"Results. {snippet} See Table 2.", "pdf")}, + ): + CalibrationTarget.model_validate( + golden_calibration_target_data, + context={"model_structure": model_structure, "papers_dir": tmp_path}, + ) + + def test_mechanistic_targets_exempt( + self, model_structure, golden_calibration_target_data, mock_crossref_success, tmp_path + ): + """A mechanistic target encodes reasoning, not measured text.""" + data = copy.deepcopy(golden_calibration_target_data) + data["epistemic_basis"] = "mechanistic" + with patch( + "maple.core.calibration.snippet_validator.load_paper_texts", + return_value={"smith_2020": ("Nothing matching.", "pdf")}, + ): + CalibrationTarget.model_validate( + data, context={"model_structure": model_structure, "papers_dir": tmp_path} + ) + + +class TestPopulationSummarizeSemScale: + """population.summarize(n=...) must narrow the CENTRE only.""" + + def _samples(self): + import numpy as np + + from maple.core.unit_registry import ureg + + rng = np.random.default_rng(7) + return rng.lognormal(0.0, 0.8, 20000) * ureg.dimensionless + + def test_bare_summarize_returns_population_range(self): + import numpy as np + + from maple.core.calibration import population as pop + + s = self._samples() + out = pop.summarize(s) + assert np.isclose(out["ci95_lower"].magnitude, np.percentile(s.magnitude, 2.5)) + assert np.isclose(out["ci95_upper"].magnitude, np.percentile(s.magnitude, 97.5)) + + def test_n_narrows_center_and_leaves_samples_alone(self): + import numpy as np + + from maple.core.calibration import population as pop + + s = self._samples() + wide = pop.summarize(s) + tight = pop.summarize(s, n=100) + + # The population channel is untouched, in both value and centre. + assert np.array_equal(tight["samples"].magnitude, s.magnitude) + assert np.isclose(tight["median_obs"].magnitude, wide["median_obs"].magnitude) + + # The centre channel narrows by roughly sqrt(n). + w = np.log(wide["ci95_upper"].magnitude / wide["ci95_lower"].magnitude) + t = np.log(tight["ci95_upper"].magnitude / tight["ci95_lower"].magnitude) + assert 5.0 < w / t < 15.0, f"expected ~sqrt(100)=10x narrowing, got {w / t:.1f}x" + + def test_center_interval_brackets_the_median(self): + from maple.core.calibration import population as pop + + s = self._samples() + out = pop.summarize(s, n=50) + assert ( + out["ci95_lower"].magnitude < out["median_obs"].magnitude < out["ci95_upper"].magnitude + ) + + def test_units_are_preserved(self): + from maple.core.calibration import population as pop + + s = self._samples() + out = pop.summarize(s, n=30) + assert out["ci95_lower"].units == s.units + assert out["ci95_upper"].units == s.units + + def test_rejects_nonpositive_n(self): + from maple.core.calibration import population as pop + + with pytest.raises(ValueError, match="positive sample size"): + pop.summarize(self._samples(), n=0) From 83e47589ddb3d71ecc9e5f836ba26f3e5c4d9e8f Mon Sep 17 00:00:00 2001 From: Joel Eliason Date: Mon, 27 Jul 2026 07:29:28 -0600 Subject: [PATCH 3/3] style: clear the 7 lint errors this branch added CI's ruff runs a wider rule set than pre-commit's (634 errors repo-wide, red on main for at least three merges). This branch added 7 on top; clearing them keeps the touched files at main's baseline of 51. One was a real leftover, not style: summarize() had `flat = mag if mag.ndim == 1 else mag`, a no-op branch from an earlier draft that treated the 1-D and 2-D cases separately. Patients index axis 0 either way, so the indexing collapses to one path. --- .../calibration/calibration_target_models.py | 8 +++---- src/maple/core/calibration/population.py | 8 +++---- .../test_calibration_target_validators.py | 21 +++++++++++-------- 3 files changed, 20 insertions(+), 17 deletions(-) diff --git a/src/maple/core/calibration/calibration_target_models.py b/src/maple/core/calibration/calibration_target_models.py index e714c4e..be27893 100644 --- a/src/maple/core/calibration/calibration_target_models.py +++ b/src/maple/core/calibration/calibration_target_models.py @@ -1203,7 +1203,7 @@ def warn_observation_sd_unreasonable(self) -> "CalibrationTarget": return self - def _all_source_relevances(self) -> List[tuple]: + def _all_source_relevances(self) -> list[tuple]: """Yield ``(source_tag, SourceRelevanceAssessment)`` for every declared source. Parity with ``SubmodelTarget._all_source_relevances``. The calibration-target @@ -1211,7 +1211,7 @@ def _all_source_relevances(self) -> List[tuple]: source could carry ``non_peer_reviewed`` quality, an unjustified perturbation, or a low TME compatibility with no notes and never be seen. """ - out: List[tuple] = [] + out: list[tuple] = [] if self.primary_data_source is not None and self.primary_data_source.source_relevance: out.append( (self.primary_data_source.source_tag, self.primary_data_source.source_relevance) @@ -1664,7 +1664,7 @@ def validate_derivation_code(self) -> "CalibrationTarget": @staticmethod def _check_center_channel_is_not_population( - finite: "np.ndarray", ci95_pair: List[float], sample_size: int + finite: "np.ndarray", ci95_pair: list[float], sample_size: int ) -> None: """The two channels must not both carry the population spread. @@ -2246,7 +2246,7 @@ def validate_no_hardcoded_values_in_distribution_code(self) -> "CalibrationTarge return self # reported by validate_derivation_code src = code.splitlines() - offenders: Dict[float, str] = {} + offenders: dict[float, str] = {} for node in ast.walk(tree): if not isinstance(node, ast.Constant): continue diff --git a/src/maple/core/calibration/population.py b/src/maple/core/calibration/population.py index 27854b1..567b733 100644 --- a/src/maple/core/calibration/population.py +++ b/src/maple/core/calibration/population.py @@ -347,11 +347,11 @@ def summarize(samples, *, n=None, rng=None, n_boot=2_000): rng = np.random.default_rng(_SHUFFLE_SEED) units = getattr(samples, "units", None) + # Patients index axis 0 whether the observable is scalar (1-D) or joint (2-D), + # so one indexing path covers both. mag = np.asarray(getattr(samples, "magnitude", samples), dtype=float) - flat = mag if mag.ndim == 1 else mag # index patients on axis 0 either way - n_pop = flat.shape[0] - idx = rng.integers(0, n_pop, size=(n_boot, n)) - boots = np.median(flat[idx], axis=1) # (n_boot,) or (n_boot, k) + idx = rng.integers(0, mag.shape[0], size=(n_boot, n)) + boots = np.median(mag[idx], axis=1) # (n_boot,) or (n_boot, k) lo = np.percentile(boots, 2.5, axis=0) hi = np.percentile(boots, 97.5, axis=0) if units is not None: diff --git a/tests/unit/core/test_calibration_target_validators.py b/tests/unit/core/test_calibration_target_validators.py index 3c0ce97..db5b0ad 100644 --- a/tests/unit/core/test_calibration_target_validators.py +++ b/tests/unit/core/test_calibration_target_validators.py @@ -33,6 +33,7 @@ import copy import warnings +from typing import ClassVar import pytest from unittest.mock import Mock, patch from pydantic import ValidationError @@ -1621,7 +1622,7 @@ class TestCalibrationTargetPopulationSample: " return pop.summarize(samples, n=42)" ) # The centre interval _GOOD_CODE produces, for fixtures that use it. - _GOOD_CI95 = [[0.829794, 1.196636]] + _GOOD_CI95: ClassVar[list[list[float]]] = [[0.829794, 1.196636]] def _with_code(self, golden, code=None, **ed_overrides): data = copy.deepcopy(golden) @@ -2170,15 +2171,17 @@ def test_skipped_without_papers_dir( def test_snippet_absent_from_paper_rejected( self, model_structure, golden_calibration_target_data, mock_crossref_success, tmp_path ): - with patch( - "maple.core.calibration.snippet_validator.load_paper_texts", - return_value={"smith_2020": ("The paper says nothing of the sort.", "pdf")}, + with ( + patch( + "maple.core.calibration.snippet_validator.load_paper_texts", + return_value={"smith_2020": ("The paper says nothing of the sort.", "pdf")}, + ), + pytest.raises(ValidationError, match="value_snippet not found"), ): - with pytest.raises(ValidationError, match="value_snippet not found"): - CalibrationTarget.model_validate( - golden_calibration_target_data, - context={"model_structure": model_structure, "papers_dir": tmp_path}, - ) + CalibrationTarget.model_validate( + golden_calibration_target_data, + context={"model_structure": model_structure, "papers_dir": tmp_path}, + ) def test_snippet_present_in_paper_passes( self, model_structure, golden_calibration_target_data, mock_crossref_success, tmp_path