xfr/longitudinal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(lee): exclude SSFP sequences and enhance T1c exclusion notes

Browse source 
Fiesta/CISS/SSFP are balanced-SSFP (T2-dominant) sequences whose
short TE fools the seq/TE fallback, causing T1c misclassification.
Add them to the exclusion regex and guard against empty candidates.

Expand write_excluded_notes to enumerate all series in a timepoint
(not just T1c candidates) and parse DICOM txt metadata for each
series' description and protocol, producing richer no-T1c reports.
This commit is contained in:
Furen Xiao 2026-09-27 15:29:02 +08:00
parent b495cf0a80
commit 0de7655153
2 changed files with 91 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
scripts/03_scan_lee_t1c.py
View file

@ -19,7 +19,9 @@ from src.common import ROOT, DATA, save_jsonl, load_jsonl, is_head_series, path
BASE = path("lee")
T1_NAME_RE = re.compile(r"t1|tfl|spgr|mp2rage|tse3d|vfl|mpage", re.I)
EXCL_RE = re.compile(r"\bt2\b|dwi|dti|mra|mrv|angi|swi|bold|\bpp2d|\bpp3d|perf|t2\*|t2star", re.I)
# fiesta/ciss/ssfp: balanced-SSFP (fluid-bright, T2-dominant) — their short TE
# fools the seq/TE fallback below, so they are excluded by name, not T1c
EXCL_RE = re.compile(r"\bt2\b|dwi|dti|mra|mrv|angi|swi|bold|\bpp2d|\bpp3d|perf|t2\*|t2star|fiesta|ciss|ssfp", re.I)
def parse_txt(p):
@ -132,6 +134,8 @@ def scan_timepoint(sid, date, tpd):
# timepoint has no thinner valid T1c; the others are flagged
# thick_dropped: out of the selected manifest, but kept in raw so 04 can
# fall back to them when every other T1c of the timepoint is lost
if not cand:
return []
thin = [c for c in cand if c["max_sp"] <= MAX_SPACING]
keep = thin if thin else [max(cand, key=lambda c: (c["acq"], c["n_slices"]))]
dropped = [c for c in cand if c not in keep]

96
scripts/04_reconstruct_lee.py
View file

@ -286,27 +286,103 @@ def zero_tp_rows(uniq):
and r["sid"] in sub_ok and (r["sid"], r["date"]) not in tp_ok]
def scan_timepoint_series(sid, date, subj_dir):
"""All series present in every <date>_MR_* timepoint dir of a subject.
Returns {ser: {"jpg": n_jpg_files, "txt": [(slice, path), ...]}} covering
not just the T1c candidates but every series the exam exported (T2 /
FLAIR / angio / ... that the scan never considered)."""
out = {}
if not subj_dir or not os.path.isdir(subj_dir):
return out
try:
tps = [d for d in os.listdir(subj_dir)
if d.startswith(date + "_MR_") and os.path.isdir(os.path.join(subj_dir, d))]
except OSError:
return out
pj = re.compile(rf"^{re.escape(sid)}_{date}_MR_(\d+)_(\d+)_(\d+)\.jpg$")
pt = re.compile(rf"^{re.escape(sid)}_{date}_MR_(\d+)_(\d+)\.txt$")
for tp in tps:
tpd = os.path.join(subj_dir, tp)
try:
entries = os.listdir(tpd)
except OSError:
continue
jpg_dir = next((os.path.join(tpd, e) for e in entries
if os.path.isdir(os.path.join(tpd, e))), None)
jpg_files = []
if jpg_dir is not None:
try:
jpg_files = os.listdir(jpg_dir)
except OSError:
pass
for f in entries:
m = pt.match(f)
if m:
out.setdefault(m.group(1), {"jpg": 0, "txt": []})["txt"].append(
(int(m.group(2)), os.path.join(tpd, f)))
for jf in jpg_files:
m = pj.match(jf)
if m:
out.setdefault(m.group(1), {"jpg": 0, "txt": []})["jpg"] += 1
return out
def write_excluded_notes(zero_rows, all_rows, thick_rows, dyn_excl, reasons):
"""Write data/qa/<sid>/<date>_no_t1c.md for every timepoint with no
surviving T1c volume, listing each candidate series and its rejection reason."""
surviving T1c volume. Lists the study description and, for ALL series in
the timepoint (including ones the scan never considered as T1c), the
series description, protocol, and rejection reason (from the DICOM txt
metadata of each series' earliest slice sample)."""
tps = {(r["sid"], r["date"]) for r in zero_rows}
written = []
def esc(s):
return (s or "").strip().replace("|", "/").replace("\n", " ")
for sid, date in sorted(tps):
lst = [r for r in all_rows if (r["sid"], r["date"]) == (sid, date)]
lst += [r for r in thick_rows if (r["sid"], r["date"]) == (sid, date)]
lst.sort(key=lambda r: int(r["ser"]))
lines = [f"# {sid} {date}: no T1c volume", "",
"This timepoint has no reconstructed T1c volume: every T1c "
"candidate series was rejected.", "",
"| series | slices | description | reason |",
"|---|---|---|---|"]
rows_by_ser = {int(r["ser"]): r for r in lst}
cand = {r["key"] for r in lst}
subj_dir = None
for r in lst:
if r["key"] in dyn_excl:
if r.get("jpg_dir"):
subj_dir = os.path.dirname(os.path.dirname(r["jpg_dir"]))
break
union = scan_timepoint_series(sid, date, subj_dir)
for r in lst: # manifest series whose timepoint dir is gone
union.setdefault(str(int(r["ser"])), {"jpg": 0, "txt": []})
metas = {}
for ser, e in union.items():
r = rows_by_ser.get(int(ser))
src = min(e["txt"])[1] if e["txt"] else (r.get("txt_first") if r else "")
p = parse_txt(src) if src else None
metas[ser] = (p[5], p[6], p[7]) if p else ("", "", "")
study = next((m[0] for m in metas.values() if m[0]), "")
lines = [f"# {sid} {date}: no T1c volume", ""]
if study:
lines += [f"Study description: {study}", ""]
lines += [
"This timepoint has no reconstructed T1c volume: every T1c "
"candidate series was rejected. All series in this timepoint are "
"listed below.", "",
"| series | slices | series description | protocol | reason |",
"|---|---|---|---|---|"]
for ser in sorted(union, key=int):
e = union[ser]
r = rows_by_ser.get(int(ser))
key = f"lee_{sid}_{date}_s{ser}"
if key in dyn_excl:
why = "excluded: dynamic-frame series (other T1c in this timepoint)"
elif key in cand:
why = reasons.get(key, "no output")
else:
why = reasons.get(r["key"], "no output")
desc = (r.get("why") or "").strip().replace("|", "/")
lines.append(f"| s{r['ser']} | {r.get('n_slices', '?')} | {desc} | {why} |")
why = "not a T1c candidate"
n = e["jpg"] or (r.get("n_slices") if r else None) or "?"
_, sdesc, sproto = metas[ser]
lines.append(f"| s{ser} | {n} | {esc(sdesc)} | {esc(sproto)} | {esc(why)} |")
out = os.path.join(DATA, "qa", sid, f"{date}_no_t1c.md")
os.makedirs(os.path.dirname(out), exist_ok=True)
with open(out, "w") as f: