|
|
import gradio as gr |
|
|
import cv2 |
|
|
import numpy as np |
|
|
import pandas as pd |
|
|
import pydicom |
|
|
import io |
|
|
from PIL import Image |
|
|
import openpyxl |
|
|
from openpyxl.utils import get_column_letter, column_index_from_string |
|
|
import logging |
|
|
import time |
|
|
import traceback |
|
|
from functools import wraps |
|
|
import sys |
|
|
|
|
|
print("Starting imports completed...") |
|
|
|
|
|
|
|
|
logging.basicConfig( |
|
|
level=logging.DEBUG, |
|
|
format='%(asctime)s - %(levelname)s - %(message)s', |
|
|
handlers=[ |
|
|
logging.FileHandler('dicom_analyzer_debug.log'), |
|
|
logging.StreamHandler(sys.stdout) |
|
|
] |
|
|
) |
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
def debug_decorator(func): |
|
|
@wraps(func) |
|
|
def wrapper(*args, **kwargs): |
|
|
logger.debug(f"Entering {func.__name__}") |
|
|
start_time = time.time() |
|
|
try: |
|
|
result = func(*args, **kwargs) |
|
|
logger.debug(f"Function {func.__name__} completed successfully") |
|
|
return result |
|
|
except Exception as e: |
|
|
logger.error(f"Error in {func.__name__}: {str(e)}") |
|
|
logger.error(traceback.format_exc()) |
|
|
raise |
|
|
finally: |
|
|
end_time = time.time() |
|
|
logger.debug(f"Execution time: {end_time - start_time:.4f} seconds") |
|
|
return wrapper |
|
|
|
|
|
class DicomAnalyzer: |
|
|
def __init__(self): |
|
|
self.results = [] |
|
|
self.circle_diameter = 9.0 |
|
|
self.zoom_factor = 1.0 |
|
|
self.current_image = None |
|
|
self.dicom_data = None |
|
|
self.display_image = None |
|
|
self.marks = [] |
|
|
self.original_image = None |
|
|
self.original_display = None |
|
|
|
|
|
self.pan_x = 0 |
|
|
self.pan_y = 0 |
|
|
self.max_pan_x = 0 |
|
|
self.max_pan_y = 0 |
|
|
|
|
|
self.CIRCLE_COLOR = (0, 255, 255) |
|
|
print("DicomAnalyzer initialized...") |
|
|
|
|
|
def load_dicom(self, file): |
|
|
try: |
|
|
if file is None: |
|
|
return None, "No file uploaded" |
|
|
|
|
|
if hasattr(file, 'name'): |
|
|
dicom_data = pydicom.dcmread(file.name) |
|
|
else: |
|
|
dicom_data = pydicom.dcmread(file) |
|
|
|
|
|
image = dicom_data.pixel_array.astype(np.float32) |
|
|
|
|
|
|
|
|
self.original_image = image.copy() |
|
|
|
|
|
|
|
|
rescale_slope = getattr(dicom_data, 'RescaleSlope', 1) |
|
|
rescale_intercept = getattr(dicom_data, 'RescaleIntercept', 0) |
|
|
image = (image * rescale_slope) + rescale_intercept |
|
|
|
|
|
self.current_image = image |
|
|
self.dicom_data = dicom_data |
|
|
|
|
|
self.display_image = self.normalize_image(image) |
|
|
self.original_display = self.display_image.copy() |
|
|
|
|
|
|
|
|
self.reset_view() |
|
|
print("DICOM file loaded successfully") |
|
|
|
|
|
return self.display_image, "DICOM file loaded successfully" |
|
|
except Exception as e: |
|
|
print(f"Error loading DICOM file: {str(e)}") |
|
|
return None, f"Error loading DICOM file: {str(e)}" |
|
|
|
|
|
def normalize_image(self, image): |
|
|
try: |
|
|
normalized = cv2.normalize( |
|
|
image, |
|
|
None, |
|
|
alpha=0, |
|
|
beta=255, |
|
|
norm_type=cv2.NORM_MINMAX, |
|
|
dtype=cv2.CV_8U |
|
|
) |
|
|
if len(normalized.shape) == 2: |
|
|
normalized = cv2.cvtColor(normalized, cv2.COLOR_GRAY2BGR) |
|
|
return normalized |
|
|
except Exception as e: |
|
|
print(f"Error normalizing image: {str(e)}") |
|
|
return None |
|
|
|
|
|
def reset_view(self): |
|
|
self.zoom_factor = 1.0 |
|
|
self.pan_x = 0 |
|
|
self.pan_y = 0 |
|
|
if self.original_display is not None: |
|
|
return self.update_display() |
|
|
return None |
|
|
|
|
|
def zoom_in(self, image): |
|
|
print("Zooming in...") |
|
|
self.zoom_factor = min(20.0, self.zoom_factor + 0.5) |
|
|
return self.update_display() |
|
|
|
|
|
def zoom_out(self, image): |
|
|
print("Zooming out...") |
|
|
self.zoom_factor = max(1.0, self.zoom_factor - 0.5) |
|
|
return self.update_display() |
|
|
|
|
|
def handle_keyboard(self, key): |
|
|
try: |
|
|
print(f"Handling key press: {key}") |
|
|
pan_amount = int(5 * self.zoom_factor) |
|
|
|
|
|
original_pan_x = self.pan_x |
|
|
original_pan_y = self.pan_y |
|
|
|
|
|
if key == 'ArrowLeft': |
|
|
self.pan_x = max(0, self.pan_x - pan_amount) |
|
|
elif key == 'ArrowRight': |
|
|
self.pan_x = min(self.max_pan_x, self.pan_x + pan_amount) |
|
|
elif key == 'ArrowUp': |
|
|
self.pan_y = max(0, self.pan_y - pan_amount) |
|
|
elif key == 'ArrowDown': |
|
|
self.pan_y = min(self.max_pan_y, self.pan_y + pan_amount) |
|
|
|
|
|
print(f"Pan X: {self.pan_x} (was {original_pan_x})") |
|
|
print(f"Pan Y: {self.pan_y} (was {original_pan_y})") |
|
|
print(f"Max Pan X: {self.max_pan_x}") |
|
|
print(f"Max Pan Y: {self.max_pan_y}") |
|
|
|
|
|
return self.update_display() |
|
|
except Exception as e: |
|
|
print(f"Error handling keyboard input: {str(e)}") |
|
|
return self.display_image |
|
|
def analyze_roi(self, evt: gr.SelectData): |
|
|
try: |
|
|
if self.current_image is None: |
|
|
return None, "No image loaded" |
|
|
|
|
|
clicked_x = evt.index[0] |
|
|
clicked_y = evt.index[1] |
|
|
|
|
|
x = clicked_x + self.pan_x |
|
|
y = clicked_y + self.pan_y |
|
|
if self.zoom_factor != 1.0: |
|
|
x = x / self.zoom_factor |
|
|
y = y / self.zoom_factor |
|
|
|
|
|
x = int(round(x)) |
|
|
y = int(round(y)) |
|
|
|
|
|
height, width = self.original_image.shape[:2] |
|
|
|
|
|
Y, X = np.ogrid[:height, :width] |
|
|
|
|
|
radius = self.circle_diameter / 2.0 |
|
|
r_squared = radius * radius |
|
|
|
|
|
dx = X - x |
|
|
dy = Y - y |
|
|
dist_squared = dx*dx + dy*dy |
|
|
|
|
|
mask = np.zeros((height, width), dtype=bool) |
|
|
mask[dist_squared <= r_squared] = True |
|
|
|
|
|
roi_pixels = self.original_image[mask] |
|
|
|
|
|
if len(roi_pixels) == 0: |
|
|
return self.display_image, "Error: No pixels selected" |
|
|
|
|
|
pixel_spacing = float(self.dicom_data.PixelSpacing[0]) |
|
|
|
|
|
n_pixels = np.sum(mask) |
|
|
area = n_pixels * (pixel_spacing ** 2) |
|
|
|
|
|
mean_value = np.mean(roi_pixels) |
|
|
std_dev = np.std(roi_pixels, ddof=1) |
|
|
min_val = np.min(roi_pixels) |
|
|
max_val = np.max(roi_pixels) |
|
|
|
|
|
rescale_slope = getattr(self.dicom_data, 'RescaleSlope', 1) |
|
|
rescale_intercept = getattr(self.dicom_data, 'RescaleIntercept', 0) |
|
|
|
|
|
mean_value = (mean_value * rescale_slope) + rescale_intercept |
|
|
std_dev = std_dev * rescale_slope |
|
|
min_val = (min_val * rescale_slope) + rescale_intercept |
|
|
max_val = (max_val * rescale_slope) + rescale_intercept |
|
|
|
|
|
result = { |
|
|
'Area (mm²)': f"{area:.3f}", |
|
|
'Mean': f"{mean_value:.3f}", |
|
|
'StdDev': f"{std_dev:.3f}", |
|
|
'Min': f"{min_val:.3f}", |
|
|
'Max': f"{max_val:.3f}", |
|
|
'Point': f"({x}, {y})" |
|
|
} |
|
|
|
|
|
self.results.append(result) |
|
|
self.marks.append((x, y, self.circle_diameter)) |
|
|
|
|
|
return self.update_display(), self.format_results() |
|
|
except Exception as e: |
|
|
print(f"Error analyzing ROI: {str(e)}") |
|
|
return self.display_image, f"Error analyzing ROI: {str(e)}" |
|
|
|
|
|
def update_display(self): |
|
|
try: |
|
|
if self.original_display is None: |
|
|
return None |
|
|
|
|
|
height, width = self.original_display.shape[:2] |
|
|
new_height = int(height * self.zoom_factor) |
|
|
new_width = int(width * self.zoom_factor) |
|
|
|
|
|
zoomed = cv2.resize(self.original_display, (new_width, new_height), |
|
|
interpolation=cv2.INTER_CUBIC) |
|
|
|
|
|
zoomed_bgr = cv2.cvtColor(zoomed, cv2.COLOR_RGB2BGR) |
|
|
|
|
|
for x, y, diameter in self.marks: |
|
|
zoomed_x = int(x * self.zoom_factor) |
|
|
zoomed_y = int(y * self.zoom_factor) |
|
|
zoomed_radius = int((diameter/2.0) * self.zoom_factor) |
|
|
|
|
|
cv2.circle(zoomed_bgr, |
|
|
(zoomed_x, zoomed_y), |
|
|
zoomed_radius, |
|
|
self.CIRCLE_COLOR, |
|
|
1, |
|
|
lineType=cv2.LINE_AA) |
|
|
|
|
|
num_points = 8 |
|
|
for i in range(num_points): |
|
|
angle = 2 * np.pi * i / num_points |
|
|
point_x = int(zoomed_x + zoomed_radius * np.cos(angle)) |
|
|
point_y = int(zoomed_y + zoomed_radius * np.sin(angle)) |
|
|
cv2.circle(zoomed_bgr, |
|
|
(point_x, point_y), |
|
|
1, |
|
|
self.CIRCLE_COLOR, |
|
|
-1, |
|
|
lineType=cv2.LINE_AA) |
|
|
|
|
|
zoomed = cv2.cvtColor(zoomed_bgr, cv2.COLOR_BGR2RGB) |
|
|
|
|
|
self.max_pan_x = max(0, new_width - width) |
|
|
self.max_pan_y = max(0, new_height - height) |
|
|
self.pan_x = min(max(0, self.pan_x), self.max_pan_x) |
|
|
self.pan_y = min(max(0, self.pan_y), self.max_pan_y) |
|
|
|
|
|
visible = zoomed[ |
|
|
int(self.pan_y):int(self.pan_y + height), |
|
|
int(self.pan_x):int(self.pan_x + width) |
|
|
] |
|
|
|
|
|
return visible |
|
|
except Exception as e: |
|
|
print(f"Error updating display: {str(e)}") |
|
|
return self.original_display |
|
|
|
|
|
def format_results(self): |
|
|
if not self.results: |
|
|
return "No measurements yet" |
|
|
df = pd.DataFrame(self.results) |
|
|
columns_order = ['Area (mm²)', 'Mean', 'StdDev', 'Min', 'Max', 'Point'] |
|
|
df = df[columns_order] |
|
|
return df.to_string(index=False) |
|
|
|
|
|
def save_results(self): |
|
|
try: |
|
|
if not self.results: |
|
|
return None, "No results to save" |
|
|
|
|
|
|
|
|
wb = openpyxl.Workbook() |
|
|
ws = wb.active |
|
|
|
|
|
|
|
|
equation_slots = [ |
|
|
('B', 'F'), ('H', 'L'), ('N', 'R'), ('T', 'X'), ('Z', 'AD'), |
|
|
('AF', 'AJ'), ('AL', 'AP'), ('AR', 'AV'), ('AX', 'BB'), ('BD', 'BH'), |
|
|
('BJ', 'BN'), ('BP', 'BT'), ('BV', 'BZ'), |
|
|
] |
|
|
|
|
|
|
|
|
row_groups = [ |
|
|
(2, 3), (5, 6), (8, 9), (11, 12), (14, 15), |
|
|
(17, 18), (20, 21), (23, 24), (26, 27), (29, 30), |
|
|
] |
|
|
|
|
|
|
|
|
phantom_sizes = ['(7mm)', '(6.5mm)', '(6mm)', '(5.5mm)', '(5mm)', '(4.5mm)'] |
|
|
for i, size in enumerate(phantom_sizes): |
|
|
row_index = row_groups[i][0] - 1 |
|
|
ws.cell(row=row_index, column=1, value=size) |
|
|
|
|
|
|
|
|
result_pairs = [self.results[i:i+2] for i in range(0, len(self.results), 2)] |
|
|
|
|
|
for pair_idx, result_pair in enumerate(result_pairs): |
|
|
if pair_idx >= len(equation_slots) * len(row_groups): |
|
|
break |
|
|
|
|
|
slot_idx = pair_idx % len(equation_slots) |
|
|
group_idx = pair_idx // len(equation_slots) |
|
|
|
|
|
if group_idx >= len(row_groups): |
|
|
break |
|
|
|
|
|
start_col, _ = equation_slots[slot_idx] |
|
|
dest_rows = row_groups[group_idx] |
|
|
|
|
|
|
|
|
for row_idx, result in enumerate(result_pair): |
|
|
if row_idx < 2: |
|
|
dest_row = dest_rows[row_idx] |
|
|
|
|
|
|
|
|
ws.cell(row=dest_row, column=1, value=row_idx + 1) |
|
|
|
|
|
|
|
|
ws.cell(row=dest_row, column=openpyxl.utils.column_index_from_string(start_col), |
|
|
value=float(result['Area (mm²)'])) |
|
|
ws.cell(row=dest_row, column=openpyxl.utils.column_index_from_string(start_col) + 1, |
|
|
value=float(result['Mean'])) |
|
|
ws.cell(row=dest_row, column=openpyxl.utils.column_index_from_string(start_col) + 2, |
|
|
value=float(result['StdDev'])) |
|
|
ws.cell(row=dest_row, column=openpyxl.utils.column_index_from_string(start_col) + 3, |
|
|
value=float(result['Min'])) |
|
|
ws.cell(row=dest_row, column=openpyxl.utils.column_index_from_string(start_col) + 4, |
|
|
value=float(result['Max'])) |
|
|
|
|
|
|
|
|
output_file = "analysis_results.xlsx" |
|
|
wb.save(output_file) |
|
|
|
|
|
return output_file, "Results saved successfully in the required format" |
|
|
except Exception as e: |
|
|
print(f"Error saving results: {str(e)}") |
|
|
return None, f"Error saving results: {str(e)}" |
|
|
|
|
|
def add_blank_row(self, image): |
|
|
self.results.append({ |
|
|
'Area (mm²)': '', |
|
|
'Mean': '', |
|
|
'StdDev': '', |
|
|
'Min': '', |
|
|
'Max': '', |
|
|
'Point': '' |
|
|
}) |
|
|
return image, self.format_results() |
|
|
|
|
|
def add_zero_row(self, image): |
|
|
self.results.append({ |
|
|
'Area (mm²)': '0.000', |
|
|
'Mean': '0.000', |
|
|
'StdDev': '0.000', |
|
|
'Min': '0.000', |
|
|
'Max': '0.000', |
|
|
'Point': '(0, 0)' |
|
|
}) |
|
|
return image, self.format_results() |
|
|
|
|
|
def undo_last(self, image): |
|
|
if self.results: |
|
|
self.results.pop() |
|
|
if self.marks: |
|
|
self.marks.pop() |
|
|
return self.update_display(), self.format_results() |
|
|
|
|
|
|
|
|
def create_interface(): |
|
|
print("Creating interface...") |
|
|
analyzer = DicomAnalyzer() |
|
|
|
|
|
with gr.Blocks(css="#image_display { outline: none; }") as interface: |
|
|
gr.Markdown("# DICOM Image Analyzer") |
|
|
|
|
|
with gr.Row(): |
|
|
with gr.Column(): |
|
|
file_input = gr.File(label="Upload DICOM file") |
|
|
diameter_slider = gr.Slider( |
|
|
minimum=1, |
|
|
maximum=20, |
|
|
value=9, |
|
|
step=1, |
|
|
label="ROI Diameter (pixels)" |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
zoom_in_btn = gr.Button("Zoom In (+)") |
|
|
zoom_out_btn = gr.Button("Zoom Out (-)") |
|
|
reset_btn = gr.Button("Reset View") |
|
|
|
|
|
with gr.Column(): |
|
|
image_display = gr.Image( |
|
|
label="DICOM Image", |
|
|
interactive=True, |
|
|
elem_id="image_display" |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
blank_btn = gr.Button("Add Blank Row") |
|
|
zero_btn = gr.Button("Add Zero Row") |
|
|
undo_btn = gr.Button("Undo Last") |
|
|
save_btn = gr.Button("Save Results") |
|
|
|
|
|
results_display = gr.Textbox(label="Results", interactive=False) |
|
|
file_output = gr.File(label="Download Results") |
|
|
key_press = gr.Textbox(visible=False, elem_id="key_press") |
|
|
|
|
|
gr.Markdown(""" |
|
|
### Controls: |
|
|
- Use arrow keys to pan when zoomed in |
|
|
- Click points to measure |
|
|
- Use Zoom In/Out buttons or Reset View to adjust zoom level |
|
|
- Results will be saved in ImageJ-compatible format |
|
|
""") |
|
|
|
|
|
def update_diameter(x): |
|
|
analyzer.circle_diameter = float(x) |
|
|
print(f"Diameter updated to: {x}") |
|
|
return f"Diameter set to {x} pixels" |
|
|
|
|
|
|
|
|
file_input.change( |
|
|
fn=analyzer.load_dicom, |
|
|
inputs=file_input, |
|
|
outputs=[image_display, results_display] |
|
|
) |
|
|
|
|
|
image_display.select( |
|
|
fn=analyzer.analyze_roi, |
|
|
outputs=[image_display, results_display] |
|
|
) |
|
|
|
|
|
diameter_slider.change( |
|
|
fn=update_diameter, |
|
|
inputs=diameter_slider, |
|
|
outputs=gr.Textbox(label="Status") |
|
|
) |
|
|
|
|
|
zoom_in_btn.click( |
|
|
fn=analyzer.zoom_in, |
|
|
inputs=image_display, |
|
|
outputs=image_display, |
|
|
queue=False |
|
|
) |
|
|
|
|
|
zoom_out_btn.click( |
|
|
fn=analyzer.zoom_out, |
|
|
inputs=image_display, |
|
|
outputs=image_display, |
|
|
queue=False |
|
|
) |
|
|
|
|
|
reset_btn.click( |
|
|
fn=analyzer.reset_view, |
|
|
outputs=image_display |
|
|
) |
|
|
|
|
|
key_press.change( |
|
|
fn=analyzer.handle_keyboard, |
|
|
inputs=key_press, |
|
|
outputs=image_display |
|
|
) |
|
|
|
|
|
blank_btn.click( |
|
|
fn=analyzer.add_blank_row, |
|
|
inputs=image_display, |
|
|
outputs=[image_display, results_display] |
|
|
) |
|
|
|
|
|
zero_btn.click( |
|
|
fn=analyzer.add_zero_row, |
|
|
inputs=image_display, |
|
|
outputs=[image_display, results_display] |
|
|
) |
|
|
|
|
|
undo_btn.click( |
|
|
fn=analyzer.undo_last, |
|
|
inputs=image_display, |
|
|
outputs=[image_display, results_display] |
|
|
) |
|
|
|
|
|
save_btn.click( |
|
|
fn=analyzer.save_results, |
|
|
outputs=[file_output, results_display] |
|
|
) |
|
|
|
|
|
js = """ |
|
|
<script> |
|
|
document.addEventListener('keydown', function(e) { |
|
|
if (['ArrowUp', 'ArrowDown', 'ArrowLeft', 'ArrowRight'].includes(e.key)) { |
|
|
e.preventDefault(); |
|
|
const keyPressElement = document.querySelector('#key_press textarea'); |
|
|
if (keyPressElement) { |
|
|
keyPressElement.value = e.key; |
|
|
keyPressElement.dispatchEvent(new Event('input')); |
|
|
} |
|
|
} |
|
|
}); |
|
|
</script> |
|
|
""" |
|
|
gr.HTML(js) |
|
|
|
|
|
print("Interface created successfully") |
|
|
return interface |
|
|
|
|
|
if __name__ == "__main__": |
|
|
try: |
|
|
print("Starting application...") |
|
|
interface = create_interface() |
|
|
print("Launching interface...") |
|
|
interface.launch( |
|
|
server_name="0.0.0.0", |
|
|
server_port=7860, |
|
|
share=True, |
|
|
debug=True |
|
|
) |
|
|
except Exception as e: |
|
|
print(f"Error launching application: {str(e)}") |
|
|
logger.error(f"Error launching application: {str(e)}") |
|
|
logger.error(traceback.format_exc()) |
|
|
raise e |