def wrap(lines, width=34):
out = []
for line in lines:
out.extend(textwrap.wrap(line, width) or [""])
return out
def box_height(lines):
return 0.30 * len(lines) + 0.35
def draw_box(ax, cx, cy, lines, width):
h = box_height(lines)
ax.add_patch(FancyBboxPatch(
(cx - width / 2, cy - h / 2), width, h,
boxstyle="round,pad=0.02,rounding_size=0.08", fc="white", ec="black", lw=1,
))
ax.text(cx, cy, "\n".join(lines), ha="center", va="center", fontsize=9, linespacing=1.4)
return h
def arrow(ax, start, end):
ax.annotate("", xy=end, xytext=start, arrowprops=dict(arrowstyle="-|>", lw=1, color="black"))
def flow_rows(c):
ident, scr, inc = c["identification"], c["screening"], c["included"]
dbs, others = ident.get("databases", []), ident.get("other_methods", [])
total = sum(d["n"] for d in dbs) + sum(d["n"] for d in others)
first = ["Records identified", f"(n = {total})"]
first += [f"{d['name']}{' [' + d['block'] + ']' if d.get('block') else ''} (n = {d['n']})" for d in dbs]
first += [f"other: {d['name']} (n = {d['n']})" for d in others]
removed = [
"Records removed before screening",
f"Duplicate records removed (n = {ident.get('duplicates_removed', 0)})",
f"Marked ineligible by automation tools (n = {ident.get('marked_ineligible_by_automation', 0)})",
f"Removed for other reasons (n = {ident.get('removed_other_reasons', 0)})",
]
reasons = scr.get("reports_excluded", [])
excl_total = sum(r["n"] for r in reasons)
return [
(first, removed),
(["Records screened", f"(n = {scr['records_screened']})"], ["Records excluded", f"(n = {scr['records_excluded']})"]),
(["Reports sought for retrieval", f"(n = {scr['reports_sought']})"], ["Reports not retrieved", f"(n = {scr['reports_not_retrieved']})"]),
(["Reports assessed for eligibility", f"(n = {scr['reports_assessed']})"],
["Reports excluded", f"(n = {excl_total})"] + [f"{r['reason']} (n = {r['n']})" for r in reasons]),
(["Sources of evidence included in the review", f"(n = {inc['sources']})",
"Reports of included sources", f"(n = {inc['reports']})"], None),
]
def draw_flow(c, path):
rows = [(wrap(m), wrap(s, 38) if s else None) for m, s in flow_rows(c)]
main_w, side_w, gap = 4.4, 5.0, 0.8
x_main, x_side = 3.0, 9.0
heights = [max(box_height(m), box_height(s) if s else 0) for m, s in rows]
total = sum(heights) + gap * (len(rows) - 1) + 1
fig, ax = plt.subplots(figsize=(13.2 * 0.7, total * 0.7))
ax.set_xlim(-1.2, 12)
ax.set_ylim(0, total)
ax.set_aspect("equal")
ax.axis("off")
y, previous, centers = total - 0.5, None, []
for (m, s), h in zip(rows, heights):
cy = y - h / 2
mh = draw_box(ax, x_main, cy, m, main_w)
if previous:
arrow(ax, previous, (x_main, cy + mh / 2))
if s:
draw_box(ax, x_side, cy, s, side_w)
arrow(ax, (x_main + main_w / 2, cy), (x_side - side_w / 2, cy))
previous = (x_main, cy - mh / 2)
centers.append((cy, h))
y -= h + gap
for label, (a, b) in {"Identification": (0, 0), "Screening": (1, 3), "Included": (4, 4)}.items():
top = centers[a][0] + centers[a][1] / 2
bottom = centers[b][0] - centers[b][1] / 2
ax.plot([-0.4, -0.4], [bottom, top], color="black", lw=1)
ax.text(-0.8, (top + bottom) / 2, label, rotation=90, ha="center", va="center", fontsize=10, fontweight="bold")
fig.savefig(path, dpi=200, bbox_inches="tight")
plt.show()
if have_flow:
draw_flow(counts, OUT / "prisma_flow.png")