"""Optional NVIDIA CUDA/CuPy engine. CPU fitting/Canny and DICOM codecs retained. No global monkey-patching: CPU and GPU sessions can coexist safely. """ import functools,time import numpy as np from scipy.signal.windows import gaussian from skimage import morphology import openbreast @functools.lru_cache(maxsize=1) def gpu_status(): try: import cupy as cp from cupyx.scipy import ndimage as ndi if cp.cuda.runtime.getDeviceCount()<1:raise RuntimeError('NVIDIA CUDA GPU 없음') # Force runtime/kernel initialization; importing CuPy alone is insufficient. a=cp.arange(16,dtype=cp.float32) ndi.minimum_filter(a,size=3).sum().item() cp.cuda.get_current_stream().synchronize() prop=cp.cuda.runtime.getDeviceProperties(cp.cuda.runtime.getDevice()) name=prop['name'];name=name.decode() if isinstance(name,bytes) else str(name) return {'available':True,'name':name,'cupy_version':cp.__version__,'reason':None} except Exception as exc: return {'available':False,'name':None,'cupy_version':None,'reason':f'{type(exc).__name__}: {exc}'} class GPUProcessor: def __init__(self): import cupy as cp from cupyx.scipy import ndimage as ndi self.cp=cp;self.ndi=ndi def largest(self,mask): cp=self.cp labels,n=self.ndi.label(mask,structure=cp.ones((3,3),dtype=cp.uint8)) if int(n)==0:raise ValueError('유방 영역을 찾지 못했습니다.') counts=cp.bincount(labels.ravel());counts[0]=0 return labels==cp.argmax(counts) def foreground(self,im): cp=self.cp;ndi=self.ndi threshold,method=openbreast.foreground_threshold(im) # small Gaussian fit remains CPU mask=cp.asarray(im)>=max(threshold,0);mask[0,:]=False;mask[-1,:]=False structure=cp.ones((5,5),dtype=cp.bool_) cleaned=ndi.binary_erosion(mask,structure=structure) if not bool(cleaned.any().item()):raise ValueError('윤곽 추출 실패; threshold 또는 mask를 수정하세요.') cleaned=self.largest(cleaned) cleaned=ndi.binary_dilation(cleaned,structure=structure) cleaned=ndi.binary_closing(cleaned,structure=structure) cleaned=ndi.binary_fill_holes(cleaned) fraction=float(cleaned.mean().item());warnings=[] if fraction<.03 or fraction>.95:warnings.append('유방 면적 비율이 비정상입니다. 윤곽을 확인하세요.') return cp.asnumpy(cleaned),threshold,method,warnings def pectoral_smooth(self,im): cp=self.cp x=self.ndi.grey_dilation(cp.asarray(im),footprint=cp.asarray(morphology.disk(8))) return cp.asnumpy(self.ndi.gaussian_filter(x,1)) def mag_density(self,im,mask,step_mm,skin_gap_mm=4.,min_area_mm2=16.,skin_mask=None,offset=0.): cp=self.cp;ndi=self.ndi gm=cp.asarray(mask,dtype=cp.bool_);gi=cp.asarray(im) if not bool(gm.any().item()):raise ValueError('빈 유방 mask') lo=float(gi[gm].min().item());hi=float(gi[gm].max().item()) if hi-lo<1e-6:raise ValueError('유방 내부 대비가 없어 밀도를 계산할 수 없습니다.') # NumPy thresholds use strong float64 scalars. Explicit float64 comparisons keep boundary semantics. values=gi.astype(cp.float64);levels=np.linspace(lo,hi,25) area_gpu=cp.empty(25,dtype=cp.float64) for k,level in enumerate(levels):area_gpu[k]=cp.count_nonzero((values>=float(level))&gm) area=cp.asnumpy(area_gpu) gradient=np.convolve(np.diff(area),gaussian(5,.8),'same') auto,threshold=openbreast.dense_threshold(levels[int(np.argmin(gradient))+1],lo,hi,offset) dense=(values>=threshold)&gm if step_mm is None:raise ValueError('MAG 밀도 계산에는 mm pixel spacing이 필요합니다. 태그 또는 수동 값을 입력하세요.') diameter=max(1,int(np.floor(2*skin_gap_mm/step_mm+.5))) source=cp.asarray(openbreast.skin_source(mask,skin_mask),dtype=cp.uint8) eroded=(ndi.minimum_filter(source,size=diameter,mode='constant',cval=1)>0)&gm if not bool(eroded.any().item()):raise ValueError('피부 제외 후 분석 영역이 없습니다. 윤곽 / spacing / skin gap 확인') minimum=max(1,int(np.floor(min_area_mm2/(step_mm**2)+.5))) labels,n=ndi.label(dense&eroded,structure=cp.ones((3,3),dtype=cp.uint8)) counts=cp.bincount(labels.ravel());keep=counts>=minimum;keep[0]=False dense=keep[labels] pd=float(100*dense.sum().item()/gm.sum().item());warnings=[] if pd<.1 or pd>90:warnings.append('극단적인 MAG 밀도 값: overlay와 전처리를 검토하세요.') return cp.asnumpy(dense),pd,{'threshold':threshold,'automatic_threshold':auto,'threshold_offset':offset, 'skin_gap_mm':skin_gap_mm,'minimum_component_mm2':min_area_mm2,'method':openbreast.VERSION,'denominator':'whole breast mask, including skin gap', 'levels':25,'area_curve':area.tolist(),'warnings':warnings} def analyze_accelerated(im,ismlo,step_mm,*args,backend='auto',**kwargs): """auto uses available CUDA; gpu explicitly requests CUDA but falls back with visible metadata.""" if backend not in ('auto','cpu','gpu'):raise ValueError('backend: auto/cpu/gpu') status=gpu_status() if backend!='cpu' else {'available':False,'name':None,'reason':None} use_gpu=backend!='cpu' and status['available'];fallback=None start=time.perf_counter() if use_gpu: try: processor=GPUProcessor() result=openbreast.analyze(im,ismlo,step_mm,*args,processor=processor,**kwargs) processor.cp.cuda.get_current_stream().synchronize() actual='gpu' except Exception as exc: # Retry whole analysis consistently; do not mix partial GPU masks with CPU data. fallback=f'{type(exc).__name__}: {exc}' result=openbreast.analyze(im,ismlo,step_mm,*args,**kwargs);actual='cpu' else: if backend=='gpu':fallback=status['reason'] result=openbreast.analyze(im,ismlo,step_mm,*args,**kwargs);actual='cpu' elapsed=time.perf_counter()-start if fallback:result['warnings'].append('GPU 사용 실패로 CPU 재분석: '+fallback) result['execution']={'requested_backend':backend,'actual_backend':actual,'gpu_name':status.get('name'), 'analysis_seconds':elapsed,'fallback_reason':fallback, 'scope':'analysis only; excludes DICOM decode/resize/render and GPU availability probe'} return result if __name__=='__main__': import json print(json.dumps(gpu_status(),ensure_ascii=False,indent=2)) if not gpu_status()['available']:raise SystemExit(1)