Spaces:
Running on Zero
Running on Zero
| """Tests for the validation section: object-by-object comparison with the annotation.""" | |
| import os | |
| import zipfile | |
| import numpy as np | |
| import pytest | |
| from PIL import Image | |
| from cellquant.annotations import RasterAnnotation | |
| from cellquant.postprocess import Result | |
| from cellquant.validation import (COMPARISON_HEADERS, DIFF_HEADERS, DIFF_COLORS, ValidationSet, compare, gt_result, | |
| legend_html, write_bundle) | |
| XML = """<?xml version="1.0" encoding="utf-8"?> | |
| <annotations><version>1.1</version><meta><task><size>1</size></task></meta> | |
| <image id="0" name="{name}" width="{w}" height="{h}"> | |
| <polyline label="Cell, Antibody +" source="manual" occluded="0" points="10,10;30,10;30,30;10,30;10,10" z_order="0"></polyline> | |
| <polyline label="Nostril External of Nucleus" source="manual" occluded="0" points="40,40;44,40;44,44;40,44;40,40" z_order="0"></polyline> | |
| </image></annotations>""" | |
| def _disc(h, w, cy, cx, r): | |
| yy, xx = np.mgrid[:h, :w] | |
| return (yy - cy) ** 2 + (xx - cx) ** 2 <= r * r | |
| def _case(): | |
| """Four annotated nuclei: 1 matched with the same label, 2 matched with a different label, | |
| 3 and 4 merged into one prediction (3 matches, 4 is missed); one extra predicted nucleus. | |
| Puncta: one matched, one missed, one extra.""" | |
| H = W = 128 | |
| gt = np.zeros((H, W), np.int32) | |
| gt[_disc(H, W, 30, 30, 12)] = 1 | |
| gt[_disc(H, W, 30, 90, 12)] = 2 | |
| gt[_disc(H, W, 90, 30, 12)] = 3 | |
| gt[_disc(H, W, 90, 52, 10)] = 4 | |
| gn = np.zeros((H, W), np.int32) | |
| gn[_disc(H, W, 30, 30, 2)] = 1 | |
| gn[_disc(H, W, 110, 110, 2)] = 2 | |
| R = RasterAnnotation(gt, ["", "Cell, Antibody +", "Cell, Antibody +", "Cell, Antibody +", "Cell, Antibody -"], gn, | |
| ["", "Nostril in Nucleus", "Nostril External of Nucleus"]) | |
| pr = np.zeros((H, W), np.int32) | |
| pr[_disc(H, W, 30, 30, 12)] = 1 | |
| pr[_disc(H, W, 30, 90, 12)] = 2 | |
| pr[_disc(H, W, 90, 30, 12) | _disc(H, W, 90, 52, 10)] = 3 | |
| pr[_disc(H, W, 110, 100, 8)] = 4 | |
| pn = np.zeros((H, W), np.int32) | |
| pn[_disc(H, W, 30, 30, 2)] = 1 | |
| pn[_disc(H, W, 60, 60, 2)] = 2 | |
| res = Result(pr, pn, [{"id": 1, "label": "Cell, Antibody +"}, {"id": 2, "label": "Cell, Antibody -"}, | |
| {"id": 3, "label": "Cell, Antibody -"}, {"id": 4, "label": "Cell, Antibody +"}], | |
| [{"id": 1, "label": "Nostril in Nucleus"}, {"id": 2, "label": "Nostril External of Nucleus"}], {}, {}) | |
| img = np.zeros((H, W, 3), np.uint8) | |
| return img, R, res | |
| def test_compare_counts_and_differences(): | |
| img, R, res = _case() | |
| c = compare("T", img, R, res) | |
| rows = {r[0]: r for r in c.rows} | |
| assert list(rows) == ["Cell, Antibody +", "Cell, Antibody -", "Cell, Nostril, Antibody +", "Cell, Nostril, Antibody -", | |
| "All nuclei", "Nostril in Nucleus", "Nostril External of Nucleus", "All puncta"] | |
| assert len(c.rows[0]) == len(COMPARISON_HEADERS) | |
| # All nuclei: 4 annotated, 4 predicted, 1 matched same label, 2 label differs, 1 missed, 1 extra | |
| assert rows["All nuclei"][1:] == [4, 4, 0, 1, 2, 1, 1] | |
| assert rows["Cell, Antibody +"][1:] == [3, 2, -1, 1, 2, 0, 1] | |
| assert rows["All puncta"][1:] == [2, 2, 0, 1, 0, 1, 1] | |
| kinds = [(d[1], d[2]) for d in c.diffs] | |
| assert kinds == [("missed", "nucleus"), ("extra", "nucleus"), ("label differs", "nucleus"), ("label differs", "nucleus"), | |
| ("missed", "punctum"), ("extra", "punctum")] | |
| assert [d[0] for d in c.diffs] == [1, 2, 3, 4, 5, 6] and len(c.diffs[0]) == len(DIFF_HEADERS) | |
| missed = c.diffs[0] | |
| assert missed[3] == "Cell, Antibody -" and abs(missed[5] - 52) < 1 and abs(missed[6] - 90) < 1 | |
| assert "merged" in missed[8] and "predicted nucleus 3" in missed[8] | |
| extra = c.diffs[1] | |
| assert extra[4] == "Cell, Antibody +" and abs(extra[5] - 100) < 1 and extra[8] == "no annotated nucleus here" | |
| assert c.metrics["nuclei"]["f1"] == 0.75 and c.metrics["n_missed_merged"] == 1 | |
| assert "4 annotated nuclei, 4 predicted" in c.summary and "nucleus F1 75.0%" in c.summary | |
| for ov in (c.overlay_gt, c.overlay_pred, c.overlay_diff, c.overlay_missed, c.overlay_extra): | |
| assert ov.shape == img.shape and ov.dtype == np.uint8 and ov.any() | |
| # the missed nucleus is outlined in the 'missed' colour, the extra one in the 'extra' colour | |
| assert (c.overlay_diff[_disc(128, 128, 90, 52, 10)] == DIFF_COLORS["missed"]).all(axis=-1).any() | |
| assert (c.overlay_diff[_disc(128, 128, 110, 100, 8)] == DIFF_COLORS["extra"]).all(axis=-1).any() | |
| # the missed-only panel shows the missed nucleus and punctum and nothing else; extra-only the reverse | |
| missed_px = (c.overlay_missed == DIFF_COLORS["missed"]).all(axis=-1) | |
| extra_px = (c.overlay_extra == DIFF_COLORS["extra"]).all(axis=-1) | |
| # (the number drawn next to the missed punctum at (110, 110) reaches into the right half of the extra nucleus, | |
| # so only its left half must be free of the missed colour) | |
| left_half = _disc(128, 128, 110, 100, 8) & (np.arange(128)[None, :] < 104) | |
| assert missed_px[_disc(128, 128, 90, 52, 10)].any() and not missed_px[left_half].any() | |
| assert missed_px[_disc(128, 128, 110, 110, 5)].any() # ring around the missed punctum | |
| assert not (c.overlay_missed == DIFF_COLORS["extra"]).all(axis=-1).any() # no extra objects drawn | |
| assert extra_px[_disc(128, 128, 110, 100, 8)].any() and not extra_px[_disc(128, 128, 90, 52, 10)].any() | |
| assert extra_px[_disc(128, 128, 60, 60, 5)].any() # ring around the extra punctum | |
| assert not (c.overlay_extra == DIFF_COLORS["missed"]).all(axis=-1).any() | |
| # matched objects are untouched on both panels (the matched nucleus keeps its raw pixels) | |
| assert (c.overlay_missed[_disc(128, 128, 30, 30, 9)] == img[_disc(128, 128, 30, 30, 9)]).all() | |
| def test_perfect_prediction_has_no_differences(): | |
| img, R, _ = _case() | |
| res = gt_result(R) | |
| c = compare("P", img, R, res) | |
| assert c.diffs == [] and c.metrics["nuclei"]["f1"] == 1.0 and c.metrics["puncta"]["f1"] == 1.0 | |
| assert all(r[3] == 0 and r[6] == 0 and r[7] == 0 for r in c.rows) | |
| def test_empty_prediction_lists_every_annotated_object_as_missed(): | |
| img, R, _ = _case() | |
| res = Result(np.zeros_like(R.cell_inst), np.zeros_like(R.nostril_inst), [], [], {}, {}) | |
| c = compare("E", img, R, res) | |
| assert sum(1 for d in c.diffs if d[1] == "missed") == R.n_cells + R.n_nostrils == len(c.diffs) | |
| assert all(d[8] == "no prediction here" for d in c.diffs if d[2] == "nucleus") | |
| def test_validation_set_and_bundle(tmp_path): | |
| name = "K7 F 35D SC Iba W001 - RGB.png" | |
| Image.fromarray(np.zeros((60, 50, 3), np.uint8)).save(tmp_path / name) | |
| with zipfile.ZipFile(tmp_path / "ann.zip", "w") as zf: | |
| zf.writestr("annotations.xml", XML.format(name=name, w=50, h=60)) | |
| vs = ValidationSet(str(tmp_path)) | |
| assert vs.keys == ["K7"] and vs.choices() == [("K7 (1 nuclei, 1 puncta annotated)", "K7")] | |
| assert vs.image("K7").shape == (60, 50, 3) | |
| R = vs.raster("K7") | |
| assert R.n_cells == 1 and R.n_nostrils == 1 | |
| c = compare("K7", vs.image("K7"), R, gt_result(R)) | |
| zp = write_bundle(c, str(tmp_path / "out")) | |
| with zipfile.ZipFile(zp) as zf: | |
| names = sorted(zf.namelist()) | |
| assert names == ["K7_all_differences.png", "K7_comparison.csv", "K7_differences.csv", "K7_extra_only.png", | |
| "K7_manual.png", "K7_missed_only.png", "K7_predicted.png"] | |
| assert "missed" in legend_html() and "extra" in legend_html() | |
| # thumbnails for the image strip | |
| g = vs.gallery(size=32) | |
| assert len(g) == 1 and os.path.exists(g[0][0]) and g[0][1] == "K7: 1 nuclei, 1 puncta annotated" | |
| assert max(Image.open(g[0][0]).size) == 32 and vs.thumbnail("K7") == g[0][0] | |
| assert vs.counts("K7") == (1, 1) | |
| # a missing or empty folder is not an error | |
| assert ValidationSet(str(tmp_path / "nothing")).keys == [] | |
| assert ValidationSet("").keys == [] | |
| def test_app_builds_with_and_without_validation_bundle(tmp_path, monkeypatch): | |
| pytest.importorskip("gradio") | |
| import importlib | |
| import app | |
| monkeypatch.setattr(app, "VALIDATION", ValidationSet(str(tmp_path))) | |
| assert app.build_demo() is not None | |
| name = "K7 F 35D SC Iba W001 - RGB.png" | |
| Image.fromarray(np.zeros((60, 50, 3), np.uint8)).save(tmp_path / name) | |
| with zipfile.ZipFile(tmp_path / "ann.zip", "w") as zf: | |
| zf.writestr("annotations.xml", XML.format(name=name, w=50, h=60)) | |
| monkeypatch.setattr(app, "VALIDATION", ValidationSet(str(tmp_path))) | |
| demo = app.build_demo() | |
| assert demo is not None | |
| with pytest.raises(Exception): | |
| app.validate_image("nope") | |
| assert app.COMPARISON_NOTE.startswith("**Matched**") | |
| if app.MODEL is not None: # trained weights present: the whole section-2 path on the one-image bundle | |
| monkeypatch.delenv("GEMINI_API_KEY", raising=False) | |
| summary, manual, predicted, missed, rows, diffs, zp = app.validate_image("K7") | |
| assert summary.startswith("### K7: 1 annotated nuclei") and "Run time" in summary and "Gemini summary is off" in summary | |
| assert manual.label.startswith("1. Manual annotation of K7: 1 nuclei, 1 puncta") | |
| v = manual.value | |
| assert os.path.exists(v["path"] if isinstance(v, dict) else v) # gradio wraps file values in a dict | |
| assert predicted.label.startswith("2. Model prediction") and missed.label.startswith("3. Missed only") | |
| assert rows[0][0] == "Cell, Antibody +" and len(rows[0]) == len(COMPARISON_HEADERS) and os.path.exists(zp) | |
| importlib.reload(app) | |
| def test_gemini_prompts_and_disabled_paths(monkeypatch): | |
| from cellquant import gemini_summary as g | |
| monkeypatch.delenv("GEMINI_API_KEY", raising=False) | |
| assert not g.enabled() | |
| rows = [["All nuclei", 178, 185, 7, 169, 2, 7, 14], ["All puncta", 20, 20, 0, 16, 0, 4, 4]] | |
| txt = g._comparison_prompt("C100", rows, COMPARISON_HEADERS) | |
| assert "C100" in txt and "All nuclei, Annotated = 178, Predicted = 185" in txt.replace("Label = ", "") and "Missed = 7" in txt | |
| assert "do not round" in txt and "training set" in txt | |
| assert g.summarize_comparison("C100", rows, COMPARISON_HEADERS).startswith("Gemini summary disabled") | |
| assert g.summarize([{"image": "a", "counts": {"Cell, Antibody +": 1}}]).startswith("Gemini summary disabled") | |
| assert g._retryable(Exception("404 NOT_FOUND. model x")) and g._retryable(Exception("429 RESOURCE_EXHAUSTED. {'error': {}}")) | |
| assert not g._retryable(Exception("400 INVALID_ARGUMENT. bad key")) | |
| assert g._short_error("ClientError: 429 RESOURCE_EXHAUSTED. {'error': {'code': 429}}") == "ClientError: 429 RESOURCE_EXHAUSTED" | |
| cap = ("ClientError: 429 RESOURCE_EXHAUSTED. {'error': {'code': 429, 'message': 'Your billing account has exceeded its " | |
| "monthly spending cap. Please go to AI Studio. Learn more at https://x', 'status': 'RESOURCE_EXHAUSTED'}}") | |
| assert g._short_error(cap) == "ClientError: 429 RESOURCE_EXHAUSTED: Your billing account has exceeded its monthly spending cap. Please go to AI Studio." | |