dsa.py

# -*- coding: utf-8 -*-
"""
DSA (Donor-Specific Antibody) parser for LABScreen Single Antigen xls.
Cutoffs:
  Strong  : Normal MFI > 4000
  Weak    : 1000 <= Normal MFI <= 3999
  Negative: Normal MFI < 1000

Specificity rendering reuses PRA's _build_allele_to_sero / generate_specificity
so colours and Sero(Allele) format stay consistent across pages.
"""

import re
import warnings
from collections import OrderedDict
from pathlib import Path

warnings.filterwarnings('ignore')

STRONG_CUT = 4000
WEAK_CUT = 1000


def _str(v):
    if v is None:
        return ''
    s = str(v)
    return '' if s.lower() in ('nan',) else s


def _flt(v):
    try:
        return float(v)
    except (TypeError, ValueError):
        return 0.0


def parse_xls_file(raw_bytes):
    """
    Parse a HLA Fusion Single Antigen xls (LS1A* / LS2A*).
    Returns dict with same shape as PRA's parse_xls_file:
      name, overall, pra (=%SA), pra_all, beads, confident_alleles,
      specificity, sero_mfi, comment, qc, _pra_class, _date, _batch, _nc_raw, _pc_raw
    `_pra_class` here uses 'DSA1' / 'DSA2' so callers can dispatch.
    """
    import xlrd
    wb = xlrd.open_workbook(file_contents=raw_bytes)
    sh = wb.sheet_by_index(0)

    sample_name = _str(sh.cell_value(0, 0)).strip()
    if not sample_name and sh.ncols > 1:
        sample_name = _str(sh.cell_value(0, 1)).strip()

    catalog = ''
    session = ''
    date_val = ''
    max_hdr = min(sh.nrows, 15)
    for r in range(max_hdr):
        if catalog and session and date_val:
            break
        row = [_str(sh.cell_value(r, c)).strip() for c in range(sh.ncols)]
        for i, v in enumerate(row):
            if not catalog and v in ('CATALOG :', 'CATALOG:'):
                for j in range(i + 1, len(row)):
                    if row[j]:
                        catalog = row[j]; break
            if not session and v in ('SESSION :', 'SESSION:'):
                for j in range(i + 1, len(row)):
                    if row[j]:
                        session = row[j]; break
            if not date_val and v in ('TEST DATE :', 'TEST DATE:'):
                for j in range(i + 1, len(row)):
                    if row[j] and '/' in row[j]:
                        date_val = row[j]; break

    # fallback date from session digits
    if not date_val and session:
        m = re.search(r'\b(\d{8})\b', session)
        if m:
            d = m.group(1)
            date_val = f'{d[:4]}/{d[4:6]}/{d[6:8]}'

    if not date_val:
        date_re = re.compile(r'\b(20\d{2})[/-](\d{1,2})[/-](\d{1,2})\b')
        for r in range(min(sh.nrows, 20)):
            for c in range(sh.ncols):
                m = date_re.search(_str(sh.cell_value(r, c)))
                if m:
                    date_val = f'{m.group(1)}/{int(m.group(2))}/{int(m.group(3))}'
                    break
            if date_val:
                break

    cu = catalog.upper()
    if cu.startswith('LS1') or 'LS1A' in cu:
        dsa_class = 'DSA1'
    elif cu.startswith('LS2') or 'LS2A' in cu:
        dsa_class = 'DSA2'
    else:
        dsa_class = 'DSA1'

    # Find BeadID header row
    header_row = None
    for r in range(sh.nrows):
        if _str(sh.cell_value(r, 0)).strip() == 'BeadID':
            header_row = r
            break
    if header_row is None:
        return None

    beads_detail = []
    nc_raw = 0
    pc_raw = 0
    for r in range(header_row + 1, sh.nrows):
        bid_cell = sh.cell_value(r, 0)
        if bid_cell == '' or bid_cell is None:
            continue
        try:
            bid = f'{int(float(bid_cell)):03d}'
        except (TypeError, ValueError):
            continue
        raw_val = _flt(sh.cell_value(r, 3) if sh.ncols > 3 else 0)
        ns_raw = _flt(sh.cell_value(r, 11) if sh.ncols > 11 else 0)
        normal = _flt(sh.cell_value(r, 15) if sh.ncols > 15 else 0)
        ratio = _flt(sh.cell_value(r, 20) if sh.ncols > 20 else 0)
        rxn_val = _str(sh.cell_value(r, 22) if sh.ncols > 22 else '').strip()
        count_val = sh.cell_value(r, 26) if sh.ncols > 26 else 0
        try:
            count_val = int(float(count_val)) if count_val != '' else 0
        except (TypeError, ValueError):
            count_val = 0
        sero_raw = _str(sh.cell_value(r, 29) if sh.ncols > 29 else '').strip()
        allele_raw = _str(sh.cell_value(r, 35) if sh.ncols > 35 else '').strip()

        if rxn_val.upper() == 'NC':
            nc_raw = raw_val
            continue
        if rxn_val.upper() == 'PC':
            pc_raw = raw_val
            continue
        try:
            rxn_int = int(float(rxn_val))
        except (TypeError, ValueError):
            continue
        if not sero_raw and not allele_raw:
            continue

        from app import clean_sero, clean_allele
        beads_detail.append({
            'bead': bid,
            'rxn': rxn_int,
            'raw': round(raw_val, 1),
            'ns_raw': round(ns_raw, 1),
            'normal': round(normal, 2),
            'ratio': round(ratio, 2),
            'count': count_val,
            'sero': clean_sero(sero_raw),
            'allele': clean_allele(allele_raw),
        })

    if not beads_detail:
        return None

    # Apply DSA cutoffs to mark strength
    total = len(beads_detail)
    strong_n = sum(1 for b in beads_detail if b['normal'] > STRONG_CUT)
    weak_n = sum(1 for b in beads_detail if WEAK_CUT <= b['normal'] <= 3999)
    pos_n = strong_n + weak_n
    pct_sa = round(pos_n / total * 100) if total else 0
    overall = 'Positive' if pos_n > 0 else 'Negative'

    # Build bead_hla_map keyed by bead
    bead_hla_map = {b['bead']: {'sero': b['sero'], 'allele': b['allele']} for b in beads_detail}

    # Confident alleles = alleles from beads that pass at least the WEAK cutoff
    from app import _build_allele_to_sero, generate_specificity, _parse_allele_list

    def _alleles_from(bead):
        out = set()
        for a in _parse_allele_list(bead.get('allele', '')):
            a_clean = a.split('/')[0].split('=')[0]
            out.add(a_clean)
        return out

    strong_alleles = set()
    weak_alleles = set()
    for b in beads_detail:
        if b['normal'] > STRONG_CUT:
            strong_alleles |= _alleles_from(b)
        elif b['normal'] >= WEAK_CUT:
            weak_alleles |= _alleles_from(b)

    a2s = _build_allele_to_sero(bead_hla_map)
    strong_seros = {a2s.get(a) for a in strong_alleles if a2s.get(a)}
    if strong_seros:
        weak_alleles = {a for a in weak_alleles if a2s.get(a) not in strong_seros}

    confident = sorted(strong_alleles | weak_alleles)

    # Mark each bead's strength label (DSA = single-antigen beads, no confident-allele concept)
    for b in beads_detail:
        if b['normal'] > STRONG_CUT:
            b['strength'] = 'Strong'
        elif b['normal'] >= WEAK_CUT:
            b['strength'] = 'Weak'
        else:
            b['strength'] = ''

    # Strong / Weak HTML (PRA-style colour render)
    strong_html = generate_specificity(strong_alleles, bead_hla_map) if strong_alleles else '(-)'
    weak_html = generate_specificity(weak_alleles, bead_hla_map) if weak_alleles else '(-)'

    # Combined specificity: "Strong: ... \n Weak: ..."
    specificity = (
        f'<div><b>Strong (MFI&gt;4000):</b> {strong_html}</div>'
        f'<div><b>Weak (MFI 1000–3999):</b> {weak_html}</div>'
    ) if overall == 'Positive' else '(-)'

    # sero_mfi list (for DB antibody_strength) — group beads by sero key
    sero_groups = OrderedDict()
    for b in beads_detail:
        if b['normal'] < WEAK_CUT:
            continue
        b_alleles = list(_alleles_from(b))
        key = None
        for a in b_alleles:
            sero = a2s.get(a)
            if sero:
                key = sero
                break
        if not key and b_alleles:
            key = b_alleles[0]
        if not key:
            continue
        g = sero_groups.setdefault(key, {
            'sero': key, 'alleles': set(), 'normals': [], 'beads': [],
            'strength': b['strength'],
        })
        g['alleles'].update(b_alleles)
        g['normals'].append(b['normal'])
        g['beads'].append(b['bead'])
        if b['strength'] == 'Strong':
            g['strength'] = 'Strong'

    sero_mfi = []
    for g in sero_groups.values():
        normals = g['normals']
        sero_mfi.append({
            'sero': g['sero'],
            'alleles': ', '.join(sorted(g['alleles'])),
            'max_mfi': round(max(normals), 1),
            'mean_mfi': round(sum(normals) / len(normals), 1),
            'count': len(normals),
            'beads': ', '.join(g['beads']),
            'strength': g['strength'],
        })

    # QC: same general rules as PRA
    from app import build_qc_comment as _qc
    qc_comment = _qc(nc_raw, pc_raw, beads_detail)

    pc_nc_ratio = round(pc_raw / nc_raw, 1) if nc_raw > 0 else 0

    return {
        'name': sample_name,
        'overall': overall,
        'pra': pct_sa,                # reuse "pra" field for %SA so templates can be shared
        'pra_all': {'Strong': strong_n, 'Weak': weak_n, 'Total': total},
        'strong_n': strong_n,
        'weak_n': weak_n,
        'beads': beads_detail,
        'confident_alleles': list(confident),
        'specificity': specificity,
        'strong_html': strong_html,
        'weak_html': weak_html,
        'sero_mfi': sero_mfi,
        'comment': qc_comment,
        'qc': {
            'nc_raw': round(nc_raw, 1),
            'pc_raw': round(pc_raw, 1),
            'pc_nc_ratio': pc_nc_ratio,
            'low_beads': [{'bead': b['bead'], 'count': b['count']}
                          for b in beads_detail if b.get('count', 0) > 0 and b['count'] < 80],
        },
        '_pra_class': dsa_class,      # 'DSA1' / 'DSA2'
        '_date': date_val,
        '_batch': session,
        '_nc_raw': nc_raw,
        '_pc_raw': pc_raw,
    }


if __name__ == '__main__':
    import sys, glob, os
    sys.stdout.reconfigure(encoding='utf-8')
    for f in sorted(glob.glob('Database/DSA/*.xls')):
        print('=' * 70)
        print('FILE:', os.path.basename(f))
        try:
            with open(f, 'rb') as fh:
                p = parse_xls_file(fh.read())
            if not p:
                print('  (no beads)'); continue
            print(f"  name={p['name']} class={p['_pra_class']} date={p['_date']}")
            print(f"  Overall={p['overall']} %SA={p['pra']} Strong={p['strong_n']} Weak={p['weak_n']}")
            print(f"  Strong: {re.sub(r'<[^>]+>', '', p['strong_html'])}")
            print(f"  Weak  : {re.sub(r'<[^>]+>', '', p['weak_html'])}")
        except Exception as e:
            print('ERROR:', e)