123 lines
3.4 KiB
Python
123 lines
3.4 KiB
Python
"""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()
|