import os, threading, time, sqlite3, webbrowser, random, cv2, json
import tkinter as tk
from tkinter import ttk
import tkinter.font as tkFont
from tkinter import filedialog
from tkinter import font
from queue import Queue
from tkinter import Label, Frame, Button
import numpy as np
import pandas as pd
from PIL import Image, ImageOps, ImageTk, ImageDraw, ImageFont, ImageEnhance
from concurrent.futures import ThreadPoolExecutor
from IPython.display import display, HTML
import imageio.v2 as imageio
from collections import deque
from skimage.filters import threshold_otsu
from skimage.exposure import rescale_intensity
from skimage.draw import polygon, line
from skimage.transform import resize
from skimage.morphology import dilation, disk
from skimage.segmentation import find_boundaries
from skimage.util import img_as_ubyte
from scipy.ndimage import binary_fill_holes, label, gaussian_filter
from tkinter import ttk, scrolledtext
from sklearn.model_selection import train_test_split
from sympy import root
from xgboost import XGBClassifier
from sklearn.metrics import classification_report, confusion_matrix
from screeninfo import get_monitors
import subprocess, platform
def _register_open_sans():
"""Register OpenSans with the system font config so tkinter renders it smoothly."""
try:
base_dir = os.path.dirname(__file__)
font_dir = os.path.join(base_dir, 'resources', 'font', 'open_sans', 'static')
system = platform.system()
if system == 'Linux':
fonts_dir = os.path.expanduser('~/.local/share/fonts')
os.makedirs(fonts_dir, exist_ok=True)
copied = 0
for ttf in ['OpenSans-Regular.ttf', 'OpenSans-Bold.ttf', 'OpenSans-Italic.ttf']:
src = os.path.join(font_dir, ttf)
dst = os.path.join(fonts_dir, ttf)
if os.path.exists(src) and not os.path.exists(dst):
try:
import shutil
shutil.copy2(src, dst)
copied += 1
except Exception as e:
print(f"Warning: Could not copy {ttf}: {e}")
if copied > 0:
print(f"Installed {copied} OpenSans font(s) to {fonts_dir}")
try:
subprocess.run(['fc-cache', '-f', fonts_dir], capture_output=True, timeout=10)
print("Font cache updated successfully")
except Exception as e:
print(f"Warning: Could not update font cache: {e}")
xresources = os.path.expanduser('~/.Xresources')
needs_update = True
if os.path.exists(xresources):
try:
with open(xresources, 'r') as f:
if 'Xft.antialias' in f.read():
needs_update = False
except Exception as e:
print(f"Warning: Could not read {xresources}: {e}")
if needs_update:
try:
with open(xresources, 'a') as f:
f.write('\nXft.antialias: 1\nXft.hinting: 1\nXft.hintstyle: hintslight\nXft.rgba: rgb\nXft.lcdfilter: lcddefault\n')
subprocess.run(['xrdb', '-merge', xresources], capture_output=True, timeout=5)
print("Xft anti-aliasing configured successfully")
except Exception as e:
print(f"Warning: Could not configure Xft anti-aliasing: {e}")
else:
print("Xft anti-aliasing already configured")
elif system == 'Windows':
try:
import ctypes
gdi32 = ctypes.windll.gdi32
loaded = 0
for ttf in ['OpenSans-Regular.ttf', 'OpenSans-Bold.ttf']:
path = os.path.join(font_dir, ttf)
if os.path.exists(path):
if gdi32.AddFontResourceW(path):
loaded += 1
print(f"Loaded {loaded} OpenSans font(s) into Windows GDI")
except Exception as e:
print(f"Warning: Could not register fonts on Windows: {e}")
elif system == 'Darwin':
fonts_dir = os.path.expanduser('~/Library/Fonts')
os.makedirs(fonts_dir, exist_ok=True)
copied = 0
for ttf in ['OpenSans-Regular.ttf', 'OpenSans-Bold.ttf', 'OpenSans-Italic.ttf']:
src = os.path.join(font_dir, ttf)
dst = os.path.join(fonts_dir, ttf)
if os.path.exists(src) and not os.path.exists(dst):
try:
import shutil
shutil.copy2(src, dst)
copied += 1
except Exception as e:
print(f"Warning: Could not copy {ttf}: {e}")
if copied > 0:
print(f"Installed {copied} OpenSans font(s) to {fonts_dir}")
else:
print("OpenSans fonts already installed on macOS")
else:
print(f"Warning: Font registration not implemented for {system}")
except Exception as e:
print(f"Warning: Font registration failed: {e}")
try:
_register_open_sans()
except Exception as e:
print(f"Warning: Could not register OpenSans fonts: {e}")
_register_open_sans()
[docs]
def restart_gui_app(root):
"""Restart the GUI by destroying ``root`` and launching a fresh instance.
:param root: the Tk root window to tear down before relaunching.
"""
try:
# Destroy the current root window
root.destroy()
# Import and launch a new instance of the application
from .gui import gui_app
new_root = tk.Tk() # Create a fresh Tkinter root instance
gui_app()
except Exception as e:
print(f"Error restarting GUI application: {e}")
[docs]
def set_element_size():
"""Return cached default sizes for GUI elements derived from screen dimensions.
:returns: dict with ``btn_size``, ``bar_size``, ``settings_width``,
``panel_width``, and ``panel_height`` in pixels.
"""
global _cached_element_size
if _cached_element_size is not None:
return _cached_element_size
m = get_monitors()[0]
screen_width, screen_height = m.width, m.height
screen_area = screen_width * screen_height
# Calculate sizes based on screen dimensions
btn_size = int((screen_area * 0.002) ** 0.5) # Button size as a fraction of screen area
bar_size = screen_height // 20 # Bar size based on screen height
settings_width = screen_width // 4 # Settings panel width as a fraction of screen width
panel_width = screen_width - settings_width # Panel width as a fraction of screen width
panel_height = screen_height // 6 # Panel height as a fraction of screen height
size_dict = {
'btn_size': btn_size,
'bar_size': bar_size,
'settings_width': settings_width,
'panel_width': panel_width,
'panel_height': panel_height
}
_cached_element_size = size_dict
return size_dict
_cached_dark_style = None
_cached_element_size = None
[docs]
def set_dark_style(style, parent_frame=None, containers=None, widgets=None,
font_family="OpenSans", font_size=12, bg_color='black',
fg_color='white', active_color='blue', inactive_color='dark_gray'):
"""Configure ttk/tk widgets with the spacr dark theme and return the palette.
Named colors (``'black'``, ``'white'``, ``'blue'``, ``'dark_gray'``,
``'teal'``) resolve to their hex equivalents; explicit hex strings pass
through. When ``parent_frame``, ``containers``, or ``widgets`` are given,
those widgets are re-styled in place.
:param style: a ``ttk.Style`` instance to configure.
:param parent_frame: optional root frame to color-match.
:param containers: optional iterable of frames to restyle.
:param widgets: optional iterable of widgets to restyle.
:param font_family: font family name; ``'OpenSans'`` loads via ``spacrFont``.
:param font_size: base font size in points.
:param bg_color: primary background color.
:param fg_color: primary text color.
:param active_color: accent color for active/pressed states.
:param inactive_color: secondary/panel color.
:returns: dict of resolved style values (colors, fonts, spacing).
"""
global _cached_dark_style
# If no side effects needed, return cache
if parent_frame is None and containers is None and widgets is None and _cached_dark_style is not None:
return _cached_dark_style
# ------------------------------------------------------------------
# Pure-black palette (user preference — cleaner than the soft-dark
# GitHub-inspired scheme this originally landed with). Named-color
# aliases resolve to these values; explicit hex passed by the caller
# is preserved.
# ------------------------------------------------------------------
if active_color == 'teal':
active_color = '#008080'
if inactive_color == 'dark_gray':
inactive_color = '#2B2B2B' # secondary surface / subtle panel
if bg_color == 'black':
bg_color = '#000000' # primary background — clean black
if fg_color == 'white':
fg_color = '#ffffff' # primary text
if active_color == 'blue':
active_color = '#007BFF' # accent
# Extended palette exposed on the returned style dict.
border_color = '#2B2B2B' # subtle dividers (matches inactive)
muted_color = '#8b949e' # secondary text / hints
success_color = '#3fb950'
warning_color = '#d29922'
error_color = '#f85149'
# 4/8-based spacing scale, exported on the returned style dict so all
# custom widgets/screens can pull from a single source of truth.
spacing = {'xs': 4, 'sm': 8, 'md': 12, 'lg': 16, 'xl': 24}
# Internal widget padding: 'left top right bottom' -> comfortable 8x6.
padding = f"{spacing['sm']} {spacing['xs'] + 2} {spacing['sm']} {spacing['xs'] + 2}"
# Font size hierarchy — pick one from style_out['font_sizes'] rather than
# hard-coding numbers scattered across the GUI.
font_sizes = {
'small': max(font_size - 1, 9),
'body': font_size,
'header': font_size + 2,
'title': font_size + 6,
}
font_style = tkFont.Font(family=font_family, size=font_size)
if font_family == 'OpenSans':
font_loader = spacrFont(font_name='OpenSans', font_style='Regular', font_size=12)
else:
font_loader = None
style.theme_use('clam')
style.configure('TEntry', padding=padding)
style.configure('TCombobox', padding=padding)
style.configure('Spacr.TEntry', padding=padding)
style.configure('TEntry', padding=padding)
style.configure('Spacr.TEntry', padding=padding)
style.configure('Custom.TLabel', padding=padding)
style.configure('TButton', padding=padding)
style.configure('TFrame', background=bg_color)
style.configure('TPanedwindow', background=bg_color)
# ------------------------------------------------------------------
# Themed ttk.Scrollbar — hide the default grey-Windows/native look.
# A slim dark trough with a soft-blue thumb on hover.
# ------------------------------------------------------------------
style.configure(
'Spacr.Vertical.TScrollbar',
background=inactive_color, # thumb (default)
troughcolor=bg_color,
bordercolor=bg_color,
arrowcolor=fg_color,
gripcount=0,
relief='flat',
)
style.map(
'Spacr.Vertical.TScrollbar',
background=[('active', active_color), ('pressed', active_color)],
arrowcolor=[('disabled', border_color)],
)
style.configure(
'Spacr.Horizontal.TScrollbar',
background=inactive_color,
troughcolor=bg_color,
bordercolor=bg_color,
arrowcolor=fg_color,
gripcount=0,
relief='flat',
)
style.map(
'Spacr.Horizontal.TScrollbar',
background=[('active', active_color), ('pressed', active_color)],
arrowcolor=[('disabled', border_color)],
)
# Also apply to the default TScrollbar so ScrolledText / ScrolledFrame
# pick it up without needing style= arguments everywhere.
style.configure('Vertical.TScrollbar',
background=inactive_color, troughcolor=bg_color,
bordercolor=bg_color, arrowcolor=fg_color, relief='flat')
style.map('Vertical.TScrollbar',
background=[('active', active_color), ('pressed', active_color)])
style.configure('Horizontal.TScrollbar',
background=inactive_color, troughcolor=bg_color,
bordercolor=bg_color, arrowcolor=fg_color, relief='flat')
style.map('Horizontal.TScrollbar',
background=[('active', active_color), ('pressed', active_color)])
# ------------------------------------------------------------------
# Themed ttk.Progressbar — dark trough, accent-colored bar.
# ------------------------------------------------------------------
style.configure(
'Spacr.Horizontal.TProgressbar',
troughcolor=inactive_color,
background=active_color,
bordercolor=bg_color,
lightcolor=active_color,
darkcolor=active_color,
thickness=8,
)
style.configure(
'Horizontal.TProgressbar',
troughcolor=inactive_color,
background=active_color,
bordercolor=bg_color,
lightcolor=active_color,
darkcolor=active_color,
)
if font_loader:
style.configure('TLabel', background=bg_color, foreground=fg_color, font=font_loader.get_font(size=font_size))
else:
style.configure('TLabel', background=bg_color, foreground=fg_color, font=(font_family, font_size))
if parent_frame:
parent_frame.configure(bg=bg_color)
parent_frame.grid_rowconfigure(0, weight=1)
parent_frame.grid_columnconfigure(0, weight=1)
if containers:
for container in containers:
if isinstance(container, ttk.Frame):
container_style = ttk.Style()
container_style.configure(f'{container.winfo_class()}.TFrame', background=bg_color)
container.configure(style=f'{container.winfo_class()}.TFrame')
else:
container.configure(bg=bg_color)
if widgets:
for widget in widgets:
if isinstance(widget, (tk.Label, tk.Button, tk.Frame, ttk.LabelFrame, tk.Canvas)):
widget.configure(bg=bg_color)
if isinstance(widget, (tk.Label, tk.Button)):
if font_loader:
widget.configure(fg=fg_color, font=font_loader.get_font(size=font_size))
else:
widget.configure(fg=fg_color, font=(font_family, font_size))
if isinstance(widget, scrolledtext.ScrolledText):
widget.configure(bg=bg_color, fg=fg_color, insertbackground=fg_color)
if isinstance(widget, tk.OptionMenu):
if font_loader:
widget.configure(bg=bg_color, fg=fg_color, font=font_loader.get_font(size=font_size))
else:
widget.configure(bg=bg_color, fg=fg_color, font=(font_family, font_size))
menu = widget['menu']
if font_loader:
menu.configure(bg=bg_color, fg=fg_color, font=font_loader.get_font(size=font_size))
else:
menu.configure(bg=bg_color, fg=fg_color, font=(font_family, font_size))
#return {'font_loader':font_loader, 'font_family': font_family, 'font_size': font_size, 'bg_color': bg_color, 'fg_color': fg_color, 'active_color': active_color, 'inactive_color': inactive_color}
result = {'font_loader': font_loader, 'font_family': font_family,
'font_size': font_size, 'font_sizes': font_sizes,
'bg_color': bg_color, 'fg_color': fg_color,
'active_color': active_color, 'inactive_color': inactive_color,
'border_color': border_color, 'muted_color': muted_color,
'success_color': success_color, 'warning_color': warning_color,
'error_color': error_color, 'spacing': spacing}
if parent_frame is None and containers is None and widgets is None:
_cached_dark_style = result
return result
_font_cache = {}
[docs]
class spacrFont:
"""Loader that resolves a bundled ``.ttf`` and registers it with Tk.
:param font_name: font family name (e.g. ``'OpenSans'``).
:param font_style: font style variant (e.g. ``'Regular'``, ``'Bold'``).
:param font_size: default size in points.
"""
def __init__(self, font_name, font_style, font_size=12):
"""Resolve the font file and register the family with Tk.
:param font_name: font family name.
:param font_style: font style variant.
:param font_size: default point size.
"""
[docs]
self.font_name = font_name
[docs]
self.font_style = font_style
[docs]
self.font_size = font_size
# Determine the path based on the font name and style
[docs]
self.font_path = self.get_font_path(font_name, font_style)
# Register the font with Tkinter
self.load_font()
[docs]
def get_font_path(self, font_name, font_style):
"""Return the on-disk path to the ``.ttf`` for a given family + style.
:param font_name: font family name.
:param font_style: font style variant.
:returns: absolute path to the font file.
:raises ValueError: if the combination is not bundled.
"""
base_dir = os.path.dirname(__file__)
if font_name == 'OpenSans':
if font_style == 'Regular':
return os.path.join(base_dir, 'resources/font/open_sans/static/OpenSans-Regular.ttf')
elif font_style == 'Bold':
return os.path.join(base_dir, 'resources/font/open_sans/static/OpenSans-Bold.ttf')
elif font_style == 'Italic':
return os.path.join(base_dir, 'resources/font/open_sans/static/OpenSans-Italic.ttf')
# Add more styles as needed
# Add more fonts as needed
raise ValueError(f"Font '{font_name}' with style '{font_style}' not found.")
[docs]
def load_font(self):
"""Register the resolved font file with Tkinter's font system."""
try:
font.Font(family=self.font_name, size=self.font_size)
except tk.TclError:
# Load the font manually if it's not already loaded
self.tk_font = font.Font(
name=self.font_name,
file=self.font_path,
size=self.font_size
)
[docs]
def get_font(self, size=None):
"""Return a ``tkFont.Font`` for this family at the requested size.
:param size: point size; defaults to the size given at construction.
:returns: ``tkFont.Font`` instance.
"""
if size is None:
size = self.font_size
return font.Font(family=self.font_name, size=size)
[docs]
class spacrContainer(tk.Frame):
"""Resizable multi-pane container with draggable sashes between panes.
:param parent: parent widget.
:param orient: ``tk.VERTICAL`` or ``tk.HORIZONTAL`` split direction.
:param bg: background color for panes and sashes.
"""
def __init__(self, parent, orient=tk.VERTICAL, bg=None, *args, **kwargs):
"""Initialize the container and bind resize behavior.
:param parent: parent widget.
:param orient: split direction.
:param bg: pane/sash background color.
"""
super().__init__(parent, *args, **kwargs)
[docs]
self.bg = bg if bg else 'lightgrey'
[docs]
self.sash_thickness = 10
self.bind("<Configure>", self.on_configure)
self.grid_rowconfigure(0, weight=1)
self.grid_columnconfigure(0, weight=1)
[docs]
def add(self, widget, stretch='always'):
"""Insert ``widget`` as a new pane and repartition the layout.
:param widget: widget to embed as a pane.
:param stretch: stretch policy (currently informational).
"""
print(f"Adding widget: {widget} with stretch: {stretch}")
pane = tk.Frame(self, bg=self.bg)
pane.grid_propagate(False)
widget.grid(in_=pane, sticky="nsew") # Use grid for the widget within the pane
self.panes.append((pane, widget))
if len(self.panes) > 1:
self.create_sash()
self.reposition_panes()
[docs]
def create_sash(self):
"""Create and register a new draggable sash between panes."""
sash = tk.Frame(self, bg=self.bg, cursor='sb_v_double_arrow' if self.orient == tk.VERTICAL else 'sb_h_double_arrow', height=self.sash_thickness, width=self.sash_thickness)
sash.bind("<Enter>", self.on_enter_sash)
sash.bind("<Leave>", self.on_leave_sash)
sash.bind("<ButtonPress-1>", self.start_resize)
self.sashes.append(sash)
[docs]
def reposition_panes(self):
"""Re-grid panes and sashes to fill the current container size."""
if not self.panes:
return
total_size = self.winfo_height() if self.orient == tk.VERTICAL else self.winfo_width()
pane_size = total_size // len(self.panes)
print(f"Total size: {total_size}, Pane size: {pane_size}, Number of panes: {len(self.panes)}")
for i, (pane, widget) in enumerate(self.panes):
if self.orient == tk.VERTICAL:
pane.grid(row=i * 2, column=0, sticky="nsew", pady=(0, self.sash_thickness if i < len(self.panes) - 1 else 0))
else:
pane.grid(row=0, column=i * 2, sticky="nsew", padx=(0, self.sash_thickness if i < len(self.panes) - 1 else 0))
for i, sash in enumerate(self.sashes):
if self.orient == tk.VERTICAL:
sash.grid(row=(i * 2) + 1, column=0, sticky="ew")
else:
sash.grid(row=0, column=(i * 2) + 1, sticky="ns")
[docs]
def on_enter_sash(self, event):
"""Highlight a sash on mouse enter."""
event.widget.config(bg='blue')
[docs]
def on_leave_sash(self, event):
"""Restore sash color on mouse leave."""
event.widget.config(bg=self.bg)
[docs]
def start_resize(self, event):
"""Begin a drag-resize gesture on the sash under the pointer."""
sash = event.widget
self.start_pos = event.y_root if self.orient == tk.VERTICAL else event.x_root
self.start_size = sash.winfo_y() if self.orient == tk.VERTICAL else sash.winfo_x()
sash.bind("<B1-Motion>", self.perform_resize)
[docs]
class spacrEntry(tk.Frame):
"""Pill-shaped themed entry with a focus ring, backed by a ``tk.Entry``.
:param parent: parent widget.
:param textvariable: ``tk.StringVar`` bound to the entry text.
:param outline: reserved; enables a subtle outline stroke when True.
:param width: unused legacy parameter (canvas expands to fill).
"""
def __init__(self, parent, textvariable=None, outline=False, width=None, *args, **kwargs):
"""Build the canvas-drawn entry and bind focus handlers."""
super().__init__(parent, *args, **kwargs)
style_out = set_dark_style(ttk.Style())
[docs]
self.bg_color = style_out['inactive_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.fg_color = style_out['fg_color']
[docs]
self.outline = outline
[docs]
self.font_family = style_out['font_family']
[docs]
self.font_size = style_out['font_size']
[docs]
self.font_loader = style_out['font_loader']
self.configure(bg=style_out['bg_color'])
[docs]
self.canvas_height = 40
[docs]
self.canvas = tk.Canvas(self, height=self.canvas_height, bd=0, highlightthickness=0, relief='ridge', bg=style_out['bg_color'])
self.canvas.pack(fill=tk.X, expand=True)
if self.font_loader:
self.entry = tk.Entry(self, textvariable=textvariable, bd=0, highlightthickness=0, fg=self.fg_color, font=self.font_loader.get_font(size=self.font_size), bg=self.bg_color)
else:
self.entry = tk.Entry(self, textvariable=textvariable, bd=0, highlightthickness=0, fg=self.fg_color, font=(self.font_family, self.font_size), bg=self.bg_color)
self.entry.place(relx=0.5, rely=0.5, anchor=tk.CENTER, relwidth=0.9, height=20)
self.entry.bind("<FocusIn>", self.on_focus_in)
self.entry.bind("<FocusOut>", self.on_focus_out)
self.canvas.bind("<Configure>", self._on_resize)
self.draw_rounded_rectangle(self.bg_color)
def _on_resize(self, event):
"""Redraw the pill background when the canvas is resized."""
self.draw_rounded_rectangle(self.bg_color)
[docs]
def draw_rounded_rectangle(self, color, focus_ring=False):
"""Draw the pill-shaped background, optionally with an accent focus ring.
The interior always fills with ``color`` so the field stays legible;
``focus_ring=True`` adds a 2 px accent outline instead of recoloring
the interior.
:param color: fill color for the pill interior.
:param focus_ring: draw the accent-colored outline when True.
"""
radius = 15
x0, y0 = 5, 5
x1 = self.canvas.winfo_width() - 5
if x1 <= x0:
x1 = 200
y1 = self.canvas_height - 5
self.canvas.delete("all")
# Filled body (interior stays consistent regardless of focus).
self.canvas.create_arc((x0, y0, x0 + radius, y0 + radius), start=90, extent=90, fill=color, outline=color)
self.canvas.create_arc((x1 - radius, y0, x1, y0 + radius), start=0, extent=90, fill=color, outline=color)
self.canvas.create_arc((x0, y1 - radius, x0 + radius, y1), start=180, extent=90, fill=color, outline=color)
self.canvas.create_arc((x1 - radius, y1 - radius, x1, y1), start=270, extent=90, fill=color, outline=color)
self.canvas.create_rectangle((x0 + radius / 2, y0, x1 - radius / 2, y1), fill=color, outline=color)
self.canvas.create_rectangle((x0, y0 + radius / 2, x1, y1 - radius / 2), fill=color, outline=color)
if focus_ring:
# Draw an outline stroke ~1 px outside the fill for a subtle ring.
ring = self.active_color
width = 2
self.canvas.create_arc((x0 - 1, y0 - 1, x0 - 1 + radius, y0 - 1 + radius),
start=90, extent=90, style='arc',
outline=ring, width=width)
self.canvas.create_arc((x1 + 1 - radius, y0 - 1, x1 + 1, y0 - 1 + radius),
start=0, extent=90, style='arc',
outline=ring, width=width)
self.canvas.create_arc((x0 - 1, y1 + 1 - radius, x0 - 1 + radius, y1 + 1),
start=180, extent=90, style='arc',
outline=ring, width=width)
self.canvas.create_arc((x1 + 1 - radius, y1 + 1 - radius, x1 + 1, y1 + 1),
start=270, extent=90, style='arc',
outline=ring, width=width)
# Straight edges of the ring.
self.canvas.create_line(x0 + radius / 2, y0 - 1, x1 - radius / 2, y0 - 1,
fill=ring, width=width)
self.canvas.create_line(x0 + radius / 2, y1 + 1, x1 - radius / 2, y1 + 1,
fill=ring, width=width)
self.canvas.create_line(x0 - 1, y0 + radius / 2, x0 - 1, y1 - radius / 2,
fill=ring, width=width)
self.canvas.create_line(x1 + 1, y0 + radius / 2, x1 + 1, y1 - radius / 2,
fill=ring, width=width)
[docs]
def on_focus_in(self, event):
"""Show the accent focus ring when the entry gains keyboard focus."""
# Interior stays inactive_color; add an accent-colored focus ring
# around the outside instead of flooding the whole field.
self.draw_rounded_rectangle(self.bg_color, focus_ring=True)
[docs]
def on_focus_out(self, event):
"""Remove the focus ring when the entry loses keyboard focus."""
self.draw_rounded_rectangle(self.bg_color)
self.entry.config(bg=self.bg_color)
[docs]
class spacrCheck(tk.Frame):
"""Themed rounded-square checkbox bound to a ``tk.BooleanVar``.
:param parent: parent widget.
:param text: unused caption (reserved).
:param variable: ``tk.BooleanVar`` whose value drives the check state.
"""
def __init__(self, parent, text="", variable=None, *args, **kwargs):
"""Build the canvas-drawn checkbox and wire it to ``variable``."""
super().__init__(parent, *args, **kwargs)
style_out = set_dark_style(ttk.Style())
[docs]
self.bg_color = style_out['bg_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.fg_color = style_out['fg_color']
[docs]
self.inactive_color = style_out['inactive_color']
[docs]
self.variable = variable
self.configure(bg=self.bg_color)
# Create a canvas for the rounded square background
[docs]
self.canvas_height = 20
[docs]
self.canvas = tk.Canvas(self, width=self.canvas_width, height=self.canvas_height, bd=0, highlightthickness=0, relief='ridge', bg=self.bg_color)
self.canvas.pack()
# Draw the initial rounded square based on the variable's value
self.draw_rounded_square(self.active_color if self.variable.get() else self.inactive_color)
# Bind variable changes to update the checkbox
self.variable.trace_add('write', self.update_check)
# Bind click event to toggle the variable
self.canvas.bind("<Button-1>", self.toggle_variable)
[docs]
def draw_rounded_square(self, color):
"""Draw the checkbox square in the given fill color.
:param color: fill color reflecting the current check state.
"""
radius = 5 # Adjust the radius for more rounded corners
x0, y0 = 2, 2
x1, y1 = 18, 18
self.canvas.delete("all")
self.canvas.create_arc((x0, y0, x0 + radius, y0 + radius), start=90, extent=90, fill=color, outline=self.fg_color)
self.canvas.create_arc((x1 - radius, y0, x1, y0 + radius), start=0, extent=90, fill=color, outline=self.fg_color)
self.canvas.create_arc((x0, y1 - radius, x0 + radius, y1), start=180, extent=90, fill=color, outline=self.fg_color)
self.canvas.create_arc((x1 - radius, y1 - radius, x1, y1), start=270, extent=90, fill=color, outline=self.fg_color)
self.canvas.create_rectangle((x0 + radius / 2, y0, x1 - radius / 2, y1), fill=color, outline=color)
self.canvas.create_rectangle((x0, y0 + radius / 2, x1, y1 - radius / 2), fill=color, outline=color)
self.canvas.create_line(x0 + radius / 2, y0, x1 - radius / 2, y0, fill=self.fg_color)
self.canvas.create_line(x0 + radius / 2, y1, x1 - radius / 2, y1, fill=self.fg_color)
self.canvas.create_line(x0, y0 + radius / 2, x0, y1 - radius / 2, fill=self.fg_color)
self.canvas.create_line(x1, y0 + radius / 2, x1, y1 - radius / 2, fill=self.fg_color)
[docs]
def update_check(self, *args):
"""Redraw the checkbox when the bound variable changes."""
self.draw_rounded_square(self.active_color if self.variable.get() else self.inactive_color)
[docs]
def toggle_variable(self, event):
"""Flip the bound ``BooleanVar`` in response to a click."""
self.variable.set(not self.variable.get())
[docs]
class spacrCombo(tk.Frame):
"""Themed dropdown combobox backed by a Toplevel selection popup.
:param parent: parent widget.
:param textvariable: ``tk.StringVar`` bound to the selected value.
:param values: iterable of selectable values.
:param width: unused legacy parameter (canvas expands to fill).
"""
def __init__(self, parent, textvariable=None, values=None, width=None, *args, **kwargs):
"""Build the closed-state pill and prepare the dropdown popup."""
super().__init__(parent, *args, **kwargs)
style_out = set_dark_style(ttk.Style())
[docs]
self.bg_color = style_out['bg_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.fg_color = style_out['fg_color']
[docs]
self.inactive_color = style_out['inactive_color']
[docs]
self.font_family = style_out['font_family']
[docs]
self.font_size = style_out['font_size']
[docs]
self.font_loader = style_out['font_loader']
self.configure(bg=self.bg_color)
[docs]
self.values = values or []
[docs]
self.canvas_height = 40
[docs]
self.canvas = tk.Canvas(self, height=self.canvas_height, bd=0, highlightthickness=0, relief='ridge', bg=self.bg_color)
self.canvas.pack(fill=tk.X, expand=True)
[docs]
self.var = textvariable if textvariable else tk.StringVar()
[docs]
self.selected_value = self.var.get()
if self.font_loader:
self.label = tk.Label(self, text=self.selected_value, bg=self.inactive_color, fg=self.fg_color, font=self.font_loader.get_font(size=self.font_size))
else:
self.label = tk.Label(self, text=self.selected_value, bg=self.inactive_color, fg=self.fg_color, font=(self.font_family, self.font_size))
self.label.place(relx=0.5, rely=0.5, anchor=tk.CENTER)
self.canvas.bind("<Button-1>", self.on_click)
self.label.bind("<Button-1>", self.on_click)
self.canvas.bind("<Configure>", self._on_resize)
self.draw_rounded_rectangle(self.inactive_color)
def _on_resize(self, event):
"""Redraw the combobox background when the canvas is resized."""
# Keep interior consistent; add focus ring only when open.
self.draw_rounded_rectangle(self.inactive_color,
focus_ring=(self.dropdown_menu is not None))
[docs]
def draw_rounded_rectangle(self, color, focus_ring=False):
"""Draw the pill-shaped background, optionally with an accent focus ring.
:param color: fill color for the pill interior.
:param focus_ring: draw the accent-colored outline when True.
"""
radius = 15
x0, y0 = 5, 5
x1 = self.canvas.winfo_width() - 5
if x1 <= x0:
x1 = 200
y1 = self.canvas_height - 5
self.canvas.delete("all")
self.canvas.create_arc((x0, y0, x0 + radius, y0 + radius), start=90, extent=90, fill=color, outline=color)
self.canvas.create_arc((x1 - radius, y0, x1, y0 + radius), start=0, extent=90, fill=color, outline=color)
self.canvas.create_arc((x0, y1 - radius, x0 + radius, y1), start=180, extent=90, fill=color, outline=color)
self.canvas.create_arc((x1 - radius, y1 - radius, x1, y1), start=270, extent=90, fill=color, outline=color)
self.canvas.create_rectangle((x0 + radius / 2, y0, x1 - radius / 2, y1), fill=color, outline=color)
self.canvas.create_rectangle((x0, y0 + radius / 2, x1, y1 - radius / 2), fill=color, outline=color)
self.label.config(bg=color)
if focus_ring:
ring = self.active_color
width = 2
self.canvas.create_arc((x0 - 1, y0 - 1, x0 - 1 + radius, y0 - 1 + radius),
start=90, extent=90, style='arc',
outline=ring, width=width)
self.canvas.create_arc((x1 + 1 - radius, y0 - 1, x1 + 1, y0 - 1 + radius),
start=0, extent=90, style='arc',
outline=ring, width=width)
self.canvas.create_arc((x0 - 1, y1 + 1 - radius, x0 - 1 + radius, y1 + 1),
start=180, extent=90, style='arc',
outline=ring, width=width)
self.canvas.create_arc((x1 + 1 - radius, y1 + 1 - radius, x1 + 1, y1 + 1),
start=270, extent=90, style='arc',
outline=ring, width=width)
self.canvas.create_line(x0 + radius / 2, y0 - 1, x1 - radius / 2, y0 - 1,
fill=ring, width=width)
self.canvas.create_line(x0 + radius / 2, y1 + 1, x1 - radius / 2, y1 + 1,
fill=ring, width=width)
self.canvas.create_line(x0 - 1, y0 + radius / 2, x0 - 1, y1 - radius / 2,
fill=ring, width=width)
self.canvas.create_line(x1 + 1, y0 + radius / 2, x1 + 1, y1 - radius / 2,
fill=ring, width=width)
[docs]
def on_click(self, event):
"""Toggle the dropdown popup open/closed on click."""
if self.dropdown_menu is None:
self.open_dropdown()
else:
self.close_dropdown()
[docs]
def open_dropdown(self):
"""Open the Toplevel popup showing selectable values."""
# Keep interior color; add focus ring while the dropdown is open.
self.draw_rounded_rectangle(self.inactive_color, focus_ring=True)
self.dropdown_menu = tk.Toplevel(self)
self.dropdown_menu.wm_overrideredirect(True)
self.dropdown_menu.configure(bg=self.inactive_color)
x, y, width, height = self.winfo_rootx(), self.winfo_rooty(), self.winfo_width(), self.winfo_height()
for index, value in enumerate(self.values):
display_text = value if value is not None else 'None'
if self.font_loader:
item = tk.Label(self.dropdown_menu, text=display_text, bg=self.inactive_color, fg=self.fg_color, font=self.font_loader.get_font(size=self.font_size), anchor='w')
else:
item = tk.Label(self.dropdown_menu, text=display_text, bg=self.inactive_color, fg=self.fg_color, font=(self.font_family, self.font_size), anchor='w')
item.pack(fill='both')
item.bind("<Button-1>", lambda e, v=value: self.on_select(v))
item.bind("<Enter>", lambda e, w=item: w.config(bg=self.active_color))
item.bind("<Leave>", lambda e, w=item: w.config(bg=self.inactive_color))
self.dropdown_menu.update_idletasks()
actual_height = self.dropdown_menu.winfo_reqheight()
self.dropdown_menu.geometry(f"{width}x{actual_height}+{x}+{y + height}")
[docs]
def close_dropdown(self):
"""Destroy the dropdown popup and remove the focus ring."""
self.draw_rounded_rectangle(self.inactive_color)
if self.dropdown_menu:
self.dropdown_menu.destroy()
self.dropdown_menu = None
[docs]
def on_select(self, value):
"""Commit ``value`` as the current selection and close the popup.
:param value: selected value from the popup list.
"""
display_text = value if value is not None else 'None'
self.var.set(value)
self.label.config(text=display_text)
self.selected_value = value
self.close_dropdown()
[docs]
def set(self, value):
"""Programmatically set the current selection without opening the popup.
:param value: value to display and store.
"""
display_text = value if value is not None else 'None'
self.var.set(value)
self.label.config(text=display_text)
self.selected_value = value
[docs]
class spacrProgressBar(ttk.Progressbar):
"""Themed ``ttk.Progressbar`` with an optional companion status label.
:param parent: parent widget.
:param label: when True, create a paired label showing progress text.
"""
def __init__(self, parent, label=True, *args, **kwargs):
"""Style the progress bar and (optionally) create the status label."""
super().__init__(parent, *args, **kwargs)
# Get the style colors
style_out = set_dark_style(ttk.Style())
[docs]
self.fg_color = style_out['fg_color']
[docs]
self.bg_color = style_out['bg_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.inactive_color = style_out['inactive_color']
[docs]
self.font_size = style_out['font_size']
[docs]
self.font_loader = style_out['font_loader']
# Configure the style for the progress bar
[docs]
self.style = ttk.Style()
# Remove any borders and ensure the active color fills the entire space
self.style.configure(
"spacr.Horizontal.TProgressbar",
troughcolor=self.inactive_color, # Set the trough to bg color
background=self.active_color, # Active part is the active color
borderwidth=0, # Remove border width
pbarrelief="flat", # Flat relief for the progress bar
troughrelief="flat", # Flat relief for the trough
thickness=20, # Set the thickness of the progress bar
darkcolor=self.active_color, # Ensure darkcolor matches the active color
lightcolor=self.active_color, # Ensure lightcolor matches the active color
bordercolor=self.bg_color # Set the border color to the background color to hide it
)
self.configure(style="spacr.Horizontal.TProgressbar")
# Set initial value to 0
self['value'] = 0
# Track whether to show the progress label
# Create the progress label with text wrapping
if self.label:
self.progress_label = tk.Label(
parent,
text="Processing: 0/0",
anchor='w',
justify='left',
bg=self.inactive_color,
fg=self.fg_color,
wraplength=300,
font=self.font_loader.get_font(size=self.font_size)
)
self.progress_label.grid_forget()
# Initialize attributes for time and operation
[docs]
self.operation_type = None
[docs]
self.additional_info = None
[docs]
def set_label_position(self):
"""Grid the status label directly beneath the progress bar."""
if self.label and self.progress_label:
row_info = self.grid_info().get('rowID', 0)
col_info = self.grid_info().get('columnID', 0)
col_span = self.grid_info().get('columnspan', 1)
self.progress_label.grid(row=row_info + 1, column=col_info, columnspan=col_span, pady=5, padx=5, sticky='ew')
[docs]
def update_label(self):
"""Refresh the label text from current value, operation, and info."""
if self.label and self.progress_label:
# Start with the base progress information
label_text = f"Processing: {self['value']}/{self['maximum']}"
# Include the operation type if it exists
if self.operation_type:
label_text += f", {self.operation_type}"
# Handle additional info without adding newlines
if hasattr(self, 'additional_info') and self.additional_info:
# Join all additional info items with a space and ensure they're on the same line
items = self.additional_info.split(", ")
formatted_additional_info = " ".join(items)
# Append the additional info to the label_text, ensuring it's all in one line
label_text += f" {formatted_additional_info.strip()}"
# Update the progress label
self.progress_label.config(text=label_text)
[docs]
class spacrSlider(tk.Frame):
"""Themed horizontal slider with a canvas-drawn knob and optional entry.
:param master: parent widget.
:param length: fixed pixel length; ``None`` for dynamic (90% of canvas).
:param thickness: line thickness for the slider track.
:param knob_radius: knob radius in pixels.
:param position: alignment when ``length`` is fixed
(``'left'``, ``'center'``, ``'right'``).
:param from_: minimum value.
:param to: maximum value.
:param value: initial value; defaults to ``from_``.
:param show_index: when True, show a companion ``tk.Entry`` for the value.
:param command: callback receiving the value on knob release.
"""
def __init__(self, master=None, length=None, thickness=2, knob_radius=10, position="center", from_=0, to=100, value=None, show_index=False, command=None, **kwargs):
"""Build the slider canvas, knob, and (optional) index entry."""
super().__init__(master, **kwargs)
[docs]
self.specified_length = length # Store the specified length, if any
[docs]
self.knob_radius = knob_radius
[docs]
self.thickness = thickness
[docs]
self.knob_position = knob_radius # Start at the beginning of the slider
[docs]
self.slider_line = None
[docs]
self.position = position.lower() # Store the position option
[docs]
self.offset = 0 # Initialize offset
[docs]
self.from_ = from_ # Minimum value of the slider
[docs]
self.to = to # Maximum value of the slider
[docs]
self.value = value if value is not None else from_ # Initial value of the slider
[docs]
self.show_index = show_index # Whether to show the index Entry widget
[docs]
self.command = command # Callback function to handle value changes
# Initialize the style and colors
style_out = set_dark_style(ttk.Style())
[docs]
self.fg_color = style_out['fg_color']
[docs]
self.bg_color = style_out['bg_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.inactive_color = style_out['inactive_color']
# Configure the frame's background color
self.configure(bg=self.bg_color)
# Create a frame for the slider and entry if needed
self.grid_columnconfigure(1, weight=1)
# Entry widget for showing and editing index, if enabled
if self.show_index:
self.index_var = tk.StringVar(value=str(int(self.value)))
self.index_entry = tk.Entry(self, textvariable=self.index_var, width=5, bg=self.bg_color, fg=self.fg_color, insertbackground=self.fg_color)
self.index_entry.grid(row=0, column=0, padx=5)
# Bind the entry to update the slider on change
self.index_entry.bind("<Return>", self.update_slider_from_entry)
# Create the slider canvas
[docs]
self.canvas = tk.Canvas(self, height=knob_radius * 2, bg=self.bg_color, highlightthickness=0)
self.canvas.grid(row=0, column=1, sticky="ew")
# Set initial length to specified length or default value
[docs]
self.length = self.specified_length if self.specified_length is not None else self.canvas.winfo_reqwidth()
# Calculate initial knob position based on the initial value
self.knob_position = self.value_to_position(self.value)
# Bind resize event to dynamically adjust the slider length if no length is specified
self.canvas.bind("<Configure>", self.resize_slider)
# Draw the slider components
self.draw_slider(inactive=True)
# Bind mouse events to the knob and slider
self.canvas.bind("<B1-Motion>", self.move_knob)
self.canvas.bind("<Button-1>", self.activate_knob) # Activate knob on click
self.canvas.bind("<ButtonRelease-1>", self.release_knob) # Trigger command on release
[docs]
def resize_slider(self, event):
"""Recompute slider length/offset when the canvas is resized."""
if self.specified_length is not None:
self.length = self.specified_length
else:
self.length = int(event.width * 0.9) # 90% of the container width
# Calculate the horizontal offset based on the position
if self.position == "center":
self.offset = (event.width - self.length) // 2
elif self.position == "right":
self.offset = event.width - self.length
else: # position is "left"
self.offset = 0
# Update the knob position after resizing
self.knob_position = self.value_to_position(self.value)
self.draw_slider(inactive=True)
[docs]
def value_to_position(self, value):
"""Map a slider value onto its knob position in canvas pixels.
:param value: value in ``[from_, to]``.
:returns: knob center x-coordinate.
"""
if self.to == self.from_:
return self.knob_radius
relative_value = (value - self.from_) / (self.to - self.from_)
return self.knob_radius + relative_value * (self.length - 2 * self.knob_radius)
[docs]
def position_to_value(self, position):
"""Map a knob position back to a slider value.
:param position: knob center x-coordinate in canvas pixels.
:returns: value in ``[from_, to]``.
"""
if self.to == self.from_:
return self.from_
relative_position = (position - self.knob_radius) / (self.length - 2 * self.knob_radius)
return self.from_ + relative_position * (self.to - self.from_)
[docs]
def draw_slider(self, inactive=False):
"""Redraw the slider track and knob.
:param inactive: when True, render the knob in the inactive color.
"""
self.canvas.delete("all")
self.slider_line = self.canvas.create_line(
self.offset + self.knob_radius,
self.knob_radius,
self.offset + self.length - self.knob_radius,
self.knob_radius,
fill=self.fg_color,
width=self.thickness
)
knob_color = self.inactive_color if inactive else self.active_color
self.knob = self.canvas.create_oval(
self.offset + self.knob_position - self.knob_radius,
self.knob_radius - self.knob_radius,
self.offset + self.knob_position + self.knob_radius,
self.knob_radius + self.knob_radius,
fill=knob_color,
outline=""
)
[docs]
def move_knob(self, event):
"""Move the knob to follow the pointer during a drag."""
new_position = min(max(event.x - self.offset, self.knob_radius), self.length - self.knob_radius)
self.knob_position = new_position
self.value = self.position_to_value(self.knob_position)
self.canvas.coords(
self.knob,
self.offset + self.knob_position - self.knob_radius,
self.knob_radius - self.knob_radius,
self.offset + self.knob_position + self.knob_radius,
self.knob_radius + self.knob_radius
)
if self.show_index:
self.index_var.set(str(int(self.value)))
[docs]
def activate_knob(self, event):
"""Switch the knob to active color and start following the pointer."""
self.draw_slider(inactive=False)
self.move_knob(event)
[docs]
def release_knob(self, event):
"""Deactivate the knob on button release and fire ``command``."""
self.draw_slider(inactive=True)
if self.command:
self.command(self.value) # Call the command with the final value when the knob is released
[docs]
def set_to(self, new_to):
"""Change the slider's maximum value and redraw the knob.
:param new_to: new upper bound.
"""
self.to = new_to
self.knob_position = self.value_to_position(self.value)
self.draw_slider(inactive=False)
[docs]
def get(self):
"""Return the current slider value."""
return self.value
[docs]
def set(self, value):
"""Set the slider's value and update the knob position.
:param value: value to display; clamped to ``[from_, to]``.
"""
self.value = max(self.from_, min(value, self.to)) # Ensure the value is within bounds
self.knob_position = self.value_to_position(self.value)
self.draw_slider(inactive=False)
if self.show_index:
self.index_var.set(str(int(self.value)))
[docs]
def jump_to_click(self, event):
"""Move the knob to the clicked position."""
self.activate_knob(event)
[docs]
def update_slider_from_entry(self, event):
"""Update the slider's value from the companion entry widget."""
try:
index = int(self.index_var.get())
self.set(index)
if self.command:
self.command(self.value)
except ValueError:
pass
[docs]
class spacrFrame(ttk.Frame):
"""Scrollable themed frame that hosts either a widget grid or a text box.
:param container: parent widget.
:param width: frame width in pixels; defaults to a quarter of screen width.
:param bg: background color.
:param radius: corner radius for the decorative rounded rectangle.
:param scrollbar: when True, attach a themed vertical scrollbar.
:param textbox: when True, use a ``tk.Text`` as the scrollable child
instead of a ``ttk.Frame``.
"""
def __init__(self, container, width=None, *args, bg='black', radius=20, scrollbar=True, textbox=False, **kwargs):
"""Build the scrollable canvas and expose ``self.scrollable_frame``."""
super().__init__(container, *args, **kwargs)
self.configure(style='TFrame')
if width is None:
screen_width = self.winfo_screenwidth()
width = screen_width // 4
# Create the canvas
canvas = tk.Canvas(self, bg=bg, width=width, highlightthickness=0)
self.rounded_rectangle(canvas, 0, 0, width, self.winfo_screenheight(), radius, fill=bg)
# Define scrollbar styles
style_out = set_dark_style(ttk.Style())
[docs]
self.inactive_color = style_out['inactive_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.fg_color = style_out['fg_color'] # Foreground color for text
# Set custom scrollbar style
style = ttk.Style()
spacrScrollbarStyle(style, self.inactive_color, self.active_color)
# Create scrollbar with custom style if scrollbar option is True
if scrollbar:
scrollbar_widget = ttk.Scrollbar(self, orient="vertical", command=canvas.yview, style='Custom.Vertical.TScrollbar')
if textbox:
self.scrollable_frame = tk.Text(canvas, bg=bg, fg=self.fg_color, wrap=tk.WORD)
else:
self.scrollable_frame = ttk.Frame(canvas, style='TFrame')
self.scrollable_frame.bind(
"<Configure>",
lambda e: canvas.configure(scrollregion=canvas.bbox("all"))
)
canvas.create_window((0, 0), window=self.scrollable_frame, anchor="nw")
if scrollbar:
canvas.configure(yscrollcommand=scrollbar_widget.set)
canvas.grid(row=0, column=0, sticky="nsew")
if scrollbar:
scrollbar_widget.grid(row=0, column=1, sticky="ns")
self.grid_rowconfigure(0, weight=1)
self.grid_columnconfigure(0, weight=1)
if scrollbar:
self.grid_columnconfigure(1, weight=0)
_ = set_dark_style(style, containers=[self], widgets=[canvas, self.scrollable_frame])
if scrollbar:
_ = set_dark_style(style, widgets=[scrollbar_widget])
[docs]
def rounded_rectangle(self, canvas, x1, y1, x2, y2, radius=20, **kwargs):
"""Draw a rounded rectangle on ``canvas`` and return its item id.
:param canvas: target ``tk.Canvas``.
:param x1: left edge.
:param y1: top edge.
:param x2: right edge.
:param y2: bottom edge.
:param radius: corner radius.
:returns: canvas item id.
"""
points = [
x1 + radius, y1,
x2 - radius, y1,
x2 - radius, y1,
x2, y1,
x2, y1 + radius,
x2, y2 - radius,
x2, y2 - radius,
x2, y2,
x2 - radius, y2,
x1 + radius, y2,
x1 + radius, y2,
x1, y2,
x1, y2 - radius,
x1, y2 - radius,
x1, y1 + radius,
x1, y1 + radius,
x1, y1
]
return canvas.create_polygon(points, **kwargs, smooth=True)
[docs]
class spacrLabel(tk.Frame):
"""Canvas-based themed text label supporting right or center alignment.
:param parent: parent widget.
:param text: label text.
:param font: fallback font when the shared font loader is unavailable.
:param style: optional ttk style name to use instead of canvas text.
:param align: ``'right'`` (default) or ``'center'``.
:param height: label height in pixels; defaults to a screen-derived size.
"""
def __init__(self, parent, text="", font=None, style=None, align="right", height=None, **kwargs):
"""Build the label canvas and render ``text``."""
valid_kwargs = {k: v for k, v in kwargs.items() if k not in ['foreground', 'background', 'font', 'anchor', 'justify', 'wraplength']}
super().__init__(parent, **valid_kwargs)
if height is None:
screen_height = self.winfo_screenheight()
label_height = screen_height // 50
label_width = label_height * 10
else:
label_height = height
label_width = label_height * 10
[docs]
self.style_out = set_dark_style(ttk.Style())
[docs]
self.font_style = self.style_out['font_family']
[docs]
self.font_size = self.style_out['font_size']
[docs]
self.font_family = self.style_out['font_family']
[docs]
self.font_loader = self.style_out['font_loader']
[docs]
self.canvas = tk.Canvas(self, width=label_width, height=label_height, highlightthickness=0, bg=self.style_out['bg_color'])
self.canvas.grid(row=0, column=0, sticky="ew")
if self.style_out['font_family'] != 'OpenSans':
self.font_style = font if font else tkFont.Font(family=self.style_out['font_family'], size=self.style_out['font_size'], weight=tkFont.NORMAL)
if self.align == "center":
anchor_value = tk.CENTER
text_anchor = 'center'
else: # default to right alignment
anchor_value = tk.E
text_anchor = 'e'
if self.style:
ttk_style = ttk.Style()
if self.font_loader:
ttk_style.configure(self.style, font=self.font_loader.get_font(size=self.font_size), background=self.style_out['bg_color'], foreground=self.style_out['fg_color'])
else:
ttk_style.configure(self.style, font=self.font_style, background=self.style_out['bg_color'], foreground=self.style_out['fg_color'])
self.label_text = ttk.Label(self.canvas, text=self.text, style=self.style, anchor=text_anchor)
self.label_text.pack(fill=tk.BOTH, expand=True)
else:
if self.font_loader:
self.label_text = self.canvas.create_text(label_width // 2 if self.align == "center" else label_width - 5,
label_height // 2, text=self.text, fill=self.style_out['fg_color'],
font=self.font_loader.get_font(size=self.font_size), anchor=anchor_value, justify=tk.RIGHT)
else:
self.label_text = self.canvas.create_text(label_width // 2 if self.align == "center" else label_width - 5,
label_height // 2, text=self.text, fill=self.style_out['fg_color'],
font=self.font_style, anchor=anchor_value, justify=tk.RIGHT)
_ = set_dark_style(ttk.Style(), containers=[self], widgets=[self.canvas])
[docs]
def set_text(self, text):
"""Replace the label's displayed text.
:param text: new text to render.
"""
if self.style:
self.label_text.config(text=text)
else:
self.canvas.itemconfig(self.label_text, text=text)
[docs]
class spacrSwitch(ttk.Frame):
"""Animated two-state toggle switch bound to a ``tk.BooleanVar``.
:param parent: parent widget.
:param text: caption shown next to the switch.
:param variable: ``tk.BooleanVar`` bound to the switch state.
:param command: callback fired after each toggle.
"""
def __init__(self, parent, text="", variable=None, command=None, *args, **kwargs):
"""Build the switch canvas and label, and wire click handlers."""
super().__init__(parent, *args, **kwargs)
[docs]
self.variable = variable if variable else tk.BooleanVar()
[docs]
self.canvas = tk.Canvas(self, width=40, height=20, highlightthickness=0, bd=0)
self.canvas.grid(row=0, column=1, padx=(10, 0))
[docs]
self.switch_bg = self.create_rounded_rectangle(2, 2, 38, 18, radius=9, outline="", fill="#fff")
[docs]
self.switch = self.canvas.create_oval(4, 4, 16, 16, outline="", fill="#800080")
[docs]
self.label = spacrLabel(self, text=self.text)
self.label.grid(row=0, column=0, padx=(0, 10))
self.bind("<Button-1>", self.toggle)
self.canvas.bind("<Button-1>", self.toggle)
self.label.bind("<Button-1>", self.toggle)
self.update_switch()
style = ttk.Style()
_ = set_dark_style(style, containers=[self], widgets=[self.canvas, self.label])
[docs]
def toggle(self, event=None):
"""Flip the bound variable, animate the knob, and fire ``command``."""
self.variable.set(not self.variable.get())
self.animate_switch()
if self.command:
self.command()
[docs]
def update_switch(self):
"""Redraw the switch knob to reflect the current bound value."""
if self.variable.get():
self.canvas.itemconfig(self.switch, fill="#008080")
self.canvas.coords(self.switch, 24, 4, 36, 16)
else:
self.canvas.itemconfig(self.switch, fill="#800080")
self.canvas.coords(self.switch, 4, 4, 16, 16)
[docs]
def animate_switch(self):
"""Animate the knob toward its new position and target color."""
if self.variable.get():
start_x, end_x = 4, 24
final_color = "#008080"
else:
start_x, end_x = 24, 4
final_color = "#800080"
self.animate_movement(start_x, end_x, final_color)
[docs]
def animate_movement(self, start_x, end_x, final_color):
"""Slide the knob from ``start_x`` to ``end_x`` and then set its color.
:param start_x: starting x-coordinate of the knob.
:param end_x: final x-coordinate of the knob.
:param final_color: fill color to apply after the animation.
"""
step = 1 if start_x < end_x else -1
for i in range(start_x, end_x, step):
self.canvas.coords(self.switch, i, 4, i + 12, 16)
self.canvas.update()
self.after(10)
self.canvas.itemconfig(self.switch, fill=final_color)
[docs]
def get(self):
"""Return the current switch state."""
return self.variable.get()
[docs]
def set(self, value):
"""Set the switch state without animation.
:param value: new boolean value.
"""
self.variable.set(value)
self.update_switch()
[docs]
def create_rounded_rectangle(self, x1, y1, x2, y2, radius=9, **kwargs):
"""Draw a rounded rectangle on the canvas and return its item id.
:param x1: left edge.
:param y1: top edge.
:param x2: right edge.
:param y2: bottom edge.
:param radius: corner radius.
:returns: canvas item id.
"""
points = [x1 + radius, y1,
x1 + radius, y1,
x2 - radius, y1,
x2 - radius, y1,
x2, y1,
x2, y1 + radius,
x2, y1 + radius,
x2, y2 - radius,
x2, y2 - radius,
x2, y2,
x2 - radius, y2,
x2 - radius, y2,
x1 + radius, y2,
x1 + radius, y2,
x1, y2,
x1, y2 - radius,
x1, y2 - radius,
x1, y1 + radius,
x1, y1 + radius,
x1, y1]
return self.canvas.create_polygon(points, **kwargs, smooth=True)
[docs]
class spacrCard(tk.Frame):
"""Themed container with optional title bar and consistent internal padding.
Produces a "lifted-panel" look that matches the soft dark palette. Add
child widgets to ``card.body`` — the outer frame reserves space for the
optional title bar, divider, and border.
:param parent: parent widget.
:param title: optional title bar text; empty string hides the title row.
:param padding: spacing key from the shared palette (``xs``/``sm``/``md``/
``lg``/``xl``).
:param show_border: when True, draw a 1 px lifted border around the card.
:ivar body: ``tk.Frame`` clients pack content into.
"""
def __init__(self, parent, title="", padding="md", show_border=False, **kwargs):
"""Build the outer border, optional title bar/divider, and body frame.
Defaults to ``show_border=False`` so on a pure-black background the
card blends in — only the title text and thin divider mark the
section. Pass ``show_border=True`` when stacked cards need visible
separation.
:param parent: parent widget.
:param title: title bar text.
:param padding: spacing key from the shared palette.
:param show_border: draw a 1 px lifted border when True.
"""
style_out = set_dark_style(ttk.Style())
[docs]
self.style_out = style_out
[docs]
self.bg_color = style_out['bg_color']
[docs]
self.border_color = style_out.get('border_color', style_out['inactive_color'])
[docs]
self.muted_color = style_out.get('muted_color', style_out['fg_color'])
[docs]
self.fg_color = style_out['fg_color']
spacing = style_out.get('spacing', {'sm': 8, 'md': 12, 'lg': 16})
font_loader = style_out.get('font_loader')
font_sizes = style_out.get('font_sizes', {'header': style_out['font_size'] + 2,
'body': style_out['font_size']})
pad = spacing.get(padding, spacing['md'])
highlight_thickness = 1 if show_border else 0
outer_bg = self.border_color if show_border else self.bg_color
super().__init__(parent, bg=outer_bg, bd=0,
highlightbackground=self.border_color,
highlightthickness=highlight_thickness, **kwargs)
# Interior always uses the bg_color so on the pure-black palette
# the card blends in; the optional 1-px border comes from the
# outer frame's highlightbackground.
interior = tk.Frame(self, bg=self.bg_color, bd=0)
interior.pack(fill=tk.BOTH, expand=True,
padx=1 if show_border else 0,
pady=1 if show_border else 0)
if title:
title_font = (
font_loader.get_font(size=font_sizes.get('header', style_out['font_size']))
if font_loader
else (style_out['font_family'], font_sizes.get('header', style_out['font_size']), "bold")
)
title_bar = tk.Frame(interior, bg=self.bg_color)
title_bar.pack(fill=tk.X, padx=pad, pady=(pad, spacing['xs']))
tk.Label(title_bar, text=title, bg=self.bg_color, fg=self.muted_color,
font=title_font, anchor="w").pack(fill=tk.X)
# Thin divider under the title.
tk.Frame(interior, bg=self.border_color, height=1).pack(fill=tk.X, padx=pad)
# Public body attribute — clients pack content into this.
[docs]
self.body = tk.Frame(interior, bg=self.bg_color)
self.body.pack(fill=tk.BOTH, expand=True, padx=pad, pady=pad)
[docs]
class spacrToggle(tk.Frame):
"""iOS-style animated toggle switch bound to a ``tk.BooleanVar``.
Modern replacement for the small check square in ``spacrCheck``.
Clicking the canvas or the caption toggles the variable and animates
the knob to its new position.
:param parent: parent widget.
:param text: caption shown to the left of the toggle; empty hides it.
:param variable: ``tk.BooleanVar`` bound to the toggle state.
:param command: callback fired after each toggle.
"""
_TRACK_W = 44
_TRACK_H = 22
_KNOB_D = 16
_ANIM_STEPS = 6
_ANIM_MS = 120
def __init__(self, parent, text="", variable=None, command=None, **kwargs):
"""Build the toggle canvas and (optional) caption label.
:param parent: parent widget.
:param text: caption shown to the left of the toggle.
:param variable: ``tk.BooleanVar`` bound to the toggle state.
:param command: callback fired after each toggle.
"""
style_out = set_dark_style(ttk.Style())
[docs]
self.bg_color = style_out['bg_color']
[docs]
self.inactive_color = style_out['inactive_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.fg_color = style_out['fg_color']
[docs]
self.muted_color = style_out.get('muted_color', style_out['fg_color'])
spacing = style_out.get('spacing', {'sm': 8})
super().__init__(parent, bg=self.bg_color, **kwargs)
[docs]
self.variable = variable if variable is not None else tk.BooleanVar(value=False)
self._anim_id = None
# Label
font_loader = style_out.get('font_loader')
if text:
font_style = (
font_loader.get_font(size=style_out['font_size'])
if font_loader
else (style_out['font_family'], style_out['font_size'])
)
self._label = tk.Label(self, text=text, bg=self.bg_color,
fg=self.fg_color, font=font_style, cursor="hand2")
self._label.pack(side=tk.LEFT, padx=(0, spacing['sm']))
self._label.bind("<Button-1>", lambda e: self.toggle())
else:
self._label = None
# Track canvas
self._canvas = tk.Canvas(
self, width=self._TRACK_W, height=self._TRACK_H,
bg=self.bg_color, highlightthickness=0, bd=0, cursor="hand2",
)
self._canvas.pack(side=tk.LEFT)
self._canvas.bind("<Button-1>", lambda e: self.toggle())
# Draw track + knob
self._track = self._round_rect(
0, 0, self._TRACK_W, self._TRACK_H, radius=self._TRACK_H // 2,
fill=self._track_color(), outline="",
)
self._knob = self._canvas.create_oval(
*self._knob_bbox(), fill=self.fg_color, outline="",
)
# Sync from variable if user rebinds it later.
self.variable.trace_add("write", lambda *_: self._sync_from_var())
# -- helpers ----------------------------------------------------------
def _round_rect(self, x1, y1, x2, y2, radius=8, **kw):
"""Draw a rounded rectangle on the toggle canvas and return its id."""
pts = [
x1 + radius, y1,
x2 - radius, y1,
x2, y1,
x2, y1 + radius,
x2, y2 - radius,
x2, y2,
x2 - radius, y2,
x1 + radius, y2,
x1, y2,
x1, y2 - radius,
x1, y1 + radius,
x1, y1,
]
return self._canvas.create_polygon(pts, smooth=True, **kw)
def _track_color(self):
"""Return the current track color for the bound state."""
return self.active_color if self.variable.get() else self.inactive_color
def _knob_bbox(self):
"""Return the knob's ``(x0, y0, x1, y1)`` bbox for the current state."""
pad = (self._TRACK_H - self._KNOB_D) // 2
if self.variable.get():
x = self._TRACK_W - self._KNOB_D - pad
else:
x = pad
return x, pad, x + self._KNOB_D, pad + self._KNOB_D
def _sync_from_var(self):
"""Redraw the toggle when the bound variable is set externally."""
self._canvas.itemconfig(self._track, fill=self._track_color())
self._canvas.coords(self._knob, *self._knob_bbox())
[docs]
def toggle(self):
"""Flip the bound variable, animate the knob, and invoke ``command``."""
self.variable.set(not self.variable.get())
self._animate()
if self.command is not None:
try:
self.command()
except Exception:
pass
def _animate(self):
"""Animate the knob sliding to its target position."""
if self._anim_id is not None:
try:
self.after_cancel(self._anim_id)
except Exception:
pass
self._anim_id = None
pad = (self._TRACK_H - self._KNOB_D) // 2
target_x = (self._TRACK_W - self._KNOB_D - pad) if self.variable.get() else pad
current_bbox = self._canvas.coords(self._knob)
current_x = current_bbox[0] if current_bbox else pad
target_track = self._track_color()
# We only animate the knob position; color swaps instantly.
self._canvas.itemconfig(self._track, fill=target_track)
step_x = (target_x - current_x) / self._ANIM_STEPS
delay = max(1, self._ANIM_MS // self._ANIM_STEPS)
def _tick(i=1, x=current_x):
x = x + step_x
self._canvas.coords(self._knob, x, pad, x + self._KNOB_D, pad + self._KNOB_D)
if i < self._ANIM_STEPS:
self._anim_id = self.after(delay, _tick, i + 1, x)
else:
self._anim_id = None
self._canvas.coords(self._knob, *self._knob_bbox())
_tick()
[docs]
def get(self):
"""Return the current toggle state."""
return self.variable.get()
[docs]
def set(self, value):
"""Set the toggle state without animation.
:param value: coerced to bool.
"""
self.variable.set(bool(value))
[docs]
class spacrDivider(tk.Frame):
"""Thin themed section separator, optionally captioned.
Pulls colors and spacing from the shared style dict so the look stays
consistent with the rest of the GUI. Renders as a plain horizontal or
vertical rule; when ``text`` is provided the horizontal variant embeds
the caption between two short rule segments.
:param parent: parent widget.
:param text: caption to embed in the rule (horizontal orientation only).
:param orient: ``'horizontal'`` or ``'vertical'``.
:param thickness: rule thickness in pixels (minimum 1).
"""
def __init__(self, parent, text="", orient='horizontal', thickness=1, **kwargs):
"""Build the rule and, when ``text`` is given, its caption row.
:param parent: parent widget.
:param text: optional caption rendered inside the rule.
:param orient: ``'horizontal'`` or ``'vertical'``.
:param thickness: rule thickness in pixels (min 1).
"""
super().__init__(parent, **kwargs)
[docs]
self.thickness = max(1, int(thickness))
style_out = set_dark_style(ttk.Style())
bg = style_out['bg_color']
border = style_out.get('border_color', style_out['inactive_color'])
muted = style_out.get('muted_color', style_out['fg_color'])
spacing = style_out.get('spacing', {'sm': 8, 'md': 12})
font_sizes = style_out.get('font_sizes', {'small': max(style_out['font_size'] - 1, 9)})
font_loader = style_out.get('font_loader')
self.configure(bg=bg)
if orient == 'vertical':
# Vertical rule, ignores text.
rule = tk.Frame(self, bg=border, width=self.thickness)
rule.pack(fill=tk.Y, expand=True)
return
if not text:
rule = tk.Frame(self, bg=border, height=self.thickness)
rule.pack(fill=tk.X, expand=True, padx=0, pady=(spacing['sm'], spacing['sm']))
return
# Captioned rule: [--- text ---------------------]
self.grid_columnconfigure(1, weight=0)
self.grid_columnconfigure(2, weight=1)
left = tk.Frame(self, bg=border, height=self.thickness, width=spacing['md'])
left.grid(row=0, column=0, sticky='ew', padx=(0, spacing['sm']))
# Force a small fixed-width leading rule.
left.grid_propagate(False)
if font_loader:
font = font_loader.get_font(size=font_sizes.get('small', style_out['font_size']))
else:
font = (style_out['font_family'], font_sizes.get('small', style_out['font_size']))
label = tk.Label(self, text=text, bg=bg, fg=muted, font=font)
label.grid(row=0, column=1, sticky='w')
right = tk.Frame(self, bg=border, height=self.thickness)
right.grid(row=0, column=2, sticky='ew', padx=(spacing['sm'], 0))
[docs]
class ModifyMaskApp:
"""Tkinter app for hand-editing segmentation masks over a set of images.
Supports zoom, draw, brush, erase, magic-wand, and dividing-line
operations, plus per-object cleanup (fill/relabel/remove small).
Masks are loaded from and saved back to ``<folder_path>/masks``.
:param root: parent Tk root or Toplevel.
:param folder_path: directory of image files to edit.
:param scale_factor: pre-canvas resize factor (applied before stretching
to the canvas so brush strokes stay proportional).
"""
def __init__(self, root, folder_path, scale_factor):
"""Discover images, prepare toolbars, and load the first pair."""
[docs]
self.folder_path = folder_path
[docs]
self.scale_factor = scale_factor
[docs]
self.image_filenames = sorted([f for f in os.listdir(folder_path) if f.endswith(('.png', '.jpg', '.jpeg', '.tif', '.tiff'))])
[docs]
self.masks_folder = os.path.join(folder_path, 'masks')
[docs]
self.current_image_index = 0
self.initialize_flags()
[docs]
self.canvas_width = self.root.winfo_screenheight() -100
[docs]
self.canvas_height = self.root.winfo_screenheight() -100
self.root.configure(bg='black')
self.setup_navigation_toolbar()
self.setup_mode_toolbar()
self.setup_function_toolbar()
self.setup_zoom_toolbar()
self.setup_canvas()
self.load_first_image()
####################################################################################################
# Helper functions#
####################################################################################################
[docs]
def update_display(self):
"""Repaint the canvas using the zoomed or full-image view."""
if self.zoom_active:
self.display_zoomed_image()
else:
self.display_image()
[docs]
def update_original_mask_from_zoom(self):
"""Write the current zoomed-region mask back into the full mask."""
y0, y1, x0, x1 = self.zoom_y0, self.zoom_y1, self.zoom_x0, self.zoom_x1
zoomed_mask_resized = resize(self.zoom_mask, (y1 - y0, x1 - x0), order=0, preserve_range=True).astype(np.uint8)
self.mask[y0:y1, x0:x1] = zoomed_mask_resized
[docs]
def update_original_mask(self, zoomed_mask, x0, x1, y0, y1):
"""Merge ``zoomed_mask`` into the full mask at box ``[x0:x1, y0:y1]``.
:param zoomed_mask: mask patch in the zoomed coordinate system.
:param x0: left edge in original image pixels.
:param x1: right edge in original image pixels.
:param y0: top edge in original image pixels.
:param y1: bottom edge in original image pixels.
"""
actual_mask_region = self.mask[y0:y1, x0:x1]
target_shape = actual_mask_region.shape
resized_mask = resize(zoomed_mask, target_shape, order=0, preserve_range=True).astype(np.uint8)
if resized_mask.shape != actual_mask_region.shape:
raise ValueError(f"Shape mismatch: resized_mask {resized_mask.shape}, actual_mask_region {actual_mask_region.shape}")
self.mask[y0:y1, x0:x1] = np.maximum(actual_mask_region, resized_mask)
self.mask = self.mask.copy()
self.mask[y0:y1, x0:x1] = np.maximum(self.mask[y0:y1, x0:x1], resized_mask)
self.mask = self.mask.copy()
[docs]
def get_scaling_factors(self, img_width, img_height, canvas_width, canvas_height):
"""Return ``(x_scale, y_scale)`` mapping canvas pixels to image pixels.
:param img_width: image width in pixels.
:param img_height: image height in pixels.
:param canvas_width: canvas width in pixels.
:param canvas_height: canvas height in pixels.
:returns: tuple ``(x_scale, y_scale)``.
"""
x_scale = img_width / canvas_width
y_scale = img_height / canvas_height
return x_scale, y_scale
[docs]
def canvas_to_image(self, x_canvas, y_canvas):
"""Convert canvas coordinates to full-image coordinates.
:param x_canvas: canvas x pixel.
:param y_canvas: canvas y pixel.
:returns: tuple ``(x_image, y_image)`` in image pixels.
"""
x_scale, y_scale = self.get_scaling_factors(
self.image.shape[1], self.image.shape[0],
self.canvas_width, self.canvas_height
)
x_image = int(x_canvas * x_scale)
y_image = int(y_canvas * y_scale)
return x_image, y_image
[docs]
def apply_zoom_on_enter(self, event):
"""Finalize the zoom rectangle when the pointer re-enters the canvas."""
if self.zoom_active and self.zoom_rectangle_start is not None:
self.set_zoom_rectangle_end(event)
[docs]
def normalize_image(self, image, lower_quantile, upper_quantile):
"""Percentile-clip ``image`` and rescale into its original dtype range.
:param image: 2D image array.
:param lower_quantile: lower percentile (0-100).
:param upper_quantile: upper percentile (0-100).
:returns: normalized image of the same dtype.
"""
lower_bound = np.percentile(image, lower_quantile)
upper_bound = np.percentile(image, upper_quantile)
normalized = np.clip(image, lower_bound, upper_bound)
normalized = (normalized - lower_bound) / (upper_bound - lower_bound)
max_value = np.iinfo(image.dtype).max
normalized = (normalized * max_value).astype(image.dtype)
return normalized
[docs]
def resize_arrays(self, img, mask):
"""Scale image + mask to fit the canvas while preserving their dtypes.
:param img: source intensity image.
:param mask: source label mask.
:returns: tuple ``(scaled_img, scaled_mask)`` sized to the canvas.
"""
original_dtype = img.dtype
scaled_height = int(img.shape[0] * self.scale_factor)
scaled_width = int(img.shape[1] * self.scale_factor)
scaled_img = resize(img, (scaled_height, scaled_width), anti_aliasing=True, preserve_range=True)
scaled_mask = resize(mask, (scaled_height, scaled_width), order=0, anti_aliasing=False, preserve_range=True)
stretched_img = resize(scaled_img, (self.canvas_height, self.canvas_width), anti_aliasing=True, preserve_range=True)
stretched_mask = resize(scaled_mask, (self.canvas_height, self.canvas_width), order=0, anti_aliasing=False, preserve_range=True)
return stretched_img.astype(original_dtype), stretched_mask.astype(original_dtype)
####################################################################################################
#Initiate canvas elements#
####################################################################################################
[docs]
def load_first_image(self):
"""Load the first image/mask pair and paint the canvas."""
self.image, self.mask = self.load_image_and_mask(self.current_image_index)
self.original_size = self.image.shape
self.image, self.mask = self.resize_arrays(self.image, self.mask)
self.display_image()
[docs]
def setup_canvas(self):
"""Create the drawing canvas and attach the mouse-info binding."""
self.canvas = tk.Canvas(self.root, width=self.canvas_width, height=self.canvas_height, bg='black')
self.canvas.pack()
self.canvas.bind("<Motion>", self.update_mouse_info)
[docs]
def initialize_flags(self):
"""Reset all interaction-mode flags and per-image state."""
self.zoom_rectangle_start = None
self.zoom_rectangle_end = None
self.zoom_rectangle_id = None
self.zoom_x0 = None
self.zoom_y0 = None
self.zoom_x1 = None
self.zoom_y1 = None
self.zoom_mask = None
self.zoom_image = None
self.zoom_image_orig = None
self.zoom_scale = 1
self.drawing = False
self.zoom_active = False
self.magic_wand_active = False
self.brush_active = False
self.dividing_line_active = False
self.dividing_line_coords = []
self.current_dividing_line = None
self.lower_quantile = tk.StringVar(value="1.0")
self.upper_quantile = tk.StringVar(value="99.9")
self.magic_wand_tolerance = tk.StringVar(value="1000")
[docs]
def update_mouse_info(self, event):
"""Update the status labels with intensity, mask value, and area."""
x, y = event.x, event.y
intensity = "N/A"
mask_value = "N/A"
pixel_count = "N/A"
if self.zoom_active:
if 0 <= x < self.canvas_width and 0 <= y < self.canvas_height:
intensity = self.zoom_image_orig[y, x] if self.zoom_image_orig is not None else "N/A"
mask_value = self.zoom_mask[y, x] if self.zoom_mask is not None else "N/A"
else:
if 0 <= x < self.image.shape[1] and 0 <= y < self.image.shape[0]:
intensity = self.image[y, x]
mask_value = self.mask[y, x]
if mask_value != "N/A" and mask_value != 0:
pixel_count = np.sum(self.mask == mask_value)
self.intensity_label.config(text=f"Intensity: {intensity}")
self.mask_value_label.config(text=f"Mask: {mask_value}, Area: {pixel_count}")
self.mask_value_label.config(text=f"Mask: {mask_value}")
if mask_value != "N/A" and mask_value != 0:
self.pixel_count_label.config(text=f"Area: {pixel_count}")
else:
self.pixel_count_label.config(text="Area: N/A")
[docs]
def load_image_and_mask(self, index):
"""Load the ``index``-th image and its mask (creating an empty one if absent).
:param index: index into ``self.image_filenames``.
:returns: tuple ``(image, mask)`` — image as uint16, mask as uint8.
"""
# Load the image
image_path = os.path.join(self.folder_path, self.image_filenames[index])
image = imageio.imread(image_path)
print(f"Original Image shape: {image.shape}, dtype: {image.dtype}")
# Handle multi-channel or transparency issues
if image.ndim == 3:
if image.shape[2] == 4: # If the image has an alpha channel (RGBA)
image = image[..., :3] # Remove the alpha channel
# Convert RGB to grayscale using weighted average
image = np.dot(image[..., :3], [0.2989, 0.5870, 0.1140]).astype(np.uint8)
print(f"Converted to grayscale: {image.shape}")
# Ensure the shape is (height, width) without extra channel
if image.ndim == 3 and image.shape[2] == 1:
image = np.squeeze(image, axis=-1)
if image.dtype != np.uint16:
# Scale the image to fit the 16-bit range (0–65535)
image = (image / image.max() * 65535).astype(np.uint16)
# eventually remove this images should not have to be 16 bit look into downstream function (non 16bit images are jsut black)
# Load the corresponding mask
mask_path = os.path.join(self.masks_folder, self.image_filenames[index])
if os.path.exists(mask_path):
print(f'Loading mask: {mask_path} for image: {image_path}')
mask = imageio.imread(mask_path)
# Ensure mask is uint8
if mask.dtype != np.uint8:
mask = (mask / mask.max() * 255).astype(np.uint8)
else:
# Create a new mask with the same size as the image
mask = np.zeros(image.shape[:2], dtype=np.uint8)
print(f'Loaded new mask for image: {image_path}')
return image, mask
####################################################################################################
# Image Display functions#
####################################################################################################
[docs]
def display_image(self):
"""Render the full image + mask overlay on the canvas."""
if self.zoom_rectangle_id is not None:
self.canvas.delete(self.zoom_rectangle_id)
self.zoom_rectangle_id = None
lower_quantile = float(self.lower_quantile.get()) if self.lower_quantile.get() else 1.0
upper_quantile = float(self.upper_quantile.get()) if self.upper_quantile.get() else 99.9
normalized = self.normalize_image(self.image, lower_quantile, upper_quantile)
combined = self.overlay_mask_on_image(normalized, self.mask)
self.tk_image = ImageTk.PhotoImage(image=Image.fromarray(combined))
self.canvas.create_image(0, 0, anchor='nw', image=self.tk_image)
[docs]
def display_zoomed_image(self):
"""Render the current zoomed region with mask overlay on the canvas."""
if self.zoom_rectangle_start and self.zoom_rectangle_end:
# Convert canvas coordinates to image coordinates
x0, y0 = self.canvas_to_image(*self.zoom_rectangle_start)
x1, y1 = self.canvas_to_image(*self.zoom_rectangle_end)
x0, x1 = min(x0, x1), max(x0, x1)
y0, y1 = min(y0, y1), max(y0, y1)
self.zoom_x0 = x0
self.zoom_y0 = y0
self.zoom_x1 = x1
self.zoom_y1 = y1
# Normalize the entire image
lower_quantile = float(self.lower_quantile.get()) if self.lower_quantile.get() else 1.0
upper_quantile = float(self.upper_quantile.get()) if self.upper_quantile.get() else 99.9
normalized_image = self.normalize_image(self.image, lower_quantile, upper_quantile)
# Extract the zoomed portion of the normalized image and mask
self.zoom_image = normalized_image[y0:y1, x0:x1]
self.zoom_image_orig = self.image[y0:y1, x0:x1]
self.zoom_mask = self.mask[y0:y1, x0:x1]
original_mask_area = self.mask.shape[0] * self.mask.shape[1]
zoom_mask_area = self.zoom_mask.shape[0] * self.zoom_mask.shape[1]
if original_mask_area > 0:
self.zoom_scale = original_mask_area/zoom_mask_area
# Resize the zoomed image and mask to fit the canvas
canvas_height = self.canvas.winfo_height()
canvas_width = self.canvas.winfo_width()
if self.zoom_image.size > 0 and canvas_height > 0 and canvas_width > 0:
self.zoom_image = resize(self.zoom_image, (canvas_height, canvas_width), preserve_range=True).astype(self.zoom_image.dtype)
self.zoom_image_orig = resize(self.zoom_image_orig, (canvas_height, canvas_width), preserve_range=True).astype(self.zoom_image_orig.dtype)
#self.zoom_mask = resize(self.zoom_mask, (canvas_height, canvas_width), preserve_range=True).astype(np.uint8)
self.zoom_mask = resize(self.zoom_mask, (canvas_height, canvas_width), order=0, preserve_range=True).astype(np.uint8)
combined = self.overlay_mask_on_image(self.zoom_image, self.zoom_mask)
self.tk_image = ImageTk.PhotoImage(image=Image.fromarray(combined))
self.canvas.create_image(0, 0, anchor='nw', image=self.tk_image)
[docs]
def overlay_mask_on_image(self, image, mask, alpha=0.5):
"""Blend a colored label mask over an intensity image.
:param image: 2D or RGB intensity image.
:param mask: integer label mask; each label gets a random color.
:param alpha: mask opacity in ``[0, 1]``.
:returns: uint8 RGB overlay image.
"""
if len(image.shape) == 2:
image = np.stack((image,) * 3, axis=-1)
mask = mask.astype(np.int32)
max_label = np.max(mask)
np.random.seed(0)
colors = np.random.randint(0, 255, size=(max_label + 1, 3), dtype=np.uint8)
colors[0] = [0, 0, 0] # background color
colored_mask = colors[mask]
image_8bit = (image / 256).astype(np.uint8)
# Blend the mask and the image with transparency
combined_image = np.where(mask[..., None] > 0,
np.clip(image_8bit * (1 - alpha) + colored_mask * alpha, 0, 255),
image_8bit)
# Convert the final image back to uint8
combined_image = combined_image.astype(np.uint8)
return combined_image
####################################################################################################
# Navigation functions#
####################################################################################################
[docs]
def previous_image(self):
"""Load the previous image/mask pair (no-op at the start of the list)."""
if self.current_image_index > 0:
self.current_image_index -= 1
self.initialize_flags()
self.image, self.mask = self.load_image_and_mask(self.current_image_index)
self.original_size = self.image.shape
self.image, self.mask = self.resize_arrays(self.image, self.mask)
self.display_image()
[docs]
def next_image(self):
"""Load the next image/mask pair (no-op at the end of the list)."""
if self.current_image_index < len(self.image_filenames) - 1:
self.current_image_index += 1
self.initialize_flags()
self.image, self.mask = self.load_image_and_mask(self.current_image_index)
self.original_size = self.image.shape
self.image, self.mask = self.resize_arrays(self.image, self.mask)
self.display_image()
[docs]
def save_mask(self):
"""Relabel connected components and write the mask to ``masks/*.tif``."""
if self.current_image_index < len(self.image_filenames):
original_size = self.original_size
if self.mask.shape != original_size:
resized_mask = resize(self.mask, original_size, order=0, preserve_range=True).astype(np.uint16)
else:
resized_mask = self.mask
resized_mask, _ = label(resized_mask > 0)
save_folder = os.path.join(self.folder_path, 'masks')
if not os.path.exists(save_folder):
os.makedirs(save_folder)
image_filename = os.path.splitext(self.image_filenames[self.current_image_index])[0] + '.tif'
save_path = os.path.join(save_folder, image_filename)
print(f"Saving mask to: {save_path}") # Debug print
imageio.imwrite(save_path, resized_mask)
####################################################################################################
# Zoom Functions #
####################################################################################################
[docs]
def set_zoom_rectangle_start(self, event):
"""Record the first corner of the zoom rectangle."""
if self.zoom_active:
self.zoom_rectangle_start = (event.x, event.y)
[docs]
def set_zoom_rectangle_end(self, event):
"""Commit the second corner of the zoom rectangle and render the zoom."""
if self.zoom_active:
self.zoom_rectangle_end = (event.x, event.y)
if self.zoom_rectangle_id is not None:
self.canvas.delete(self.zoom_rectangle_id)
self.zoom_rectangle_id = None
self.display_zoomed_image()
self.canvas.unbind("<Motion>")
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<Button-3>")
self.canvas.bind("<Motion>", self.update_mouse_info)
[docs]
def update_zoom_box(self, event):
"""Redraw the live zoom-selection rectangle as the pointer moves."""
if self.zoom_active and self.zoom_rectangle_start is not None:
if self.zoom_rectangle_id is not None:
self.canvas.delete(self.zoom_rectangle_id)
# Assuming event.x and event.y are already in image coordinates
self.zoom_rectangle_end = (event.x, event.y)
x0, y0 = self.zoom_rectangle_start
x1, y1 = self.zoom_rectangle_end
self.zoom_rectangle_id = self.canvas.create_rectangle(x0, y0, x1, y1, outline="red", width=2)
####################################################################################################
# Mode activation#
####################################################################################################
[docs]
def toggle_zoom_mode(self):
"""Enter or exit zoom-selection mode, rebinding mouse handlers."""
if not self.zoom_active:
self.brush_btn.config(text="Brush")
self.canvas.unbind("<B1-Motion>")
self.canvas.unbind("<B3-Motion>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<ButtonRelease-3>")
self.zoom_active = True
self.drawing = False
self.magic_wand_active = False
self.erase_active = False
self.brush_active = False
self.dividing_line_active = False
self.draw_btn.config(text="Draw")
self.erase_btn.config(text="Erase")
self.magic_wand_btn.config(text="Magic Wand")
self.zoom_btn.config(text="Zoom ON")
self.dividing_line_btn.config(text="Dividing Line")
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<Button-3>")
self.canvas.unbind("<Motion>")
self.canvas.bind("<Button-1>", self.set_zoom_rectangle_start)
self.canvas.bind("<Button-3>", self.set_zoom_rectangle_end)
self.canvas.bind("<Motion>", self.update_zoom_box)
else:
self.zoom_active = False
self.zoom_btn.config(text="Zoom")
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<Button-3>")
self.canvas.unbind("<Motion>")
self.zoom_rectangle_start = self.zoom_rectangle_end = None
self.zoom_rectangle_id = None
self.display_image()
self.canvas.bind("<Motion>", self.update_mouse_info)
self.zoom_rectangle_start = None
self.zoom_rectangle_end = None
self.zoom_rectangle_id = None
self.zoom_x0 = None
self.zoom_y0 = None
self.zoom_x1 = None
self.zoom_y1 = None
self.zoom_mask = None
self.zoom_image = None
self.zoom_image_orig = None
[docs]
def toggle_brush_mode(self):
"""Enter or exit brush painting mode, rebinding mouse handlers."""
self.brush_active = not self.brush_active
if self.brush_active:
self.drawing = False
self.magic_wand_active = False
self.erase_active = False
self.brush_btn.config(text="Brush ON")
self.draw_btn.config(text="Draw")
self.erase_btn.config(text="Erase")
self.magic_wand_btn.config(text="Magic Wand")
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<Button-3>")
self.canvas.unbind("<Motion>")
self.canvas.bind("<B1-Motion>", self.apply_brush) # Left click and drag to apply brush
self.canvas.bind("<B3-Motion>", self.erase_brush) # Right click and drag to erase with brush
self.canvas.bind("<ButtonRelease-1>", self.apply_brush_release) # Left button release
self.canvas.bind("<ButtonRelease-3>", self.erase_brush_release) # Right button release
else:
self.brush_active = False
self.brush_btn.config(text="Brush")
self.canvas.unbind("<B1-Motion>")
self.canvas.unbind("<B3-Motion>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<ButtonRelease-3>")
[docs]
def image_to_canvas(self, x_image, y_image):
"""Convert full-image coordinates to canvas coordinates.
:param x_image: image x pixel.
:param y_image: image y pixel.
:returns: tuple ``(x_canvas, y_canvas)`` in canvas pixels.
"""
x_scale, y_scale = self.get_scaling_factors(
self.image.shape[1], self.image.shape[0],
self.canvas_width, self.canvas_height
)
x_canvas = int(x_image / x_scale)
y_canvas = int(y_image / y_scale)
return x_canvas, y_canvas
[docs]
def toggle_dividing_line_mode(self):
"""Enter or exit dividing-line mode, rebinding mouse handlers."""
self.dividing_line_active = not self.dividing_line_active
if self.dividing_line_active:
self.drawing = False
self.magic_wand_active = False
self.erase_active = False
self.brush_active = False
self.draw_btn.config(text="Draw")
self.erase_btn.config(text="Erase")
self.magic_wand_btn.config(text="Magic Wand")
self.brush_btn.config(text="Brush")
self.dividing_line_btn.config(text="Dividing Line ON")
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<Motion>")
self.canvas.bind("<Button-1>", self.start_dividing_line)
self.canvas.bind("<ButtonRelease-1>", self.finish_dividing_line)
self.canvas.bind("<Motion>", self.update_dividing_line_preview)
else:
print("Dividing Line Mode: OFF")
self.dividing_line_active = False
self.dividing_line_btn.config(text="Dividing Line")
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<Motion>")
self.display_image()
[docs]
def start_dividing_line(self, event):
"""Begin a dividing-line stroke at the pointer position."""
if self.dividing_line_active:
self.dividing_line_coords = [(event.x, event.y)]
self.current_dividing_line = self.canvas.create_line(event.x, event.y, event.x, event.y, fill="red", width=2)
[docs]
def finish_dividing_line(self, event):
"""Close the dividing-line stroke and apply it to the mask."""
if self.dividing_line_active:
self.dividing_line_coords.append((event.x, event.y))
if self.zoom_active:
self.dividing_line_coords = [self.canvas_to_image(x, y) for x, y in self.dividing_line_coords]
self.apply_dividing_line()
self.canvas.delete(self.current_dividing_line)
self.current_dividing_line = None
[docs]
def update_dividing_line_preview(self, event):
"""Extend and redraw the in-progress dividing-line preview stroke."""
if self.dividing_line_active and self.dividing_line_coords:
x, y = event.x, event.y
if self.zoom_active:
x, y = self.canvas_to_image(x, y)
self.dividing_line_coords.append((x, y))
canvas_coords = [(self.image_to_canvas(*pt) if self.zoom_active else pt) for pt in self.dividing_line_coords]
flat_canvas_coords = [coord for pt in canvas_coords for coord in pt]
self.canvas.coords(self.current_dividing_line, *flat_canvas_coords)
[docs]
def apply_dividing_line(self):
"""Cut the mask along the recorded dividing-line polyline and relabel."""
if self.dividing_line_coords:
coords = self.dividing_line_coords
if self.zoom_active:
coords = [self.canvas_to_image(x, y) for x, y in coords]
rr, cc = [], []
for (x0, y0), (x1, y1) in zip(coords[:-1], coords[1:]):
line_rr, line_cc = line(y0, x0, y1, x1)
rr.extend(line_rr)
cc.extend(line_cc)
rr, cc = np.array(rr), np.array(cc)
mask_copy = self.mask.copy()
if self.zoom_active:
# Update the zoomed mask
self.zoom_mask[rr, cc] = 0
# Reflect changes to the original mask
y0, y1, x0, x1 = self.zoom_y0, self.zoom_y1, self.zoom_x0, self.zoom_x1
zoomed_mask_resized_back = resize(self.zoom_mask, (y1 - y0, x1 - x0), order=0, preserve_range=True).astype(np.uint8)
self.mask[y0:y1, x0:x1] = zoomed_mask_resized_back
else:
# Directly update the original mask
mask_copy[rr, cc] = 0
self.mask = mask_copy
labeled_mask, num_labels = label(self.mask > 0)
self.mask = labeled_mask
self.update_display()
self.dividing_line_coords = []
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<Motion>")
self.dividing_line_active = False
self.dividing_line_btn.config(text="Dividing Line")
[docs]
def toggle_draw_mode(self):
"""Enter or exit freehand polygon draw mode, rebinding mouse handlers."""
self.drawing = not self.drawing
if self.drawing:
self.brush_btn.config(text="Brush")
self.canvas.unbind("<B1-Motion>")
self.canvas.unbind("<B3-Motion>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<ButtonRelease-3>")
self.magic_wand_active = False
self.erase_active = False
self.brush_active = False
self.draw_btn.config(text="Draw ON")
self.magic_wand_btn.config(text="Magic Wand")
self.erase_btn.config(text="Erase")
self.draw_coordinates = []
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<Motion>")
self.canvas.bind("<B1-Motion>", self.draw)
self.canvas.bind("<ButtonRelease-1>", self.finish_drawing)
else:
self.drawing = False
self.draw_btn.config(text="Draw")
self.canvas.unbind("<B1-Motion>")
self.canvas.unbind("<ButtonRelease-1>")
[docs]
def toggle_magic_wand_mode(self):
"""Enter or exit magic-wand mode, rebinding mouse handlers."""
self.magic_wand_active = not self.magic_wand_active
if self.magic_wand_active:
self.brush_btn.config(text="Brush")
self.canvas.unbind("<B1-Motion>")
self.canvas.unbind("<B3-Motion>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<ButtonRelease-3>")
self.drawing = False
self.erase_active = False
self.brush_active = False
self.draw_btn.config(text="Draw")
self.erase_btn.config(text="Erase")
self.magic_wand_btn.config(text="Magic Wand ON")
self.canvas.bind("<Button-1>", self.use_magic_wand)
self.canvas.bind("<Button-3>", self.use_magic_wand)
else:
self.magic_wand_btn.config(text="Magic Wand")
self.canvas.unbind("<Button-1>")
self.canvas.unbind("<Button-3>")
[docs]
def toggle_erase_mode(self):
"""Enter or exit whole-object erase mode, rebinding mouse handlers."""
self.erase_active = not self.erase_active
if self.erase_active:
self.brush_btn.config(text="Brush")
self.canvas.unbind("<B1-Motion>")
self.canvas.unbind("<B3-Motion>")
self.canvas.unbind("<ButtonRelease-1>")
self.canvas.unbind("<ButtonRelease-3>")
self.erase_btn.config(text="Erase ON")
self.canvas.bind("<Button-1>", self.erase_object)
self.drawing = False
self.magic_wand_active = False
self.brush_active = False
self.draw_btn.config(text="Draw")
self.magic_wand_btn.config(text="Magic Wand")
else:
self.erase_active = False
self.erase_btn.config(text="Erase")
self.canvas.unbind("<Button-1>")
####################################################################################################
# Mode functions#
####################################################################################################
[docs]
def apply_brush_release(self, event):
"""Commit the accumulated brush path into the mask on button release."""
if hasattr(self, 'brush_path'):
for x, y, brush_size in self.brush_path:
img_x, img_y = (x, y) if self.zoom_active else self.canvas_to_image(x, y)
x0 = max(img_x - brush_size // 2, 0)
y0 = max(img_y - brush_size // 2, 0)
x1 = min(img_x + brush_size // 2, self.zoom_mask.shape[1] if self.zoom_active else self.mask.shape[1])
y1 = min(img_y + brush_size // 2, self.zoom_mask.shape[0] if self.zoom_active else self.mask.shape[0])
if self.zoom_active:
self.zoom_mask[y0:y1, x0:x1] = 255
self.update_original_mask_from_zoom()
else:
self.mask[y0:y1, x0:x1] = 255
del self.brush_path
self.canvas.delete("temp_line")
self.update_display()
[docs]
def erase_brush_release(self, event):
"""Commit the accumulated erase-brush path into the mask on release."""
if hasattr(self, 'erase_path'):
for x, y, brush_size in self.erase_path:
img_x, img_y = (x, y) if self.zoom_active else self.canvas_to_image(x, y)
x0 = max(img_x - brush_size // 2, 0)
y0 = max(img_y - brush_size // 2, 0)
x1 = min(img_x + brush_size // 2, self.zoom_mask.shape[1] if self.zoom_active else self.mask.shape[1])
y1 = min(img_y + brush_size // 2, self.zoom_mask.shape[0] if self.zoom_active else self.mask.shape[0])
if self.zoom_active:
self.zoom_mask[y0:y1, x0:x1] = 0
self.update_original_mask_from_zoom()
else:
self.mask[y0:y1, x0:x1] = 0
del self.erase_path
self.canvas.delete("temp_line")
self.update_display()
[docs]
def apply_brush(self, event):
"""Record a brush stroke segment and draw a preview line."""
brush_size = int(self.brush_size_entry.get())
x, y = event.x, event.y
if not hasattr(self, 'brush_path'):
self.brush_path = []
self.last_brush_coord = (x, y)
if self.last_brush_coord:
last_x, last_y = self.last_brush_coord
rr, cc = line(last_y, last_x, y, x)
for ry, rx in zip(rr, cc):
self.brush_path.append((rx, ry, brush_size))
self.canvas.create_line(self.last_brush_coord[0], self.last_brush_coord[1], x, y, width=brush_size, fill="blue", tag="temp_line")
self.last_brush_coord = (x, y)
[docs]
def erase_brush(self, event):
"""Record an erase-brush stroke segment and draw a preview line."""
brush_size = int(self.brush_size_entry.get())
x, y = event.x, event.y
if not hasattr(self, 'erase_path'):
self.erase_path = []
self.last_erase_coord = (x, y)
if self.last_erase_coord:
last_x, last_y = self.last_erase_coord
rr, cc = line(last_y, last_x, y, x)
for ry, rx in zip(rr, cc):
self.erase_path.append((rx, ry, brush_size))
self.canvas.create_line(self.last_erase_coord[0], self.last_erase_coord[1], x, y, width=brush_size, fill="white", tag="temp_line")
self.last_erase_coord = (x, y)
[docs]
def erase_object(self, event):
"""Erase the whole labeled object under the click position."""
x, y = event.x, event.y
if self.zoom_active:
canvas_x, canvas_y = x, y
zoomed_x = int(canvas_x * (self.zoom_image.shape[1] / self.canvas_width))
zoomed_y = int(canvas_y * (self.zoom_image.shape[0] / self.canvas_height))
orig_x = int(zoomed_x * ((self.zoom_x1 - self.zoom_x0) / self.canvas_width) + self.zoom_x0)
orig_y = int(zoomed_y * ((self.zoom_y1 - self.zoom_y0) / self.canvas_height) + self.zoom_y0)
if orig_x < 0 or orig_y < 0 or orig_x >= self.image.shape[1] or orig_y >= self.image.shape[0]:
print("Point is out of bounds in the original image.")
return
else:
orig_x, orig_y = x, y
label_to_remove = self.mask[orig_y, orig_x]
if label_to_remove > 0:
self.mask[self.mask == label_to_remove] = 0
self.update_display()
[docs]
def use_magic_wand(self, event):
"""Run a magic-wand add (left) or erase (right) at the click position."""
x, y = event.x, event.y
tolerance = int(self.magic_wand_tolerance.get())
maximum = int(self.max_pixels_entry.get())
action = 'add' if event.num == 1 else 'erase'
if self.zoom_active:
self.magic_wand_zoomed((x, y), tolerance, action)
else:
self.magic_wand_normal((x, y), tolerance, action)
[docs]
def apply_magic_wand(self, image, mask, seed_point, tolerance, maximum, action='add'):
"""Flood-fill mask from ``seed_point`` while intensity delta stays within tolerance.
:param image: intensity image used for the tolerance check.
:param mask: mask array to modify in place.
:param seed_point: tuple ``(x, y)`` of the starting pixel.
:param tolerance: maximum L2 distance from the seed intensity.
:param maximum: cap on newly added pixels.
:param action: ``'add'`` sets mask to 255, ``'erase'`` sets it to 0.
:returns: the updated ``mask``.
"""
x, y = seed_point
initial_value = image[y, x].astype(np.float32)
visited = np.zeros_like(image, dtype=bool)
queue = deque([(x, y)])
added_pixels = 0
while queue and added_pixels < maximum:
cx, cy = queue.popleft()
if visited[cy, cx]:
continue
visited[cy, cx] = True
current_value = image[cy, cx].astype(np.float32)
if np.linalg.norm(abs(current_value - initial_value)) <= tolerance:
if mask[cy, cx] == 0:
added_pixels += 1
mask[cy, cx] = 255 if action == 'add' else 0
if added_pixels >= maximum:
break
for dx, dy in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
nx, ny = cx + dx, cy + dy
if 0 <= nx < image.shape[1] and 0 <= ny < image.shape[0] and not visited[ny, nx]:
queue.append((nx, ny))
return mask
[docs]
def magic_wand_normal(self, seed_point, tolerance, action):
"""Apply the magic wand to the full-image mask and repaint the canvas.
:param seed_point: starting pixel in image coordinates.
:param tolerance: intensity tolerance for flood expansion.
:param action: ``'add'`` or ``'erase'``.
"""
try:
maximum = int(self.max_pixels_entry.get())
except ValueError:
print("Invalid maximum value; using default of 1000")
maximum = 1000
self.mask = self.apply_magic_wand(self.image, self.mask, seed_point, tolerance, maximum, action)
self.display_image()
[docs]
def magic_wand_zoomed(self, seed_point, tolerance, action):
"""Apply the magic wand within the current zoom and reflect it in the full mask.
:param seed_point: starting pixel in canvas coordinates.
:param tolerance: intensity tolerance for flood expansion.
:param action: ``'add'`` or ``'erase'``.
"""
if self.zoom_image_orig is None or self.zoom_mask is None:
print("Zoomed image or mask not initialized")
return
try:
maximum = int(self.max_pixels_entry.get())
maximum = maximum * self.zoom_scale
except ValueError:
print("Invalid maximum value; using default of 1000")
maximum = 1000
canvas_x, canvas_y = seed_point
if canvas_x < 0 or canvas_y < 0 or canvas_x >= self.zoom_image_orig.shape[1] or canvas_y >= self.zoom_image_orig.shape[0]:
print("Selected point is out of bounds in the zoomed image.")
return
self.zoom_mask = self.apply_magic_wand(self.zoom_image_orig, self.zoom_mask, (canvas_x, canvas_y), tolerance, maximum, action)
y0, y1, x0, x1 = self.zoom_y0, self.zoom_y1, self.zoom_x0, self.zoom_x1
zoomed_mask_resized_back = resize(self.zoom_mask, (y1 - y0, x1 - x0), order=0, preserve_range=True).astype(np.uint8)
if action == 'erase':
self.mask[y0:y1, x0:x1] = np.where(zoomed_mask_resized_back == 0, 0, self.mask[y0:y1, x0:x1])
else:
self.mask[y0:y1, x0:x1] = np.where(zoomed_mask_resized_back > 0, zoomed_mask_resized_back, self.mask[y0:y1, x0:x1])
self.update_display()
[docs]
def draw(self, event):
"""Append the pointer position to the current freehand polygon."""
if self.drawing:
x, y = event.x, event.y
if self.draw_coordinates:
last_x, last_y = self.draw_coordinates[-1]
self.current_line = self.canvas.create_line(last_x, last_y, x, y, fill="yellow", width=3)
self.draw_coordinates.append((x, y))
[docs]
def draw_on_zoomed_mask(self, draw_coordinates):
"""Rasterize a polygon (in canvas coords) into a fresh zoomed-mask array.
:param draw_coordinates: list of ``(x, y)`` tuples defining the polygon.
:returns: uint8 canvas-sized mask with the polygon filled.
"""
canvas_height = self.canvas.winfo_height()
canvas_width = self.canvas.winfo_width()
zoomed_mask = np.zeros((canvas_height, canvas_width), dtype=np.uint8)
rr, cc = polygon(np.array(draw_coordinates)[:, 1], np.array(draw_coordinates)[:, 0], shape=zoomed_mask.shape)
zoomed_mask[rr, cc] = 255
return zoomed_mask
[docs]
def finish_drawing(self, event):
"""Close the polygon and rasterize it into the mask."""
if len(self.draw_coordinates) > 2:
self.draw_coordinates.append(self.draw_coordinates[0])
if self.zoom_active:
x0, x1, y0, y1 = self.zoom_x0, self.zoom_x1, self.zoom_y0, self.zoom_y1
zoomed_mask = self.draw_on_zoomed_mask(self.draw_coordinates)
self.update_original_mask(zoomed_mask, x0, x1, y0, y1)
else:
rr, cc = polygon(np.array(self.draw_coordinates)[:, 1], np.array(self.draw_coordinates)[:, 0], shape=self.mask.shape)
self.mask[rr, cc] = np.maximum(self.mask[rr, cc], 255)
self.mask = self.mask.copy()
self.canvas.delete(self.current_line)
self.draw_coordinates.clear()
self.update_display()
[docs]
def finish_drawing_if_active(self, event):
"""Close the polygon only if draw mode is active with enough vertices."""
if self.drawing and len(self.draw_coordinates) > 2:
self.finish_drawing(event)
####################################################################################################
# Single function butons#
####################################################################################################
[docs]
def apply_normalization(self):
"""Read the percentile entries and repaint with the new normalization."""
self.lower_quantile.set(self.lower_entry.get())
self.upper_quantile.set(self.upper_entry.get())
self.update_display()
[docs]
def fill_objects(self):
"""Fill holes inside all mask objects and relabel."""
binary_mask = self.mask > 0
filled_mask = binary_fill_holes(binary_mask)
self.mask = filled_mask.astype(np.uint8) * 255
labeled_mask, _ = label(filled_mask)
self.mask = labeled_mask
self.update_display()
[docs]
def relabel_objects(self):
"""Assign fresh consecutive labels to the mask's connected components."""
mask = self.mask
labeled_mask, num_labels = label(mask > 0)
self.mask = labeled_mask
self.update_display()
[docs]
def clear_objects(self):
"""Zero the entire mask and repaint."""
self.mask = np.zeros_like(self.mask)
self.update_display()
[docs]
def invert_mask(self):
"""Invert the binary mask and relabel connected components."""
self.mask = np.where(self.mask > 0, 0, 1)
self.relabel_objects()
self.update_display()
[docs]
def remove_small_objects(self):
"""Delete labeled objects below the ``Min Area`` threshold."""
try:
min_area = int(self.min_area_entry.get())
except ValueError:
print("Invalid minimum area value; using default of 100")
min_area = 100
labeled_mask, num_labels = label(self.mask > 0)
for i in range(1, num_labels + 1): # Skip background
if np.sum(labeled_mask == i) < min_area:
self.mask[labeled_mask == i] = 0 # Remove small objects
self.update_display()
[docs]
class AnnotateApp:
"""Grid-based annotation viewer backed by an SQLite measurements database.
Renders a paginated grid of PNGs (with optional colored outlines and
normalization), lets the user click-annotate each cell, and streams
updates back to ``png_list.<annotation_column>`` via a background writer
thread. Supports pre-filtering by measurement thresholds and training a
lightweight XGBoost classifier on the collected labels.
:param root: parent Tk root or Toplevel.
:param db_path: path to the measurements SQLite database.
:param src: source directory containing the ``measurements/`` folder.
:param image_type: substring filter on ``png_path`` (or a list of them).
:param channels: list of channels (subset of ``'r','g','b'``) to display.
:param image_size: grid tile size in pixels (int or ``[w, h]``).
:param annotation_column: ``png_list`` column that stores user labels.
:param percentiles: ``(low, high)`` percentiles for per-image normalization.
:param measurement: column name, list, or list-of-lists driving prefilter.
:param threshold: numeric or quantile-code (``q1``..``q9``) threshold(s).
:param threshold_direction: ``'lower'`` or ``'higher'`` (or a list).
:param normalize_channels: channels to normalize (subset of ``'r','g','b'``).
:param outline: channels to overlay outlines on.
:param outline_threshold_factor: multiplier on the Otsu threshold.
:param outline_sigma: Gaussian sigma for outline extraction.
:param edge_thickness: outline thickness in output pixels.
:param edge_transparency: outline opacity in ``[0, 100]``.
:param edge_image: when True, composite the outline image on display.
:param object_size: ``(min_px, max_px)`` connected-component filter; 0 disables.
"""
def __init__(self, root, db_path, src, image_type=None, channels=None, image_size=200, annotation_column='annotate', percentiles=(1, 99), measurement=None, threshold=None, threshold_direction = "higher", normalize_channels=None, outline=None, outline_threshold_factor=1, outline_sigma=1, edge_thickness=1, edge_transparency=100, edge_image=False, object_size=(0,0)):
"""Build the annotation UI, start the DB worker, and load the first page."""
[docs]
self.SENTINEL = object()
if isinstance(image_size, list):
self.image_size = (int(image_size[0]), int(image_size[0]))
elif isinstance(image_size, int):
self.image_size = (image_size, image_size)
else:
raise ValueError("Invalid image size")
# Cross-platform right-click event: 'aqua' (macOS) uses Button-2, others use Button-3
windowing = self.root.tk.call('tk', 'windowingsystem')
self._right_click_event = '<Button-2>' if windowing == 'aqua' else '<Button-3>'
[docs]
self.orig_annotation_columns = annotation_column
[docs]
self.annotation_column = annotation_column
self._ensure_annotation_column()
self._ensure_png_path_index()
[docs]
self.image_type = image_type
[docs]
self.channels = channels
[docs]
self.percentiles = percentiles
[docs]
self.pending_updates = {}
[docs]
self.adjusted_to_original_paths = {}
[docs]
self.update_queue = Queue()
[docs]
self.measurement = measurement
[docs]
self.threshold = threshold
[docs]
self.threshold_direction = threshold_direction
[docs]
self.normalize_channels = normalize_channels
[docs]
self.outline_threshold_factor = outline_threshold_factor
[docs]
self.outline_sigma = outline_sigma
[docs]
self.edge_thickness = edge_thickness
[docs]
self.edge_transparency = edge_transparency
[docs]
self.edge_image = edge_image
[docs]
self.object_size = tuple(object_size) if object_size else (0, 0)
style_out = set_dark_style(ttk.Style())
[docs]
self.font_loader = style_out['font_loader']
[docs]
self.font_size = style_out['font_size']
[docs]
self.bg_color = style_out['bg_color']
[docs]
self.fg_color = style_out['fg_color']
[docs]
self.active_color = style_out['active_color']
[docs]
self.inactive_color = style_out['inactive_color']
# --- save-status UI & state ---
self._spinner_frames = ["⠋","⠙","⠸","⠴","⠦","⠇"]
self._spinner_idx = 0
[docs]
self.worker_busy = False
self._unsaved_batches = 0
self._batch_lock = threading.Lock()
self._last_save_ts = None
if self.font_loader:
self.font_style = self.font_loader.get_font(size=self.font_size)
else:
self.font_style = ("Arial", 12)
self.root.configure(bg=style_out['inactive_color'])
# Defer data loading until grid dimensions are known
[docs]
self.filtered_paths_annotations = []
self._total_filtered = 0
[docs]
self.db_update_thread = threading.Thread(target=self.update_database_worker)
self.db_update_thread.start()
# Set the initial window size and make it fit the screen size
self.root.geometry(f"{self.root.winfo_screenwidth()}x{self.root.winfo_screenheight()}")
self.root.update_idletasks()
# grid at top
[docs]
self.grid_frame = Frame(root, bg=self.root.cget('bg'))
self.grid_frame.grid(row=0, column=0, columnspan=2, padx=0, pady=0, sticky="nsew")
# status (left) + buttons (right) on the same bottom row
[docs]
self.status_label = Label(root, text="", font=self.font_style, bg=self.bg_color, fg=self.fg_color)
self.status_label.grid(row=2, column=0, padx=10, pady=8, sticky="w")
# begin polling the status 6–10 times/sec
self._poll_save_status()
self.button_frame.grid(row=2, column=1, padx=10, pady=8, sticky="e")
# macOS tk.Button ignores fg; force contrast explicitly
_is_mac = platform.system() == 'Darwin'
def _make_button(parent, text, command):
if _is_mac:
# We use a Label because macOS completely ignores background locks on standard Buttons
btn = tk.Label(parent, text=text,
bg='#1a1a1a', fg='white',
padx=12, pady=6, # Generates the button padding
relief='flat', cursor='hand2') # Makes it look clickable
# Mimic the button click behavior and active state color switches
def on_press(event):
"""Darken the label to mimic a pressed button."""
btn.config(bg='#333333')
def on_release(event):
"""Restore label color and fire the wrapped command."""
btn.config(bg='#1a1a1a')
command() # Triggers the button's actual function
# Bind the mouse clicks directly to the label
btn.bind("<ButtonPress-1>", on_press)
btn.bind("<ButtonRelease-1>", on_release)
else:
btn = Button(parent, text=text, command=command,
bg=self.bg_color, fg=self.fg_color,
highlightbackground=self.fg_color,
highlightcolor=self.fg_color,
highlightthickness=1)
return btn
# pack (right to left)
self.next_button.pack(side="right", padx=5)
self.previous_button.pack(side="right", padx=5)
self.skip_to_last_annotated_button.pack(side="right", padx=5)
self.exit_button.pack(side="right", padx=5)
self.settings_button.pack(side="right", padx=5)
self.clear_button.pack(side="right", padx=5)
self.count_button.pack(side="right", padx=5)
self.dl_train_button.pack(side="right", padx=5)
# Arrow key bindings to previous and next
self.root.bind('<Left>', lambda event: self.previous_page())
self.root.bind('<Right>', lambda event: self.next_page())
# compute grid size (after buttons exist with real height)
self.button_frame.update_idletasks()
needed = self.button_frame.winfo_reqwidth()
self.root.grid_columnconfigure(1, minsize=needed + 10, weight=0)
self.root.grid_columnconfigure(0, weight=1)
self.root.update_idletasks()
self.calculate_grid_dimensions()
self.prefilter_paths_annotations()
for i in range(self.grid_rows * self.grid_cols):
label = Label(self.grid_frame, bg=self.root.cget('bg'))
label.grid(row=i // self.grid_cols, column=i % self.grid_cols, padx=2, pady=2, sticky="nsew")
self.labels.append(label)
# column/row weights
self.root.grid_rowconfigure(0, weight=1)
self.root.grid_rowconfigure(2, weight=0)
for row in range(self.grid_rows):
self.grid_frame.grid_rowconfigure(row, weight=1)
for col in range(self.grid_cols):
self.grid_frame.grid_columnconfigure(col, weight=1)
def _ensure_png_path_index(self):
"""Create the ``png_list.png_path`` index if it is missing."""
import sqlite3
with sqlite3.connect(self.db_path, timeout=30) as conn:
conn.execute('CREATE INDEX IF NOT EXISTS idx_png_path ON "png_list" (png_path)')
def _int_to_color(self, k, s=0.65, v=0.95):
"""
Deterministically map any non-negative integer k -> hex color using
the golden-ratio conjugate to distribute hues around the color wheel.
s,v control saturation and value (brightness).
"""
import colorsys
# Golden ratio conjugate (~0.618...) spreads hues evenly
phi = 0.618033988749895
# Wrap k around the unit interval
h = (k * phi) % 1.0
r, g, b = colorsys.hsv_to_rgb(h, float(s), float(v))
return "#{:02x}{:02x}{:02x}".format(int(r * 255 + 0.5),
int(g * 255 + 0.5),
int(b * 255 + 0.5))
def _label_to_color(self, val):
"""
Public helper: for an integer label return a hex color.
- None/0/invalid -> None (no border).
- 1 -> blue, 2 -> red.
- 3+ -> infinite distinct colors, deterministic.
Caches results so colors stay stable across the session.
"""
# Lazy-init cache on the instance
if not hasattr(self, "_class_color_cache"):
self._class_color_cache = {}
try:
if val is None:
return None
iv = int(val)
if iv <= 0:
return None
except Exception:
return None
if iv in self._class_color_cache:
return self._class_color_cache[iv]
# Fixed starters
if iv == 1:
color = "#1f77b4" # blue
elif iv == 2:
color = "#d62728" # red
else:
# Map 3 -> k=0, 4 -> k=1, ... so early classes are well separated
k = iv - 3
color = self._int_to_color(k)
self._class_color_cache[iv] = color
return color
def _embed_figure_in(self, parent, fig):
"""Replace ``parent``'s contents with a Tk canvas rendering ``fig``."""
# Clear parent
for w in parent.winfo_children():
try: w.destroy()
except Exception: pass
try:
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
except Exception as e:
lab = tk.Label(parent, text=f"Matplotlib not available: {e}", bg=self.bg_color, fg="red")
lab.pack()
return None
canvas = FigureCanvasTkAgg(fig, master=parent)
widget = canvas.get_tk_widget()
widget.pack(fill="both", expand=True)
canvas.draw()
return canvas
[docs]
def open_umap_window(self):
"""Open a settings + live-plot window for image UMAP + hyperparam search."""
import tkinter as tk
from tkinter import ttk, messagebox
import threading
import ast
win = tk.Toplevel(self.root)
win.title("Image UMAP & Hyperparameter Search")
win.configure(bg=self.bg_color)
win.geometry("1200x800")
outer = tk.Frame(win, bg=self.bg_color)
outer.pack(fill=tk.BOTH, expand=True)
# Left: settings
left = tk.Frame(outer, bg=self.bg_color)
left.pack(side=tk.LEFT, fill=tk.Y, padx=10, pady=10)
# Right: live plot
right = tk.Frame(outer, bg=self.bg_color)
right.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0,10), pady=10)
# --- Settings widgets (a practical subset; add more if you like) ---
def _row(lbl, widget):
r = tk.Frame(left, bg=self.bg_color)
tk.Label(r, text=lbl, bg=self.bg_color, fg=self.fg_color, font=self.font_style).pack(side=tk.TOP, anchor="w")
widget.pack(in_=r, fill=tk.X, expand=True)
r.pack(fill=tk.X, pady=6)
src_entry = tk.Entry(left)
src_entry.insert(0, self.src)
tables_entry = tk.Entry(left)
tables_entry.insert(0, "cell,cytoplasm,nucleus,pathogen") # sensible default
row_limit_entry = tk.Entry(left)
row_limit_entry.insert(0, "") # blank = no limit
# UMAP params
n_neighbors_entry = tk.Entry(left); n_neighbors_entry.insert(0, "15")
min_dist_entry = tk.Entry(left); min_dist_entry.insert(0, "0.1")
metric_entry = tk.Entry(left); metric_entry.insert(0, "euclidean")
# Clustering params
clustering_cbx = ttk.Combobox(left, state="readonly", values=["dbscan","kmeans"])
clustering_cbx.set("dbscan")
eps_entry = tk.Entry(left); eps_entry.insert(0, "0.5") # DBSCAN
min_samples_entry = tk.Entry(left); min_samples_entry.insert(0, "5")
kmeans_k_entry = tk.Entry(left); kmeans_k_entry.insert(0, "8") # KMeans
color_by_entry = tk.Entry(left); color_by_entry.insert(0, "") # e.g. columnID or cond
dot_size_entry = tk.Entry(left); dot_size_entry.insert(0, "6")
fig_size_entry = tk.Entry(left); fig_size_entry.insert(0, "10") # inches
img_nr_entry = tk.Entry(left); img_nr_entry.insert(0, "200") # images in overlay (if plotting images)
plot_images_var = tk.BooleanVar(value=False)
tk.Checkbutton(left, text="plot_images (heavy)", variable=plot_images_var,
bg=self.bg_color, fg=self.fg_color, selectcolor=self.bg_color).pack(anchor="w", pady=(4,0))
# Hyperparam grids (comma-separated lists interpreted as Python literals)
# UMAP grid: list of dicts like {"n_neighbors": 10, "min_dist": 0.1}
red_grid_entry = tk.Entry(left)
red_grid_entry.insert(0, """[{"n_neighbors":10,"min_dist":0.05},{"n_neighbors":15,"min_dist":0.1},{"n_neighbors":30,"min_dist":0.3}]""")
# DBSCAN grid: list of dicts like {"eps": 0.5, "min_samples":5}
dbscan_grid_entry = tk.Entry(left)
dbscan_grid_entry.insert(0, """[{"eps":0.3,"min_samples":5},{"eps":0.5,"min_samples":5},{"eps":0.7,"min_samples":3}]""")
# KMeans grid: list of dicts like {"n_clusters": 6}
kmeans_grid_entry = tk.Entry(left)
kmeans_grid_entry.insert(0, """[{"n_clusters":6},{"n_clusters":8},{"n_clusters":10}]""")
# pack rows
_row("src", src_entry)
_row("tables (csv)", tables_entry)
_row("row_limit (blank = all)", row_limit_entry)
ttk.Separator(left, orient="horizontal").pack(fill=tk.X, pady=6)
_row("UMAP n_neighbors", n_neighbors_entry)
_row("UMAP min_dist", min_dist_entry)
_row("metric", metric_entry)
ttk.Separator(left, orient="horizontal").pack(fill=tk.X, pady=6)
_row("clustering", clustering_cbx)
_row("DBSCAN eps", eps_entry)
_row("DBSCAN min_samples", min_samples_entry)
_row("KMeans n_clusters", kmeans_k_entry)
ttk.Separator(left, orient="horizontal").pack(fill=tk.X, pady=6)
_row("color_by (optional)", color_by_entry)
_row("dot_size", dot_size_entry)
_row("figsize (inches)", fig_size_entry)
_row("image_nr (if plotting images)", img_nr_entry)
ttk.Separator(left, orient="horizontal").pack(fill=tk.X, pady=6)
_row("UMAP grid (JSON list of dicts)", red_grid_entry)
_row("DBSCAN grid (JSON list of dicts)", dbscan_grid_entry)
_row("KMeans grid (JSON list of dicts)", kmeans_grid_entry)
# Status + buttons
status = tk.Label(left, text="", bg=self.bg_color, fg=self.fg_color, font=self.font_style)
status.pack(fill=tk.X, pady=(8,4))
btn_row = tk.Frame(left, bg=self.bg_color)
btn_row.pack(fill=tk.X, pady=(0,6))
run_umap_btn = ttk.Button(btn_row, text="Run UMAP")
run_grid_btn = ttk.Button(btn_row, text="Run Hyperparam Search")
run_umap_btn.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=(0,4))
run_grid_btn.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=(4,0))
# ----- runners -----
def _collect_common_settings():
# parse helpers
def _csv_list(s):
s = (s or "").strip()
if not s: return []
return [p.strip() for p in s.split(",") if p.strip()]
def _int_or_none(s):
s = (s or "").strip()
return None if s == "" else int(float(s))
def _float(s, default):
try: return float(str(s).strip())
except Exception: return default
tables = _csv_list(tables_entry.get())
row_limit = _int_or_none(row_limit_entry.get())
settings = {
"src": src_entry.get().strip(),
"tables": tables if tables else ["cell","cytoplasm","nucleus","pathogen"],
"row_limit": row_limit,
"reduction_method": "umap",
"n_neighbors": _int_or_none(n_neighbors_entry.get()) or 15,
"min_dist": _float(min_dist_entry.get(), 0.1),
"metric": metric_entry.get().strip() or "euclidean",
"clustering": clustering_cbx.get().strip().upper(), # DBSCAN or KMEANS
"eps": _float(eps_entry.get(), 0.5),
"min_samples": _int_or_none(min_samples_entry.get()) or 5,
"image_nr": _int_or_none(img_nr_entry.get()) or 200,
"dot_size": _int_or_none(dot_size_entry.get()) or 6,
"figuresize": _float(fig_size_entry.get(), 10.0),
"plot_images": bool(plot_images_var.get()),
"color_by": (color_by_entry.get().strip() or None),
# defaults you already support in set_default_umap_image_settings:
"verbose": True,
"black_background": False,
"remove_image_canvas": False,
"plot_outlines": False,
"plot_points": True,
"smooth_lines": False,
"embedding_by_controls": False,
"exclude": [],
"save_figure": False,
"plot_cluster_grids": False,
"analyze_clusters": False,
"n_jobs": max(1, (os.cpu_count() or 8) - 2),
# clustering-specific extra:
"kmeans_k": _int_or_none(kmeans_k_entry.get()) or 8,
}
return settings
def _run_umap():
"""Kick off a background UMAP run and embed the returned figure."""
settings = _collect_common_settings()
def worker():
"""Thread body: run UMAP and embed the figure or an error."""
try:
status.config(text="Running UMAP…")
# Call your function; ask it to return a Figure (see tweak below)
from spacr.core import generate_image_umap as _gen
fig = _gen(settings=settings, return_fig=True)
status.config(text="Done.")
self._embed_figure_in(right, fig)
except Exception as e:
status.config(text=f"Error: {e}")
threading.Thread(target=worker, daemon=True).start()
def _run_grid():
# parse JSON-ish lists safely
def _parse_list(s):
txt = (s or "").strip()
if not txt: return []
try:
return ast.literal_eval(txt)
except Exception:
return []
settings = _collect_common_settings()
red_grid = _parse_list(red_grid_entry.get())
dbscan_grid = _parse_list(dbscan_grid_entry.get())
kmeans_grid = _parse_list(kmeans_grid_entry.get())
def worker():
"""Thread body: run the hyperparameter search and embed the figure or an error."""
try:
status.config(text="Running hyperparameter search…")
from spacr.core import reducer_hyperparameter_search as _search
fig = _search(
settings=settings,
reduction_params=red_grid or [{"n_neighbors":15,"min_dist":0.1}],
dbscan_params=dbscan_grid or [{"eps":0.5,"min_samples":5}],
kmeans_params=kmeans_grid or [{"n_clusters":settings["kmeans_k"]}],
show=False, return_fig=True
)
status.config(text="Done.")
self._embed_figure_in(right, fig)
except Exception as e:
status.config(text=f"Error: {e}")
threading.Thread(target=worker, daemon=True).start()
run_umap_btn.configure(command=_run_umap)
run_grid_btn.configure(command=_run_grid)
def _poll_save_status(self):
"""Update the status label with saving progress; re-schedules itself."""
with self._batch_lock:
unsaved = self._unsaved_batches
saving = unsaved > 0 or bool(self.pending_updates)
if saving:
self._spinner_idx = (self._spinner_idx + 1) % len(self._spinner_frames)
spin = self._spinner_frames[self._spinner_idx]
msg = f"{spin} Saving… pending={unsaved}"
else:
if self._last_save_ts:
msg = "✓ All changes saved"
else:
msg = ""
self.status_label.config(text=msg)
self.root.after(125, self._poll_save_status)
[docs]
def open_settings_window(self):
"""Open the Toplevel that edits annotation display and filter settings."""
from .gui_utils import generate_annotate_fields, convert_to_number
# ---- Local tooltip implementation (kept here so it is always defined
# ---- when the method runs, regardless of import or reload state) -----
class _ToolTip:
"""Local settings-window tooltip helper (kept inline for reload safety)."""
def __init__(self, widget, text):
"""Bind enter/leave handlers that show/hide the tooltip label."""
self.widget = widget
self.text = text
self.tipwindow = None
# add='+' preserves any existing <Enter>/<Leave> handlers
widget.bind('<Enter>', self._show, add='+')
widget.bind('<Leave>', self._hide, add='+')
def _show(self, _event=None):
"""Create the tooltip Toplevel on pointer enter."""
if self.tipwindow or not self.text:
return
x = self.widget.winfo_rootx() + 20
y = self.widget.winfo_rooty() + self.widget.winfo_height() + 4
tw = tk.Toplevel(self.widget)
tw.wm_overrideredirect(True)
tw.wm_geometry(f"+{x}+{y}")
tk.Label(
tw, text=self.text, justify='left',
background='#222', foreground='#eee',
relief='solid', borderwidth=1, wraplength=420,
font=('TkDefaultFont', 9),
).pack(ipadx=6, ipady=3)
self.tipwindow = tw
def _hide(self, _event=None):
"""Destroy the tooltip Toplevel on pointer leave."""
if self.tipwindow:
self.tipwindow.destroy()
self.tipwindow = None
_SETTING_TOOLTIPS = {
'src': "Path to the experiment source directory. The measurements database is read from <src>/measurements/measurements.db.",
'db_path': "Full path to the SQLite measurements database. Filled automatically from src.",
'image_type': "Substring filter on png_path (e.g. 'cell', 'nucleus'). Leave empty for no filter. A comma-separated list applies all substrings.",
'channels': "Channel names as comma-separated lowercase letters (e.g. 'r,g,b'). Empty disables channel selection.",
'img_size': "Crop or display size as 'width,height' in pixels (e.g. '224,224').",
'annotation_column': "Name of the column in png_list used to store the user annotation.",
'percentiles': "Two comma-separated percentiles for per-image normalization (e.g. '1,99').",
'normalize_channels': "Channels to normalize, as comma-separated 'r','g','b'. Empty disables normalization.",
'outline': "Channels to draw object outlines on, as comma-separated 'r','g','b'. Empty disables outlines.",
'outline_threshold_factor': "Multiplicative factor applied to the outline detection threshold (float, default 1.0).",
'outline_sigma': "Gaussian sigma used during outline extraction (float, default 1.0).",
'edge_thickness': "Outline thickness in pixels (float, default 1).",
'edge_transparency': "Outline transparency in [0, 100]. 0 is fully opaque.",
'edge_image': "Boolean. If true, the outline image is composited on display.",
'object_size': "Object area bounds as 'min,max' in pixels. 0 disables that bound.",
'measurement': (
"Measurement(s) used for prefiltering. Three accepted shapes:\n"
" - single column: e.g. area\n"
" - flat list: e.g. area, intensity_mean (each filtered with the same-index threshold and direction)\n"
" - list of lists (JSON): e.g. [[\"area\"], [\"int_a\", \"int_b\"]] "
"(an inner pair is filtered as a ratio numerator/denominator)\n"
"Use JSON syntax when nesting is needed."
),
'threshold': (
"Threshold value(s). Accepted shapes mirror 'measurement':\n"
" - single value: int, float, or quantile string 'q1'..'q9' (e.g. q3 = 30th percentile)\n"
" - list: e.g. 100, q5 or as JSON [100, \"q5\"]\n"
"Use 'none' to disable threshold filtering."
),
'threshold_direction': (
"Direction(s) for thresholding: 'lower' keeps values <= threshold, 'higher' keeps values >= threshold.\n"
"Single value or comma-separated list matching 'measurement'."
),
}
def _find_label_for(entry_widget, key):
"""Return the Label widget that pairs with this entry.
Looks at the entry's siblings (entry.master.winfo_children()). If
more than one Label is present, prefers the one whose text resembles
the key. Returns None if no Label is found.
"""
try:
siblings = list(entry_widget.master.winfo_children())
except Exception:
return None
labels = [w for w in siblings if isinstance(w, (tk.Label, ttk.Label))]
if not labels:
return None
if len(labels) == 1:
return labels[0]
kname = key.lower().replace('_', ' ')
for lab in labels:
try:
if kname in str(lab.cget('text')).lower():
return lab
except Exception:
continue
return labels[0]
# ---- window ----------------------------------------------------------
settings_window = tk.Toplevel(self.root)
settings_window.title("Modify Annotation Settings")
style_out = set_dark_style(ttk.Style())
settings_window.configure(bg=style_out['bg_color'])
settings_frame = tk.Frame(settings_window, bg=style_out['bg_color'])
settings_frame.pack(fill=tk.BOTH, expand=True)
vars_dict = generate_annotate_fields(settings_frame)
# Add 'threshold_direction' manually if generate_annotate_fields does
# not include it. Keep the label and entry as siblings so the tooltip
# lookup can find the label later.
if 'threshold_direction' not in vars_dict:
row = tk.Frame(settings_frame, bg=style_out['bg_color'])
row.pack(fill=tk.X, padx=6, pady=2)
tk.Label(
row, text='threshold_direction',
bg=style_out['bg_color'], fg=style_out['fg_color'],
width=22, anchor='w'
).pack(side=tk.LEFT)
entry = tk.Entry(
row, bg=style_out['bg_color'], fg=style_out['fg_color'],
insertbackground=style_out['fg_color']
)
entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
vars_dict['threshold_direction'] = {'entry': entry}
# ---- serializers for the flexible shapes -----------------------------
def _serialize_measurement(m):
if m is None or m == '':
return ''
if isinstance(m, str):
return m
if isinstance(m, (list, tuple)):
if any(isinstance(x, (list, tuple)) for x in m):
return json.dumps(m)
return ','.join(map(str, m))
return str(m)
def _serialize_threshold(t):
if t is None:
return ''
if isinstance(t, (list, tuple)):
return json.dumps(list(t))
return str(t)
def _serialize_direction(d):
if d is None or d == '':
return ''
if isinstance(d, (list, tuple)):
return ','.join(map(str, d))
return str(d)
current_settings = {
'image_type': self.image_type or '',
'channels': ','.join(self.channels) if self.channels else '',
'img_size': f"{self.image_size[0]},{self.image_size[1]}",
'annotation_column': self.annotation_column or '',
'percentiles': ','.join(map(str, self.percentiles)),
'measurement': _serialize_measurement(self.measurement),
'threshold': _serialize_threshold(self.threshold),
'threshold_direction': _serialize_direction(getattr(self, 'threshold_direction', None)),
'normalize_channels': ','.join(
[s for s in (self.normalize_channels or []) if isinstance(s, str) and s.strip()]
),
'outline': ','.join(self.outline) if self.outline else '',
'outline_threshold_factor': str(getattr(self, 'outline_threshold_factor', 1.0)),
'outline_sigma': str(getattr(self, 'outline_sigma', 1.0)),
'edge_thickness': str(getattr(self, 'edge_thickness', 1)),
'edge_transparency': str(getattr(self, 'edge_transparency', 0.0)),
'edge_image': str(getattr(self, 'edge_image', False)),
'object_size': f"{getattr(self, 'object_size', (0, 0))[0]},{getattr(self, 'object_size', (0, 0))[1]}",
'src': self.src,
'db_path': self.db_path,
}
# Fill entries and attach tooltips to LABELS (fall back to the entry
# only when no sibling label is found)
for key, data in vars_dict.items():
if key in current_settings:
data['entry'].delete(0, tk.END)
data['entry'].insert(0, current_settings[key])
tip = _SETTING_TOOLTIPS.get(key)
if not tip:
continue
target = _find_label_for(data['entry'], key) or data['entry']
spacrToolTip(target, tip)
# ---- parsers for the flexible shapes ---------------------------------
QUANTILE_TOKENS = {f'q{i}' for i in range(1, 10)}
def _parse_one_threshold(token):
s = str(token).strip()
if s.lower() in QUANTILE_TOKENS:
return s.lower()
try:
return int(s)
except ValueError:
return float(s.replace(',', '.'))
def _parse_measurement(raw):
s = (raw or '').strip()
if not s:
return None
if s.startswith('['):
return json.loads(s)
parts = [p.strip() for p in s.split(',') if p.strip()]
if not parts:
return None
return parts[0] if len(parts) == 1 else parts
def _parse_threshold(raw):
s = (raw or '').strip()
if not s or s.lower() == 'none':
return None
if s.startswith('['):
parsed = json.loads(s)
if isinstance(parsed, list):
return [_parse_one_threshold(x) if not isinstance(x, (int, float)) else x for x in parsed]
return parsed
parts = [p.strip() for p in s.split(',') if p.strip()]
if len(parts) == 1:
return _parse_one_threshold(parts[0])
return [_parse_one_threshold(p) for p in parts]
def _parse_direction(raw):
s = (raw or '').strip()
if not s:
return None
if s.startswith('['):
return json.loads(s)
parts = [p.strip().lower() for p in s.split(',') if p.strip()]
if not parts:
return None
return parts[0] if len(parts) == 1 else parts
def apply_new_settings():
"""Read the settings entries, parse types, and call ``update_settings``."""
settings = {key: data['entry'].get() for key, data in vars_dict.items()}
settings['channels'] = (
[s.strip().lower() for s in (settings.get('channels') or '').split(',') if s.strip()]
or None
)
settings['img_size'] = list(map(int, settings['img_size'].split(',')))
settings['percentiles'] = (
list(map(convert_to_number, settings['percentiles'].split(',')))
if settings['percentiles'] else [1, 99]
)
for key in ('normalize_channels', 'outline'):
raw = settings.get(key)
if raw is None or raw.strip() == '':
settings[key] = []
else:
vals = [s.strip().lower() for s in raw.split(',') if s.strip()]
settings[key] = [s for s in vals if s in {'r', 'g', 'b'}]
def _parse_object_size(s):
if not s:
return (0, 0)
s = s.replace(';', ',')
parts = [p.strip() for p in s.split(',') if p.strip() != '']
nums = []
for p in parts[:2]:
try:
nums.append(max(0, int(float(p))))
except Exception:
nums.append(0)
while len(nums) < 2:
nums.append(0)
mn, mx = nums[0], nums[1]
if mn and mx and mn > mx:
mn, mx = mx, mn
return (mn, mx)
settings['object_size'] = _parse_object_size((settings.get('object_size') or '').strip())
settings['outline_threshold_factor'] = (
float(settings['outline_threshold_factor'].replace(',', '.'))
if settings['outline_threshold_factor'] else 1.0
)
settings['outline_sigma'] = (
float(settings['outline_sigma'].replace(',', '.'))
if settings['outline_sigma'] else 1.0
)
settings['edge_thickness'] = (
float(settings['edge_thickness'].replace(',', '.'))
if settings['edge_thickness'] else 1
)
et = settings.get('edge_transparency')
if et is None or et == '':
settings['edge_transparency'] = 0.0
else:
try:
settings['edge_transparency'] = float(str(et).replace(',', '.'))
except Exception:
settings['edge_transparency'] = 0.0
settings['edge_transparency'] = max(0.0, min(100.0, settings['edge_transparency']))
ei_raw = str(settings.get('edge_image', 'true')).strip().lower()
settings['edge_image'] = ei_raw in ('1', 'true', 't', 'yes', 'y')
try:
settings['measurement'] = _parse_measurement(settings.get('measurement'))
settings['threshold'] = _parse_threshold(settings.get('threshold'))
settings['threshold_direction'] = _parse_direction(settings.get('threshold_direction'))
except (ValueError, json.JSONDecodeError) as e:
print(f"Warning: could not parse measurement/threshold/threshold_direction ({e}); disabling threshold filtering.")
settings['measurement'] = None
settings['threshold'] = None
settings['threshold_direction'] = None
for k, v in list(settings.items()):
if isinstance(v, list):
settings[k] = [x for x in v if x not in (None, '')]
elif v == '':
settings[k] = None
self.db_path = os.path.join(settings.get('src'), 'measurements', 'measurements.db')
self.update_settings(**{
'image_type': settings.get('image_type'),
'channels': settings.get('channels'),
'image_size': settings.get('img_size'),
'annotation_column': settings.get('annotation_column'),
'percentiles': settings.get('percentiles'),
'measurement': settings.get('measurement'),
'threshold': settings.get('threshold'),
'threshold_direction': settings.get('threshold_direction'),
'normalize_channels': settings.get('normalize_channels'),
'outline': settings.get('outline'),
'outline_threshold_factor': settings.get('outline_threshold_factor'),
'outline_sigma': settings.get('outline_sigma'),
'edge_thickness': settings.get('edge_thickness'),
'edge_transparency': settings.get('edge_transparency'),
'edge_image': settings.get('edge_image'),
'object_size': settings.get('object_size'),
'src': settings.get('src'),
'db_path': self.db_path,
})
settings_window.destroy()
apply_button = spacrButton(
settings_window, text="Apply Settings",
command=apply_new_settings, show_text=False, icon_name="annotate"
)
apply_button.pack(pady=10)
def _ensure_annotation_column(self):
"""Add the current annotation column to ``png_list`` if it is missing."""
import sqlite3
if not getattr(self, "annotation_column", None):
return
col = (self.annotation_column or "").replace('"', '""')
with sqlite3.connect(self.db_path, timeout=30) as conn:
cur = conn.cursor()
cur.execute('PRAGMA table_info("png_list")')
cols = {row[1] for row in cur.fetchall()}
if self.annotation_column not in cols:
# NULL allowed; values will be 1/2 per your app
try:
cur.execute(f'ALTER TABLE "png_list" ADD COLUMN "{col}" INTEGER')
# commit occurs automatically on exiting the context if no exception
except sqlite3.OperationalError:
pass # column already exists
[docs]
def update_settings(self, **kwargs):
"""Apply changed settings, coerce types, and re-prime the DB worker.
Only keys in the internal ``allowed_attributes`` set are honored;
``None`` values are ignored. Handles cross-cutting side effects
(rebuilding the grid when ``image_size`` changes, restarting the
writer thread when ``db_path`` changes, resetting pagination when
``src`` changes).
:param kwargs: attribute names mapped to their new values.
"""
import threading
allowed_attributes = {
'image_type', 'channels', 'image_size', 'annotation_column', 'src', 'db_path',
'percentiles', 'measurement', 'threshold', 'normalize_channels',
'outline', 'outline_threshold_factor', 'outline_sigma',
'edge_thickness', 'edge_transparency', 'edge_image', 'object_size'
}
old_db = getattr(self, 'db_path', None)
old_src = getattr(self, 'src', None)
updated = False
for attr, value in kwargs.items():
if attr in allowed_attributes and value is not None:
if attr == 'normalize_channels':
if isinstance(value, (list, tuple)):
value = [str(s).strip().lower() for s in value if s is not None and str(s).strip()]
value = [s for s in value if s in {'r','g','b'}]
value = value or None
elif isinstance(value, str):
parts = [s.strip().lower() for s in value.split(',') if s.strip()]
parts = [s for s in parts if s in {'r','g','b'}]
value = parts or None
else:
value = None
elif attr == 'outline':
if isinstance(value, (list, tuple)):
value = [str(s).strip().lower() for s in value if s is not None and str(s).strip()]
elif isinstance(value, str):
value = [s.strip().lower() for s in value.split(',') if s.strip()]
else:
value = []
value = [s for s in value if s in {'r','g','b'}]
value = value or None
elif attr == 'outline_threshold_factor':
value = float(value)
elif attr == 'outline_sigma':
value = float(value)
# **CHANGED: keep fractional thickness**
elif attr == 'edge_thickness':
value = float(value)
elif attr == 'edge_transparency':
try:
value = float(value)
except Exception:
value = 0.0
value = max(0.0, min(100.0, value))
elif attr == 'edge_image':
value = bool(value)
elif attr == 'object_size':
# normalize to a 2-tuple of non-negative ints; (0,0) means no bounds
v = value
if v in (None, '', []):
v = (0, 0)
elif isinstance(v, str):
# reuse the same parsing logic as above, inline:
s = v.replace(';', ',')
parts = [p.strip() for p in s.split(',') if p.strip() != '']
a = []
for p in parts[:2]:
try:
a.append(max(0, int(float(p))))
except Exception:
a.append(0)
while len(a) < 2:
a.append(0)
mn, mx = a
elif isinstance(v, (list, tuple)):
mn = max(0, int(v[0])) if len(v) > 0 else 0
mx = max(0, int(v[1])) if len(v) > 1 else 0
else:
mn, mx = (0, 0)
if mn and mx and mn > mx:
mn, mx = mx, mn
value = (mn, mx)
setattr(self, attr, value)
updated = True
if ('annotation_column' in kwargs and kwargs['annotation_column']) or ('db_path' in kwargs and kwargs['db_path']):
self._ensure_annotation_column()
if 'image_size' in kwargs:
if isinstance(self.image_size, list):
self.image_size = (int(self.image_size[0]), int(self.image_size[0]))
elif isinstance(self.image_size, int):
self.image_size = (self.image_size, self.image_size)
elif isinstance(self.image_size, tuple) and len(self.image_size) == 2:
self.image_size = tuple(map(int, self.image_size))
else:
raise ValueError("Invalid image size")
self.calculate_grid_dimensions()
self.recreate_image_grid()
if self.src != old_src:
self.adjusted_to_original_paths.clear()
self.index = 0
if self.db_path != old_db:
if self.pending_updates:
with self._batch_lock:
self._unsaved_batches += 1
self.update_queue.put(self.pending_updates.copy())
self.pending_updates.clear()
self.update_queue.put(self.SENTINEL)
self.update_queue.join()
try:
if getattr(self, 'db_update_thread', None):
self.db_update_thread.join()
except Exception:
pass
self.terminate = False
self.worker_busy = False
self._last_save_ts = None
self.db_update_thread = threading.Thread(target=self.update_database_worker, daemon=True)
self.db_update_thread.start()
if updated:
current_index = self.index
self.prefilter_paths_annotations()
max_index = len(self.filtered_paths_annotations) - 1
self.index = min(current_index, max(0, max(len(self.filtered_paths_annotations) - self.grid_rows * self.grid_cols, 0)))
self.load_images()
[docs]
def recreate_image_grid(self):
"""Rebuild the label grid to match current ``grid_rows``/``grid_cols``."""
# Remove current labels
for label in self.labels:
label.destroy()
self.labels.clear()
# Recreate the labels grid with updated dimensions
for i in range(self.grid_rows * self.grid_cols):
label = Label(self.grid_frame, bg=self.root.cget('bg'))
label.grid(row=i // self.grid_cols, column=i % self.grid_cols, padx=2, pady=2, sticky="nsew")
self.labels.append(label)
# Reconfigure grid weights
for row in range(self.grid_rows):
self.grid_frame.grid_rowconfigure(row, weight=1)
for col in range(self.grid_cols):
self.grid_frame.grid_columnconfigure(col, weight=1)
[docs]
def update_display(self):
"""Re-run the prefilter and reload the visible grid."""
self.prefilter_paths_annotations()
self.load_images()
[docs]
def swich_back_annotation_column(self):
"""Restore the originally configured annotation column and refresh."""
self.annotation_column = self.orig_annotation_columns
self._ensure_annotation_column()
self.prefilter_paths_annotations()
self.update_display()
[docs]
def calculate_grid_dimensions(self):
"""Derive ``grid_rows`` and ``grid_cols`` from the current window size."""
self.root.update_idletasks()
w, h = self.root.winfo_width(), self.root.winfo_height()
status_h = self.status_label.winfo_height()
buttons_h = self.button_frame.winfo_height()
bottom_h = max(status_h, buttons_h) + 8 # same row => max
self.grid_cols = max(1, w // (self.image_size[0] + 4))
self.grid_rows = max(1, (h - bottom_h) // (self.image_size[1] + 4))
def _normalize_filter_inputs(self):
"""Coerce self.measurement, self.threshold, self.threshold_direction into compatible shapes.
Returns a structure tag: 'scalar', 'list', or 'list_of_lists'.
Prints a warning whenever an attribute is broadcast or otherwise adjusted.
"""
from .utils import is_list_of_lists
m, t, d = self.measurement, self.threshold, self.threshold_direction
# Scalar measurement
if isinstance(m, str):
if isinstance(t, (list, tuple)):
print(f"Warning: threshold is a list but measurement is a single string; using threshold[0] = {t[0]}.")
t = t[0]
if isinstance(d, (list, tuple)):
print(f"Warning: threshold_direction is a list but measurement is a single string; using threshold_direction[0] = {d[0]}.")
d = d[0]
if d not in ('lower', 'higher'):
raise ValueError(f"threshold_direction must be 'lower' or 'higher', got {d!r}.")
self.measurement, self.threshold, self.threshold_direction = m, t, d
return 'scalar'
# List or list of lists
if isinstance(m, (list, tuple)):
m = list(m)
n = len(m)
if n == 0:
raise ValueError("measurement is an empty list.")
if not isinstance(t, (list, tuple)):
print(f"Warning: threshold is scalar but measurement is a list; broadcasting threshold to length {n}.")
t = [t] * n
else:
t = list(t)
if len(t) != n:
raise ValueError(f"len(threshold) = {len(t)} does not match len(measurement) = {n}.")
if isinstance(d, str):
print(f"Warning: threshold_direction is a string but measurement is a list; broadcasting threshold_direction to length {n}.")
d = [d] * n
else:
d = list(d)
if len(d) != n:
raise ValueError(f"len(threshold_direction) = {len(d)} does not match len(measurement) = {n}.")
for i, di in enumerate(d):
if di not in ('lower', 'higher'):
raise ValueError(f"threshold_direction[{i}] must be 'lower' or 'higher', got {di!r}.")
if is_list_of_lists(m):
for i, inner in enumerate(m):
if not isinstance(inner, (list, tuple)) or len(inner) not in (1, 2):
raise ValueError(
f"measurement[{i}] must be a list of 1 or 2 column names, got {inner!r}."
)
self.measurement, self.threshold, self.threshold_direction = m, t, d
return 'list_of_lists'
self.measurement, self.threshold, self.threshold_direction = m, t, d
return 'list'
raise TypeError(f"measurement must be a string or a list, got {type(m).__name__}.")
def _apply_threshold(self, df, col, threshold, direction):
"""Apply a single threshold filter to df on column col."""
threshold = self._resolve_threshold_value(threshold, df[col])
before = len(df)
if direction == 'lower':
df = df[df[col] <= threshold]
else:
df = df[df[col] >= threshold]
print(f"Filter on '{col}' {direction} {threshold}: removed {before - len(df)} rows, retained {len(df)}.")
return df
def _resolve_threshold_value(self, threshold, series):
"""Resolve a quantile string ('q1'..'q9') against the given series; otherwise return as is."""
if isinstance(threshold, str):
quantile_map = {f'q{i}': i / 10 for i in range(1, 10)}
if threshold in quantile_map:
return series.quantile(quantile_map[threshold])
raise ValueError(
f"Unknown threshold string {threshold!r}. Expected 'q1'..'q9' or a numeric value."
)
return threshold
[docs]
def prefilter_paths_annotations(self):
"""Populate ``filtered_paths_annotations`` from the DB using current filters.
When a measurement + threshold is configured, joins the measurement
tables and applies each configured filter; otherwise pages directly
against ``png_list``. Also honors ``image_type`` substring filtering.
"""
from .io import _read_and_join_tables, _read_db
self._ensure_annotation_column()
if self.measurement and self.threshold is not None:
structure = self._normalize_filter_inputs()
df = _read_and_join_tables(self.db_path)
# Bring in png_path only if the join did not already provide it
if 'png_path' not in df.columns:
png_list_df = _read_db(self.db_path, tables=['png_list'])[0]
# Match on the prcfo column explicitly, regardless of where it lives
# (index vs. column) in either frame
if 'prcfo' not in df.columns and df.index.name == 'prcfo':
df = df.reset_index()
if 'prcfo' not in png_list_df.columns and png_list_df.index.name == 'prcfo':
png_list_df = png_list_df.reset_index()
df = df.merge(
png_list_df[['prcfo', 'png_path']],
on='prcfo',
how='left', # keep all measurement rows
suffixes=('', '_dup'), # never silently rename png_path
)
df[self.annotation_column] = None
print(f"df after merge: {len(df)} rows, columns include png_path: {'png_path' in df.columns}, cell_area range: {df['cell_area'].min()}..{df['cell_area'].max()}")
if structure == 'scalar':
df = self._apply_threshold(df, self.measurement, self.threshold, self.threshold_direction)
elif structure == 'list':
for col, thr, direction in zip(self.measurement, self.threshold, self.threshold_direction):
df = self._apply_threshold(df, col, thr, direction)
elif structure == 'list_of_lists':
for i, (inner, thr, direction) in enumerate(
zip(self.measurement, self.threshold, self.threshold_direction)
):
if len(inner) == 1:
col = inner[0]
else:
col = f'ratio_{i}_{inner[0]}_over_{inner[1]}'
df[col] = df[inner[0]] / df[inner[1]]
df = self._apply_threshold(df, col, thr, direction)
df = df.dropna(subset=['png_path'])
if self.image_type:
image_types = self.image_type if isinstance(self.image_type, list) else [self.image_type]
before = len(df)
for tpe in image_types:
df = df[df['png_path'].str.contains(tpe)]
print(f"image_type '{tpe}': retained {len(df)} entries.")
print(f"image_type filter: removed {before - len(df)} rows, retained {len(df)}.")
self.filtered_paths_annotations = df[['png_path', self.annotation_column]].values.tolist()
self._total_filtered = len(self.filtered_paths_annotations)
else:
col = (self.annotation_column or "").replace('"', '""')
page_size = getattr(self, 'grid_rows', 5) * getattr(self, 'grid_cols', 5)
with sqlite3.connect(self.db_path, timeout=30) as conn:
c = conn.cursor()
if self.image_type:
c.execute(
'SELECT COUNT(*) FROM "png_list" WHERE png_path LIKE ?',
(f"%{self.image_type}%",),
)
else:
c.execute('SELECT COUNT(*) FROM "png_list"')
self._total_filtered = c.fetchone()[0]
if self.image_type:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'WHERE png_path LIKE ? LIMIT ? OFFSET ?',
(f"%{self.image_type}%", page_size, self.index),
)
else:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" LIMIT ? OFFSET ?',
(page_size, self.index),
)
self.filtered_paths_annotations = c.fetchall()
[docs]
def load_images(self):
"""Load and paint the current page of PNGs into the grid labels."""
for label in self.labels:
label.config(image='')
self.images = {}
page_size = self.grid_rows * self.grid_cols
if self.measurement and self.threshold is not None:
paths_annotations = self.filtered_paths_annotations[self.index:self.index + page_size]
else:
paths_annotations = self.filtered_paths_annotations
adjusted_paths = []
for path, annotation in paths_annotations:
if not path.startswith(self.src):
parts = path.split('/data/')
if len(parts) > 1:
new_path = os.path.join(self.src, 'data', parts[1])
self.adjusted_to_original_paths[new_path] = path
adjusted_paths.append((new_path, annotation))
else:
adjusted_paths.append((path, annotation))
else:
adjusted_paths.append((path, annotation))
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
loaded_images = list(executor.map(self.load_single_image, adjusted_paths))
for i, (img, annotation) in enumerate(loaded_images):
border_color = self._label_to_color(annotation)
if border_color:
img = self.add_colored_border(img, border_width=5, border_color=border_color)
from PIL import ImageTk
photo = ImageTk.PhotoImage(img)
label = self.labels[i]
self.images[label] = photo
label.config(image=photo)
path = adjusted_paths[i][0]
label.bind('<Button-1>', self.get_on_image_click(path, label, img))
label.bind(self._right_click_event, self.get_on_image_click(path, label, img))
self.root.update()
[docs]
def show_class_counts(self):
"""Open a window summarizing counts per class in the current column."""
import tkinter as tk
from tkinter import ttk, messagebox
if not self.annotation_column:
messagebox.showerror("Error", "No annotation column is set.")
return
self._ensure_annotation_column()
col = (self.annotation_column or "").replace('"', '""')
with sqlite3.connect(self.db_path, timeout=30) as conn:
cur = conn.cursor()
cur.execute(
f'SELECT "{col}" AS cls, COUNT(*) '
f'FROM "png_list" '
f'WHERE "{col}" IS NOT NULL '
f'GROUP BY "{col}" '
f'ORDER BY 1'
)
rows = cur.fetchall()
win = tk.Toplevel(self.root)
win.title("Class counts (all)")
win.configure(bg=self.root.cget('bg'))
frame = tk.Frame(win, bg=self.root.cget('bg'))
frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
tree = ttk.Treeview(frame, columns=("cls","count","color"), show="headings", height=10)
for cid, text, width in (("cls","Class",80), ("count","Count",100), ("color","Color",120)):
tree.heading(cid, text=text)
tree.column(cid, anchor="center", width=width)
tree.pack(fill=tk.BOTH, expand=True)
# Insert rows with hex color in the last column
for cls, cnt in rows:
try:
cls_int = int(cls)
except Exception:
continue
hexcol = self._label_to_color(cls_int) or ""
tree.insert('', 'end', values=(cls_int, int(cnt), hexcol))
# Simple legend of colored squares
legend = tk.Frame(win, bg=self.root.cget('bg'))
legend.pack(fill=tk.X, padx=10, pady=8)
for cls, _ in rows[:20]: # avoid over-long legends
try:
c = int(cls)
except Exception:
continue
hx = self._label_to_color(c) or "#888888"
sw = tk.Canvas(legend, width=18, height=18, highlightthickness=0, bg=self.root.cget('bg'))
sw.create_rectangle(2, 2, 16, 16, outline=hx, fill=hx)
tk.Label(legend, text=str(c), bg=self.root.cget('bg'), fg=self.fg_color).pack(side="left", padx=(2,8))
sw.pack(side="left")
[docs]
def load_single_image(self, path_annotation_tuple):
"""Load one PNG, apply normalization/channel filtering/outlines, and resize.
:param path_annotation_tuple: ``(path, annotation)`` pair from the DB.
:returns: tuple ``(PIL.Image, annotation)`` sized to ``self.image_size``.
"""
path, annotation = path_annotation_tuple
if not os.path.exists(path):
blank = Image.new('RGB', self.image_size, color=(30, 30, 30))
print(f"Could not find image: {path}")
return blank, annotation
img = Image.open(path)
# Normalize (optionally) – returns RGB ndarray-equivalent in a PIL Image
img = self.normalize_image(img, self.percentiles, self.normalize_channels)
#img = self.normalize_image(img, self.normalize, self.percentiles, self.normalize_channels)
img = img.convert('RGB')
# Keep a copy BEFORE filtering – used for edge generation and for edge_image=True blending
full_img = img
# Apply channel filter for the visible base
img = self.filter_channels(img)
if self.outline:
img = self.outline_image(
base_img=img,
full_img=full_img,
edge_sigma=self.outline_sigma,
edge_thickness=self.edge_thickness,
fill_holes=True,
object_size=getattr(self, "object_size", (0, 0))
)
img = img.resize(self.image_size)
return img, annotation
@staticmethod
[docs]
def fill_holes(mask, min_size=0):
"""Fill interior holes inside True regions of a binary mask.
:param mask: ndarray[bool] mask where True denotes foreground.
:param min_size: minimum hole area in pixels to fill; ``<= 0`` fills
all interior holes, ``> 0`` fills only holes smaller than this
and re-opens larger ones.
:returns: hole-filled boolean mask.
"""
import numpy as np
from scipy.ndimage import binary_fill_holes, label
m = mask.astype(bool)
filled = binary_fill_holes(m)
if min_size <= 0:
return filled
# Pixels that were holes and got filled
filled_holes = filled & ~m
# Reopen (unfill) holes whose area >= min_size
lbl, n = label(filled_holes)
if n == 0:
return filled
reopen = np.zeros_like(m, dtype=bool)
for i in range(1, n + 1):
if (lbl == i).sum() >= int(min_size):
reopen |= (lbl == i)
return filled & ~reopen
@staticmethod
def _filter_objects_by_area(mask, min_size=0, max_size=0):
"""Keep only connected components with area in ``[min_size, max_size]``."""
import numpy as np
from scipy.ndimage import label
m = mask.astype(bool)
if not m.any():
return m
lbl, n = label(m)
if n == 0:
return m
counts = np.bincount(lbl.ravel())
lo = int(min_size) if int(min_size) > 0 else 0
hi = int(max_size) if int(max_size) > 0 else np.iinfo(np.int64).max
keep = np.zeros_like(counts, dtype=bool)
for i in range(1, len(counts)):
area = counts[i]
if lo <= area <= hi:
keep[i] = True
return keep[lbl]
[docs]
def outline_image(self, base_img, full_img, edge_sigma=1, edge_thickness=1, fill_holes=True, object_size=(0, 0)):
"""Composite anti-aliased outlines onto ``base_img`` using ``full_img`` for detection.
Peak-normalizes the outline alpha so brightness is thickness-invariant;
only the global ``edge_transparency`` attribute then attenuates it.
:param base_img: PIL image after channel filtering (visible base).
:param full_img: normalized RGB image before filtering (used for detection).
:param edge_sigma: Gaussian smoothing sigma applied before thresholding.
:param edge_thickness: outline thickness in output pixels (sub-pixel OK).
:param fill_holes: fill internal foreground holes before boundary extraction.
:param object_size: ``(min_px, max_px)`` area filter; 0 disables that bound.
:returns: PIL image with outlines composited into the outline channels.
"""
import numpy as np
from PIL import Image
from scipy.ndimage import gaussian_filter, binary_closing
from skimage.filters import threshold_otsu
from skimage.segmentation import find_boundaries
base_arr = np.asarray(base_img).copy()
full_arr = np.asarray(full_img)
if base_arr.ndim != 3 or base_arr.shape[2] != 3:
return base_img
out_img = base_arr
channel_map = {'r': 0, 'g': 1, 'b': 2}
factor = float(getattr(self, 'outline_threshold_factor', 1.0))
# global opacity 0..1 (100 => fully bright)
transp = float(getattr(self, 'edge_transparency', 0.0))
opacity_global = max(0.0, min(1.0, transp / 100.0))
outline_channels = [ch for ch in (self.outline or []) if ch in channel_map]
show_underlay = bool(getattr(self, 'edge_image', True))
if not show_underlay and outline_channels:
for ch in outline_channels:
out_img[:, :, channel_map[ch]] = 0
if opacity_global == 0.0 or not outline_channels:
from PIL import Image as _Image
return _Image.fromarray(out_img)
# Supersampling factor (AA quality; does NOT widen geometry)
SS = 8
H, W = out_img.shape[:2]
upW, upH = W * SS, H * SS
# unpack object_size bounds
try:
min_px, max_px = object_size if object_size is not None else (0, 0)
except Exception:
min_px, max_px = (0, 0)
for ch in outline_channels:
idx = channel_map[ch]
if show_underlay:
out_img[:, :, idx] = full_arr[:, :, idx]
# Smooth & threshold (original grid)
ch_sm = gaussian_filter(full_arr[:, :, idx].astype(np.float32), sigma=float(edge_sigma))
try:
otsu = threshold_otsu(ch_sm)
except Exception:
otsu = np.percentile(ch_sm, 50.0)
thr = float(min(255.0, max(0.0, otsu * factor)))
fg_mask = (ch_sm > thr)
# Bridge tiny gaps + fill internal holes
fg_mask = binary_closing(fg_mask, structure=np.ones((3, 3), dtype=bool))
if fill_holes:
fg_mask = self.fill_holes(fg_mask, min_size=0)
# Area filtering (keep only sizes within [min_px, max_px], with 0 => no bound)
if (min_px and min_px > 0) or (max_px and max_px > 0):
fg_mask = self._filter_objects_by_area(fg_mask, min_size=min_px, max_size=max_px)
# 1-px boundary (original grid)
edge = find_boundaries(fg_mask, mode='inner').astype(np.uint8)
# Supersample WITHOUT widening: keep a crisp hi-res line
edge_img = Image.fromarray((edge * 255).astype(np.uint8), mode='L')
edge_hi = edge_img.resize((upW, upH), resample=Image.NEAREST)
edge_hi_arr = np.asarray(edge_hi, dtype=np.float32) / 255.0 # {0,1} in hi-res
# Thickness mapping (output px -> hi-res px); only dilate if >= 1 px
desired = max(0.0, float(edge_thickness))
hi_radius = desired * SS
if hi_radius >= 1.0:
from skimage.morphology import dilation, disk
r_int = int(np.floor(hi_radius))
if r_int >= 1:
thick = dilation(edge_hi_arr > 0.5, disk(r_int)).astype(np.float32)
edge_hi_arr = np.maximum(edge_hi_arr, thick)
# Downsample → anti-aliased coverage (0..1)
alpha_lo = Image.fromarray((edge_hi_arr * 255).astype(np.uint8), mode='L') \
.resize((W, H), resample=Image.LANCZOS)
alpha = np.asarray(alpha_lo, dtype=np.float32) / 255.0
# NEVER-DIM: normalize to unit peak
peak = float(alpha.max())
if peak > 0:
alpha = alpha / peak
# Apply global opacity
alpha = np.clip(alpha * opacity_global, 0.0, 1.0)
# Alpha blend onto the channel
orig = out_img[:, :, idx].astype(np.float32)
blended = alpha * 255.0 + (1.0 - alpha) * orig
out_img[:, :, idx] = np.clip(blended, 0, 255).astype(np.uint8)
return Image.fromarray(out_img)
@staticmethod
[docs]
def normalize_image(img, percentiles=(1, 99), normalize_channels=None):
"""Percentile-normalize selected channels of ``img`` and return a PIL image.
No-op when ``normalize_channels`` is falsy.
:param img: input PIL image or array.
:param percentiles: ``(low, high)`` percentiles for rescaling.
:param normalize_channels: iterable subset of ``'r','g','b'``.
:returns: uint8 PIL image.
"""
img_array = np.array(img)
img_array = np.clip(img_array, 0, 255)
if not normalize_channels: # None or []
return Image.fromarray(img_array.astype('uint8'))
if img_array.ndim == 2:
p2, p98 = np.percentile(img_array, percentiles)
out = rescale_intensity(img_array, in_range=(p2, p98), out_range=(0, 255))
return Image.fromarray(np.clip(out, 0, 255).astype('uint8'))
channel_map = {'r': 0, 'g': 1, 'b': 2}
out = img_array.astype(np.float32).copy()
for ch in normalize_channels:
idx = channel_map.get(str(ch).lower())
if idx is None:
continue
p2, p98 = np.percentile(out[:, :, idx], percentiles)
out[:, :, idx] = rescale_intensity(out[:, :, idx], in_range=(p2, p98), out_range=(0, 255))
return Image.fromarray(np.clip(out, 0, 255).astype('uint8'))
[docs]
def add_colored_border(self, img, border_width, border_color):
"""Return ``img`` framed by a solid colored border of the given width.
:param img: source PIL image.
:param border_width: border thickness in pixels on every side.
:param border_color: RGB tuple or hex string for the border fill.
:returns: new PIL image with the border pasted around the source.
"""
top_border = Image.new('RGB', (img.width, border_width), color=border_color)
bottom_border = Image.new('RGB', (img.width, border_width), color=border_color)
left_border = Image.new('RGB', (border_width, img.height), color=border_color)
right_border = Image.new('RGB', (border_width, img.height), color=border_color)
bordered_img = Image.new('RGB', (img.width + 2 * border_width, img.height + 2 * border_width), color=self.fg_color)
bordered_img.paste(top_border, (border_width, 0))
bordered_img.paste(bottom_border, (border_width, img.height + border_width))
bordered_img.paste(left_border, (0, border_width))
bordered_img.paste(right_border, (img.width + border_width, border_width))
bordered_img.paste(img, (border_width, border_width))
return bordered_img
[docs]
def filter_channels(self, img):
"""Zero out channels not present in ``self.channels`` and return an RGB image.
:param img: input PIL image.
:returns: RGB PIL image with unselected channels zeroed.
"""
r, g, b = img.split()
if self.channels:
# normalize and sanitize input like ['R', ' g ', None] -> {'r','g'}
chset = {str(c).strip().lower() for c in self.channels if c is not None and str(c).strip()}
if 'r' not in chset:
r = r.point(lambda _: 0)
if 'g' not in chset:
g = g.point(lambda _: 0)
if 'b' not in chset:
b = b.point(lambda _: 0)
# always return RGB; never collapse to grayscale
return Image.merge("RGB", (r, g, b))
[docs]
def get_on_image_click(self, path, label, img):
"""Return a click handler that toggles the annotation for ``path``.
Left-click sets class 1, right-click sets class 2; clicking the same
button on an already-annotated tile clears the label.
:param path: image path used as the DB row key.
:param label: Tk ``Label`` widget hosting the tile.
:param img: PIL image displayed in the tile.
:returns: event handler callable.
"""
from PIL import ImageTk, ImageOps
import os
def on_image_click(event):
"""Toggle the annotation for the clicked tile and update its border."""
new_annotation = 1 if event.num == 1 else 2
original_path = self.adjusted_to_original_paths.get(path, path)
if original_path in self.pending_updates and self.pending_updates[original_path] == new_annotation:
self.pending_updates[original_path] = None
new_annotation = None
else:
self.pending_updates[original_path] = new_annotation
print(f"Image {os.path.split(path)[1]} annotated: {new_annotation}")
img_ = img.crop((5, 5, img.width - 5, img.height - 5))
border_fill = self._label_to_color(new_annotation)
if border_fill:
img_ = ImageOps.expand(img_, border=5, fill=border_fill)
photo = ImageTk.PhotoImage(img_)
self.images[label] = photo
label.config(image=photo)
self.root.update()
def on_image_click_v1(event):
"""Legacy click handler: left=1, right=2, middle clears (unused)."""
new_annotation = 1 if event.num == 1 else (2 if event.num == 3 else None)
original_path = self.adjusted_to_original_paths.get(path, path)
if original_path in self.pending_updates and self.pending_updates[original_path] == new_annotation:
self.pending_updates[original_path] = None
new_annotation = None
else:
self.pending_updates[original_path] = new_annotation
print(f"Image {os.path.split(path)[1]} annotated: {new_annotation}")
# Remove existing 5px border then reapply with new color (if any)
img_ = img.crop((5, 5, img.width - 5, img.height - 5))
border_fill = self._label_to_color(new_annotation)
if border_fill:
img_ = ImageOps.expand(img_, border=5, fill=border_fill)
photo = ImageTk.PhotoImage(img_)
self.images[label] = photo
label.config(image=photo)
self.root.update()
return on_image_click
@staticmethod
[docs]
def update_html(text):
"""Inject ``text`` into the ``#unique_id`` element via IPython display.
:param text: HTML-safe string to render.
"""
display(HTML(f"""
<script>
document.getElementById('unique_id').innerHTML = '{text}';
</script>
"""))
[docs]
def clear_current_annotation(self):
"""Null every value in the current annotation column after user confirm."""
import sqlite3, queue
from tkinter import messagebox
# Confirm
if not messagebox.askyesno(
"Confirm",
f'This will clear all annotations in "{self.annotation_column}".'
):
return # cancel
# Ensure column exists
self._ensure_annotation_column()
# Null the entire column (context manager => fast close/unlock)
col = (self.annotation_column or "").replace('"', '""')
with sqlite3.connect(self.db_path, timeout=30) as conn:
cur = conn.cursor()
cur.execute(f'UPDATE "png_list" SET "{col}" = NULL')
# Clear any pending updates and drain the queue
self.pending_updates.clear()
try:
while True:
self.update_queue.get_nowait()
except queue.Empty:
pass
with self._batch_lock:
self._unsaved_batches = 0
# Refresh UI (no borders now)
self.prefilter_paths_annotations()
self.load_images()
[docs]
def update_database_worker(self):
"""Background thread that batches pending updates into SQLite commits.
Consumes ``self.update_queue`` until it sees the sentinel, coalescing
multiple queued batches into single WAL-mode transactions.
"""
import sqlite3, queue, time
conn = sqlite3.connect(self.db_path, timeout=30)
cur = conn.cursor()
try:
try:
cur.execute("PRAGMA journal_mode=WAL;")
cur.execute("PRAGMA synchronous=NORMAL;")
conn.commit()
except Exception:
pass
while True:
try:
item = self.update_queue.get(timeout=0.1)
except queue.Empty:
if self.terminate:
break
continue
if item is self.SENTINEL:
self.update_queue.task_done()
break
pending_updates = item
if not pending_updates:
self.update_queue.task_done()
continue
# Coalesce: grab any additional queued batches into one commit
while True:
try:
extra = self.update_queue.get_nowait()
if extra is self.SENTINEL:
self.update_queue.task_done()
self.update_queue.put(self.SENTINEL)
break
if extra:
pending_updates.update(extra)
with self._batch_lock:
self._unsaved_batches -= 1
self.update_queue.task_done()
except queue.Empty:
break
self.worker_busy = True
col = (self.annotation_column or "").replace('"', '""')
to_null = [p for p, v in pending_updates.items() if v is None]
to_set = [(int(v), p) for p, v in pending_updates.items() if v is not None]
try:
if to_null:
cur.executemany(
f'UPDATE "png_list" SET "{col}" = NULL WHERE png_path = ?',
[(p,) for p in to_null]
)
if to_set:
cur.executemany(
f'UPDATE "png_list" SET "{col}" = ? WHERE png_path = ?',
to_set
)
conn.commit()
finally:
with self._batch_lock:
self._unsaved_batches -= 1
self.worker_busy = False
self._last_save_ts = time.time()
self.update_queue.task_done()
finally:
try:
cur.close()
except Exception:
pass
conn.close()
[docs]
def shutdown(self):
"""Flush pending annotations, stop the DB worker, and close the app."""
from tkinter import messagebox
with self._batch_lock:
unsaved = self._unsaved_batches
if unsaved > 0 or bool(self.pending_updates):
if not messagebox.askyesno(
"Updating Database",
"Annotations are still being saved to the database.\n\n"
"Do you want to exit before saving is complete?\n"
"Unsaved annotations will be lost."
):
return # user chose to stay
# push any pending UI updates first
if self.pending_updates:
self.update_queue.put(self.pending_updates.copy())
self.pending_updates.clear()
# signal termination and sentinel
self.terminate = True
self.update_queue.put(self.SENTINEL)
# wait for ALL tasks (including the sentinel) to be marked done
self.update_queue.join()
# now the worker has exited; join without timeout
try:
self.db_update_thread.join()
except Exception:
pass
# close UI
try:
self.root.quit()
finally:
try:
self.root.destroy()
except Exception:
pass
print("Quit application")
[docs]
def skip_to_last_annotated(self):
"""Jump directly to the page containing the last annotated image.
Flushes any pending updates first, then scans the ordered ``png_list``
for the highest-indexed row whose annotation is non-null and non-zero.
"""
print(f"[skip_to_last_annotated] using column: '{self.annotation_column}'")
if self.pending_updates:
with self._batch_lock:
self._unsaved_batches += 1
self.update_queue.put(self.pending_updates.copy())
self.pending_updates.clear()
page_size = self.grid_rows * self.grid_cols
col = (self.annotation_column or "").replace('"', '""')
with sqlite3.connect(self.db_path, timeout=30) as conn:
c = conn.cursor()
# Fetch all png_paths in the same order used by next/previous page,
# then find the index of the last annotated one.
if self.image_type:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'WHERE png_path LIKE ?',
(f"%{self.image_type}%",)
)
else:
c.execute(f'SELECT png_path, "{col}" FROM "png_list"')
all_rows = c.fetchall()
# Find the last row (by position) that has a non-NULL, non-zero annotation
last_annotated_index = None
for i, (path, annotation) in enumerate(all_rows):
if annotation is not None and annotation != 0:
last_annotated_index = i
if last_annotated_index is None:
self.update_gui_text("No annotated images found.")
return
print(f"[skip_to_last_annotated] last annotated at row {last_annotated_index} of {len(all_rows)}")
new_index = (last_annotated_index // page_size) * page_size
self.index = new_index
if not (self.measurement and self.threshold is not None):
with sqlite3.connect(self.db_path, timeout=30) as conn:
c = conn.cursor()
if self.image_type:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'WHERE png_path LIKE ? LIMIT ? OFFSET ?',
(f"%{self.image_type}%", page_size, self.index)
)
else:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'LIMIT ? OFFSET ?',
(page_size, self.index)
)
self.filtered_paths_annotations = c.fetchall()
self.load_images()
[docs]
def next_page(self):
"""Advance to the next page of the grid, flushing pending annotations first."""
if self.pending_updates:
with self._batch_lock:
self._unsaved_batches += 1
self.update_queue.put(self.pending_updates.copy())
self.pending_updates.clear()
page_size = self.grid_rows * self.grid_cols
total = getattr(self, '_total_filtered', len(self.filtered_paths_annotations))
new_index = self.index + page_size
if new_index >= total:
new_index = self.index # already at last page
self.index = new_index
# For the simple (non-measurement) branch, re-fetch the new page from DB
if not (self.measurement and self.threshold is not None):
col = (self.annotation_column or "").replace('"', '""')
with sqlite3.connect(self.db_path, timeout=30) as conn:
c = conn.cursor()
if self.image_type:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'WHERE png_path LIKE ? LIMIT ? OFFSET ?',
(f"%{self.image_type}%", page_size, self.index)
)
else:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'LIMIT ? OFFSET ?',
(page_size, self.index)
)
self.filtered_paths_annotations = c.fetchall()
self.load_images()
[docs]
def previous_page(self):
"""Step back to the previous page of the grid, flushing pending annotations."""
if self.pending_updates:
with self._batch_lock:
self._unsaved_batches += 1
self.update_queue.put(self.pending_updates.copy())
self.pending_updates.clear()
page_size = self.grid_rows * self.grid_cols
self.index = max(0, self.index - page_size)
if not (self.measurement and self.threshold is not None):
col = (self.annotation_column or "").replace('"', '""')
with sqlite3.connect(self.db_path, timeout=30) as conn:
c = conn.cursor()
if self.image_type:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'WHERE png_path LIKE ? LIMIT ? OFFSET ?',
(f"%{self.image_type}%", page_size, self.index)
)
else:
c.execute(
f'SELECT png_path, "{col}" FROM "png_list" '
f'LIMIT ? OFFSET ?',
(page_size, self.index)
)
self.filtered_paths_annotations = c.fetchall()
self.load_images()
[docs]
def update_gui_text(self, text):
"""Update the status label with ``text`` and flush the UI.
:param text: message to display.
"""
self.status_label.config(text=text)
self.root.update()
[docs]
def train_and_classify(self):
"""Train an XGBoost classifier on manual annotations and write predictions.
Merges measurement tables, uses manual labels from
``png_list.<annotation_column>`` (mapping 1->1 and 2->0), fabricates
the missing class by sampling unlabeled rows when only one is present,
trains an ``XGBClassifier``, and writes ``XGboost_score`` /
``XGboost_annotation`` back to ``png_list``.
"""
import sqlite3
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix
from xgboost import XGBClassifier
# Optionally, update your GUI status label
self.update_gui_text("Merging data...")
from .io import _read_and_merge_data
# (1) Merge data
merged_df, obj_df_ls = _read_and_merge_data(
locs=[self.db_path],
tables=['cell', 'cytoplasm', 'nucleus', 'pathogen', 'png_list'],
verbose=False
)
# (2) Load manual annotations from the DB (with context manager)
self._ensure_annotation_column()
colq = (self.annotation_column or "").replace('"', '""')
with sqlite3.connect(self.db_path, timeout=30) as conn:
c = conn.cursor()
c.execute(
f'SELECT png_path, "{colq}" FROM "png_list" '
f'WHERE "{colq}" IS NOT NULL'
)
annotated_rows = c.fetchall()
annot_dict = dict(annotated_rows)
merged_df['manual_annotation'] = merged_df['png_path'].map(annot_dict)
# Subset with manual labels
annotated_df = merged_df.dropna(subset=['manual_annotation']).copy()
annotated_df['manual_annotation'] = annotated_df['manual_annotation'].replace({2: 0}).astype(int)
# (3) Handle single-class scenario
class_counts = annotated_df['manual_annotation'].value_counts()
if len(class_counts) == 1:
single_class = class_counts.index[0] # 0 or 1
needed = class_counts.iloc[0]
other_class = 1 if single_class == 0 else 0
unannotated_df_all = merged_df[merged_df['manual_annotation'].isna()].copy()
if len(unannotated_df_all) == 0:
print("No unannotated rows to sample for the other class. Cannot proceed.")
self.update_gui_text("Not enough data to train (no second class).")
return
sample_size = min(needed, len(unannotated_df_all))
artificially_labeled = unannotated_df_all.sample(n=sample_size, replace=False).copy()
artificially_labeled['manual_annotation'] = other_class
annotated_df = pd.concat([annotated_df, artificially_labeled], ignore_index=True)
print(f"Only one class was present => randomly labeled {sample_size} unannotated rows as {other_class}.")
if len(annotated_df) < 2:
print("Not enough annotated data to train (need at least 2).")
self.update_gui_text("Not enough data to train.")
return
# (4) Train XGBoost
self.update_gui_text("Training XGBoost model...")
# Identify numeric columns
ignore_cols = {'png_path', 'manual_annotation'}
feature_cols = [
col for col in annotated_df.columns
if col not in ignore_cols
and (annotated_df[col].dtype == float or annotated_df[col].dtype == int)
]
X_data = annotated_df[feature_cols].fillna(0).values
y_data = annotated_df['manual_annotation'].values
X_train, X_test, y_train, y_test = train_test_split(
X_data, y_data, test_size=0.1, random_state=42
)
model = XGBClassifier(use_label_encoder=False, eval_metric='logloss')
model.fit(X_train, y_train)
preds = model.predict(X_test)
print("=== Classification Report ===")
print(classification_report(y_test, preds))
print("=== Confusion Matrix ===")
print(confusion_matrix(y_test, preds))
# (5) Classify ALL rows
all_df = merged_df.copy()
X_all = all_df[feature_cols].fillna(0).values
probs_all = model.predict_proba(X_all)[:, 1]
all_df['XGboost_score'] = probs_all
def get_annotation_from_prob(prob):
"""Return 1 above 0.9, 0 below 0.1, else None (uncertain)."""
if prob > 0.9:
return 1
elif prob < 0.1:
return 0
return None
xgb_anno_col = [get_annotation_from_prob(p) for p in probs_all]
xgb_anno_col = [2 if x == 0 else x for x in xgb_anno_col] # convert 0->2
all_df['XGboost_annotation'] = xgb_anno_col
# (6) Write results back (context manager + WAL tuning)
self.update_gui_text("Updating the database with XGBoost predictions...")
with sqlite3.connect(self.db_path, timeout=30) as conn:
c = conn.cursor()
try:
c.execute("ALTER TABLE png_list ADD COLUMN XGboost_annotation INTEGER")
except sqlite3.OperationalError:
pass
try:
c.execute("ALTER TABLE png_list ADD COLUMN XGboost_score FLOAT")
except sqlite3.OperationalError:
pass
c.execute("PRAGMA journal_mode=WAL;")
c.execute("PRAGMA synchronous=NORMAL;")
for _, row in all_df.iterrows():
score_val = float(row['XGboost_score'])
anno_val = row['XGboost_annotation']
the_path = row['png_path']
if pd.isna(the_path):
continue
if pd.isna(anno_val):
c.execute("""
UPDATE png_list
SET XGboost_annotation = NULL,
XGboost_score = ?
WHERE png_path = ?
""", (score_val, the_path))
else:
c.execute("""
UPDATE png_list
SET XGboost_annotation = ?,
XGboost_score = ?
WHERE png_path = ?
""", (int(anno_val), score_val, the_path))
# switch to the new column and (optionally) refresh the view
self.annotation_column = 'XGboost_annotation'
def _get_png_list_columns(self):
"""Return all columns from png_list; caller can decide which are 'annotation'."""
import sqlite3
cols = []
with sqlite3.connect(self.db_path, timeout=30) as conn:
cur = conn.cursor()
cur.execute('PRAGMA table_info("png_list")')
for _, name, coltype, *_ in cur.fetchall():
cols.append((name, (coltype or "").upper()))
return cols
def _parse_field_value(self, key, raw):
"""
Convert string from Entry to the right Python type using defaults as a hint.
Handles bools, ints, floats, lists (comma-separated), and passthrough strings.
"""
if raw is None:
return None
s = str(raw).strip()
if s == "":
return None
low = s.lower()
# booleans
if low in ("true", "t", "1", "yes", "y", "on"):
return True
if low in ("false", "f", "0", "no", "n", "off"):
return False
# numbers
try:
if "." in s or "e" in low:
return float(s)
return int(s)
except Exception:
pass
# lists (comma-separated) for known list keys
listy = {
"classes", "annotated_classes", "class_metadata", "train_channels",
"tables", "file_metadata"
}
if key in listy or ("," in s):
# split, trim, coerce numbers if possible
out = []
for token in s.split(","):
token = token.strip()
if token == "":
continue
try:
if "." in token or "e" in token.lower():
out.append(float(token))
else:
out.append(int(token))
except Exception:
out.append(token)
return out
return s
@staticmethod
[docs]
def convert_settings_dict_for_gui(settings):
"""Classify each setting into a GUI widget spec.
Each entry becomes ``(kind, options, initial)`` where ``kind`` is
``'check'`` for bools, ``'combo'`` for known-choice fields, or
``'entry'`` for free-form values.
:param settings: mapping of setting name to current value.
:returns: dict of ``key -> (kind, options, initial)``.
"""
try:
from torchvision import models as torch_models
torchvision_models = sorted({name for name, obj in torch_models.__dict__.items() if callable(obj)})
except Exception:
torchvision_models = ['resnet18', 'resnet34', 'resnet50', 'densenet121', 'mobilenet_v2']
chan_list = [
'[0,1,2,3,4,5,6,7,8]',
'[0,1,2,3,4,5,6,7]',
'[0,1,2,3,4,5,6]',
'[0,1,2,3,4,5]',
'[0,1,2,3,4]',
'[0,1,2,3]',
'[0,1,2]',
'[0,1]',
'[0]',
'[0,0]'
]
variables = {}
special_cases = {
'metadata_type': ('combo', ['cellvoyager', 'cq1', 'auto', 'custom'], 'cellvoyager'),
'channels': ('combo', chan_list, '[0,1,2,3]'),
'train_channels': ('combo', ["['r','g','b']", "['r','g']", "['r','b']", "['g','b']", "['r']", "['g']", "['b']"], "['r','g','b']"),
'channel_dims': ('combo', chan_list, '[0,1,2,3]'),
'dataset_mode': ('combo', ['annotation', 'metadata', 'measurement'], 'metadata'),
'cov_type': ('combo', ['HC0', 'HC1', 'HC2', 'HC3', None], None),
'crop_mode': ('combo', ["['cell']", "['nucleus']", "['pathogen']", "['organelle']", "['cell', 'nucleus']", "['cell', 'pathogen']", "['cell', 'organelle']", "['nucleus', 'pathogen']", "['cell', 'nucleus', 'pathogen']", "['cell', 'nucleus', 'pathogen', 'organelle']"], "['cell']"),
'timelapse_mode': ('combo', ['trackpy', 'iou', 'btrack'], 'trackpy'),
'train_mode': ('combo', ['erm', 'irm'], 'erm'),
'clustering': ('combo', ['dbscan', 'kmean'], 'dbscan'),
'reduction_method': ('combo', ['umap', 'tsne'], 'umap'),
'model_name': ('combo', ['cyto', 'cyto_2', 'cyto_3', 'nuclei'], 'cyto'),
'regression_type': ('combo', ['ols','gls','wls','rlm','glm','mixed','quantile','logit','probit','poisson','lasso','ridge'], 'ols'),
'timelapse_objects': ('combo', ["['cell']", "['nucleus']", "['pathogen']", "['organelle']", "['cell', 'nucleus']", "['cell', 'pathogen']", "['cell', 'organelle']", "['nucleus', 'pathogen']", "['nucleus', 'organelle']", "['cell', 'nucleus', 'pathogen']", "['cell', 'nucleus', 'organelle']", "['cell', 'nucleus', 'pathogen', 'organelle']"], "['cell']"),
'model_type': ('combo', torchvision_models, 'resnet50'),
'optimizer_type': ('combo', ['adamw', 'adam'], 'adamw'),
'schedule': ('combo', ['cosine','reduce_lr_on_plateau', 'step_lr'], 'cosine'),
'loss_type': ('combo', ['focal_loss', 'binary_cross_entropy_with_logits'], 'focal_loss'),
'normalize_by': ('combo', ['fov', 'png'], 'png'),
'agg_type': ('combo', ['mean', 'median'], 'mean'),
'grouping': ('combo', ['mean', 'median'], 'mean'),
'min_max': ('combo', ['allq', 'all'], 'allq'),
'transform': ('combo', ['log', 'sqrt', 'square', None], None),
'organelle_morphology': ('combo', ['spots', 'network', 'irregular', 'ring'], 'spots'),
'organelle_method': ('combo', ['otsu', 'adaptive', 'log', 'dog', 'ridge', 'hysteresis', 'cellpose', 'unet'], 'otsu'),
'organelle_model_name': ('combo', ['cyto', 'cyto2', 'cyto3', 'nuclei'], 'cyto3'),
'organelle_ridge_filter': ('combo', ['frangi', 'sato', 'meijering'], 'frangi'),
'organelle_network_threshold': ('combo', ['otsu', 'adaptive'], 'otsu'),
'organelle_ring_fill_method': ('combo', ['flood', 'convex'], 'flood'),
'summarize_organelles_by': ('combo', ["['cell']","['nucleus']","['pathogen']","['cytoplasm']","['cell', 'nucleus']","['cell', 'pathogen']","['cell', 'cytoplasm']","['cell', 'nucleus', 'pathogen']","['cell', 'nucleus', 'pathogen', 'cytoplasm']",None], None)
}
for key, value in settings.items():
if key in special_cases:
variables[key] = special_cases[key]
elif isinstance(value, bool):
variables[key] = ('check', None, value)
elif isinstance(value, (int, float)):
variables[key] = ('entry', None, value)
elif isinstance(value, list):
variables[key] = ('entry', None, str(value))
else: # str / None / other
variables[key] = ('entry', None, "" if value is None else value)
return variables
[docs]
def build_multi_annotation(self, source_columns, target_column="multi_annot"):
"""Consolidate several ``{1,2,NULL}`` columns into a single integer code.
Each source contributes a base-3 digit (NULL->0, 1->1, 2->2) so every
combination becomes a unique code ``1 + sum(digit_i * 3**i)``; the
all-zero combination stores NULL. Sets ``self.annotation_column`` to
``target_column`` and refreshes the grid.
:param source_columns: iterable of column names in ``png_list``.
:param target_column: name of the derived column to write.
:raises ValueError: when ``source_columns`` is empty.
"""
import sqlite3
if not source_columns or not isinstance(source_columns, (list, tuple)):
raise ValueError("build_multi_annotation: provide a non-empty list of source columns")
# precompute multipliers 3^i in Python (SQLite lacks POWER())
multipliers = [1]
for _ in range(1, len(source_columns)):
multipliers.append(multipliers[-1] * 3)
# safe identifiers
src_q = [f'"{c.replace(chr(34), chr(34)*2)}"' for c in source_columns]
tgt_q = f'"{target_column.replace(chr(34), chr(34)*2)}"'
# CASE to map each source to {0,1,2}
digits = [f"(CASE {c} WHEN 1 THEN 1 WHEN 2 THEN 2 ELSE 0 END)" for c in src_q]
# sum_i digit_i * 3^i
weighted_sum = " + ".join(f"{digits[i]} * {multipliers[i]}" for i in range(len(digits))) or "0"
# final value: NULL if all zero; else 1 + sum
final_expr = f"CASE WHEN ({weighted_sum}) = 0 THEN NULL ELSE (1 + {weighted_sum}) END"
with sqlite3.connect(self.db_path, timeout=30) as conn:
cur = conn.cursor()
# ensure all source columns exist (as INTEGER, NULL) so SQL won't fail
cur.execute('PRAGMA table_info("png_list")')
have = {row[1] for row in cur.fetchall()}
for col in source_columns:
if col not in have:
cq = col.replace('"','""')
cur.execute(f'ALTER TABLE "png_list" ADD COLUMN "{cq}" INTEGER')
# ensure target column exists
if target_column not in have:
tq = target_column.replace('"','""')
cur.execute(f'ALTER TABLE "png_list" ADD COLUMN "{tq}" INTEGER')
# compute in-place
cur.execute(f'UPDATE "png_list" SET {tgt_q} = {final_expr};')
conn.commit()
# make it the working annotation column and refresh view
self.annotation_column = target_column
self._ensure_annotation_column()
self.prefilter_paths_annotations()
self.load_images()
[docs]
def ensure_multi_annot_from_selection(self, source_columns, target_column="class_column", force_rebuild=True):
"""Pick or build the effective annotation column from a user selection.
A single-column selection is used directly. Multi-column selections
build a consolidated ``target_column``; if that name already exists,
an auto-bumped ``target_column_1``, ``_2``, ... is used instead.
:param source_columns: iterable of column names in ``png_list``.
:param target_column: base name for the consolidated column.
:param force_rebuild: rebuild the consolidated column even if it exists.
:returns: the effective annotation column name that was activated.
:raises ValueError: when ``source_columns`` is empty.
"""
import sqlite3
if not source_columns or not isinstance(source_columns, (list, tuple)):
raise ValueError("ensure_multi_annot_from_selection: provide a non-empty list of source columns")
# Single column => just use it directly
if len(source_columns) == 1:
self.annotation_column = source_columns[0]
self._ensure_annotation_column()
self.prefilter_paths_annotations()
self.load_images()
return self.annotation_column
# Multi-column consolidation: pick a free target name (auto-bump)
with sqlite3.connect(self.db_path, timeout=30) as conn:
cur = conn.cursor()
cur.execute('PRAGMA table_info("png_list")')
existing = {row[1] for row in cur.fetchall()}
base = (str(target_column).strip() or "class_column")
effective = base
suffix = 1
while effective in existing:
effective = f"{base}_{suffix}"
suffix += 1
# Build / refresh the consolidated column under 'effective'
# (build_multi_annotation will set self.annotation_column and refresh UI)
if force_rebuild or self.annotation_column != effective:
self.build_multi_annotation(source_columns, target_column=effective)
else:
self.build_multi_annotation(source_columns, target_column=effective)
# Ensure local state reflects the chosen column name
self.annotation_column = effective
self._ensure_annotation_column()
self.prefilter_paths_annotations()
self.load_images()
return effective
[docs]
def open_deep_spacr_window(self):
"""Open the Deep-SPACR train/apply configuration window.
Presents a notebook of tabs (dataset generation, training, inference)
whose 'Run' hands the resolved settings dict to ``deep_spacr`` on a
background thread.
"""
import tkinter as tk
from tkinter import ttk, messagebox
import sqlite3, threading, ast, json, os
from spacr.settings import deep_spacr_defaults
# ---- defaults ---------------------------------------------------------
defaults = deep_spacr_defaults({})
defaults['src'] = self.src or defaults.get('src')
defaults['dataset'] = defaults.get('dataset', defaults['src'])
defaults['annotation_column'] = self.annotation_column or defaults.get('annotation_column')
# keep your app-wide style usage
style_out = set_dark_style(ttk.Style())
bg = self.bg_color
fg = self.fg_color
font = self.font_style
# ---- window -----------------------------------------------------------
win = tk.Toplevel(self.root)
win.title("Deep SPACR — Train")
win.configure(bg=bg)
win.geometry("1120x760")
outer = tk.Frame(win, bg=bg)
outer.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
# ---- header master toggles (govern tab enablement) --------------------
header = tk.Frame(outer, bg=bg)
header.pack(fill=tk.X, pady=(0,8))
gen_var = tk.BooleanVar(value=bool(defaults.get('generate_training_dataset', True)))
#train_var = tk.BooleanVar(value=bool(defaults.get('train_DL_model', True)))
train_var = tk.BooleanVar(value=bool(defaults.get('train', False) or defaults.get('test', False)))
apply_var = tk.BooleanVar(value=bool(defaults.get('apply_model_to_dataset', True)))
def _chk(label, var):
return tk.Checkbutton(
header, text=label, variable=var,
bg=bg, fg=fg, selectcolor=bg, font=font, activebackground=bg, activeforeground=fg
)
_chk("Generate training dataset", gen_var).pack(side=tk.LEFT, padx=(0,12))
_chk("Train", train_var).pack(side=tk.LEFT, padx=(0,12))
_chk("Apply model to dataset", apply_var).pack(side=tk.LEFT, padx=(0,12))
# ---- notebook ---------------------------------------------------------
nb = ttk.Notebook(outer)
nb.pack(fill=tk.BOTH, expand=True)
# ---- helpers ----------------------------------------------------------
def _label(parent, text):
return tk.Label(parent, text=text, bg=bg, fg=fg, anchor='w', font=font)
def _row(parent, r, label_text, widget):
_label(parent, label_text).grid(row=r, column=0, sticky="w", padx=6, pady=4)
widget.grid(row=r, column=1, sticky="ew", padx=6, pady=4)
parent.grid_columnconfigure(1, weight=1)
def _parse_list_literal(s, fallback=None):
if s is None or str(s).strip() == "":
return fallback
try:
return ast.literal_eval(str(s))
except Exception:
return fallback
def _parse_csv_list(s, fallback=None):
if s is None or str(s).strip() == "":
return fallback
parts = [p.strip() for p in str(s).split(",") if p.strip() != ""]
return parts if parts else fallback
def _set_disabled_state(frame, disabled=True):
# toggle "viability": enable/disable inputs inside the frame
state = tk.DISABLED if disabled else tk.NORMAL
for child in frame.winfo_children():
try:
child.configure(state=state)
except Exception:
pass
# ======================================================================
# TAB 1: Generate training dataset
# ======================================================================
tab_gen = tk.Frame(nb, bg=bg)
nb.add(tab_gen, text="Generate training dataset")
gen_split = tk.PanedWindow(tab_gen, orient=tk.HORIZONTAL, sashwidth=6, bg=bg)
gen_split.pack(fill=tk.BOTH, expand=True)
gen_form = tk.Frame(gen_split, bg=bg)
gen_right = tk.Frame(gen_split, bg=bg)
gen_split.add(gen_form)
gen_split.add(gen_right)
# --- Left column (general) --------------------------------------------
r = 0
dataset_mode_cbx = ttk.Combobox(gen_form, values=['annotation','metadata','measurement'], state='readonly')
dataset_mode_cbx.set(defaults.get('dataset_mode', 'annotation'))
_row(gen_form, r, "dataset_mode", dataset_mode_cbx); r += 1
size_sp = ttk.Spinbox(gen_form, from_=16, to=4096, increment=16)
size_sp.set(int(defaults.get('size', 224)))
_row(gen_form, r, "size (cropped PNG side)", size_sp); r += 1
img_size_sp = ttk.Spinbox(gen_form, from_=16, to=4096, increment=16)
img_size_sp.set(int(defaults.get('image_size', 224)))
_row(gen_form, r, "image_size (model input)", img_size_sp); r += 1
test_split_sp = ttk.Spinbox(gen_form, from_=0.0, to=0.9, increment=0.01)
test_split_sp.set(float(defaults.get('test_split', 0.1)))
_row(gen_form, r, "test_split", test_split_sp); r += 1
sample_sp = ttk.Spinbox(gen_form, from_=0, to=10**9, increment=1)
sample_val = defaults.get('sample', None)
sample_sp.delete(0, tk.END)
sample_sp.insert(0, "" if sample_val in (None, "") else str(sample_val))
_row(gen_form, r, "sample (rows, optional)", sample_sp); r += 1
# FILE TYPE: free text (any string)
file_type_entry = tk.Entry(gen_form)
file_type_entry.insert(0, str(defaults.get('file_type', defaults.get('png_type','cell_png'))))
_row(gen_form, r, "file_type / png_type", file_type_entry); r += 1
tables_entry = tk.Entry(gen_form)
tables_entry.insert(0, "" if defaults.get('tables') in (None, []) else ",".join(defaults.get('tables')))
_row(gen_form, r, "tables (csv)", tables_entry); r += 1
file_metadata_entry = tk.Entry(gen_form)
if defaults.get('file_metadata') not in (None, []):
file_metadata_entry.insert(0, ",".join(defaults['file_metadata']) if isinstance(defaults['file_metadata'], list) else str(defaults['file_metadata']))
_row(gen_form, r, "file_metadata (csv)", file_metadata_entry); r += 1
metadata_type_by_cbx = ttk.Combobox(gen_form, state='readonly', values=['columnID','something_else'])
metadata_type_by_cbx.set(defaults.get('metadata_type_by','columnID'))
_row(gen_form, r, "metadata_type_by", metadata_type_by_cbx); r += 1
class_metadata_entry = tk.Entry(gen_form)
class_metadata_entry.insert(0, str(defaults.get('class_metadata', [['c1'],['c2']])))
_row(gen_form, r, "class_metadata (list-of-lists)", class_metadata_entry); r += 1
classes_entry = tk.Entry(gen_form)
classes_entry.insert(0, str(defaults.get('classes', ['nc','pc'])))
_row(gen_form, r, "classes (list)", classes_entry); r += 1
annotated_classes_entry = tk.Entry(gen_form)
annotated_classes_entry.insert(0, str(defaults.get('annotated_classes', [1,2])))
_row(gen_form, r, "annotated_classes (list)", annotated_classes_entry); r += 1
ch_interest_sp = ttk.Spinbox(gen_form, from_=1, to=5, increment=1)
ch_interest_sp.set(int(defaults.get('channel_of_interest', 3)))
_row(gen_form, r, "channel_of_interest", ch_interest_sp); r += 1
custom_measurement_entry = tk.Entry(gen_form)
if defaults.get('custom_measurement'):
custom_measurement_entry.insert(0, str(defaults['custom_measurement']))
_row(gen_form, r, "custom_measurement (optional)", custom_measurement_entry); r += 1
balance_var = tk.BooleanVar(value=bool(defaults.get('balance_to_smallest', True)))
balance_chk = tk.Checkbutton(gen_form, text="Balance classes to smallest",
variable=balance_var, bg=bg, fg=fg, selectcolor=bg, font=font,
activebackground=bg, activeforeground=fg)
_row(gen_form, r, "", balance_chk); r += 1
# --- Right column: three MODE-SPECIFIC panels -------------------------
# 1) Annotation panel
ann_frame = tk.LabelFrame(gen_right, text="Annotation columns", bg=bg, fg=fg, font=font, labelanchor='n')
ann_inner = tk.Frame(ann_frame, bg=bg)
_label(ann_inner, "Use DB Annotation Columns").grid(row=0, column=0, sticky="w", padx=6, pady=(8,2))
use_db_var = tk.BooleanVar(value=True)
tk.Checkbutton(ann_inner, text="Use selected DB columns as classes",
variable=use_db_var, bg=bg, fg=fg, selectcolor=bg, font=font,
activebackground=bg, activeforeground=fg).grid(row=1, column=0, sticky="w", padx=6, pady=(0,6))
lb = tk.Listbox(ann_inner, selectmode=tk.EXTENDED, height=10,
bg=self.inactive_color, fg=fg, highlightbackground=fg, selectbackground=self.active_color)
lb.grid(row=2, column=0, sticky="nsew", padx=6, pady=(0,8))
ann_inner.grid_columnconfigure(0, weight=1)
ann_inner.grid_rowconfigure(2, weight=1)
try:
with sqlite3.connect(self.db_path, timeout=10) as conn:
cur = conn.cursor()
cur.execute('PRAGMA table_info("png_list")')
for _, name, coltype, *_ in cur.fetchall():
nm = str(name)
if nm.lower() in ('png_path', 'prcfo'):
continue
if (coltype or '').upper().startswith('INT') or nm not in ('png_path',):
lb.insert(tk.END, nm)
except Exception:
pass
ann_inner.pack(fill=tk.BOTH, expand=True)
# 2) Metadata panel
meta_grp = tk.LabelFrame(gen_right, text="Metadata rules (JSON)", bg=bg, fg=fg, font=font, labelanchor='n')
meta_inner = tk.Frame(meta_grp, bg=bg)
meta_rules_entry = tk.Entry(meta_inner)
meta_rules_entry.pack(fill=tk.X, padx=6, pady=(6,4))
ex = tk.Frame(meta_inner, bg=bg)
tk.Label(
ex,
text=("Example:\n"
"[\n"
" {\"name\":\"test_1\", \"where\":[{\"column\":\"test\",\"op\":\"==\",\"value\":1}]},\n"
" {\"name\":\"test_2\", \"where\":[{\"column\":\"test\",\"op\":\"==\",\"value\":2}]},\n"
" {\"name\":\"parasite_1\", \"where\":[{\"column\":\"parasite\",\"op\":\"==\",\"value\":1}]}\n"
"]"),
justify='left', anchor='w', bg=bg, fg=fg
).pack(side=tk.LEFT, fill=tk.X, expand=True)
def _insert_meta_example():
meta_rules_entry.delete(0, tk.END)
meta_rules_entry.insert(
0,
'[{"name":"test_1","where":[{"column":"test","op":"==","value":1}]},'
' {"name":"test_2","where":[{"column":"test","op":"==","value":2}]},'
' {"name":"parasite_1","where":[{"column":"parasite","op":"==","value":1}]}]'
)
ttk.Button(ex, text="Insert example", command=_insert_meta_example).pack(side=tk.RIGHT, padx=6)
ex.pack(fill=tk.X, padx=6, pady=(0,6))
meta_inner.pack(fill=tk.BOTH, expand=True)
# 3) Measurement panel
meas_grp = tk.LabelFrame(gen_right, text="Measurement selection", bg=bg, fg=fg, font=font, labelanchor='n')
meas_inner = tk.Frame(meas_grp, bg=bg)
_label(meas_inner, "measurement (csv: columns)").grid(row=0, column=0, sticky="w", padx=6, pady=(8,2))
meas_cols_entry = tk.Entry(meas_inner)
meas_cols_entry.insert(0, "" if defaults.get('measurement') in (None, []) else (
",".join(defaults['measurement']) if isinstance(defaults['measurement'], list) else str(defaults['measurement'])
))
meas_cols_entry.grid(row=0, column=1, sticky="ew", padx=6, pady=(8,2))
_label(meas_inner, "threshold (float or q1..q9)").grid(row=1, column=0, sticky="w", padx=6, pady=(4,2))
threshold_entry = tk.Entry(meas_inner)
threshold_entry.insert(0, str(defaults.get('threshold', 'q8')))
threshold_entry.grid(row=1, column=1, sticky="ew", padx=6, pady=(4,2))
tk.Label(meas_inner, text="Examples: 0.42 or q7", bg=bg, fg=fg, font=font)\
.grid(row=2, column=1, sticky="w", padx=6, pady=(0,6))
meas_inner.grid_columnconfigure(1, weight=1)
meas_inner.pack(fill=tk.BOTH, expand=True)
# start with correct panel visible & viable
def _toggle_gen_right(*_):
mode = dataset_mode_cbx.get().strip().lower()
# hide all
for w in (ann_frame, meta_grp, meas_grp):
w.pack_forget()
_set_disabled_state(w, disabled=True)
# show + enable chosen
if mode == 'annotation':
ann_frame.pack(fill=tk.BOTH, expand=True, padx=6, pady=(0,8))
_set_disabled_state(ann_frame, disabled=False)
elif mode == 'metadata':
meta_grp.pack(fill=tk.BOTH, expand=True, padx=6, pady=(0,8))
_set_disabled_state(meta_grp, disabled=False)
elif mode == 'measurement':
meas_grp.pack(fill=tk.BOTH, expand=True, padx=6, pady=(0,8))
_set_disabled_state(meas_grp, disabled=False)
dataset_mode_cbx.bind("<<ComboboxSelected>>", _toggle_gen_right)
_toggle_gen_right()
# ======================================================================
# TAB 2: Train
# ======================================================================
tab_train = tk.Frame(nb, bg=bg)
nb.add(tab_train, text="Train")
tr_basic = tk.LabelFrame(tab_train, text="Basic", bg=bg, fg=fg)
tr_basic.pack(fill=tk.X, padx=8, pady=(8,6))
rr = 0
try:
import torchvision
model_names = sorted({n for n, o in getattr(torchvision.models, '__dict__', {}).items() if callable(o)})
except Exception:
model_names = ['resnet18','resnet34','resnet50','densenet121','mobilenet_v2']
model_cbx = ttk.Combobox(tr_basic, state='readonly', values=model_names)
model_cbx.set(defaults.get('model_type', 'resnet50'))
_row(tr_basic, rr, "model_type", model_cbx); rr += 1
epochs_sp = ttk.Spinbox(tr_basic, from_=1, to=2000, increment=1)
epochs_sp.set(int(defaults.get('epochs', 100)))
_row(tr_basic, rr, "epochs", epochs_sp); rr += 1
bs_sp = ttk.Spinbox(tr_basic, from_=1, to=4096, increment=1)
bs_sp.set(int(defaults.get('batch_size', 64)))
_row(tr_basic, rr, "batch_size", bs_sp); rr += 1
lr_sp = ttk.Spinbox(tr_basic, from_=1e-6, to=1e-1, increment=1e-6)
lr_sp.set(float(defaults.get('learning_rate', 1e-3)))
_row(tr_basic, rr, "learning_rate", lr_sp); rr += 1
val_split_sp = ttk.Spinbox(tr_basic, from_=0.0, to=0.9, increment=0.01)
val_split_sp.set(float(defaults.get('val_split', 0.1)))
_row(tr_basic, rr, "val_split", val_split_sp); rr += 1
loss_cbx = ttk.Combobox(tr_basic, state='readonly',
values=["auto","ce","ce_smooth","ce_weighted","focal_ce","bce","focal_bce","logit_adjust_ce", "asl"])
loss_cbx.set(defaults.get('loss_type', 'auto'))
_row(tr_basic, rr, "loss_type", loss_cbx); rr += 1
train_channels_cbx = ttk.Combobox(tr_basic, state='readonly',
values=["['r','g','b']", "['r','g']", "['r','b']", "['g','b']", "['r']", "['g']", "['b']"])
tdef = defaults.get('train_channels', ['r','g','b'])
train_channels_cbx.set(str(tdef if isinstance(tdef, list) else "['r','g','b']"))
_row(tr_basic, rr, "train_channels", train_channels_cbx); rr += 1
do_train_var = tk.BooleanVar(value=bool(defaults.get('train', True)))
do_test_var = tk.BooleanVar(value=bool(defaults.get('test', False)))
_row(tr_basic, rr, "", tk.Checkbutton(tr_basic, text="train (legacy flag)",
variable=do_train_var, bg=bg, fg=fg, selectcolor=bg, font=font)); rr += 1
_row(tr_basic, rr, "", tk.Checkbutton(tr_basic, text="test after training (legacy flag)",
variable=do_test_var, bg=bg, fg=fg, selectcolor=bg, font=font)); rr += 1
adv = tk.LabelFrame(tab_train, text="Advanced", bg=bg, fg=fg)
adv.pack(fill=tk.X, padx=8, pady=(0,8))
ra = 0
opt_cbx = ttk.Combobox(adv, state='readonly', values=['adamw','adagrad','adam'])
opt_cbx.set(defaults.get('optimizer_type', 'adamw'))
_row(adv, ra, "optimizer_type", opt_cbx); ra += 1
sched_cbx = ttk.Combobox(adv, state='readonly', values=['cosine','reduce_lr_on_plateau','step_lr'])
sched_cbx.set(defaults.get('schedule', 'cosine'))
_row(adv, ra, "schedule", sched_cbx); ra += 1
wd_sp = ttk.Spinbox(adv, from_=0.0, to=1.0, increment=1e-6)
wd_sp.set(float(defaults.get('weight_decay', 1e-5)))
_row(adv, ra, "weight_decay", wd_sp); ra += 1
dr_sp = ttk.Spinbox(adv, from_=0.0, to=0.9, increment=0.01)
dr_sp.set(float(defaults.get('dropout_rate', 0.1)))
_row(adv, ra, "dropout_rate", dr_sp); ra += 1
init_w_var = tk.BooleanVar(value=bool(defaults.get('init_weights', True)))
_row(adv, ra, "", tk.Checkbutton(adv, text="init_weights",
variable=init_w_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
use_ckpt_var = tk.BooleanVar(value=bool(defaults.get('use_checkpoint', True)))
_row(adv, ra, "", tk.Checkbutton(adv, text="use_checkpoint (activation checkpointing)",
variable=use_ckpt_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
amsgrad_var = tk.BooleanVar(value=bool(defaults.get('amsgrad', True)))
_row(adv, ra, "", tk.Checkbutton(adv, text="AMSGrad",
variable=amsgrad_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
intermed_var = tk.BooleanVar(value=bool(defaults.get('intermedeate_save', True)))
_row(adv, ra, "", tk.Checkbutton(adv, text="intermedeate_save",
variable=intermed_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
jobs_sp = ttk.Spinbox(adv, from_=0, to=max(1, os.cpu_count() or 64), increment=1)
jobs_sp.set(int(defaults.get('n_jobs', max(1, (os.cpu_count() or 8)-4))))
_row(adv, ra, "n_jobs (DataLoader workers)", jobs_sp); ra += 1
pin_var = tk.BooleanVar(value=bool(defaults.get('pin_memory', False)))
_row(adv, ra, "", tk.Checkbutton(adv, text="pin_memory",
variable=pin_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
ga_sp = ttk.Spinbox(adv, from_=1, to=64, increment=1)
ga_sp.set(int(defaults.get('gradient_accumulation_steps', 4)))
_row(adv, ra, "gradient_accumulation_steps", ga_sp); ra += 1
grad_acc_var = tk.BooleanVar(value=bool(defaults.get('gradient_accumulation', True)))
_row(adv, ra, "", tk.Checkbutton(adv, text="gradient_accumulation",
variable=grad_acc_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
augment_var = tk.BooleanVar(value=bool(defaults.get('augment', False)))
_row(adv, ra, "", tk.Checkbutton(adv, text="augment",
variable=augment_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
normalize_var = tk.BooleanVar(value=bool(defaults.get('normalize', True)))
_row(adv, ra, "", tk.Checkbutton(adv, text="normalize",
variable=normalize_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
verbose_var = tk.BooleanVar(value=bool(defaults.get('verbose', True)))
_row(adv, ra, "", tk.Checkbutton(adv, text="verbose",
variable=verbose_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
custom_model_var = tk.BooleanVar(value=bool(defaults.get('custom_model', False)))
_row(adv, ra, "", tk.Checkbutton(adv, text="custom_model",
variable=custom_model_var, bg=bg, fg=fg, selectcolor=bg, font=font)); ra += 1
custom_model_entry = tk.Entry(adv)
custom_model_entry.insert(0, str(defaults.get('custom_model_path','path')))
_row(adv, ra, "custom_model_path", custom_model_entry); ra += 1
# ======================================================================
# TAB 3: Apply model to dataset
# ======================================================================
tab_apply = tk.Frame(nb, bg=bg)
nb.add(tab_apply, text="Apply model")
apply_frame = tk.LabelFrame(tab_apply, text="Inference", bg=bg, fg=fg)
apply_frame.pack(fill=tk.X, padx=8, pady=8)
rr2 = 0
score_sp = ttk.Spinbox(apply_frame, from_=0.0, to=1.0, increment=0.01)
score_sp.set(float(defaults.get('score_threshold', 0.5)))
_row(apply_frame, rr2, "score_threshold", score_sp); rr2 += 1
dataset_entry = tk.Entry(apply_frame)
dataset_entry.insert(0, str(defaults.get('dataset', defaults['src'])))
_row(apply_frame, rr2, "dataset (apply on this path)", dataset_entry); rr2 += 1
model_path_entry = tk.Entry(apply_frame)
model_path_entry.insert(0, str(defaults.get('model_path','path')))
_row(apply_frame, rr2, "model_path (optional override)", model_path_entry); rr2 += 1
# ---- enable/disable tabs based on header toggles ----------------------
def _apply_tab_state(*_):
nb.tab(0, state='normal' if gen_var.get() else 'disabled')
nb.tab(1, state='normal' if train_var.get() else 'disabled')
nb.tab(2, state='normal' if apply_var.get() else 'disabled')
for var in (gen_var, train_var, apply_var):
var.trace_add("write", _apply_tab_state)
_apply_tab_state()
# ---- bottom buttons ---------------------------------------------------
btns = tk.Frame(win, bg=bg)
btns.pack(fill=tk.X, padx=10, pady=(0,10))
run_btn = ttk.Button(btns, text="Run")
cancel_btn = ttk.Button(btns, text="Cancel", command=win.destroy)
run_btn.pack(side=tk.RIGHT, padx=5)
cancel_btn.pack(side=tk.RIGHT, padx=5)
# ---- run handler ------------------------------------------------------
def on_run():
"""Assemble the settings dict from all tabs and launch ``deep_spacr``."""
settings = dict(defaults) # copy
# Master toggles
settings['generate_training_dataset'] = bool(gen_var.get())
#settings['train_DL_model'] = bool(train_var.get())
settings['apply_model_to_dataset'] = bool(apply_var.get())
# GENERATE / DATASET (shared)
mode = dataset_mode_cbx.get().strip()
settings['dataset_mode'] = mode
settings['size'] = int(float(size_sp.get()))
settings['image_size'] = int(float(img_size_sp.get()))
settings['test_split'] = float(test_split_sp.get())
settings['sample'] = None if str(sample_sp.get()).strip() == "" else int(float(sample_sp.get()))
ft = file_type_entry.get().strip()
settings['file_type'] = ft
settings['png_type'] = ft
settings['tables'] = _parse_csv_list(tables_entry.get(), None)
settings['file_metadata'] = _parse_csv_list(file_metadata_entry.get(), None)
settings['metadata_type_by'] = metadata_type_by_cbx.get().strip()
settings['class_metadata'] = _parse_list_literal(class_metadata_entry.get(), defaults.get('class_metadata'))
settings['classes'] = _parse_list_literal(classes_entry.get(), defaults.get('classes'))
settings['annotated_classes'] = _parse_list_literal(annotated_classes_entry.get(), defaults.get('annotated_classes'))
settings['channel_of_interest'] = int(float(ch_interest_sp.get()))
cm = custom_measurement_entry.get().strip()
settings['custom_measurement'] = (cm if cm != "" else None)
settings['balance_to_smallest'] = bool(balance_var.get())
# MODE-SPECIFIC
if mode == 'annotation':
settings['use_db_columns'] = bool(use_db_var.get())
if settings['use_db_columns']:
sel_cols = [lb.get(i) for i in lb.curselection()]
if not sel_cols:
messagebox.showwarning("No DB columns selected", "Select at least one annotation column or uncheck the DB option.")
return
# Build/choose effective consolidated column name.
# Base name is "class_column"; if it exists, you'll get class_column_1, _2, ...
base_name = "class_column" if len(sel_cols) > 1 else sel_cols[0]
effective_col = self.ensure_multi_annot_from_selection(
sel_cols, target_column=base_name, force_rebuild=True
)
settings['annotation_column'] = effective_col
else:
settings['annotation_column'] = self.annotation_column
# Remove non-annotation keys
settings.pop('metadata_rules', None)
settings.pop('measurement', None)
settings.pop('threshold', None)
elif mode == 'metadata':
raw = meta_rules_entry.get().strip()
rules = None
if raw:
try:
rules = json.loads(raw)
except Exception:
rules = _parse_list_literal(raw, None)
if not rules:
messagebox.showwarning("Metadata rules", "Provide valid JSON rules or click 'Insert example'.")
return
settings['metadata_rules'] = rules
settings.pop('measurement', None)
settings.pop('threshold', None)
settings.pop('annotation_column', None)
settings.pop('db_annotation_columns', None)
settings.pop('use_db_columns', None)
elif mode == 'measurement':
meas_cols = _parse_csv_list(meas_cols_entry.get(), None)
if not meas_cols:
messagebox.showwarning("Measurement", "Provide at least one measurement column (csv).")
return
settings['measurement'] = meas_cols if len(meas_cols) > 1 else meas_cols[0]
th_raw = threshold_entry.get().strip()
if th_raw == "":
messagebox.showwarning("Measurement", "Provide a threshold (number) or a quantile code q1..q9.")
return
try:
settings['threshold'] = float(th_raw)
except Exception:
settings['threshold'] = th_raw # e.g. "q8"
settings.pop('metadata_rules', None)
settings.pop('annotation_column', None)
settings.pop('db_annotation_columns', None)
settings.pop('use_db_columns', None)
# TRAIN
settings['model_type'] = model_cbx.get().strip()
settings['epochs'] = int(float(epochs_sp.get()))
settings['batch_size'] = int(float(bs_sp.get()))
settings['learning_rate'] = float(lr_sp.get())
settings['val_split'] = float(val_split_sp.get())
settings['loss_type'] = loss_cbx.get().strip()
settings['train_channels'] = _parse_list_literal(train_channels_cbx.get(), ['r','g','b'])
settings['train'] = bool(do_train_var.get()) # legacy flag
settings['test'] = bool(do_test_var.get()) # legacy flag
settings['optimizer_type'] = opt_cbx.get().strip()
settings['schedule'] = sched_cbx.get().strip()
settings['weight_decay'] = float(wd_sp.get())
settings['dropout_rate'] = float(dr_sp.get())
settings['init_weights'] = bool(init_w_var.get())
settings['use_checkpoint'] = bool(use_ckpt_var.get())
settings['amsgrad'] = bool(amsgrad_var.get())
settings['intermedeate_save'] = bool(intermed_var.get())
settings['n_jobs'] = int(float(jobs_sp.get()))
settings['pin_memory'] = bool(pin_var.get())
settings['gradient_accumulation_steps'] = int(float(ga_sp.get()))
settings['gradient_accumulation'] = bool(grad_acc_var.get())
settings['augment'] = bool(augment_var.get())
settings['normalize'] = bool(normalize_var.get())
settings['verbose'] = bool(verbose_var.get())
settings['custom_model'] = bool(custom_model_var.get())
settings['custom_model_path'] = custom_model_entry.get().strip() or settings.get('custom_model_path')
# APPLY
settings['score_threshold'] = float(score_sp.get())
settings['dataset'] = dataset_entry.get().strip() or self.src
mp = model_path_entry.get().strip()
if mp:
settings['model_path'] = mp
# Essentials
settings['src'] = self.src
win.destroy()
def _worker():
try:
self.update_gui_text("Deep SPACR: preparing…")
from spacr.deep_spacr import deep_spacr
deep_spacr(settings)
self.update_gui_text("Deep SPACR: done.")
except Exception as e:
import traceback
traceback.print_exc()
self.update_gui_text(f"Deep SPACR error: {e}")
threading.Thread(target=_worker, daemon=True).start()
run_btn.configure(command=on_run)
[docs]
def generate_dna_matrix(output_path='dna_matrix.gif', canvas_width=1500, canvas_height=1000, duration=30, fps=20, base_size=20, transition_frames=30, font_type='arial.ttf', enhance=None, lowercase_prob=0.3):
"""Render a Matrix-style DNA-base rain animation and save it to disk.
The output format is inferred from the ``output_path`` extension
(``.gif``, ``.mp4``, or ``.avi``); videos are written via OpenCV.
:param output_path: destination path; extension picks the format.
:param canvas_width: frame width in pixels.
:param canvas_height: frame height in pixels.
:param duration: total animation length in seconds.
:param fps: frames per second.
:param base_size: glyph size in pixels (also the column stride).
:param transition_frames: number of blended frames appended for looping.
:param font_type: font family or file used for the glyphs.
:param enhance: optional ``[brightness, sharpness, contrast, color]``
multipliers applied per frame.
:param lowercase_prob: probability a glyph is rendered lowercase.
"""
if enhance is None:
enhance = [1.1, 1.5, 1.2, 1.5]
def save_output(frames, output_path, fps, output_format):
"""Save the animation based on output format."""
if output_format in ['.mp4', '.avi']:
images = [np.array(img.convert('RGB')) for img in frames]
fourcc = cv2.VideoWriter_fourcc(*('mp4v' if output_format == '.mp4' else 'XVID'))
out = cv2.VideoWriter(output_path, fourcc, fps, (canvas_width, canvas_height))
for img in images:
out.write(cv2.cvtColor(img, cv2.COLOR_RGB2BGR))
out.release()
elif output_format == '.gif':
frames[0].save(output_path, save_all=True, append_images=frames[1:], duration=int(1000/fps), loop=0)
def draw_base(draw, col_idx, base_position, base, font, alpha=255, fill_color=None):
"""Draws a DNA base at the specified position."""
draw.text((col_idx * base_size, base_position * base_size), base, fill=(*fill_color, alpha), font=font)
# Setup variables
num_frames = duration * fps
num_columns = canvas_width // base_size
bases = ['A', 'T', 'C', 'G']
active_color = (155, 55, 155)
color = (255, 255, 255)
base_colors = {'A': color, 'T': color, 'C': color, 'G': color}
_, output_format = os.path.splitext(output_path)
# Initialize font
try:
font = ImageFont.truetype(font_type, base_size)
except IOError:
font = ImageFont.load_default()
# DNA string and positions
string_lengths = [random.randint(10, 100) for _ in range(num_columns)]
visible_bases = [0] * num_columns
base_positions = [random.randint(-canvas_height // base_size, 0) for _ in range(num_columns)]
column_strings = [[''] * 100 for _ in range(num_columns)]
random_white_sequences = [None] * num_columns
frames = []
end_frame_start = int(num_frames * 0.8)
for frame_idx in range(num_frames):
img = Image.new('RGBA', (canvas_width, canvas_height), color=(0, 0, 0, 255))
draw = ImageDraw.Draw(img)
for col_idx in range(num_columns):
if base_positions[col_idx] >= canvas_height // base_size and frame_idx < end_frame_start:
string_lengths[col_idx] = random.randint(10, 100)
base_positions[col_idx] = -string_lengths[col_idx]
visible_bases[col_idx] = 0
# Randomly choose whether to make each base lowercase
column_strings[col_idx] = [
random.choice([base.lower(), base]) if random.random() < lowercase_prob else base
for base in [random.choice(bases) for _ in range(string_lengths[col_idx])]
]
if string_lengths[col_idx] > 8:
random_start = random.randint(0, string_lengths[col_idx] - 8)
random_white_sequences[col_idx] = range(random_start, random_start + 8)
last_10_percent_start = max(0, int(string_lengths[col_idx] * 0.9))
for row_idx in range(min(visible_bases[col_idx], string_lengths[col_idx])):
base_position = base_positions[col_idx] + row_idx
if 0 <= base_position * base_size < canvas_height:
base = column_strings[col_idx][row_idx]
if base:
if row_idx == visible_bases[col_idx] - 1:
draw_base(draw, col_idx, base_position, base, font, fill_color=active_color)
elif row_idx >= last_10_percent_start:
alpha = 255 - int(((row_idx - last_10_percent_start) / (string_lengths[col_idx] - last_10_percent_start)) * 127)
draw_base(draw, col_idx, base_position, base, font, alpha=alpha, fill_color=base_colors[base.upper()])
elif random_white_sequences[col_idx] and row_idx in random_white_sequences[col_idx]:
draw_base(draw, col_idx, base_position, base, font, fill_color=active_color)
else:
draw_base(draw, col_idx, base_position, base, font, fill_color=base_colors[base.upper()])
if visible_bases[col_idx] < string_lengths[col_idx]:
visible_bases[col_idx] += 1
base_positions[col_idx] += 2
# Convert the image to numpy array to check unique pixel values
img_array = np.array(img)
if len(np.unique(img_array)) > 2: # Only append frames with more than two unique pixel values (avoid black frames)
# Enhance contrast and saturation
if enhance:
img = ImageEnhance.Brightness(img).enhance(enhance[0]) # Slightly increase brightness
img = ImageEnhance.Sharpness(img).enhance(enhance[1]) # Sharpen the image
img = ImageEnhance.Contrast(img).enhance(enhance[2]) # Enhance contrast
img = ImageEnhance.Color(img).enhance(enhance[3]) # Boost color saturation
frames.append(img)
for i in range(transition_frames):
alpha = i / float(transition_frames)
transition_frame = Image.blend(frames[-1], frames[0], alpha)
frames.append(transition_frame)
save_output(frames, output_path, fps, output_format)