# %matplotlib widget
import time
import warnings
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
import ipywidgets as widgets
from IPython.display import clear_output, display
from imperandi.qc.viewer_resample import (
DEFAULT_ISOTROPIC_RESOLUTION_MM,
load_nifti_isotropic,
validate_isotropic_resolution,
)
warnings.filterwarnings("ignore") # Ignore warnings
# List of DICOM tags to display
DICOM_TAGS_TO_DISPLAY = [
"patient_key",
"date",
"visit_order",
"phase",
"SeriesDescription",
"ImageType",
"PixelSpacing",
"SpacingBetweenSlices",
"SliceThickness",
"liver_noise",
"liver_volume",
"liver_median_hu",
"vessels_median_hu",
"tumor_median_hu",
"num_tumors",
"tumor_volume",
]
SLICE_NAV_STEP = 3
WINDOW_PRESETS = {
"Soft Tissue": (40, 400),
"Liver": (60, 150),
"Lung": (-600, 1500),
"Bone": (300, 1500),
}
COLORMAPS = ["jet", "autumn", "summer", "winter", "viridis"]
CONTOUR_COLORS = ["blue", "red", "green", "cyan", "magenta"]
DISPLAY_CANVAS_PX = 700
FIGURE_DPI = 100
JUMP_DROPDOWN_WIDTH = "250px"
JUMP_NAV_BUTTON_WIDTH = "120px"
[docs]
def load_nifti(
file_path,
orientation="LAS",
isotropic_resolution_mm=DEFAULT_ISOTROPIC_RESOLUTION_MM,
order=1,
):
"""Load a NIfTI file, orient it, and resample to isotropic display spacing."""
return load_nifti_isotropic(
file_path,
orientation=orientation,
resolution_mm=isotropic_resolution_mm,
order=order,
)
[docs]
def clip_hu_values(ct_scan, min_hu, max_hu):
"""Clip the Hounsfield Unit (HU) values of the CT scan."""
return np.clip(ct_scan, min_hu, max_hu)
[docs]
class CTScanViewer:
"""Interactive Jupyter viewer for CT volumes and segmentation overlays.
The viewer navigates cohort rows, patients, exams, phases, and anatomical
planes while applying configurable HU windows and isotropic resampling.
"""
def __init__(
self,
df,
ct_scan_col,
segmentation_cols=None,
phase_col=None,
HU_min=-100,
HU_max=400,
exploration_mode="ordered",
isotropic_resolution_mm=DEFAULT_ISOTROPIC_RESOLUTION_MM,
):
self.df = df
self.ct_scan_col = ct_scan_col
self.patient_col = "patient_key" if "patient_key" in df.columns else None
self.date_col = "date" if "date" in df.columns else None
if phase_col is not None and phase_col in df.columns:
self.phase_col = phase_col
elif "phase" in df.columns:
self.phase_col = "phase"
elif "totalseg_phase" in df.columns:
self.phase_col = "totalseg_phase"
else:
self.phase_col = None
# Handle segmentation_cols gracefully
if segmentation_cols is None:
auto_cols = [col for col in df.columns if str(col).startswith("mask_")]
if not auto_cols:
auto_cols = [
col for col in ["liver_path", "liver_tumor_path"] if col in df
]
self.segmentation_cols = [
col
for col in auto_cols
if df[col].apply(lambda v: not self._is_empty_value(v)).any()
]
elif isinstance(segmentation_cols, str):
self.segmentation_cols = [segmentation_cols]
else:
self.segmentation_cols = segmentation_cols
# Filter out missing segmentation columns
self.segmentation_cols = [
col for col in self.segmentation_cols if col in df.columns
]
if segmentation_cols and not self.segmentation_cols:
print("Warning: No valid segmentation columns found in DataFrame.")
self.seg_colormaps = {}
self.seg_contour_colors = {}
for i, seg_name in enumerate(self.segmentation_cols):
self.seg_colormaps[seg_name] = COLORMAPS[i % len(COLORMAPS)]
self.seg_contour_colors[seg_name] = CONTOUR_COLORS[i % len(CONTOUR_COLORS)]
self.HU_min = HU_min
self.HU_max = HU_max
self.current_index = 0
self.view_plane = "axial"
self.slice_idx = 0
self.ct_scan_raw = np.zeros([2, 2, 2])
self.segmentations = {}
self.seg_visibility = {}
self.fig = None
self.ax = None
self.display_widget = None
self._uses_output_fallback = False
self._suspend_jump = False
self.exploration_mode = exploration_mode
self.canvas_size_px = DISPLAY_CANVAS_PX
self.figure_dpi = FIGURE_DPI
self.image_aspect = "auto"
self.isotropic_resolution_mm = validate_isotropic_resolution(
isotropic_resolution_mm
)
if self.exploration_mode == "random":
self.explored_history = [self.current_index]
self.history_index = 0
self.init_widgets()
self.load_data()
def _option_values(self, options):
values = []
for option in options:
if isinstance(option, tuple) and len(option) == 2:
values.append(option[1])
else:
values.append(option)
return values
def _step_dropdown(self, dropdown, direction, wrap=False):
options = self._option_values(dropdown.options)
values = [opt for opt in options if opt is not None]
if not values:
return
current = dropdown.value
if current not in values:
if wrap and direction < 0:
dropdown.value = values[-1]
else:
dropdown.value = values[0]
return
idx = values.index(current)
if wrap:
next_idx = (idx + direction) % len(values)
else:
next_idx = max(0, min(len(values) - 1, idx + direction))
if next_idx != idx:
dropdown.value = values[next_idx]
def _update_jump_nav_buttons(self):
patient_values = [
opt
for opt in self._option_values(self.patient_dropdown.options)
if opt is not None
]
date_values = [
opt
for opt in self._option_values(self.date_dropdown.options)
if opt is not None
]
self.prev_patient_button.disabled = len(patient_values) <= 1
self.next_patient_button.disabled = len(patient_values) <= 1
if len(date_values) <= 1:
self.prev_date_button.disabled = True
self.next_date_button.disabled = True
return
current_date = self.date_dropdown.value
if current_date not in date_values:
self.prev_date_button.disabled = True
self.next_date_button.disabled = True
return
current_idx = date_values.index(current_date)
self.prev_date_button.disabled = current_idx == 0
self.next_date_button.disabled = current_idx == (len(date_values) - 1)
def _build_options_for_column(self, column, formatter, frame=None):
if column is None:
return [("N/A", None)]
source = self.df if frame is None else frame
seen = set()
options = []
for _, row in source.iterrows():
value = formatter(row.get(column))
if value == "":
continue
if value in seen:
continue
seen.add(value)
options.append((value, value))
if not options:
return [("N/A", None)]
return options
def _filter_frame_for_jump(self, patient_value=None, date_value=None):
frame = self.df
if self.patient_col and patient_value is not None:
frame = frame[
frame[self.patient_col].apply(
lambda v: self._format_value(v) == patient_value
)
]
if self.date_col and date_value is not None:
frame = frame[
frame[self.date_col].apply(lambda v: self._format_date(v) == date_value)
]
return frame
def _build_patient_options(self):
return self._build_options_for_column(self.patient_col, self._format_value)
def _build_date_options(self, patient_value):
if self.date_col is None:
return [("N/A", None)]
frame = self._filter_frame_for_jump(patient_value=patient_value)
return self._build_options_for_column(
self.date_col, self._format_date, frame=frame
)
def _build_phase_options(self, patient_value, date_value):
if self.phase_col is None:
return [("N/A", None)]
frame = self._filter_frame_for_jump(
patient_value=patient_value,
date_value=date_value,
)
return self._build_options_for_column(
self.phase_col, self._format_value, frame=frame
)
def _set_dropdown_options(self, dropdown, options, preferred=None, disabled=False):
values = self._option_values(options)
dropdown.options = options
if preferred in values:
dropdown.value = preferred
elif values:
dropdown.value = values[0]
else:
dropdown.value = None
dropdown.disabled = disabled or (len(values) == 1 and values[0] is None)
def _refresh_jump_dropdowns(self, *, use_current_row=False):
if not hasattr(self, "patient_dropdown"):
return
row = self.df.iloc[self.current_index]
patient_pref = (
self._format_value(row.get(self.patient_col))
if use_current_row and self.patient_col is not None
else getattr(self.patient_dropdown, "value", None)
)
date_pref = (
self._format_date(row.get(self.date_col))
if use_current_row and self.date_col is not None
else getattr(self.date_dropdown, "value", None)
)
phase_pref = (
self._format_value(row.get(self.phase_col))
if use_current_row and self.phase_col is not None
else getattr(self.phase_dropdown, "value", None)
)
self._suspend_jump = True
try:
patient_options = self._build_patient_options()
self._set_dropdown_options(
self.patient_dropdown,
patient_options,
preferred=patient_pref,
disabled=self.patient_col is None,
)
date_options = self._build_date_options(self.patient_dropdown.value)
self._set_dropdown_options(
self.date_dropdown,
date_options,
preferred=date_pref,
disabled=self.date_col is None,
)
phase_options = self._build_phase_options(
self.patient_dropdown.value,
self.date_dropdown.value,
)
self._set_dropdown_options(
self.phase_dropdown,
phase_options,
preferred=phase_pref,
disabled=self.phase_col is None,
)
finally:
self._suspend_jump = False
self._update_jump_nav_buttons()
def _jump_to_selected_filters(self):
if len(self.df) == 0:
return
patient_value = self.patient_dropdown.value if self.patient_col else None
date_value = self.date_dropdown.value if self.date_col else None
phase_value = self.phase_dropdown.value if self.phase_col else None
for pos in range(len(self.df)):
row = self.df.iloc[pos]
if self.patient_col and patient_value is not None:
if self._format_value(row.get(self.patient_col)) != patient_value:
continue
if self.date_col and date_value is not None:
if self._format_date(row.get(self.date_col)) != date_value:
continue
if self.phase_col and phase_value is not None:
if self._format_value(row.get(self.phase_col)) != phase_value:
continue
if int(pos) == int(self.current_index):
return
self.current_index = int(pos)
self.load_data()
return
def _format_date(self, value):
if value is None:
return "?"
try:
if pd.isna(value):
return "?"
except Exception:
pass
if isinstance(value, pd.Timestamp):
return value.strftime("%Y-%m-%d")
try:
dt = pd.to_datetime(value, errors="coerce")
if pd.isna(dt):
return str(value)
return dt.strftime("%Y-%m-%d")
except Exception:
return str(value)
def _format_value(self, value):
if value is None:
return ""
try:
if pd.isna(value):
return ""
except Exception:
pass
if isinstance(value, pd.Timestamp):
return value.strftime("%Y-%m-%d")
if isinstance(value, (np.integer, np.floating)):
value = value.item()
if isinstance(value, float):
return f"{value:.3f}"
return str(value)
def _is_empty_value(self, value):
if value is None:
return True
try:
if pd.isna(value):
return True
except Exception:
pass
if isinstance(value, (list, tuple, dict)) and len(value) == 0:
return True
return False
def _get_selected_segmentation(self):
if not self.segmentation_cols:
return None
seg_name = self.center_seg_dropdown.value
if seg_name in self.segmentations:
return self.segmentations[seg_name]
return None
def _compute_center_slice(self, seg):
if seg is None:
return None
if self.view_plane == "axial":
sums = np.sum(seg, axis=(0, 1))
elif self.view_plane == "sagittal":
sums = np.sum(seg, axis=(1, 2))
else:
sums = np.sum(seg, axis=(0, 2))
if sums.size == 0:
return None
return int(np.argmax(sums))
def _set_jump_value(self):
self._refresh_jump_dropdowns(use_current_row=True)
def _try_enable_widget_backend(self):
"""Best-effort switch to ipympl when available."""
backend = (plt.get_backend() or "").lower()
if "ipympl" in backend or "nbagg" in backend:
return
try:
import ipympl # noqa: F401
plt.switch_backend("module://ipympl.backend_nbagg")
except Exception:
# Keep current backend; output fallback will still be interactive.
return
def _render_output_figure(self):
if not self._uses_output_fallback:
return
if not isinstance(self.display_widget, widgets.Output):
return
with self.display_widget:
clear_output(wait=True)
display(self.fig)
def _pin_axes_to_canvas(self):
if self.fig is None or self.ax is None:
return
self.fig.subplots_adjust(left=0.0, right=1.0, bottom=0.0, top=1.0)
self.ax.set_position([0.0, 0.0, 1.0, 1.0])
self.ax.set_aspect(self.image_aspect, adjustable="box")
self.ax.margins(0)
[docs]
def on_window_preset_change(self, change):
preset = change["new"]
if preset == "Custom":
return
wl, ww = WINDOW_PRESETS[preset]
self.HU_min = wl - ww / 2.0
self.HU_max = wl + ww / 2.0
self.update_display()
[docs]
def on_resolution_change(self, change):
self.isotropic_resolution_mm = validate_isotropic_resolution(change["new"])
self.load_data()
[docs]
def on_patient_change(self, change):
if self._suspend_jump:
return
self._refresh_jump_dropdowns(use_current_row=False)
self._jump_to_selected_filters()
[docs]
def on_date_change(self, change):
if self._suspend_jump:
return
self._refresh_jump_dropdowns(use_current_row=False)
self._jump_to_selected_filters()
[docs]
def on_phase_change(self, change):
if self._suspend_jump:
return
self._jump_to_selected_filters()
[docs]
def on_prev_patient(self, button):
self._step_dropdown(self.patient_dropdown, -1, wrap=True)
[docs]
def on_next_patient(self, button):
self._step_dropdown(self.patient_dropdown, 1, wrap=True)
[docs]
def on_prev_date(self, button):
self._step_dropdown(self.date_dropdown, -1)
[docs]
def on_next_date(self, button):
self._step_dropdown(self.date_dropdown, 1)
[docs]
def on_seg_visibility_change(self, change):
self.update_display()
[docs]
def on_center_on_lesion(self, button):
seg = self._get_selected_segmentation()
if seg is None:
return
center_idx = self._compute_center_slice(seg)
if center_idx is not None:
self.slice_slider.value = int(center_idx)
[docs]
def on_key_press(self, event):
key = (event.key or "").lower()
if "shift+" in key:
if "left" in key or "up" in key:
self.on_prev(None)
elif "right" in key or "down" in key:
self.on_next(None)
return
if key in {"left", "up"}:
self.on_prev_slice(None)
elif key in {"right", "down"}:
self.on_next_slice_manual(None)
[docs]
def load_data(self):
self.progress_bar.layout.visibility = "visible"
self.progress_bar.value = 0
self.progress_bar.bar_style = "info"
self.progress_bar.description = "Loading..."
row = self.df.iloc[self.current_index]
self.progress_bar.value = 0.1
self.ct_scan_raw = load_nifti(
row[self.ct_scan_col],
isotropic_resolution_mm=self.isotropic_resolution_mm,
order=1,
)
self.segmentations = {}
if self.segmentation_cols:
for seg_col in self.segmentation_cols:
seg_path = row.get(seg_col, None)
if seg_path is None:
continue
if isinstance(seg_path, float) and np.isnan(seg_path):
continue
try:
self.segmentations[seg_col] = load_nifti(
seg_path,
isotropic_resolution_mm=self.isotropic_resolution_mm,
order=0,
)
except Exception as exc:
print(
f"Warning: failed to load segmentation {seg_col} "
f"for index {self.current_index}: {exc}"
)
self.progress_bar.value = 0.6
self.update_info_display()
self.update_slice_slider()
self._set_jump_value()
self.progress_bar.value = 1
self.progress_bar.bar_style = "success"
self.progress_bar.description = "Loaded"
time.sleep(0.5)
self.progress_bar.layout.visibility = "hidden"
[docs]
def update_slice_slider(self):
self.view_plane = self.plane_selector.value
if self.view_plane == "axial":
self.num_slices = self.ct_scan_raw.shape[2]
elif self.view_plane == "sagittal":
self.num_slices = self.ct_scan_raw.shape[0]
else:
self.num_slices = self.ct_scan_raw.shape[1]
center_idx = None
seg = self._get_selected_segmentation()
if seg is not None:
center_idx = self._compute_center_slice(seg)
if center_idx is None:
center_idx = self.num_slices // 2
self.slice_idx = int(center_idx)
self.slice_slider.unobserve(self.on_slice_change, names="value")
self.slice_slider.max = max(0, self.num_slices - 1)
self.slice_slider.value = min(self.slice_idx, self.slice_slider.max)
self.slice_slider.observe(self.on_slice_change, names="value")
self.update_display()
[docs]
def update_display(self, *_):
if self.ct_scan_raw is None or self.ax is None:
return
self.view_plane = self.plane_selector.value
slice_idx = int(self.slice_slider.value)
alpha = self.alpha_slider.value
if self.view_plane == "axial":
ct_slice = self.ct_scan_raw[:, :, slice_idx]
seg_slices = {
name: seg[:, :, slice_idx] for name, seg in self.segmentations.items()
}
elif self.view_plane == "sagittal":
ct_slice = self.ct_scan_raw[slice_idx, :, :]
seg_slices = {
name: seg[slice_idx, :, :] for name, seg in self.segmentations.items()
}
else:
ct_slice = self.ct_scan_raw[:, slice_idx, :]
seg_slices = {
name: seg[:, slice_idx, :] for name, seg in self.segmentations.items()
}
ct_slice = clip_hu_values(ct_slice, self.HU_min, self.HU_max)
self.ax.clear()
self._pin_axes_to_canvas()
self.ax.imshow(
ct_slice.T,
cmap="gray",
origin="lower",
aspect=self.image_aspect,
interpolation="nearest",
)
visible_names = []
for seg_name in self.segmentation_cols:
if seg_name not in seg_slices:
continue
cb = self.seg_visibility.get(seg_name)
if cb is not None and not cb.value:
continue
visible_names.append(seg_name)
seg_slice = seg_slices[seg_name]
cmap = self.seg_colormaps.get(seg_name, "jet")
contour_color = self.seg_contour_colors.get(seg_name, "red")
self.ax.imshow(
np.ma.masked_where(seg_slice == 0, seg_slice).T,
cmap=cmap,
alpha=alpha,
origin="lower",
aspect=self.image_aspect,
interpolation="nearest",
)
self.ax.contour(
seg_slice.T,
colors=contour_color,
linewidths=0.8,
alpha=min(1.0, alpha + 0.1),
origin="lower",
)
if visible_names:
handles = [
mpatches.Patch(
color=self.seg_contour_colors.get(name, "red"), label=name
)
for name in visible_names
]
self.ax.legend(
handles=handles,
loc="upper right",
fontsize="small",
framealpha=0.6,
)
self.ax.axis("off")
if self._uses_output_fallback:
self._render_output_figure()
else:
self.fig.canvas.draw_idle()
[docs]
def update_info_display(self):
row = self.df.iloc[self.current_index]
rows = []
for column in DICOM_TAGS_TO_DISPLAY:
if column not in row.index:
continue
value = row[column]
if self._is_empty_value(value):
continue
formatted = self._format_value(value)
if formatted == "":
continue
rows.append(
"<tr>"
f"<td style='word-wrap: break-word; overflow-wrap: anywhere;'><b>{column}</b></td>"
f"<td style='word-wrap: break-word; overflow-wrap: anywhere;'>{formatted}</td>"
"</tr>"
)
if rows:
html = (
"<table style='width: 100%; table-layout: fixed; border-collapse: collapse;'>"
+ "".join(rows)
+ "</table>"
)
else:
html = "<i>No metadata</i>"
self.info_display.value = html
[docs]
def on_slice_change(self, change):
self.slice_idx = self.slice_slider.value
self.update_display()
[docs]
def on_plane_change(self, change):
self.view_plane = self.plane_selector.value
self.update_slice_slider()
[docs]
def on_prev_slice(self, button):
new_val = max(0, self.slice_slider.value - SLICE_NAV_STEP)
self.slice_slider.value = new_val
[docs]
def on_next_slice_manual(self, button):
new_val = min(self.slice_slider.max, self.slice_slider.value + SLICE_NAV_STEP)
self.slice_slider.value = new_val
[docs]
def on_next(self, button):
if self.exploration_mode == "ordered":
self.current_index = (self.current_index + 1) % len(self.df)
else:
if self.history_index == len(self.explored_history) - 1:
unexplored = set(range(len(self.df))) - set(self.explored_history)
if unexplored:
new_index = np.random.choice(list(unexplored))
else:
new_index = np.random.choice(range(len(self.df)))
self.explored_history.append(new_index)
self.history_index += 1
self.current_index = new_index
else:
self.history_index += 1
self.current_index = self.explored_history[self.history_index]
self.load_data()
[docs]
def on_prev(self, button):
if self.exploration_mode == "ordered":
self.current_index = (self.current_index - 1) % len(self.df)
self.load_data()
else:
if self.history_index > 0:
self.history_index -= 1
self.current_index = self.explored_history[self.history_index]
self.load_data()
else:
print("Already at the first explored scan.")