| |
| """ |
| PRA Class I 分析腳本 |
| 解析 Luminex xPONENT CSV 匯出檔，計算 Normal / Ratio / Rxn / PRA% |
| 適用於 One Lambda LABScreen PRA Class I (Lot 21) |
| |
| 用法: |
| python PRA.py <csv檔案路徑> |
| python PRA.py <csv檔案路徑> --threshold X4 |
| python PRA.py <csv檔案路徑> --xlsx |
| """ |
|
|
| import csv |
| import sys |
| import os |
| import io |
| import math |
| from pathlib import Path |
| from collections import OrderedDict |
|
|
| |
| if sys.platform == 'win32': |
| sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') |
| sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') |
|
|
| |
| |
| |
|
|
| NC_BEAD = '001' |
| PC_BEAD = '002' |
| CONTROL_BEADS = [NC_BEAD, PC_BEAD] |
|
|
| |
| DEFAULT_CUTOFFS = OrderedDict([ |
| ('X2', 50), |
| ('X4', 250), |
| ('X6', 500), |
| ('X8', 1000), |
| ]) |
|
|
|
|
| def load_cutoffs(cutoff_file=None): |
| """ |
| 從 CutOff.txt 讀取閾值設定。 |
| 格式: 每行 "X2\t50" (tab 分隔) |
| 找不到檔案時使用預設值。 |
| """ |
| if cutoff_file is None: |
| cutoff_file = Path(__file__).parent / 'CutOff.txt' |
| else: |
| cutoff_file = Path(cutoff_file) |
|
|
| cutoffs = OrderedDict(DEFAULT_CUTOFFS) |
| if not cutoff_file.exists(): |
| return cutoffs |
|
|
| with open(cutoff_file, 'r', encoding='utf-8') as f: |
| for line in f: |
| line = line.strip() |
| if not line or line.startswith('#'): |
| continue |
| parts = line.split('\t') |
| if len(parts) >= 2 and parts[0].upper().startswith('X'): |
| key = parts[0].upper() |
| try: |
| cutoffs[key] = float(parts[1]) |
| except ValueError: |
| pass |
| return cutoffs |
|
|
|
|
| CUTOFFS = load_cutoffs() |
|
|
| |
| |
| |
| |
|
|
| BEAD_HLA_LOT21 = OrderedDict([ |
| ('003', {'sero': 'A2, B46, Bw6, Cw1', |
| 'allele': 'A*02:01, A*02:07, B*46:01, C*01:02'}), |
| ('004', {'sero': 'A1, A23, B49, B55, Bw4, Bw6, Cw9, Cw7', |
| 'allele': 'A*01:01, A*23:01, B*49:01, B*55:01, C*03:03, C*07:01'}), |
| ('005', {'sero': 'A11, A24, B27, B60, Bw4, Bw6, Cw10', |
| 'allele': 'A*11:02, A*24:02, B*27:06, B*40:01, C*03:04'}), |
| ('006', {'sero': 'A11, A24, B54, B59, Bw6, Bw4, Cw1', |
| 'allele': 'A*11:01, A*24:02, B*54:01, B*59:01, C*01:02'}), |
| ('007', {'sero': 'A11, A24, B60, B59, Bw6, Bw4, Cw1, Cw10', |
| 'allele': 'A*11:01, A*24:02, B*40:01, B*59:01, C*01:02, C*03:04'}), |
| ('008', {'sero': 'A1, A11, B47, B51, Bw4, Cw9, Cw6', |
| 'allele': 'A*01:01, A*11:01, B*47:01, B*51:01, C*03:03, C*06:02'}), |
| ('009', {'sero': 'A2, B44, B51, Bw4, Cw7, Cw14', |
| 'allele': 'A*02:01, B*44:02, B*51:01, C*07:674, C*14:02'}), |
| ('010', {'sero': 'A1, A2, B7, B13, Bw6, Bw4, Cw6, Cw7', |
| 'allele': 'A*01:01, A*02:01, B*07:02, B*13:02, C*06:02, C*07:02'}), |
| ('011', {'sero': 'A2, A11, B76, B51, Bw6, Bw4, Cw10, Cw14', |
| 'allele': 'A*02:06, A*11:01, B*15:12, B*51:01, C*03:04, C*14:02'}), |
| ('012', {'sero': 'A2, A24, B54, B67, Bw6, Cw1, Cw7', |
| 'allele': 'A*02:06, A*24:02, B*54:01, B*67:01, C*01:02, C*07:02'}), |
| ('013', {'sero': 'A11, A24, B62, B35, Bw6, Cw9, Cw4', |
| 'allele': 'A*11:01, A*24:02, B*15:01, B*35:01, C*03:03, C*04:01'}), |
| ('014', {'sero': 'A25, A36, B7, B44, Bw6, Bw4, Cw5, Cw7', |
| 'allele': 'A*25:01, A*36:01, B*07:02, B*44:02, C*05:01, C*07:02'}), |
| ('015', {'sero': 'A3, A32, B61, B47, Bw6, Bw4, Cw2, Cw6', |
| 'allele': 'A*03:01, A*32:01, B*40:02, B*47:01, C*02:02, C*06:02'}), |
| ('016', {'sero': 'A11, A30, B13, B75, Bw4, Bw6, Cw6, Cw8', |
| 'allele': 'A*11:01, A*30:01, B*13:02, B*15:02, C*06:02, C*08:01'}), |
| ('017', {'sero': 'A2, A29, B64, B57, Bw6, Bw4, Cw8, Cw18', |
| 'allele': 'A*02:01, A*29:02, B*14:01, B*57:02, C*08:02, C*18:02'}), |
| ('018', {'sero': 'A2, A80, B18, B27, Bw6, Bw4, Cw2', |
| 'allele': 'A*02:06, A*80:01, B*18:01, B*27:05, C*02:02'}), |
| ('019', {'sero': 'A2, A31, B39, B48, Bw6, Cw7, Cw8', |
| 'allele': 'A*02:06, A*31:01, B*39:01, B*48:01, C*07:02, C*08:22'}), |
| ('020', {'sero': 'A25, A30, B42, B44, Bw6, Bw4, Cw16, Cw17', |
| 'allele': 'A*25:01, A*30:01, B*42:01, B*44:03, C*16:01, C*17:01'}), |
| ('021', {'sero': 'A2, A66, B65, B63, Bw6, Bw4, Cw8, Cw16', |
| 'allele': 'A*02:01, A*66:01, B*14:02, B*15:16, C*08:02, C*16:01'}), |
| ('022', {'sero': 'A24, A29, B7, B65, Bw6, Cw7, Cw8', |
| 'allele': 'A*24:02, A*29:02, B*07:02, B*14:02, C*07:02, C*08:02'}), |
| ('023', {'sero': 'A2, A31, B62, B67, Bw6, Cw7', |
| 'allele': 'A*02:01, A*31:01, B*15:01, B*67:01, C*07:02'}), |
| ('024', {'sero': 'A2, B52, B73, Bw4, Bw6, Cw7, Cw15', |
| 'allele': 'A*02:01, B*52:01, B*73:01, C*07:01, C*15:05'}), |
| ('025', {'sero': 'A2, A11, B75, B38, Bw6, Bw4, Cw7, Cw8', |
| 'allele': 'A*02:03, A*11:01, B*15:02, B*38:02, C*07:02, C*08:147'}), |
| ('026', {'sero': 'A23, A32, B41, B44, Bw6, Bw4, Cw4, Cw17', |
| 'allele': 'A*23:01, A*32:01, B*41:02, B*44:03, C*04:01, C*17:03'}), |
| ('027', {'sero': 'A33, A68, B37, B58, Bw4, Cw2, Cw7', |
| 'allele': 'A*33:03, A*68:02, B*37:01, B*58:01, C*02:10, C*07:18'}), |
| ('028', {'sero': 'A24, A30, B37, B48, Bw4, Bw6, Cw6, Cw8', |
| 'allele': 'A*24:02, A*30:02, B*37:01, B*48:01, C*06:02, C*08:01'}), |
| ('029', {'sero': 'A25, A68, B65, B58, Bw6, Bw4, Cw7, Cw8', |
| 'allele': 'A*25:01, A*68:01, B*14:02, B*58:01, C*07:18, C*08:02'}), |
| ('030', {'sero': 'A26, A30, B18, B60, Bw6, Cw10, Cw5', |
| 'allele': 'A*26:01, A*30:02, B*18:01, B*40:01, C*03:04, C*05:01'}), |
| ('031', {'sero': 'A34, A66, B8, B35, Bw6, Cw7, Cw16', |
| 'allele': 'A*34:02, A*66:03, B*08:01, B*35:01, C*07:01, C*16:01'}), |
| ('032', {'sero': 'A29, A69, B39, B55, Bw6, Cw1, Cw7', |
| 'allele': 'A*29:02, A*69:01, B*39:06, B*55:01, C*01:02, C*07:02'}), |
| ('033', {'sero': 'A3, A23, B41, B49, Bw6, Bw4, Cw7, Cw8', |
| 'allele': 'A*03:01, A*23:01, B*41:01, B*49:01, C*07:01, C*08:02'}), |
| ('034', {'sero': 'A2, A68, B50, B53, Bw6, Bw4, Cw4, Cw6', |
| 'allele': 'A*02:02, A*68:01, B*50:01, B*53:01, C*04:01, C*06:02'}), |
| ('035', {'sero': 'A1, A29, B8, B45, Bw6, Cw6, Cw7', |
| 'allele': 'A*01:01, A*29:02, B*08:01, B*45:01, C*06:02, C*07:01'}), |
| ('036', {'sero': 'A3, A66, B7, B81, Bw6, Cw15, Cw18', |
| 'allele': 'A*03:01, A*66:01, B*07:02, B*81:01, C*15:05, C*18:01'}), |
| ('069', {'sero': 'A3, A26, B27, B57, Bw4, Cw6, Cw7', |
| 'allele': 'A*03:01, A*26:01, B*27:05, B*57:01, C*06:02, C*07:04'}), |
| ('070', {'sero': 'A30, A36, B71, B35, Bw6, Cw10, Cw4', |
| 'allele': 'A*30:01, A*36:01, B*15:10, B*35:01, C*03:04, C*04:01'}), |
| ('071', {'sero': 'A2, B61, B50, Bw6, Cw10, Cw6', |
| 'allele': 'A*02:01, B*40:02, B*50:01, C*03:04, C*06:02'}), |
| ('072', {'sero': 'A24, A32, B64, B18, Bw6, Cw7, Cw8', |
| 'allele': 'A*24:02, A*32:01, B*14:01, B*18:01, C*07:01, C*08:02'}), |
| ('073', {'sero': 'A2, B46, Bw6, Cw1, Cw8', |
| 'allele': 'A*02:06, A*02:07, B*46:01, C*01:02, C*08:01'}), |
| ('074', {'sero': 'A33, A36, B63, B53, Bw4, Cw4, Cw14', |
| 'allele': 'A*33:03, A*36:01, B*15:16, B*53:01, C*04:01, C*14:02'}), |
| ('075', {'sero': 'A33, A74, B63, B42, Bw4, Bw6, Cw14, Cw17', |
| 'allele': 'A*33:03, A*74:01, B*15:16, B*42:01, C*14:02, C*17:01'}), |
| ('076', {'sero': 'A34, B61, B56, Bw6, Cw1, Cw15', |
| 'allele': 'A*34:01, B*40:02, B*56:01, C*01:02, C*15:02'}), |
| ('077', {'sero': 'A1, A2, B57, B82, Bw4, Bw6, Cw10, Cw6', |
| 'allele': 'A*01:01, A*02:01, B*57:01, B*82:01, C*03:02, C*06:02'}), |
| ('078', {'sero': 'A11, A34, B62, B61, Bw6, Cw7, Cw15', |
| 'allele': 'A*11:01, A*34:01, B*15:35, B*40:02, C*07:02, C*15:02'}), |
| ('079', {'sero': 'A1, A29, B13, B81, Bw4, Bw6, Cw6, Cw18', |
| 'allele': 'A*01:01, A*29:02, B*13:02, B*81:01, C*06:02, C*18:01'}), |
| ('080', {'sero': 'A1, A30, B71, B42, Bw6, Cw10, Cw17', |
| 'allele': 'A*01:01, A*30:01, B*15:10, B*42:01, C*03:04, C*17:01'}), |
| ('081', {'sero': 'A26, A34, B75, B38, Bw6, Bw4, Cw4, Cw7', |
| 'allele': 'A*26:01, A*34:01, B*15:21, B*38:02, C*04:03, C*07:02'}), |
| ('082', {'sero': 'A1, A69, B8, B55, Bw6, Cw1, Cw7', |
| 'allele': 'A*01:01, A*69:01, B*08:01, B*55:01, C*01:02, C*07:01'}), |
| ('083', {'sero': 'A23, A80, B8, B18, Bw6, Cw2, Cw10', |
| 'allele': 'A*23:01, A*80:01, B*08:01, B*18:01, C*02:02, C*03:04'}), |
| ('084', {'sero': 'A3, A66, B72, B52, Bw6, Bw4, Cw2, Cw12', |
| 'allele': 'A*03:01, A*66:01, B*15:03, B*52:01, C*02:10, C*12:02'}), |
| ('085', {'sero': 'A3, B35, B56, Bw6, Cw1, Cw4', |
| 'allele': 'A*03:01, B*35:01, B*56:01, C*01:02, C*04:01'}), |
| ('086', {'sero': 'A11, A26, B64, B38, Bw6, Bw4, Cw8, Cw12', |
| 'allele': 'A*11:01, A*26:01, B*14:01, B*38:01, C*08:02, C*12:03'}), |
| ('087', {'sero': 'A33, A74, B72, B78, Bw6, Cw2, Cw16', |
| 'allele': 'A*33:01, A*74:01, B*15:03, B*78:01, C*02:10, C*16:01'}), |
| ('088', {'sero': 'A23, A24, B77, B45, Bw4, Bw6, Cw6, Cw8', |
| 'allele': 'A*23:01, A*24:02, B*15:13, B*45:01, C*06:02, C*08:01'}), |
| ('089', {'sero': 'A29, A68, B52, B78, Bw4, Bw6, Cw16', |
| 'allele': 'A*29:02, A*68:02, B*52:01, B*78:01, C*16:01'}), |
| ('095', {'sero': 'A68, A74, B72, B45, Bw6, Cw2, Cw16', |
| 'allele': 'A*68:02, A*74:01, B*15:03, B*45:01, C*02:10, C*16:01'}), |
| ]) |
|
|
| HLA_BEADS = list(BEAD_HLA_LOT21.keys()) |
|
|
|
|
| def _build_allele_to_sero_map(): |
| """從 bead 資料建立 allele→sero 對照表 (e.g. B*15:01→B62, C*03:02→Cw10)""" |
| mapping = {} |
| for bid, hla in BEAD_HLA_LOT21.items(): |
| sero_parts = [s.strip() for s in hla['sero'].split(',')] |
| allele_parts = [a.strip() for a in hla['allele'].split(',')] |
|
|
| |
| sero_a = [s for s in sero_parts if s and s.startswith('A')] |
| sero_b = [s for s in sero_parts if s and s.startswith('B') and s not in ('Bw4', 'Bw6')] |
| sero_c = [s for s in sero_parts if s and (s.startswith('Cw') or s.startswith('C'))] |
|
|
| |
| allele_a = [a for a in allele_parts if a and a.startswith('A') and a != '-'] |
| allele_b = [a for a in allele_parts if a and a.startswith('B') and a != '-'] |
| allele_c = [a for a in allele_parts if a and a.startswith('C') and a != '-'] |
|
|
| for i, ag in enumerate(allele_a): |
| if i < len(sero_a): |
| mapping[ag] = sero_a[i] |
| for i, ag in enumerate(allele_b): |
| if i < len(sero_b): |
| mapping[ag] = sero_b[i] |
| for i, ag in enumerate(allele_c): |
| if i < len(sero_c): |
| mapping[ag] = sero_c[i] |
| return mapping |
|
|
|
|
| ALLELE_TO_SERO = _build_allele_to_sero_map() |
|
|
|
|
| |
| |
| |
|
|
| def parse_xponent_csv(filepath): |
| """ |
| 解析 xPONENT 匯出的多區塊 CSV 檔案。 |
| 回傳 (metadata_dict, data_blocks_dict) |
| data_blocks 的 key 為 DataType 名稱 (如 'Median', 'Net MFI', 'Count' 等)， |
| value 為 dict: {sample_name: {bead_id: float_value}} |
| """ |
| with open(filepath, 'r', encoding='utf-8-sig') as f: |
| raw = f.read() |
| |
| lines = raw.replace('\r\n', '\n').replace('\r', '\n').split('\n') |
|
|
| metadata = {} |
| data_blocks = {} |
| bead_ids = [] |
|
|
| i = 0 |
| |
| while i < len(lines): |
| line = lines[i].strip() |
| if line.startswith('"Results"'): |
| i += 1 |
| break |
| |
| parts = _parse_csv_line(line) |
| if len(parts) >= 2: |
| metadata[parts[0]] = parts[1:] |
| i += 1 |
|
|
| |
| current_datatype = None |
| current_header = None |
|
|
| while i < len(lines): |
| line = lines[i].strip() |
| i += 1 |
|
|
| if not line: |
| continue |
|
|
| parts = _parse_csv_line(line) |
| if not parts: |
| continue |
|
|
| |
| if parts[0] == 'DataType:': |
| current_datatype = parts[1] if len(parts) > 1 else None |
| data_blocks[current_datatype] = {} |
| current_header = None |
| continue |
|
|
| |
| if current_datatype and current_header is None: |
| if parts[0] in ('Location', 'Sample'): |
| current_header = parts |
| if not bead_ids: |
| |
| if parts[0] == 'Location': |
| bead_ids = parts[2:] |
| else: |
| bead_ids = parts[1:] |
| |
| if bead_ids and bead_ids[-1] == 'Total Events': |
| bead_ids = bead_ids[:-1] |
| continue |
|
|
| |
| if current_datatype and current_header: |
| if current_header[0] == 'Location': |
| if len(parts) < 3: |
| continue |
| sample_name = parts[1] |
| values = parts[2:] |
| else: |
| if len(parts) < 2: |
| continue |
| sample_name = parts[0] |
| values = parts[1:] |
|
|
| |
| if len(values) > len(bead_ids): |
| values = values[:len(bead_ids)] |
|
|
| sample_data = {} |
| for j, bid in enumerate(bead_ids): |
| if j < len(values): |
| try: |
| sample_data[bid] = float(values[j]) |
| except (ValueError, TypeError): |
| sample_data[bid] = 0.0 |
| data_blocks[current_datatype][sample_name] = sample_data |
|
|
| return metadata, bead_ids, data_blocks |
|
|
|
|
| def _parse_csv_line(line): |
| """解析 CSV 行，處理引號欄位""" |
| if not line: |
| return [] |
| try: |
| reader = csv.reader([line]) |
| return next(reader) |
| except Exception: |
| return [] |
|
|
|
|
| |
| |
| |
|
|
| def get_rxn(normal): |
| """根據 Normal 值判定 Rxn 等級""" |
| if normal >= CUTOFFS['X8']: |
| return 8 |
| elif normal >= CUTOFFS['X6']: |
| return 6 |
| elif normal >= CUTOFFS['X4']: |
| return 4 |
| elif normal >= CUTOFFS['X2']: |
| return 2 |
| else: |
| return 1 |
|
|
|
|
| def analyze_sample(sample_medians, nc_medians, bead_ids): |
| """ |
| 分析單一病人樣本。 |
| 回傳 OrderedDict: bead_id -> {raw, ns_raw, normal, ratio, rxn, count} |
| """ |
| sample_nc_bead = sample_medians.get(NC_BEAD, 0) |
| nc_nc_bead = nc_medians.get(NC_BEAD, 0) |
|
|
| results = OrderedDict() |
| for bid in bead_ids: |
| if bid in CONTROL_BEADS: |
| continue |
| if bid not in BEAD_HLA_LOT21: |
| continue |
|
|
| raw = sample_medians.get(bid, 0) |
| ns_raw = nc_medians.get(bid, 0) |
|
|
| |
| normal = max(0.0, raw - sample_nc_bead - ns_raw + nc_nc_bead) |
|
|
| |
| denom = ns_raw * sample_nc_bead |
| if denom > 0: |
| ratio = (raw * nc_nc_bead) / denom |
| else: |
| ratio = 0.0 |
|
|
| rxn = get_rxn(normal) |
|
|
| results[bid] = { |
| 'raw': raw, |
| 'ns_raw': ns_raw, |
| 'sample_nc': sample_nc_bead, |
| 'nsnc': nc_nc_bead, |
| 'normal': round(normal, 2), |
| 'ratio': round(ratio, 2), |
| 'rxn': rxn, |
| } |
|
|
| return results |
|
|
|
|
| def calculate_pra(bead_results, threshold='X6'): |
| """ |
| 計算指定閾值下的 PRA%。 |
| 回傳 (pra_percent, positive_count, total_count) |
| """ |
| threshold_rxn = int(threshold[1:]) |
| total = len(bead_results) |
| positive = sum(1 for r in bead_results.values() if r['rxn'] >= threshold_rxn) |
| pra = round(positive / total * 100) if total > 0 else 0 |
| return pra, positive, total |
|
|
|
|
| def get_positive_specificities(bead_results, threshold='X6'): |
| """ |
| 取得在指定閾值下陽性 bead 的 HLA 特異性 (Sero 及 Allele)。 |
| 回傳 (sero_set, allele_set, positive_beads_list) |
| """ |
| threshold_rxn = int(threshold[1:]) |
| sero_set = set() |
| allele_set = set() |
| positive_beads = [] |
|
|
| for bid, r in bead_results.items(): |
| if r['rxn'] >= threshold_rxn: |
| positive_beads.append(bid) |
| hla = BEAD_HLA_LOT21.get(bid, {}) |
| if hla.get('sero'): |
| for ag in hla['sero'].split(','): |
| ag = ag.strip() |
| if ag and ag not in ('Bw4', 'Bw6', ''): |
| sero_set.add(ag) |
| if hla.get('allele'): |
| for ag in hla['allele'].split(','): |
| ag = ag.strip() |
| if ag and ag != '-' and ag != '': |
| allele_set.add(ag) |
|
|
| return sero_set, allele_set, positive_beads |
|
|
|
|
| def _parse_allele_list(allele_str): |
| """解析 allele 字串，回傳有效 allele 的集合。 |
| 清除命名後綴: /276N (null allele), =DPB1*105:01 (別名)""" |
| result = set() |
| if not allele_str: |
| return result |
| for ag in allele_str.split(','): |
| ag = ag.strip() |
| if ag and ag != '-': |
| ag = ag.split('/')[0] |
| ag = ag.split('=')[0] |
| result.add(ag) |
| return result |
|
|
|
|
| def get_confident_alleles(bead_results): |
| """ |
| 篩選高信度 Allele：只出現在 X6/X8 (Rxn≥6) 的 bead 上， |
| 所有 bead 都必須 Rxn≥6，任何 Rxn<6 (X4/X2/Rxn1) 都排除。 |
| 回傳 (confident_alleles, x6x8_alleles, not_x6x8_alleles, detail_list) |
| detail_list: [(allele, source_beads, max_normal)] 按 max_normal 降序 |
| """ |
| x6x8_alleles = {} |
| not_x6x8_alleles = {} |
|
|
| for bid, r in bead_results.items(): |
| hla = BEAD_HLA_LOT21.get(bid, {}) |
| alleles = _parse_allele_list(hla.get('allele', '')) |
|
|
| if r['rxn'] >= 6: |
| for ag in alleles: |
| x6x8_alleles.setdefault(ag, []).append((bid, r['normal'])) |
| else: |
| for ag in alleles: |
| not_x6x8_alleles.setdefault(ag, []).append((bid, r['normal'])) |
|
|
| |
| confident = set(x6x8_alleles.keys()) - set(not_x6x8_alleles.keys()) |
|
|
| |
| detail = [] |
| for ag in confident: |
| beads = x6x8_alleles[ag] |
| max_normal = max(n for _, n in beads) |
| bead_list = [b for b, _ in beads] |
| detail.append((ag, bead_list, max_normal)) |
| detail.sort(key=lambda x: -x[2]) |
|
|
| |
| excluded = set(x6x8_alleles.keys()) & set(not_x6x8_alleles.keys()) |
|
|
| return confident, excluded, x6x8_alleles, not_x6x8_alleles, detail |
|
|
|
|
| def generate_specificity_string(bead_results): |
| """ |
| 用 confident allele 分析生成 Specificity 字串（模擬 HLA Fusion 報告格式）。 |
| 格式: Sero名 或 Sero名(Allele名) ，空白分隔 |
| 例如: A11(A*11:02) A23 A24 B7 B62(B*15:01) Cw1 |
| """ |
| confident, _, _, _, _ = get_confident_alleles(bead_results) |
| if not confident: |
| return '(-)' |
|
|
| |
| sero_groups = {} |
| all_alleles_of_sero = {} |
|
|
| |
| for bid, hla in BEAD_HLA_LOT21.items(): |
| alleles = _parse_allele_list(hla.get('allele', '')) |
| for ag in alleles: |
| sero = ALLELE_TO_SERO.get(ag) |
| if sero: |
| all_alleles_of_sero.setdefault(sero, set()).add(ag) |
|
|
| |
| for ag in confident: |
| sero = ALLELE_TO_SERO.get(ag) |
| if sero: |
| sero_groups.setdefault(sero, set()).add(ag) |
|
|
| |
| parts = [] |
| for sero in _sort_hla(sero_groups.keys()): |
| conf_alleles = sero_groups[sero] |
| all_alleles = all_alleles_of_sero.get(sero, set()) |
|
|
| if conf_alleles >= all_alleles: |
| |
| parts.append(sero) |
| else: |
| |
| sorted_alleles = sorted(conf_alleles) |
| parts.append(f'{sero}({" ".join(sorted_alleles)})') |
|
|
| return ' '.join(parts) |
|
|
|
|
| def parse_docx_report(docx_path): |
| """ |
| 解析確認報告 docx，回傳 list of dict: |
| [{sample_id, overall, pra_percent, specificities_str, specificities_set}, ...] |
| """ |
| from docx import Document |
| doc = Document(docx_path) |
|
|
| results = [] |
| current = None |
| expect_spec = False |
|
|
| for para in doc.paragraphs: |
| text = para.text.strip() |
| if not text: |
| continue |
|
|
| if text.startswith('EL'): |
| |
| sample_id = text.split('_')[0] + '_' + text.split('_')[-1].split()[0] |
| current = {'sample_id_prefix': text.split('_')[0], |
| 'full_line': text, |
| 'overall': '', 'pra_percent': 0, |
| 'spec_str': '', 'spec_sero': set()} |
| results.append(current) |
| expect_spec = False |
|
|
| elif current and text.startswith('Overall:'): |
| current['overall'] = text.split(':')[1].strip() |
|
|
| elif current and text.startswith('%SA') or (current and text.startswith('%PRA')): |
| if ':' in text: |
| val = text.split(':')[1].strip() |
| try: |
| current['pra_percent'] = int(val) |
| except ValueError: |
| current['pra_percent'] = 0 |
|
|
| elif current and text == 'Specificity:': |
| expect_spec = True |
|
|
| elif current and expect_spec: |
| current['spec_str'] = text |
| expect_spec = False |
| |
| if text != '(-)': |
| import re |
| current['spec_sero'] = set(re.findall( |
| r'(?<!\*)(A\d+|B\d+|Cw\d+)', text)) |
|
|
| elif current and text == 'COMMENT:': |
| current = None |
|
|
| return results |
|
|
|
|
| def compare_with_docx(docx_path, all_results, threshold='X6'): |
| """比對程式分析結果與確認報告""" |
| docx_results = parse_docx_report(docx_path) |
|
|
| print(f'\n{"="*72}') |
| print(f' 比對確認報告: {Path(docx_path).name}') |
| print(f'{"="*72}') |
|
|
| for doc_r in docx_results: |
| prefix = doc_r['sample_id_prefix'] |
| |
| matched_name = None |
| matched_br = None |
| for sname, br in all_results.items(): |
| if sname.startswith(prefix): |
| matched_name = sname |
| matched_br = br |
| break |
|
|
| if not matched_br: |
| print(f'\n [!] 找不到 {prefix} 對應的分析結果') |
| continue |
|
|
| |
| pra_computed, _, _ = calculate_pra(matched_br, threshold) |
| overall_computed = 'Positive' if pra_computed > 0 else 'Negative' |
|
|
| pra_match = '✓' if pra_computed == doc_r['pra_percent'] else '✗' |
| overall_match = '✓' if overall_computed == doc_r['overall'] else '✗' |
|
|
| print(f'\n ── {matched_name} ──') |
| print(f' {"":20} {"報告":>12} {"程式":>12} 比對') |
| print(f' {"Overall":<20} {doc_r["overall"]:>12} {overall_computed:>12} {overall_match}') |
| print(f' {"PRA%":<20} {doc_r["pra_percent"]:>12} {pra_computed:>12} {pra_match}') |
|
|
| |
| computed_spec_str = generate_specificity_string(matched_br) |
|
|
| |
| import re |
| computed_sero = set() |
| if computed_spec_str != '(-)': |
| computed_sero = set(re.findall( |
| r'(?<!\*)(A\d+|B\d+|Cw\d+)', computed_spec_str)) |
|
|
| report_sero = doc_r['spec_sero'] |
|
|
| in_both = report_sero & computed_sero |
| only_report = report_sero - computed_sero |
| only_program = computed_sero - report_sero |
|
|
| print(f'\n 報告 Specificity:') |
| print(f' {doc_r["spec_str"]}') |
| print(f' 程式 Specificity:') |
| print(f' {computed_spec_str}') |
|
|
| if report_sero or computed_sero: |
| print(f'\n 吻合 ({len(in_both)}): {" ".join(_sort_hla(in_both)) if in_both else "—"}') |
| if only_report: |
| print(f' 僅報告有 ({len(only_report)}): {" ".join(_sort_hla(only_report))}') |
| if only_program: |
| print(f' 僅程式有 ({len(only_program)}): {" ".join(_sort_hla(only_program))}') |
|
|
| total_union = len(report_sero | computed_sero) |
| accuracy = len(in_both) / total_union * 100 if total_union > 0 else 100 |
| print(f' 吻合率: {accuracy:.0f}% ({len(in_both)}/{total_union})') |
|
|
|
|
| def _sort_hla(antigens): |
| """將 HLA 抗原按 locus (A, B, C) 排序""" |
| def sort_key(ag): |
| ag_upper = ag.upper() |
| if ag_upper.startswith('A'): |
| return (0, ag) |
| elif ag_upper.startswith('B'): |
| return (1, ag) |
| elif ag_upper.startswith('C'): |
| return (2, ag) |
| else: |
| return (3, ag) |
| return sorted(antigens, key=sort_key) |
|
|
|
|
| |
| |
| |
|
|
| def find_nc_sample(sample_names): |
| """從樣本名稱列表中找出 NC (Negative Control) 樣本""" |
| for name in sample_names: |
| if name.upper().startswith('NC') or name.upper().startswith('NC_'): |
| return name |
| return None |
|
|
|
|
| def print_results(sample_name, bead_results, counts=None, threshold='X6'): |
| """印出單一樣本的分析結果""" |
| |
| print(f'\n{"="*72}') |
| print(f' 樣本: {sample_name}') |
| print(f'{"="*72}') |
|
|
| sample_nc_val = next(iter(bead_results.values()))['sample_nc'] if bead_results else 0 |
| nsnc_val = next(iter(bead_results.values()))['nsnc'] if bead_results else 0 |
| print(f' Sample NC (bead 001): {sample_nc_val:.1f} ' |
| f'NSNC (bead 001): {nsnc_val:.1f}') |
|
|
| print(f'\n {"Threshold":<12} {"PRA%":>6} {"Positive":>10} / {"Total":<6}') |
| print(f' {"-"*40}') |
| for t in ['X2', 'X4', 'X6', 'X8']: |
| pra, pos, tot = calculate_pra(bead_results, t) |
| marker = ' <--' if t == threshold else '' |
| print(f' {t:<12} {pra:>5}% {pos:>10} / {tot:<6}{marker}') |
|
|
| |
| print(f'\n {"Bead":>4} {"Rxn":>4} {"Raw":>10} {"NS Raw":>10} ' |
| f'{"Normal":>10} {"Ratio":>8} {"Count":>6} Specificity') |
| print(f' {"-"*90}') |
|
|
| threshold_rxn = int(threshold[1:]) |
| for bid, r in bead_results.items(): |
| count_str = '' |
| if counts and bid in counts: |
| count_str = f'{counts[bid]:.0f}' |
|
|
| marker = '+' if r['rxn'] >= threshold_rxn else ' ' |
| sero = BEAD_HLA_LOT21.get(bid, {}).get('sero', '') |
| print(f'{marker} {bid:>4} {r["rxn"]:>4} {r["raw"]:>10.1f} {r["ns_raw"]:>10.1f} ' |
| f'{r["normal"]:>10.2f} {r["ratio"]:>8.2f} {count_str:>6} {sero}') |
|
|
| |
| sero_set, allele_set, pos_beads = get_positive_specificities(bead_results, threshold) |
| pra, pos_count, total = calculate_pra(bead_results, threshold) |
|
|
| print(f'\n 判定 (Threshold {threshold}): ', end='') |
| if pra > 0: |
| print(f'Positive | PRA% = {pra}%') |
| else: |
| print(f'Negative | PRA% = 0%') |
|
|
| if pos_beads: |
| print(f' 陽性 Beads: {", ".join(pos_beads)}') |
|
|
| |
| confident, excluded, x6x8_ag, x2x4_ag, detail = get_confident_alleles(bead_results) |
|
|
| if detail: |
| print(f'\n ── Confident Alleles (只在 X6/X8，不在 X2/X4) ──') |
| sorted_confident = _sort_hla(confident) |
| print(f' {", ".join(sorted_confident)}') |
| print(f'\n {"Allele":<16} {"Max Normal":>10} 來源 Beads') |
| print(f' {"-"*50}') |
| for ag, beads, max_n in detail: |
| print(f' {ag:<16} {max_n:>10.1f} {", ".join(beads)}') |
|
|
| if excluded: |
| sorted_excluded = _sort_hla(excluded) |
| print(f'\n ── 排除 (同時出現在 X6/X8 和 X2/X4) ──') |
| print(f' {", ".join(sorted_excluded)}') |
|
|
| print(f'\n ※ 正式報告請以 HLA Fusion 軟體 pattern analysis 為準。') |
|
|
|
|
| def export_xlsx(filepath, all_results, nc_name, metadata, bead_ids, data_blocks, threshold='X6'): |
| """匯出分析結果至 Excel 檔案""" |
| try: |
| from openpyxl import Workbook |
| from openpyxl.styles import Font, Alignment, PatternFill, Border, Side |
| except ImportError: |
| print('\n [警告] 需要 openpyxl 套件才能匯出 Excel。請執行: pip install openpyxl') |
| return None |
|
|
| wb = Workbook() |
|
|
| |
| header_font = Font(bold=True) |
| positive_fill = PatternFill(start_color='FFCCCC', end_color='FFCCCC', fill_type='solid') |
| header_fill = PatternFill(start_color='CCE5FF', end_color='CCE5FF', fill_type='solid') |
| thin_border = Border( |
| left=Side(style='thin'), right=Side(style='thin'), |
| top=Side(style='thin'), bottom=Side(style='thin') |
| ) |
|
|
| |
| count_data = data_blocks.get('Count', {}) |
|
|
| |
| ws_sum = wb.active |
| ws_sum.title = 'Summary' |
| ws_sum.append(['PRA Class I Analysis Summary']) |
| ws_sum['A1'].font = Font(bold=True, size=14) |
|
|
| batch = metadata.get('Batch', [''])[0] if 'Batch' in metadata else '' |
| date_val = metadata.get('Date', [''])[0] if 'Date' in metadata else '' |
| ws_sum.append([]) |
| ws_sum.append(['Batch:', batch]) |
| ws_sum.append(['Date:', date_val]) |
| ws_sum.append(['NC Sample:', nc_name]) |
| ws_sum.append(['Threshold:', threshold]) |
| ws_sum.append([]) |
|
|
| |
| sum_headers = ['Sample', 'PRA% (X2)', 'PRA% (X4)', 'PRA% (X6)', 'PRA% (X8)', |
| 'Result', 'Sample NC', 'PC Signal'] |
| ws_sum.append(sum_headers) |
| for c_idx, _ in enumerate(sum_headers, 1): |
| cell = ws_sum.cell(row=ws_sum.max_row, column=c_idx) |
| cell.font = header_font |
| cell.fill = header_fill |
| cell.border = thin_border |
|
|
| median_data = data_blocks.get('Median', {}) |
| for sample_name, bead_results in all_results.items(): |
| pra2, _, _ = calculate_pra(bead_results, 'X2') |
| pra4, _, _ = calculate_pra(bead_results, 'X4') |
| pra6, _, _ = calculate_pra(bead_results, 'X6') |
| pra8, _, _ = calculate_pra(bead_results, 'X8') |
| pra_main, _, _ = calculate_pra(bead_results, threshold) |
| result = 'Positive' if pra_main > 0 else 'Negative' |
| sample_nc_val = median_data.get(sample_name, {}).get(NC_BEAD, 0) |
| pc_val = median_data.get(sample_name, {}).get(PC_BEAD, 0) |
| row = [sample_name, pra2, pra4, pra6, pra8, result, |
| round(sample_nc_val, 1), round(pc_val, 1)] |
| ws_sum.append(row) |
| r = ws_sum.max_row |
| for c_idx in range(1, len(row) + 1): |
| cell = ws_sum.cell(row=r, column=c_idx) |
| cell.border = thin_border |
| if result == 'Positive': |
| cell.fill = positive_fill |
|
|
| |
| ws_sum.column_dimensions['A'].width = 25 |
| for col_letter in ['B', 'C', 'D', 'E']: |
| ws_sum.column_dimensions[col_letter].width = 12 |
| ws_sum.column_dimensions['F'].width = 12 |
| ws_sum.column_dimensions['G'].width = 12 |
| ws_sum.column_dimensions['H'].width = 12 |
|
|
| |
| threshold_rxn = int(threshold[1:]) |
| for sample_name, bead_results in all_results.items(): |
| |
| sheet_name = sample_name[:31] |
| ws = wb.create_sheet(title=sheet_name) |
|
|
| ws.append([sample_name]) |
| ws['A1'].font = Font(bold=True, size=12) |
| ws.append([]) |
|
|
| pra_main, pos_count, total = calculate_pra(bead_results, threshold) |
| result = 'Positive' if pra_main > 0 else 'Negative' |
| ws.append(['Result:', result, '', 'PRA%:', pra_main, '', 'Threshold:', threshold]) |
| ws.append([]) |
|
|
| |
| headers = ['BeadID', 'Rxn', 'Raw', 'NS Raw', 'Sample NC', 'NSNC', |
| 'Normal', 'Ratio', 'Count', 'Specificity', 'Allele Specificity'] |
| ws.append(headers) |
| for c_idx, h in enumerate(headers, 1): |
| cell = ws.cell(row=ws.max_row, column=c_idx) |
| cell.font = header_font |
| cell.fill = header_fill |
| cell.border = thin_border |
|
|
| |
| sample_counts = count_data.get(sample_name, {}) |
| for bid, r in bead_results.items(): |
| hla = BEAD_HLA_LOT21.get(bid, {}) |
| count_val = sample_counts.get(bid, '') |
| if count_val: |
| count_val = int(count_val) |
| row = [bid, r['rxn'], round(r['raw'], 2), round(r['ns_raw'], 2), |
| round(r['sample_nc'], 2), round(r['nsnc'], 2), |
| r['normal'], r['ratio'], count_val, |
| hla.get('sero', ''), hla.get('allele', '')] |
| ws.append(row) |
| row_num = ws.max_row |
| for c_idx in range(1, len(row) + 1): |
| cell = ws.cell(row=row_num, column=c_idx) |
| cell.border = thin_border |
| if r['rxn'] >= threshold_rxn: |
| for c_idx in range(1, len(row) + 1): |
| ws.cell(row=row_num, column=c_idx).fill = positive_fill |
|
|
| |
| ws.append([]) |
| sample_med = median_data.get(sample_name, {}) |
| nc_med = median_data.get(nc_name, {}) |
| nc_bead_val = sample_med.get(NC_BEAD, 0) |
| pc_bead_val = sample_med.get(PC_BEAD, 0) |
| nc_nc_val = nc_med.get(NC_BEAD, 0) |
| pc_nc_val = nc_med.get(PC_BEAD, 0) |
| ws.append([NC_BEAD, 'NC', round(nc_bead_val, 2), round(nc_nc_val, 2)]) |
| ws.append([PC_BEAD, 'PC', round(pc_bead_val, 2), round(pc_nc_val, 2)]) |
|
|
| |
| sero_set, allele_set, pos_beads = get_positive_specificities(bead_results, threshold) |
| ws.append([]) |
| if sero_set: |
| ws.append(['Candidate Antibodies (Sero):', ', '.join(_sort_hla(sero_set))]) |
| if allele_set: |
| ws.append(['Candidate Antibodies (Allele):', ', '.join(_sort_hla(allele_set))]) |
| ws.append(['Positive Beads:', ', '.join(pos_beads)]) |
|
|
| |
| col_widths = {'A': 8, 'B': 6, 'C': 12, 'D': 12, 'E': 12, 'F': 10, |
| 'G': 12, 'H': 10, 'I': 8, 'J': 35, 'K': 45} |
| for col_letter, w in col_widths.items(): |
| ws.column_dimensions[col_letter].width = w |
|
|
| wb.save(filepath) |
| return filepath |
|
|
|
|
| |
| |
| |
|
|
| def main(): |
| import argparse |
| parser = argparse.ArgumentParser( |
| description='PRA Class I 分析 — 解析 xPONENT CSV 並計算 PRA%') |
| parser.add_argument('csv_path', help='xPONENT CSV 檔案路徑 (PRA1)') |
| parser.add_argument('--threshold', default='X6', choices=['X2', 'X4', 'X6', 'X8'], |
| help='判定閾值 (預設: X6)') |
| parser.add_argument('--xlsx', action='store_true', |
| help='匯出 Excel 結果檔') |
| parser.add_argument('--nc', default=None, |
| help='指定 NC 樣本名稱 (預設: 自動偵測)') |
| parser.add_argument('--compare', default=None, |
| help='與確認報告 docx 比對 (指定 docx 路徑)') |
| args = parser.parse_args() |
|
|
| csv_path = Path(args.csv_path) |
| if not csv_path.exists(): |
| print(f'[錯誤] 找不到檔案: {csv_path}') |
| sys.exit(1) |
|
|
| |
| print(f'解析 CSV: {csv_path.name}') |
| metadata, bead_ids, data_blocks = parse_xponent_csv(str(csv_path)) |
|
|
| |
| median_data = data_blocks.get('Median') |
| if not median_data: |
| |
| median_data = data_blocks.get('Avg Net MFI') |
| if not median_data: |
| print('[錯誤] CSV 中找不到 Median 或 Avg Net MFI 資料區塊') |
| sys.exit(1) |
|
|
| |
| sample_names = list(median_data.keys()) |
| nc_name = args.nc or find_nc_sample(sample_names) |
| if not nc_name or nc_name not in median_data: |
| print(f'[錯誤] 無法找到 NC 樣本。可用樣本: {sample_names}') |
| print(' 請使用 --nc 參數指定 NC 樣本名稱') |
| sys.exit(1) |
|
|
| nc_medians = median_data[nc_name] |
|
|
| |
| count_data = data_blocks.get('Count', {}) |
|
|
| |
| batch = metadata.get('Batch', [''])[0] if 'Batch' in metadata else '' |
| protocol = metadata.get('ProtocolName', [''])[0] if 'ProtocolName' in metadata else '' |
| date_val = metadata.get('Date', [''])[0] if 'Date' in metadata else '' |
| print(f'Batch: {batch}') |
| print(f'Protocol: {protocol}') |
| print(f'Date: {date_val}') |
| print(f'NC 樣本: {nc_name}') |
| print(f'樣本數: {len(sample_names)} (含 NC)') |
|
|
| pc_nc = nc_medians.get(PC_BEAD, 0) |
| nc_nc = nc_medians.get(NC_BEAD, 0) |
| if nc_nc > 0: |
| print(f'NC QC — PC: {pc_nc:.0f}, NC: {nc_nc:.0f}, PC/NC: {pc_nc/nc_nc:.1f}') |
|
|
| |
| patient_samples = [s for s in sample_names if s != nc_name] |
| all_results = OrderedDict() |
|
|
| for sample_name in patient_samples: |
| sample_medians = median_data[sample_name] |
| bead_results = analyze_sample(sample_medians, nc_medians, bead_ids) |
| all_results[sample_name] = bead_results |
|
|
| counts = count_data.get(sample_name, {}) |
| print_results(sample_name, bead_results, counts, args.threshold) |
|
|
| |
| if args.compare: |
| compare_path = Path(args.compare) |
| if compare_path.exists(): |
| compare_with_docx(str(compare_path), all_results, args.threshold) |
| else: |
| print(f'\n[警告] 找不到比對檔案: {compare_path}') |
|
|
| |
| if args.xlsx: |
| xlsx_name = csv_path.stem + '_PRA1_results.xlsx' |
| xlsx_path = csv_path.parent / xlsx_name |
| result_path = export_xlsx( |
| str(xlsx_path), all_results, nc_name, metadata, |
| bead_ids, data_blocks, args.threshold) |
| if result_path: |
| print(f'\n已匯出 Excel: {xlsx_name}') |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
