"""Breast Density · 유방 MG CC 분석 (간단 버전 · 환자별 QC). mView 에서 MG 를 보다가 버튼을 누르면, 그 환자의 로딩된 MG 경로가 `/job_.json` 으로 전달되고 브라우저가 `http://127.0.0.1:/?job=` 로 열린다(환자마다 새 탭). 이 페이지는 query param 의 token 으로 해당 job 파일을 읽어 그 환자를 분석한다. 탭마다 URL 이 달라 환자 혼선이 없다(폴링/auto-refresh 불필요). CC(좌/우)만 분석하고, 같은 side 에 CC 가 여러 장이면 부피·밀도를 산술평균해 side 별 한 줄로 요약한다. 영상별 상세 UI 는 띄우지 않고 썸네일+요약표만 보여준다(QC 가 목적). 통계·분포·백분위·저장은 이 간단 버전의 범위가 아니다(복잡 버전 별도). """ import argparse import html import io import json import os from pathlib import Path import numpy as np import pandas as pd import pydicom import streamlit as st from PIL import Image, ImageOps from dicom_io import read_image from acceleration import analyze_accelerated from planning import LABELS, descriptors from cc_select import select_cc def get_datadir(): """job 파일 위치(datadir) 를 얻는다. 우선순위: 실행 인자 `--datadir` → 환경변수 `BREASTDENSITY_DATADIR`. (streamlit run app.py -- --datadir <경로>, 또는 AppTest/배포 시 환경변수.) """ parser = argparse.ArgumentParser() parser.add_argument('--datadir', default=None) args, _ = parser.parse_known_args() return args.datadir or os.environ.get('BREASTDENSITY_DATADIR') DATADIR = get_datadir() st.set_page_config(page_title='Breast Density · MG CC QC', layout='wide') st.title('Breast Density · 유방 MG CC 분석 (QC)') st.caption('OpenBreast 알고리즘 기반') def load_job(token): """datadir 의 job_.json 을 읽어 paths 리스트를 돌려준다. 없거나 깨졌으면 None.""" if not DATADIR or not token: return None # token 은 파일명에 들어가므로 경로 조작 방지를 위해 영숫자/대시/언더스코어만 허용. if not all(c.isalnum() or c in '-_' for c in token): return None path = Path(DATADIR) / f'job_{token}.json' if not path.exists(): return None try: return json.loads(path.read_text(encoding='utf-8')) except (ValueError, OSError): return None job_token = st.query_params.get('job') if not DATADIR: st.warning('실행 인자 --datadir 가 없습니다. mView 가 전달하는 job 파일 위치를 지정해야 합니다.') st.stop() if not job_token: st.info('mView 에서 유방촬영(MG) 을 열고 분석 버튼을 누르면 새 탭에 결과가 표시됩니다.') st.stop() job = load_job(job_token) if not job or not job.get('paths'): st.error('이 분석 요청(job) 을 찾을 수 없거나 경로가 비어 있습니다. mView 에서 다시 실행하세요.') st.stop() # --- 경로 → DICOM bytes (로컬에 실제로 존재하는 것만) --- # job json 은 Python 이 삭제하지 않는다. 경로가 mView DUMP 폴더라, mView 종료 시 원본 DICOM 과 # 함께 job 파일도 mView 가 정리한다(BreastDensityLauncher.stopServer → cleanupAllJobFiles). # 세션 중에는 rerun(재실행) 시 다시 읽을 수 있도록 남겨둔다. items = [] missing = [] for path in job['paths']: try: items.append((Path(path).name, Path(path).read_bytes())) except OSError: missing.append(path) # --- 환자 정보 (상단 고정 헤더용) --- def read_patient_info(raw): try: ds = pydicom.dcmread(io.BytesIO(raw), stop_before_pixels=True, force=True) except Exception: return {} return { 'id': str(getattr(ds, 'PatientID', '')), 'name': str(getattr(ds, 'PatientName', '')), 'sex': str(getattr(ds, 'PatientSex', '')), 'birth': str(getattr(ds, 'PatientBirthDate', '')), 'date': str(getattr(ds, 'StudyDate', '')), } pinfo = read_patient_info(items[0][1]) if items else {} # 환자 정보를 화면 상단에 '고정(fixed)'해 스크롤해도 항상 보이게 한다. Streamlit 의 position:sticky 는 # 스크롤 컨테이너 때문에 잘 안 먹으므로 viewport 기준 position:fixed 로 두고, Streamlit 상단 헤더(Deploy 등)를 # 숨긴 뒤 본문 상단에 고정바 높이만큼 여백을 준다. 검은 배경에서 잘 보이도록 글씨를 밝게. st.markdown( "", unsafe_allow_html=True, ) _pe = {k: html.escape(str(v)) for k, v in pinfo.items()} # HTML 안전(이름에 특수문자 대비) st.markdown( "
" f"환자 {_pe.get('id','')}  ·  {_pe.get('name','')} " f" ({_pe.get('sex','')}/{_pe.get('birth','')}) " f"   검사일 {_pe.get('date','')}" "
", unsafe_allow_html=True, ) def analyze_cc(rec): """CC 한 장을 기본 설정으로 분석 + 썸네일 생성. digest 로 session_state 캐시. 설정 UI 는 없다(QC 간소화). 전처리는 DICOM 태그 기준, 윤곽 threshold 는 자동, MAG 피부 제외 4mm·dense threshold +0.07 은 알고리즘 기본값(analyze_accelerated 기본)이다. CC 는 대흉근 제외(skip_pectoral)·하부 절단(inferior_cut)이 없다. """ digest = rec['digest'] cache_key = 'analysis' + digest cached = st.session_state.get(cache_key) if cached is not None: return cached image, working_spacing, step, warnings = read_image(rec['raw']) result = analyze_accelerated(image, False, step, backend='auto') meta = rec['meta'].copy() values = descriptors(result['breast_mask'], working_spacing, meta, result['density_percent']) # overlay: 녹색=유방, 분홍=고밀도 rgb = np.repeat(image[..., None], 3, axis=2) bm, dm = result['breast_mask'], result['dense_mask'] rgb[bm] = 0.75 * rgb[bm] + 0.25 * np.array([0, 1, .5]) rgb[dm] = 0.55 * rgb[dm] + 0.45 * np.array([1, .1, .5]) thumb_orig = ImageOps.pad(Image.fromarray(np.rint(np.clip(image, 0, 1) * 255).astype('uint8')), (240, 300), method=Image.Resampling.LANCZOS, color=0) thumb_over = ImageOps.pad(Image.fromarray(np.rint(np.clip(rgb, 0, 1) * 255).astype('uint8')), (240, 300), method=Image.Resampling.LANCZOS, color=0) out = {'values': values, 'thumb_orig': thumb_orig, 'thumb_over': thumb_over, 'warnings': warnings + result['warnings'], 'execution': result['execution']} st.session_state[cache_key] = out return out # --- CC 선별 --- accepted, rejected = select_cc(items) if not accepted['R'] and not accepted['L']: st.error('분석할 CC 영상이 없습니다. (MLO·tomo·AI결과 이미지만 있거나 경로 오류)') with st.expander('제외된 영상'): for name, reason in rejected: st.write(f'- {name}: {reason}') st.stop() # --- 분석 (side 별) --- # 한 장 분석 실패(윤곽 추출 실패 등 정상 임상 상황에서도 raise 가능)가 환자 전체 QC 를 # 막지 않도록 영상별로 격리한다. 실패 장은 제외 목록에 남겨 QC 투명성을 유지한다. side_results = {'R': [], 'L': []} for side in ('R', 'L'): for rec in accepted[side]: try: side_results[side].append(analyze_cc(rec)) except Exception as exc: rejected.append((rec['name'], f'분석 실패: {exc}')) if not side_results['R'] and not side_results['L']: st.error('CC 영상 분석에 모두 실패했습니다. 전처리·윤곽·spacing 을 확인하세요.') with st.expander('제외/실패 영상'): for name, reason in rejected: st.write(f'- {name}: {reason}') st.stop() # --- 썸네일: 한 줄에 4개(원본/분석 쌍은 인접 유지). 존재하는 side·복수 CC 모두 표시(편측 허용) --- st.subheader('CC 영상 (원본 / 분석 overlay)') st.caption('녹색: 유방 영역 · 분홍색: 고밀도(MAG) 영역') thumbs = [] for side in ('R', 'L'): results = side_results[side] for idx, res in enumerate(results): suffix = '' if len(results) == 1 else f' #{idx + 1}' thumbs.append((f'{side}CC{suffix} 원본', res['thumb_orig'])) thumbs.append((f'{side}CC{suffix} 분석', res['thumb_over'])) for start in range(0, len(thumbs), 4): for col, (caption, img) in zip(st.columns(4), thumbs[start:start + 4]): col.image(img, caption=caption, width=240) def side_average(results): """side 의 여러 CC 지표를 산술평균(None 제외). 전부 None 이면 None(표에서 '미측정'으로 표시). 주의: 압박두께(thickness)도 평균하지만, 복수 CC 가 재촬영이면 압박 조건이 다를 수 있어 평균 두께는 엄밀히는 '존재하지 않는 중간값'이다. 간단 QC 범위에서는 산술평균으로 둔다. 숫자/문자 혼재로 표 dtype 이 깨지지 않도록 미측정은 None 으로 두고 표시에서만 치환한다. """ row = {} for metric in LABELS: vals = [r['values'][metric] for r in results if r['values'].get(metric) is not None] row[LABELS[metric]] = round(float(np.mean(vals)), 2) if vals else None return row # --- 요약 표: side 별 1행 (복수면 산술평균) --- st.subheader('요약 (좌/우 CC)') summary_rows = [] for side in ('R', 'L'): results = side_results[side] if not results: continue summary_rows.append({'좌우': side, '장수': len(results), **side_average(results)}) summary_df = pd.DataFrame(summary_rows) # 지표 컬럼은 숫자 dtype 유지, 미측정(None/NaN)은 '미측정'으로 표시(숫자열 오염 방지). metric_cols = [c for c in summary_df.columns if c in LABELS.values()] st.dataframe(summary_df.style.format(na_rep='미측정', precision=2, subset=metric_cols), hide_index=True) st.caption('같은 side CC 가 2장 이상이면 부피·밀도 등 지표는 산술평균입니다. 체적은 CC 투영면적 × 압박두께 근사치, ' 'MAG %는 면적밀도(BI-RADS·체적밀도 아님).') # --- 경고 / 제외 영상 (QC 투명성) --- all_warnings = [w for side in ('R', 'L') for r in side_results[side] for w in r['warnings']] if all_warnings: with st.expander(f'경고 {len(all_warnings)}건'): for warning in all_warnings: st.warning(warning) if rejected or missing: with st.expander(f'제외/누락 영상 {len(rejected) + len(missing)}건'): for name, reason in rejected: st.write(f'- (제외) {name}: {reason}') for path in missing: # 전체 경로는 환자 식별 가능 → 파일명만 표시(PHI 노출 최소화). st.write(f'- (누락) {Path(path).name}: 로컬 파일 없음')