Xiao Furen
d167c1f7c7
Introduce a robust coordinate transformation system to manage the relationship between original CT space and rotated/standardized segmentation spaces. This includes a new directory hierarchy to separate unrotated crops from rotated outputs and utility functions for geometric mapping. Key changes: - Implement `imaging/transforms.py` to handle bounding box metadata, affine standardization, and coordinate mapping between spaces. - Restructure dataset output: unrotated segmentation files (binary, SDF, ROI, etc.) are now stored in a `<vol>/crop/` subdirectory to distinguish them from `<vol>/rotated/` aligned versions. - Add `level_file_path` utility to abstract file discovery across legacy (top-level) and new (crop-based) directory structures. - Enhance `seg_bone` to capture and export bounding box metadata (`bbox2`, `nn_bbox`, `bbox_orig`) into `transform.json`. - Implement `xfr_cbt_native.py` for mapping screw positions back to original CT space. - Update preprocessing and visualization scripts to support the new directory layout and transformation metadata. - Improve TinyDB metadata migration logic to prevent accidental corruption of existing database structures.
312 lines
No EOL
14 KiB
Python
312 lines
No EOL
14 KiB
Python
#!/home/xfr/.conda/envs/cbt/bin/python
|
||
"""
|
||
檢查各 (volume, level) 的棘突（midline 後側構造）是否缺如
|
||
（先前手術如 laminectomy / 棘突切除，例：0005 L5、0770 L5）。
|
||
|
||
判定原理（見 imaging.orientation.diagnose_spinous_process）：
|
||
棘突完整時，尖端是全椎體最後側的骨且位於中線 →
|
||
deficit（全骨最後側 - 中線帶最後側）≈ 0，後側中線窄帶有骨。
|
||
切除後，最後側骨偏到側方殘餘、後側中線空洞 → deficit 大 + rear3 ≈ 0。
|
||
no_spinous = deficit >= 4 vox 且 rear3 <= 20 vox（正常 level deficit 恒為 0）。
|
||
|
||
兩種模式：
|
||
quick（預設，掃全資料集用）→ 兩段式：
|
||
stage 1：axial 投影快速篩檢（~0.5s/level）。用「寬行（椎體行）」擬合
|
||
中線軌跡（對傾斜/旋轉穩健，不受後側殘片污染），找出「缺棘突候選」。
|
||
高召回（幾乎不漏）、但因純投影在傾斜脊椎上仍會有少數假候補。
|
||
stage 2：僅對候選跑 full（鏡稱面）驗證（~3s/候選），以 full 為準給出
|
||
最終判定（confirmed / false alarm）。全資料集 ~25min + 候數×3s。
|
||
full（指定單一 volume+level 時預設）：跑鏡稱面 + 棘突/椎體完整分割，
|
||
另回報 VBODY 分割 mode（nosp_gap / nosp_post_min / quantile...），
|
||
確認該 level 的椎體切分落在「椎體後側末端（體/弓最薄處）」。
|
||
|
||
識別出的 no_spinous level 在 xfr_debug.py 執行時會打 [NO-SP] 標記，
|
||
並自動改用放寬後側谷底的椎體切分（不再把殘留後側要素當棘突移除）。
|
||
|
||
Usage:
|
||
python xfr_check_spinous.py # quick，全部 volume L1~L5
|
||
python xfr_check_spinous.py 0005 # quick，該 volume L1~L5
|
||
python xfr_check_spinous.py 0005 L5 # full（精確）單一 level
|
||
python xfr_check_spinous.py --full 0005 L5 # 強制 full
|
||
python xfr_check_spinous.py --quick # 強制 quick（全資料集掃）
|
||
"""
|
||
|
||
import csv
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
import numpy as np
|
||
import SimpleITK as sitk
|
||
|
||
from imaging.orientation import (best_symmetry_plane, segment_spinous_process,
|
||
best_upper_endplate_plane, segment_vertebral_body,
|
||
diagnose_spinous_process)
|
||
from imaging.transforms import level_file_path
|
||
|
||
standardized_dir = '/mnt/1248/open2/cyrou/CBT/Seg/Resample/standardized-xfr/'
|
||
|
||
LEVELS = ('L1', 'L2', 'L3', 'L4', 'L5')
|
||
|
||
LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'results')
|
||
|
||
# 與 diagnose_spinous_process 相同的判定閾值
|
||
DEFICIT_MIN = 4.0
|
||
REAR3_MAX = 20
|
||
# quick：中線帶內一行的最小 bone voxel 數（< 此值視為殘片游離 voxel，不算「尖端在」）
|
||
MIN_MID_VOX = 3
|
||
|
||
|
||
# =====================================================================
|
||
# quick screen：不依賴鏡稱面（~0.3s/level）
|
||
# =====================================================================
|
||
def quick_screen(m):
|
||
"""axial 投影 (y, x) 快速篩檢（不跑鏡稱面，對傾斜/旋轉穩健）：
|
||
- 「寬行」= 行內骨 voxel 數 >= 最大行的 50%（即椎體行；椎體左右
|
||
大致對稱 → 寬行 x 重心 ≈ 中線上的點）。
|
||
- 中線軌跡 x(y) = p0 + p1·y 只用寬行（椎體行）擬合：
|
||
* 用全部行擬合 → laminectomy 後側殘片行污染中線，deficit 被吃掉
|
||
（0005 L5 假正常）；
|
||
* 用固定垂直中線 → 傾斜棘突尖端被誤判為側方偏移
|
||
（0555/0609/0637 假切除）。椎體行兩者都不沾，故最穩。
|
||
- deficit = 最後側骨行 (y_min) 到「中線軌跡 ± 25% 椎體寬度 band
|
||
內有骨的最後側行」的行距（尖端在中線上 → 0；切除後 → ≈切除量）。
|
||
- rear3/rear6 = 最後側 3/6 行內、中線軌跡 ± 10% 寬度（窄中線帶）的
|
||
骨 voxel 數。
|
||
- top_off = 最後側一行 x 重心到中線軌跡的距離。
|
||
回傳 dict 或 None（資料不足）。"""
|
||
proj = m.max(axis=0)
|
||
ys_p, xs_p = np.where(proj > 0)
|
||
if ys_p.size < 50:
|
||
return None
|
||
ny = proj.shape[0]
|
||
cnt = np.bincount(ys_p, minlength=ny)
|
||
max_cnt = int(cnt.max())
|
||
if max_cnt < 4:
|
||
return None
|
||
xsum = np.bincount(ys_p, weights=xs_p.astype(np.float64), minlength=ny)
|
||
centroid = xsum / cnt.astype(float)
|
||
wide = cnt >= (0.5 * max_cnt) # 椎體行
|
||
if int(wide.sum()) < 10:
|
||
return None
|
||
yv = np.where(wide)[0]
|
||
p1, p0 = np.polyfit(yv, centroid[wide], 1) # 中線軌跡（椎體行擬合）
|
||
xmax = np.full(ny, -1.0)
|
||
xmin = np.full(ny, np.inf)
|
||
np.maximum.at(xmax, ys_p, xs_p)
|
||
np.minimum.at(xmin, ys_p, xs_p)
|
||
span_c = float(np.median(xmax[yv] - xmin[yv]))
|
||
if span_c < 10:
|
||
return None
|
||
line = p0 + p1 * ys_p
|
||
band = 0.25 * span_c # central band（椎體中央區）
|
||
narrow = 0.10 * span_c # 窄中線帶
|
||
y_min = int(ys_p.min())
|
||
# 「中線有骨」需 >= MIN_MID_VOX 個 voxel 才算數：棘突尖端在帶內有
|
||
# 多個 voxel，切除後的側方殘片在帶內通常只有 1-2 個游離 voxel
|
||
# （否則 0005 L5 的單 voxel 殘端會被當成尖端 → 假正常）。
|
||
bcnt = np.bincount(ys_p[np.abs(xs_p - line) <= band], minlength=ny)
|
||
idx = np.where(bcnt >= MIN_MID_VOX)[0]
|
||
y_mid = int(idx.min()) if idx.size else ny - 1
|
||
deficit = float(y_mid - y_min)
|
||
nn = np.abs(xs_p - line) <= narrow
|
||
rear3 = int(((ys_p <= y_min + 3.0) & nn).sum())
|
||
rear6 = int(((ys_p <= y_min + 6.0) & nn).sum())
|
||
top_off = float(abs(xs_p[ys_p == y_min].mean() - (p0 + p1 * y_min)))
|
||
return {'deficit': deficit, 'rear3': rear3, 'rear6': rear6,
|
||
'top_off': top_off, 'no_spinous': bool(deficit >= DEFICIT_MIN
|
||
and rear3 <= REAR3_MAX)}
|
||
|
||
|
||
# =====================================================================
|
||
# full：鏡稱面 + 完整棘突/椎體分割（精確，較慢）
|
||
# =====================================================================
|
||
def full_check(m):
|
||
sym = best_symmetry_plane(m)
|
||
diag = diagnose_spinous_process(m, sym)
|
||
sp_mask, sp_th, sp_info = segment_spinous_process(m, sym)
|
||
ep = best_upper_endplate_plane(m)
|
||
vb, vb_th, vb_info = segment_vertebral_body(m, sym, ep, sp_th, sp_info['mode'])
|
||
row = {'deficit': float(diag['deficit']), 'rear3': diag['rear3'],
|
||
'rear6': diag['rear6'], 'top_off': diag['top_off'],
|
||
'no_spinous': bool(diag['no_spinous']),
|
||
'sp_mode': sp_info['mode'],
|
||
'sp_pct': 100.0 * sp_info['n_sp'] / max(int(m.sum()), 1),
|
||
'vb_mode': vb_info['mode'],
|
||
'vb_pct': (100.0 * vb_info['n_vb'] / max(int(m.sum()), 1))
|
||
if vb is not None else None,
|
||
'vb_ap_th': vb_th}
|
||
return row
|
||
|
||
|
||
# =====================================================================
|
||
def load_binary(volume_dir, level):
|
||
# _binary.nii.gz 現為原解析度（未插值）；0.5mm 分析優 _binary_sdf（SDF 平滑遮罩）
|
||
# 未旋轉檔：新世代在 crop/ 子資料夾、舊世代在頂層
|
||
p = level_file_path(volume_dir, level, 'binary_sdf')
|
||
if not os.path.exists(p):
|
||
p = level_file_path(volume_dir, level, 'binary')
|
||
if not os.path.exists(p):
|
||
return None
|
||
m = sitk.GetArrayFromImage(sitk.ReadImage(p, sitk.sitkUInt8)) > 0
|
||
if int(m.sum()) < 1000 or min(m.shape) < 8:
|
||
return None
|
||
return m
|
||
|
||
|
||
def check_level(vid, level, use_full):
|
||
m = load_binary(os.path.join(standardized_dir, vid), level)
|
||
if m is None:
|
||
return {'vid': vid, 'level': level, 'note': 'missing/degenerate mask'}
|
||
t0 = time.time()
|
||
row = quick_screen(m) if not use_full else full_check(m)
|
||
row.update(vid=vid.rsplit('.', 1)[-1], level=level,
|
||
n_bone=int(m.sum()), sec=round(time.time() - t0, 2))
|
||
return row
|
||
|
||
|
||
def parse_args(argv):
|
||
use_full = None
|
||
pos = []
|
||
for a in argv:
|
||
if a == '--full':
|
||
use_full = True
|
||
elif a == '--quick':
|
||
use_full = False
|
||
else:
|
||
pos.append(a)
|
||
if len(pos) > 2:
|
||
sys.exit('Usage: python xfr_check_spinous.py [--full|--quick] [volume_id] [level]')
|
||
vid_arg = pos[0] if len(pos) >= 1 else None
|
||
level_arg = pos[1] if len(pos) >= 2 else None
|
||
if level_arg and level_arg.upper() not in LEVELS:
|
||
sys.exit(f'Invalid level: {level_arg} (choose from {"/".join(LEVELS)})')
|
||
if level_arg and not vid_arg:
|
||
sys.exit('level requires volume_id')
|
||
if use_full is None:
|
||
use_full = bool(vid_arg and level_arg) # 指定單 level → 精確模式
|
||
return vid_arg, (level_arg.upper() if level_arg else None), use_full
|
||
|
||
|
||
def main():
|
||
os.makedirs(LOG_DIR, exist_ok=True)
|
||
vid_arg, level_arg, use_full = parse_args(sys.argv[1:])
|
||
|
||
volumes = [d for d in sorted(os.listdir(standardized_dir))
|
||
if os.path.isdir(os.path.join(standardized_dir, d))]
|
||
if vid_arg is not None:
|
||
key = vid_arg.lower()
|
||
vols = [v for v in volumes
|
||
if v.lower() == key or v.rsplit('.', 1)[-1] == key]
|
||
if not vols:
|
||
sys.exit(f'Volume not found: {vid_arg}')
|
||
volumes = vols
|
||
levels = (level_arg,) if level_arg else LEVELS
|
||
tasks = [(v, lvl) for v in volumes for lvl in levels]
|
||
mode = ('FULL (mirror plane)' if use_full
|
||
else 'QUICK screen -> FULL-verify candidates (2-stage)')
|
||
print(f'Mode: {mode} | {len(volumes)} volumes x {len(levels)} levels = {len(tasks)} tasks',
|
||
flush=True)
|
||
|
||
t0 = time.time()
|
||
rows = []
|
||
candidates = [] # (row 索引, volume 目錄名)
|
||
for i, (vid, lvl) in enumerate(tasks, 1):
|
||
try:
|
||
row = check_level(vid, lvl, use_full)
|
||
except Exception as e:
|
||
row = {'vid': vid.rsplit('.', 1)[-1], 'level': lvl, 'note': f'error: {e}'}
|
||
rows.append(row)
|
||
if 'note' in row:
|
||
print(f"[{i}/{len(tasks)}] {row['vid']} {row['level']}: {row['note']}", flush=True)
|
||
elif use_full:
|
||
top_str = 'n/a' if row['top_off'] is None else f"{row['top_off']:.1f}"
|
||
print(f"[{i}/{len(tasks)}] {row['vid']} {row['level']}: "
|
||
f"deficit={row['deficit']:6.1f} rear3={row['rear3']} "
|
||
f"top_off={top_str} {'<<< NO-SPINUS' if row['no_spinous'] else 'ok'} "
|
||
f"sp={row['sp_mode']} {row['sp_pct']:.1f}% "
|
||
f"vb={row['vb_mode']} {row['vb_pct']:.1f}%", flush=True)
|
||
else:
|
||
mark = 'candidate' if row['no_spinous'] else 'ok'
|
||
print(f"[{i}/{len(tasks)}] {row['vid']} {row['level']}: "
|
||
f"deficit={row['deficit']:6.1f} rear3={row['rear3']:4d} "
|
||
f"rear6={row['rear6']:5d} top_off={row['top_off']:5.1f} {mark}",
|
||
flush=True)
|
||
if not use_full and row.get('no_spinous'):
|
||
candidates.append((len(rows) - 1, vid))
|
||
|
||
# ---- Stage 2（quick 模式）：候選用 full（鏡稱面）驗證，以 full 為準 ----
|
||
flagged = []
|
||
cleared = []
|
||
if use_full:
|
||
for r in rows:
|
||
if r.get('no_spinous'):
|
||
flagged.append(f"{r['vid']} {r['level']}")
|
||
elif candidates:
|
||
print(f'\n=== stage 2: {len(candidates)} quick candidate(s) -> '
|
||
f'full (mirror plane) verification ===', flush=True)
|
||
for j, (k, vid_dir) in enumerate(candidates, 1):
|
||
row = rows[k]
|
||
try:
|
||
m = load_binary(os.path.join(standardized_dir, vid_dir), row['level'])
|
||
frow = full_check(m) if m is not None else {}
|
||
except Exception as e:
|
||
frow = {'note': f'verify error: {e}'}
|
||
for kk in ('deficit', 'rear3', 'rear6', 'top_off', 'no_spinous',
|
||
'sp_mode', 'sp_pct', 'vb_mode', 'vb_pct', 'vb_ap_th'):
|
||
if kk in frow:
|
||
row[kk] = frow[kk]
|
||
if not frow:
|
||
flag_str = '-> UNVERIFIED (mask missing/full failed)'
|
||
cleared.append(f"{row['vid']} {row['level']} (UNVERIFIED)")
|
||
elif row.get('no_spinous'):
|
||
flag_str = '<<< NO-SPINUS (confirmed)'
|
||
flagged.append(f"{row['vid']} {row['level']}")
|
||
else:
|
||
flag_str = '-> normal (quick false alarm)'
|
||
cleared.append(f"{row['vid']} {row['level']} "
|
||
f"(full deficit={row.get('deficit', float('nan')):.1f})")
|
||
to = row.get('top_off')
|
||
print(f"[verify {j}/{len(candidates)}] {row['vid']} {row['level']}: "
|
||
f"deficit={row.get('deficit', float('nan')):6.1f} "
|
||
f"rear3={row.get('rear3', '?')} "
|
||
f"top_off={'n/a' if to is None else format(to, '5.1f')} "
|
||
f"sp={row.get('sp_mode', '?')} "
|
||
f"vb={row.get('vb_mode', '?')} {row.get('vb_pct') or 0:.1f}% {flag_str}",
|
||
flush=True)
|
||
|
||
# ---- CSV ----
|
||
ts = time.strftime('%Y%m%d_%H%M%S')
|
||
fields = ['vid', 'level', 'n_bone', 'deficit', 'rear3', 'rear6', 'top_off',
|
||
'no_spinous', 'sp_mode', 'sp_pct', 'vb_mode', 'vb_pct', 'vb_ap_th',
|
||
'sec', 'note']
|
||
csv_path = os.path.join(LOG_DIR, f'spinous_check_{ts}.csv')
|
||
with open(csv_path, 'w', newline='') as f:
|
||
wtr = csv.DictWriter(f, fieldnames=fields, extrasaction='ignore')
|
||
wtr.writeheader()
|
||
for r in rows:
|
||
r['vid'] = r.get('vid', '')
|
||
wtr.writerow(r)
|
||
|
||
dt = (time.time() - t0) / 60.0
|
||
print('\n' + '=' * 64)
|
||
print(f'Done in {dt:.1f} min | {len(rows)} levels checked (mode: {mode})')
|
||
if flagged:
|
||
print(f'\n>>> {len(flagged)} level(s) with NO SPINOUS PROCESS (prior laminectomy/resection):')
|
||
for f_ in flagged:
|
||
print(f' {f_}')
|
||
print(' (xfr_debug.py 執行這些 level 時會打 [NO-SP] 並自動改用放寬後側谷底的椎體切分)')
|
||
else:
|
||
print('\nNo missing spinous process detected.')
|
||
if cleared:
|
||
print(f' (quick 候選、經 full 驗證為正常: {", ".join(cleared)})')
|
||
if use_full:
|
||
quant = [f"{r['vid']} {r['level']}" for r in rows
|
||
if r.get('vb_mode') == 'quantile']
|
||
if quant:
|
||
print(f'\nWARNING: VBODY 退回 55 百分位切分（可能切進椎體）: {", ".join(quant)}')
|
||
print(f'CSV: {csv_path}')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main() |