From fc4266a9386209fcc436d945ffd8d59b1ff7e418 Mon Sep 17 00:00:00 2001 From: kert Date: Sat, 31 Oct 2026 20:40:43 -0400 Subject: [PATCH] feat(prisma): PRISMA 2020 flow counts from tags --- src/prisma/flow.py | 160 ++++++++++++++++++++++++++++++++- tests/prisma/test_flow_2020.py | 122 +++++++++++++++++++++++++ 2 files changed, 280 insertions(+), 2 deletions(-) create mode 100644 tests/prisma/test_flow_2020.py diff --git a/src/prisma/flow.py b/src/prisma/flow.py index 7411fd9..69f69e6 100644 --- a/src/prisma/flow.py +++ b/src/prisma/flow.py @@ -22,12 +22,34 @@ reviewers can spot-check categorisation at a glance. from __future__ import annotations -from dataclasses import dataclass +from collections.abc import Sequence +from dataclasses import dataclass, field, fields from typing import TYPE_CHECKING if TYPE_CHECKING: from zot.db import Db +# The scoping review pools these four projects. The systematic review +# is the precursor strand on its own. +SCOPING_PROJECTS = ( + "lead-history", + "lead-limits", + "lead-scan", + "lead-external", +) +SYSTEMATIC_PROJECT = "lead-precursors" + +# Databases the search log knows. A source with no rows still prints as 0. +_SOURCES = ( + "pubmed", + "europepmc", + "openalex", + "crossref", + "semanticscholar", + "arxiv", +) +_OTHER_METHOD_TAGS = ("found-in:citation", "found-in:books", "found-in:expert") + @dataclass class FlowCounts: @@ -39,6 +61,11 @@ class FlowCounts: excluded_stage3: int excluded_stage3_reasons: dict[str, int] included: int + identified_by_source: dict[str, int] = field(default_factory=dict) + duplicates_removed: int = 0 + reports_sought: int = 0 + reports_not_retrieved: int = 0 + other_methods: int = 0 # ── Queries ────────────────────────────────────────────────────── @@ -95,6 +122,31 @@ def _reason_counts(db: Db, item_ids: set[int]) -> dict[str, int]: return {r[0].removeprefix("screen:reason:"): r[1] for r in rows} +def _found_in(db: Db, item_ids: set[int]) -> dict[str, int]: + if not item_ids: + return {} + placeholders = ",".join("?" * len(item_ids)) + rows = db.con.execute( + f""" + SELECT t.name, COUNT(DISTINCT it.itemID) + FROM itemTags it + JOIN tags t ON it.tagID = t.tagID + WHERE it.itemID IN ({placeholders}) + AND t.name LIKE 'found-in:%' + GROUP BY t.name + """, + tuple(item_ids), + ).fetchall() + return {name.removeprefix("found-in:"): n for name, n in rows} + + +def _source_names(counts: FlowCounts) -> tuple[str, ...]: + extra = tuple( + sorted(key for key in counts.identified_by_source if key not in _SOURCES) + ) + return _SOURCES + extra + + def count(db: Db, project: str) -> FlowCounts: """Snapshot the current PRISMA flow counts for *project*.""" # The three per-project anchor items (criteria, extraction @@ -131,6 +183,13 @@ def count(db: Db, project: str) -> FlowCounts: [f"project:{project}", "stage:3-full-text", "screen:exclude"], ) + unavailable = all_proj & _items_with_tag( + db, project, "screen:reason:unavailable-fulltext" + ) + other: set[int] = set() + for tag in _OTHER_METHOD_TAGS: + other |= all_proj & _items_with_tag(db, project, tag) + return FlowCounts( identified=len(all_proj), screened=len(screened), @@ -140,6 +199,13 @@ def count(db: Db, project: str) -> FlowCounts: excluded_stage3=len(exc3), excluded_stage3_reasons=_reason_counts(db, exc3), included=len(stage4), + identified_by_source=_found_in(db, all_proj), + duplicates_removed=len( + all_proj & _items_with_tag(db, project, "dedup:duplicate") + ), + reports_sought=len((all_proj & (stage3 | stage4)) | unavailable), + reports_not_retrieved=len(unavailable), + other_methods=len(other), ) @@ -172,13 +238,20 @@ def mermaid(counts: FlowCounts, *, project: str) -> str: f'{_reasons(counts.excluded_stage3_reasons)}"]' ), f' D["Studies included
n = {counts.included}"]', + f' DUP["Duplicates removed: {counts.duplicates_removed}"]', + f' SOUGHT["Reports sought: {counts.reports_sought}"]', + f' MISS["Reports not retrieved: {counts.reports_not_retrieved}"]', + f' OTH["Other methods: {counts.other_methods}"]', " A --> B", " B --> X2", " B --> C", " C --> X3", " C --> D", - "```", ] + for name in _source_names(counts): + n = counts.identified_by_source.get(name, 0) + lines.append(f' src_{name}["{name}: {n}"]') + lines.append("```") return "\n".join(lines) @@ -202,4 +275,87 @@ def text_summary(counts: FlowCounts) -> str: ): lines.append(f" {code:<28} {n:>6}") lines.append(f"Included: {counts.included:>6}") + lines.append(f"Duplicates removed: {counts.duplicates_removed:>6}") + lines.append(f"Reports sought: {counts.reports_sought:>6}") + lines.append(f"Reports not retrieved: {counts.reports_not_retrieved:>6}") + lines.append(f"Other methods: {counts.other_methods:>6}") + lines.append("Identified by source:") + for name in _source_names(counts): + n = counts.identified_by_source.get(name, 0) + lines.append(f" {name}: {n}") return "\n".join(lines) + + +def report(counts: FlowCounts, *, project: str) -> str: + """Markdown flow whose numbers are the fields of ``counts``.""" + lines = [f"# PRISMA 2020 — {project}", "", "## Identification"] + for name in _source_names(counts): + lines.append(f"- {name}: {counts.identified_by_source.get(name, 0)}") + lines.append(f"- other methods: {counts.other_methods}") + lines.append(f"- duplicates removed: {counts.duplicates_removed}") + lines.extend( + [ + "", + "## Screening and eligibility", + f"- identified: {counts.identified}", + f"- screened: {counts.screened}", + f"- excluded at title/abstract: {counts.excluded_stage2}", + f"- reports sought: {counts.reports_sought}", + f"- reports not retrieved: {counts.reports_not_retrieved}", + f"- full-text assessed: {counts.full_text_assessed}", + f"- excluded at full text: {counts.excluded_stage3}", + f"- included: {counts.included}", + ] + ) + return "\n".join(lines) + + +def _sum_dicts(parts: Sequence[dict[str, int]]) -> dict[str, int]: + total: dict[str, int] = {} + for part in parts: + for key, value in part.items(): + total[key] = total.get(key, 0) + value + return total + + +def combined(counts: Sequence[FlowCounts]) -> FlowCounts: + """Sum flow boxes. The scoping review is ``combined`` of four projects.""" + if not counts: + return FlowCounts( + identified=0, + screened=0, + excluded_stage2=0, + excluded_stage2_reasons={}, + full_text_assessed=0, + excluded_stage3=0, + excluded_stage3_reasons={}, + included=0, + ) + return FlowCounts( + identified=sum(item.identified for item in counts), + screened=sum(item.screened for item in counts), + excluded_stage2=sum(item.excluded_stage2 for item in counts), + excluded_stage2_reasons=_sum_dicts( + [item.excluded_stage2_reasons for item in counts] + ), + full_text_assessed=sum(item.full_text_assessed for item in counts), + excluded_stage3=sum(item.excluded_stage3 for item in counts), + excluded_stage3_reasons=_sum_dicts( + [item.excluded_stage3_reasons for item in counts] + ), + included=sum(item.included for item in counts), + identified_by_source=_sum_dicts([item.identified_by_source for item in counts]), + duplicates_removed=sum(item.duplicates_removed for item in counts), + reports_sought=sum(item.reports_sought for item in counts), + reports_not_retrieved=sum(item.reports_not_retrieved for item in counts), + other_methods=sum(item.other_methods for item in counts), + ) + + +def changed(before: FlowCounts, after: FlowCounts) -> tuple[str, ...]: + """Field names whose values differ, in declaration order.""" + return tuple( + item.name + for item in fields(before) + if getattr(before, item.name) != getattr(after, item.name) + ) diff --git a/tests/prisma/test_flow_2020.py b/tests/prisma/test_flow_2020.py new file mode 100644 index 0000000..0654302 --- /dev/null +++ b/tests/prisma/test_flow_2020.py @@ -0,0 +1,122 @@ +"""PRISMA 2020 flow counts read from Zotero tags.""" + +from __future__ import annotations + +import re + +from prisma.flow import ( + SCOPING_PROJECTS, + SYSTEMATIC_PROJECT, + FlowCounts, + changed, + combined, + count, + mermaid, + report, + text_summary, +) +from zot.db import TYPE_MAP, Db +from zot.schema import create_db + + +def _numbers_on(text: str, label: str) -> list[int]: + found: list[int] = [] + for line in text.splitlines(): + if label in line: + found.extend(int(n) for n in re.findall(r"\d+", line)) + return found + + +def _counts(**overrides: object) -> FlowCounts: + base: dict[str, object] = { + "identified": 0, + "screened": 0, + "excluded_stage2": 0, + "excluded_stage2_reasons": {}, + "full_text_assessed": 0, + "excluded_stage3": 0, + "excluded_stage3_reasons": {}, + "included": 0, + } + base.update(overrides) + return FlowCounts(**base) # type: ignore[arg-type] + + +def test_count_reads_2020_boxes_from_tags(tmp_path): + path = str(tmp_path / "z.sqlite") + con = create_db(path) + con.close() + with Db(path) as db: + _add( + db, + "ABCD2345", + ["project:lead-precursors", "found-in:pubmed"], + ) + _add( + db, + "EFGH5678", + ["project:lead-precursors", "found-in:books", "found-in:citation"], + ) + _add(db, "IJKM2345", ["project:lead-precursors", "dedup:duplicate"]) + _add( + db, + "NPQR6789", + [ + "project:lead-precursors", + "stage:3-full-text", + "screen:reason:unavailable-fulltext", + ], + ) + counts = count(db, "lead-precursors") + + assert counts.identified_by_source["pubmed"] == 1 + assert counts.identified_by_source.get("arxiv", 0) == 0 + assert counts.duplicates_removed == 1 + assert counts.reports_sought == 1 + assert counts.reports_not_retrieved == 1 + assert counts.other_methods == 1 + + summary = text_summary(counts) + assert _numbers_on(summary, "pubmed") == [counts.identified_by_source["pubmed"]] + assert _numbers_on(summary, "arxiv") == [0] + + md = report(counts, project="lead-precursors") + pubmed_numbers = _numbers_on(md, "pubmed") + assert pubmed_numbers + assert set(pubmed_numbers) == {counts.identified_by_source["pubmed"]} + + diagram = mermaid(counts, project="lead-precursors") + assert "Duplicates removed" in diagram + assert "Reports sought" in diagram + assert "Reports not retrieved" in diagram + assert "Other methods" in diagram + assert _numbers_on(diagram, "pubmed") == [1] + + +def test_combined_sums_and_changed_names_the_field(): + assert ( + combined( + [_counts(duplicates_removed=1), _counts(duplicates_removed=2)] + ).duplicates_removed + == 3 + ) + assert changed(_counts(included=0), _counts(included=1)) == ("included",) + assert changed(_counts(), _counts()) == () + + +def test_scoping_combines_four_projects_and_systematic_is_precursors(): + assert SCOPING_PROJECTS == ( + "lead-history", + "lead-limits", + "lead-scan", + "lead-external", + ) + assert SYSTEMATIC_PROJECT == "lead-precursors" + + +def _add(db: Db, key: str, tags: list[str]) -> None: + item_id = db.create_item(TYPE_MAP["journalArticle"], key=key) + db.set_fields(item_id, {"title": key}) + for tag in tags: + db.tag_item(item_id, tag) + db.commit()