homelab/stack
1
0
Fork 0
Code Issues 148 Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat(prisma): PRISMA 2020 flow counts from tags

Browse Source
This commit is contained in:
kert
2026-10-31 20:40:43 -04:00
parent 55614648f3
commit fc4266a938
2 changed files with 280 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
src/prisma/flow.py
View File

@@ -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<br/>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)
)

122
tests/prisma/test_flow_2020.py Normal file
View File

@@ -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()