Compare commits

..

No commits in common. "master" and "Stitching" have entirely different histories.

4 changed files with 315 additions and 441 deletions

BIN
Mos1.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 513 KiB

BIN
Mos2.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 194 KiB

View file

@ -88,18 +88,6 @@ class AppGUI:
on_mosaic_updated=self._on_mosaic_updated
)
# Update initial memory estimate
self._update_memory_estimate()
def _update_memory_estimate(self):
"""Update memory estimate label"""
bytes_per_pixel = 3
mem_mb = (self.stitch_config.max_mosaic_width *
self.stitch_config.max_mosaic_height *
bytes_per_pixel) / (1024 * 1024)
if hasattr(self, 'memory_label'):
self.memory_label.config(text=f"~{mem_mb:.0f} MB")
def _on_command_sent(self, cmd):
self.log_message(f"> {cmd}")
@ -289,54 +277,63 @@ class AppGUI:
command=self._show_mosaic_window).pack(side=tk.RIGHT, padx=2)
def _build_row2_stitch_settings(self, parent):
"""Row 2: Stitching settings with max dimensions"""
"""Row 2: Stitching settings"""
row = ttk.LabelFrame(parent, text="Stitch Settings")
row.pack(fill=tk.X, pady=(0, 3))
# First row: Threshold, Speed, Overlap
inner1 = ttk.Frame(row)
inner1.pack(fill=tk.X, padx=5, pady=3)
inner = ttk.Frame(row)
inner.pack(fill=tk.X, padx=5, pady=3)
# Displacement threshold
ttk.Label(inner1, text="Threshold:").pack(side=tk.LEFT)
ttk.Label(inner, text="Threshold:").pack(side=tk.LEFT)
self.disp_threshold_var = tk.DoubleVar(value=0.10)
self.disp_threshold_spinbox = ttk.Spinbox(
inner1, from_=0.05, to=0.30, increment=0.01, width=5,
inner, from_=0.05, to=0.30, increment=0.01, width=5,
textvariable=self.disp_threshold_var,
command=self._update_stitch_config
)
self.disp_threshold_spinbox.pack(side=tk.LEFT, padx=(2, 10))
# Scan speed
ttk.Label(inner1, text="Speed:").pack(side=tk.LEFT)
self.scan_speed_var = tk.IntVar(value=3)
self.scan_speed_spinbox = ttk.Spinbox(
inner1, from_=1, to=6, width=3,
textvariable=self.scan_speed_var,
# Number of rows
ttk.Label(inner, text="Rows:").pack(side=tk.LEFT)
self.num_rows_var = tk.IntVar(value=3)
self.num_rows_spinbox = ttk.Spinbox(
inner, from_=1, to=10, width=3,
textvariable=self.num_rows_var,
command=self._update_stitch_config
)
self.scan_speed_spinbox.pack(side=tk.LEFT, padx=(2, 10))
self.num_rows_spinbox.pack(side=tk.LEFT, padx=(2, 10))
# Row overlap
ttk.Label(inner1, text="Overlap:").pack(side=tk.LEFT)
ttk.Label(inner, text="Overlap:").pack(side=tk.LEFT)
self.row_overlap_var = tk.DoubleVar(value=0.15)
self.row_overlap_spinbox = ttk.Spinbox(
inner1, from_=0.05, to=0.50, increment=0.05, width=5,
inner, from_=0.05, to=0.50, increment=0.05, width=5,
textvariable=self.row_overlap_var,
command=self._update_stitch_config
)
self.row_overlap_spinbox.pack(side=tk.LEFT, padx=(2, 10))
# Scan speed
ttk.Label(inner, text="Speed:").pack(side=tk.LEFT)
self.scan_speed_var = tk.IntVar(value=3)
self.scan_speed_spinbox = ttk.Spinbox(
inner, from_=1, to=6, width=3,
textvariable=self.scan_speed_var,
command=self._update_stitch_config
)
self.scan_speed_spinbox.pack(side=tk.LEFT, padx=(2, 10))
# Autofocus toggle
self.af_every_row_var = tk.BooleanVar(value=True)
ttk.Checkbutton(
inner1, text="AF each row",
inner, text="AF each row",
variable=self.af_every_row_var,
command=self._update_stitch_config
).pack(side=tk.LEFT, padx=(10, 0))
# Row/Direction status on right
status_frame = ttk.Frame(inner1)
status_frame = ttk.Frame(inner)
status_frame.pack(side=tk.RIGHT)
ttk.Label(status_frame, text="Row:").pack(side=tk.LEFT)
@ -347,49 +344,6 @@ class AppGUI:
self.direction_label = ttk.Label(status_frame, text="--", width=6, font=('Arial', 9))
self.direction_label.pack(side=tk.LEFT)
# Second row: Max dimensions and test buttons
inner2 = ttk.Frame(row)
inner2.pack(fill=tk.X, padx=5, pady=(0, 3))
# Max Width
ttk.Label(inner2, text="Max W:").pack(side=tk.LEFT)
self.max_width_var = tk.IntVar(value=5000)
self.max_width_entry = ttk.Entry(inner2, textvariable=self.max_width_var, width=7)
self.max_width_entry.pack(side=tk.LEFT, padx=(2, 5))
self.max_width_entry.bind('<Return>', lambda e: self._update_stitch_config())
# Max Height
ttk.Label(inner2, text="Max H:").pack(side=tk.LEFT)
self.max_height_var = tk.IntVar(value=5000)
self.max_height_entry = ttk.Entry(inner2, textvariable=self.max_height_var, width=7)
self.max_height_entry.pack(side=tk.LEFT, padx=(2, 5))
self.max_height_entry.bind('<Return>', lambda e: self._update_stitch_config())
# Apply button
ttk.Button(inner2, text="Apply", width=5,
command=self._update_stitch_config).pack(side=tk.LEFT, padx=(5, 10))
ttk.Separator(inner2, orient=tk.VERTICAL).pack(side=tk.LEFT, fill=tk.Y, padx=5)
# Test buttons
ttk.Label(inner2, text="Test:").pack(side=tk.LEFT)
ttk.Button(inner2, text="Row↓", width=5,
command=self._test_row_transition).pack(side=tk.LEFT, padx=2)
ttk.Button(inner2, text="Scan→", width=5,
command=lambda: self._test_horizontal('right')).pack(side=tk.LEFT, padx=2)
ttk.Button(inner2, text="Scan←", width=5,
command=lambda: self._test_horizontal('left')).pack(side=tk.LEFT, padx=2)
ttk.Button(inner2, text="Est", width=4,
command=self._show_scan_estimate).pack(side=tk.LEFT, padx=2)
# Memory estimate label
self.memory_label = ttk.Label(inner2, text="~-- MB", font=('Arial', 9))
self.memory_label.pack(side=tk.RIGHT)
def _build_row3_movement(self, parent):
"""Row 3: Movement controls for all axes"""
row = ttk.LabelFrame(parent, text="Movement")
@ -557,15 +511,11 @@ class AppGUI:
def _update_stitch_config(self):
"""Update stitching scanner config from GUI values"""
self.stitch_config.displacement_threshold = self.disp_threshold_var.get()
self.stitch_config.max_mosaic_width = self.max_width_var.get()
self.stitch_config.max_mosaic_height = self.max_height_var.get()
self.stitch_config.rows = self.num_rows_var.get()
self.stitch_config.row_overlap = self.row_overlap_var.get()
self.stitch_config.scan_speed_index = self.scan_speed_var.get()
self.stitch_config.autofocus_every_row = self.af_every_row_var.get()
# Update memory estimate
self._update_memory_estimate()
# Reinitialize stitching scanner with new config
self.stitch_scanner = StitchingScanner(
camera=self.camera,
@ -577,54 +527,6 @@ class AppGUI:
on_mosaic_updated=self._on_mosaic_updated
)
self.log_message(f"Config updated: {self.stitch_config.max_mosaic_width}x{self.stitch_config.max_mosaic_height}")
def _test_row_transition(self):
"""Test row transition without full scan"""
if not self.stitch_scanner:
self._update_stitch_config()
self.log_message("Testing row transition...")
def run_test():
result = self.stitch_scanner.test_row_transition()
self.root.after(0, lambda: self.log_message(
f"Row transition: {'SUCCESS' if result['success'] else 'FAILED'}, "
f"Y moved: {result['y_moved']:.1f}px"))
self.root.after(0, self._update_mosaic_window)
threading.Thread(target=run_test, daemon=True).start()
def _test_horizontal(self, direction: str):
"""Test single row scan"""
if not self.stitch_scanner:
self._update_stitch_config()
self.log_message(f"Testing single row scan ({direction})...")
def run_test():
result = self.stitch_scanner.test_single_row(direction)
self.root.after(0, lambda: self.log_message(
f"Row scan: {result['stop_reason']}, {result['appends']} appends, "
f"mosaic: {result['mosaic_after'][0]}x{result['mosaic_after'][1]}"))
self.root.after(0, self._update_mosaic_window)
threading.Thread(target=run_test, daemon=True).start()
def _show_scan_estimate(self):
"""Show memory estimate based on current settings"""
if not self.stitch_scanner:
self._update_stitch_config()
est = self.stitch_scanner.get_memory_estimate()
msg = (f"Memory Estimate:\n"
f" Current: {est['current_size'][0]}x{est['current_size'][1]} = {est['current_mb']:.1f} MB\n"
f" Max: {est['max_size'][0]}x{est['max_size'][1]} = {est['max_mb']:.0f} MB")
self.log_message(msg)
self.memory_label.config(text=f"~{est['max_mb']:.0f} MB")
# =========================================================================
# Overlay Drawing
# =========================================================================
@ -1024,16 +926,10 @@ class AppGUI:
text=f"X: {state.cumulative_x:.1f} Y: {state.cumulative_y:.1f}"
)
# Update row/direction and size
# Update row/direction
if state.is_scanning:
# Show current row and progress percentage
width_pct = min(100, (state.mosaic_width / self.stitch_config.max_mosaic_width) * 100)
height_pct = min(100, (state.mosaic_height / self.stitch_config.max_mosaic_height) * 100)
self.row_label.config(text=f"R{state.current_row + 1}")
self.row_label.config(text=f"{state.current_row + 1}/{state.total_rows}")
self.direction_label.config(text=state.direction)
# Update mosaic size label with progress
self.mosaic_size_label.config(
text=f"{state.mosaic_width}x{state.mosaic_height} ({height_pct:.0f}%)")
else:
self.row_label.config(text="--")
self.direction_label.config(text="--")

View file

@ -1,59 +1,84 @@
"""
Stitching Scanner v2 - Simplified unified approach
Stitching Scanner v2 - Fixed displacement tracking
Same displacement-based stitching for both horizontal rows and vertical row transitions.
No complex visual matching - just track displacement and append strips.
Key fix: Track displacement since last APPEND, not just cumulative.
The strip width must match actual movement since we last added to the mosaic.
"""
import cv2
import numpy as np
import time
import threading
from dataclasses import dataclass
from typing import Optional, Callable, Tuple
from dataclasses import dataclass, field
from typing import List, Optional, Callable, Tuple
from enum import Enum
class ScanDirection(Enum):
RIGHT = 'right'
LEFT = 'left'
DOWN = 'down'
UP = 'up'
"""Scan direction constants"""
RIGHT = 'right' # X+ (E command)
LEFT = 'left' # X- (W command)
DOWN = 'down' # Y- (N command)
UP = 'up' # Y+ (S command)
@dataclass
class StitchConfig:
displacement_threshold: float = 0.10 # 10% of frame triggers append
movement_interval: float = 0.001
frame_interval: float = 1.00
settle_time: float = 0.75
max_scan_time: float = 300.0
"""Stitching scanner configuration"""
# Displacement threshold (percentage of frame size)
displacement_threshold: float = 0.10 # 10% of frame dimension
# Movement timing
movement_interval: float = 0.001 # Seconds of motor on time
frame_interval: float = 0.25 # Seconds between frame captures (settle time)
settle_time: float = 0.5 # Seconds to wait after stopping
max_scan_time: float = 2400.0 # Safety timeout (5 minutes)
# Scan pattern
rows: int = 3
row_overlap: float = 0.15
max_mosaic_width: int = 15000
max_mosaic_height: int = 12000
# Speed setting for scanning
scan_speed_index: int = 3
# Focus
autofocus_every_row: bool = True
# Memory management
max_mosaic_width: int = 11000
max_mosaic_height: int = 11000
# 11000, 24500, 450000
@dataclass
class StitchState:
"""Current state for visualization"""
is_scanning: bool = False
direction: str = ''
# Displacement tracking
cumulative_x: float = 0.0
cumulative_y: float = 0.0
last_displacement: Tuple[float, float] = (0.0, 0.0)
# Progress
current_row: int = 0
total_rows: int = 0
# Mosaic size
mosaic_width: int = 0
mosaic_height: int = 0
# Debug
frame_count: int = 0
append_count: int = 0
class StitchingScanner:
"""
Slide scanner using continuous stitching.
Unified approach for horizontal and vertical movement.
Slide scanner using continuous stitching with correct displacement tracking.
Key insight: We must track displacement since the LAST APPEND, and the
strip we append must exactly match that displacement.
"""
def __init__(self, camera, motion_controller, autofocus_controller=None,
@ -66,25 +91,32 @@ class StitchingScanner:
self.autofocus = autofocus_controller
self.config = config or StitchConfig()
# Callbacks
self.on_log = on_log
self.on_progress = on_progress
self.on_mosaic_updated = on_mosaic_updated
# State
self.running = False
self.paused = False
self.state = StitchState()
self._state_lock = threading.Lock()
# Mosaic data
self.mosaic: Optional[np.ndarray] = None
self._mosaic_lock = threading.Lock()
self._prev_frame: Optional[np.ndarray] = None
self._displacement_since_append_x: float = 0.0
# Frame tracking - KEY CHANGE: separate reference for displacement calc vs append
self._prev_frame: Optional[np.ndarray] = None # For frame-to-frame displacement
self._append_ref_frame: Optional[np.ndarray] = None # Reference from last append
self._displacement_since_append_x: float = 0.0 # Accumulated since last append
self._displacement_since_append_y: float = 0.0
# Thread
self._thread: Optional[threading.Thread] = None
def log(self, message: str):
"""Log a message"""
if self.on_log:
self.on_log(f"[Stitch] {message}")
print(f"[Stitch] {message}")
@ -94,12 +126,17 @@ class StitchingScanner:
# =========================================================================
def _to_grayscale(self, frame: np.ndarray) -> np.ndarray:
"""Convert frame to grayscale"""
if len(frame.shape) == 3:
return cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
return frame
def _detect_displacement(self, prev_frame: np.ndarray,
curr_frame: np.ndarray) -> Tuple[float, float]:
"""
Detect displacement between two frames using phase correlation.
Returns (dx, dy) in pixels.
"""
prev_gray = self._to_grayscale(prev_frame)
curr_gray = self._to_grayscale(curr_frame)
@ -109,16 +146,20 @@ class StitchingScanner:
prev_f = prev_gray.astype(np.float32)
curr_f = curr_gray.astype(np.float32)
# Apply window function to reduce edge effects
h, w = prev_gray.shape
window = cv2.createHanningWindow((w, h), cv2.CV_32F)
prev_f = prev_f * window
curr_f = curr_f * window
shift, _ = cv2.phaseCorrelate(prev_f, curr_f)
return shift
shift, response = cv2.phaseCorrelate(prev_f, curr_f)
dx, dy = shift
return (dx, dy)
def _detect_displacement_robust(self, prev_frame: np.ndarray,
curr_frame: np.ndarray) -> Tuple[float, float]:
"""Displacement detection with sanity checks"""
dx, dy = self._detect_displacement(prev_frame, curr_frame)
h, w = prev_frame.shape[:2]
@ -131,14 +172,17 @@ class StitchingScanner:
return (dx, dy)
# =========================================================================
# Mosaic Building
# Mosaic Building - FIXED VERSION
# =========================================================================
def _init_mosaic(self, frame: np.ndarray):
"""Initialize mosaic with first frame"""
with self._mosaic_lock:
self.mosaic = frame.copy()
# Set reference frames
self._prev_frame = frame.copy()
self._append_ref_frame = frame.copy()
self._displacement_since_append_x = 0.0
self._displacement_since_append_y = 0.0
@ -151,17 +195,20 @@ class StitchingScanner:
self.log(f"Initialized mosaic: {frame.shape[1]}x{frame.shape[0]}")
def _blend_horizontal(self, base: np.ndarray, strip: np.ndarray,
def _blend_strips_horizontal(self, base: np.ndarray, strip: np.ndarray,
blend_width: int, append_right: bool) -> np.ndarray:
"""Blend strip onto base with gradient at seam to hide discontinuities."""
if blend_width <= 0 or blend_width >= strip.shape[1]:
if append_right:
return np.hstack([base, strip])
else:
return np.hstack([strip, base])
h_base, w_base = base.shape[:2]
h_strip, w_strip = strip.shape[:2]
if h_strip != h_base:
# Height mismatch - can't blend properly
if append_right:
return np.hstack([base, strip])
return np.hstack([strip, base])
@ -169,80 +216,53 @@ class StitchingScanner:
blend_w = min(blend_width, w_strip, w_base)
if append_right:
# base | blend_zone | rest_of_strip
result_width = w_base + w_strip - blend_w
result = np.zeros((h_base, result_width, 3), dtype=np.uint8)
# Copy base
result[:, :w_base] = base
# Create gradient: 1->0 for base weight
alpha = np.linspace(1, 0, blend_w, dtype=np.float32)[np.newaxis, :, np.newaxis]
base_overlap = base[:, -blend_w:].astype(np.float32)
strip_overlap = strip[:, :blend_w].astype(np.float32)
blended = (base_overlap * alpha + strip_overlap * (1 - alpha)).astype(np.uint8)
result[:, w_base - blend_w:w_base] = blended
result[:, w_base:] = strip[:, blend_w:]
return result
else:
# rest_of_strip | blend_zone | base
result_width = w_base + w_strip - blend_w
result = np.zeros((h_base, result_width, 3), dtype=np.uint8)
result[:, :w_strip] = strip
alpha = np.linspace(0, 1, blend_w, dtype=np.float32)[np.newaxis, :, np.newaxis]
strip_overlap = strip[:, -blend_w:].astype(np.float32)
base_overlap = base[:, :blend_w].astype(np.float32)
blended = (strip_overlap * (1 - alpha) + base_overlap * alpha).astype(np.uint8)
result[:, w_strip - blend_w:w_strip] = blended
result[:, w_strip:] = base[:, blend_w:]
return result
def _blend_vertical(self, base: np.ndarray, strip: np.ndarray,
blend_height: int, append_below: bool) -> np.ndarray:
mh, mw = base.shape[:2]
sh, sw = strip.shape[:2]
def _append_to_mosaic_fixed(self, frame: np.ndarray, direction: ScanDirection):
"""
FIXED: Append with blending and fractional pixel preservation.
# Match widths
if sw > mw:
strip = strip[:, :mw]
elif sw < mw:
pad = np.zeros((sh, mw - sw, 3), dtype=np.uint8)
strip = np.hstack([strip, pad])
blend_h = min(blend_height, sh, mh)
if blend_h <= 0:
if append_below:
return np.vstack([base, strip])
return np.vstack([strip, base])
if append_below:
alpha = np.linspace(1, 0, blend_h, dtype=np.float32)[:, np.newaxis, np.newaxis]
base_overlap = base[-blend_h:].astype(np.float32)
strip_overlap = strip[:blend_h].astype(np.float32)
blended = (base_overlap * alpha + strip_overlap * (1 - alpha)).astype(np.uint8)
result_h = mh + sh - blend_h
result = np.zeros((result_h, mw, 3), dtype=np.uint8)
result[:mh - blend_h] = base[:-blend_h]
result[mh - blend_h:mh] = blended
result[mh:] = strip[blend_h:]
return result
else:
alpha = np.linspace(0, 1, blend_h, dtype=np.float32)[:, np.newaxis, np.newaxis]
strip_overlap = strip[-blend_h:].astype(np.float32)
base_overlap = base[:blend_h].astype(np.float32)
blended = (strip_overlap * (1 - alpha) + base_overlap * alpha).astype(np.uint8)
result_h = mh + sh - blend_h
result = np.zeros((result_h, mw, 3), dtype=np.uint8)
result[:sh - blend_h] = strip[:-blend_h]
result[sh - blend_h:sh] = blended
result[sh:] = base[blend_h:]
return result
def _append_strip(self, frame: np.ndarray, direction: ScanDirection):
"""Append strip to mosaic based on accumulated displacement."""
BLEND_WIDTH = 10
SAFETY_MARGIN = 2
Key improvements:
1. Gradient blending at seams to hide color discontinuities
2. Preserve fractional pixel remainder to prevent cumulative drift
3. Small safety margin for alignment tolerance
"""
BLEND_WIDTH = 10 # Pixels to blend at seam
SAFETY_MARGIN = 2 # Extra pixels as tolerance
with self._mosaic_lock:
if self.mosaic is None:
@ -255,26 +275,30 @@ class StitchingScanner:
dy = abs(self._displacement_since_append_y)
if direction in [ScanDirection.RIGHT, ScanDirection.LEFT]:
# Round and add safety margin
append_width = round(dx) + SAFETY_MARGIN
append_width = min(append_width, w - BLEND_WIDTH - 5)
if append_width < 1:
return
# Calculate fractional remainder to preserve
pixels_consumed = append_width - SAFETY_MARGIN
fractional_remainder = dx - pixels_consumed
if direction == ScanDirection.RIGHT:
# Grab strip with extra for blending
strip_start = max(0, w - append_width - BLEND_WIDTH)
new_strip = frame[:, strip_start:]
self.mosaic = self._blend_horizontal(
self.mosaic = self._blend_strips_horizontal(
self.mosaic, new_strip, BLEND_WIDTH, append_right=True)
else:
strip_end = min(w, append_width + BLEND_WIDTH)
new_strip = frame[:, :strip_end]
self.mosaic = self._blend_horizontal(
self.mosaic = self._blend_strips_horizontal(
self.mosaic, new_strip, BLEND_WIDTH, append_right=False)
# KEEP fractional remainder instead of resetting to 0!
self._displacement_since_append_x = fractional_remainder
self._displacement_since_append_y = 0.0
@ -289,34 +313,112 @@ class StitchingScanner:
fractional_remainder = dy - pixels_consumed
if direction == ScanDirection.DOWN:
strip_end = min(h, append_height + BLEND_WIDTH)
new_strip = frame[:strip_end:, :]
self.mosaic = self._blend_vertical(
self.mosaic, new_strip, BLEND_WIDTH, append_below=False)
else:
strip_start = max(0, h - append_height - BLEND_WIDTH)
new_strip = frame[:strip_start, :]
self.mosaic = self._blend_vertical(
self.mosaic, new_strip, BLEND_WIDTH, append_below=True)
new_strip = frame[strip_start:, :]
# Match widths
if new_strip.shape[1] > mw:
new_strip = new_strip[:, :mw]
elif new_strip.shape[1] < mw:
pad = np.zeros((new_strip.shape[0], mw - new_strip.shape[1], 3), dtype=np.uint8)
new_strip = np.hstack([new_strip, pad])
# Vertical blend
blend_h = min(BLEND_WIDTH, new_strip.shape[0], mh)
alpha = np.linspace(1, 0, blend_h, dtype=np.float32)[:, np.newaxis, np.newaxis]
base_overlap = self.mosaic[-blend_h:].astype(np.float32)
strip_overlap = new_strip[:blend_h].astype(np.float32)
blended = (base_overlap * alpha + strip_overlap * (1 - alpha)).astype(np.uint8)
result_h = mh + new_strip.shape[0] - blend_h
result = np.zeros((result_h, mw, 3), dtype=np.uint8)
result[:mh - blend_h] = self.mosaic[:-blend_h]
result[mh - blend_h:mh] = blended
result[mh:] = new_strip[blend_h:]
self.mosaic = result
else:
strip_end = min(h, append_height + BLEND_WIDTH)
new_strip = frame[:strip_end, :]
if new_strip.shape[1] > mw:
new_strip = new_strip[:, :mw]
elif new_strip.shape[1] < mw:
pad = np.zeros((new_strip.shape[0], mw - new_strip.shape[1], 3), dtype=np.uint8)
new_strip = np.hstack([new_strip, pad])
self.mosaic = np.vstack([new_strip, self.mosaic])
self._displacement_since_append_x = 0.0
self._displacement_since_append_y = fractional_remainder
new_mh, new_mw = self.mosaic.shape[:2]
# Update state
with self._state_lock:
self.state.mosaic_width = new_mw
self.state.mosaic_height = new_mh
self.state.append_count += 1
# Update reference frame (fractional remainder already set above - don't reset!)
self._append_ref_frame = frame.copy()
if self.on_mosaic_updated:
self.on_mosaic_updated()
def _start_new_row(self, frame: np.ndarray, direction: ScanDirection):
"""Start a new row in the mosaic"""
with self._mosaic_lock:
if self.mosaic is None:
self._init_mosaic(frame)
return
h, w = frame.shape[:2]
mh, mw = self.mosaic.shape[:2]
# Calculate overlap
overlap_pixels = int(h * self.config.row_overlap)
append_height = h - overlap_pixels
if direction == ScanDirection.DOWN:
new_strip = frame[overlap_pixels:, :]
if new_strip.shape[1] < mw:
pad = np.zeros((new_strip.shape[0], mw - new_strip.shape[1], 3), dtype=np.uint8)
new_strip = np.hstack([new_strip, pad])
elif new_strip.shape[1] > mw:
new_strip = new_strip[:, :mw]
self.mosaic = np.vstack([self.mosaic, new_strip])
else:
new_strip = frame[:append_height, :]
if new_strip.shape[1] < mw:
pad = np.zeros((new_strip.shape[0], mw - new_strip.shape[1], 3), dtype=np.uint8)
new_strip = np.hstack([new_strip, pad])
elif new_strip.shape[1] > mw:
new_strip = new_strip[:, :mw]
self.mosaic = np.vstack([new_strip, self.mosaic])
# Reset all tracking for new row
self._prev_frame = frame.copy()
self._append_ref_frame = frame.copy()
self._displacement_since_append_x = 0.0
self._displacement_since_append_y = 0.0
with self._state_lock:
self.state.mosaic_height = self.mosaic.shape[0]
self.state.mosaic_width = self.mosaic.shape[1]
self.log(f"New row started, mosaic: {self.mosaic.shape[1]}x{self.mosaic.shape[0]}")
# =========================================================================
# Scan Control
# =========================================================================
def start(self) -> bool:
"""Start the stitching scan"""
if self.running:
self.log("Already running")
return False
@ -327,11 +429,13 @@ class StitchingScanner:
with self._state_lock:
self.state = StitchState()
self.state.is_scanning = True
self.state.total_rows = self.config.rows
with self._mosaic_lock:
self.mosaic = None
self._prev_frame = None
self._append_ref_frame = None
self._displacement_since_append_x = 0.0
self._displacement_since_append_y = 0.0
@ -342,6 +446,7 @@ class StitchingScanner:
return True
def stop(self):
"""Stop the scan"""
self.running = False
self.paused = False
self.motion.stop_all()
@ -352,24 +457,26 @@ class StitchingScanner:
self.log("Scan stopped")
def pause(self):
"""Pause the scan"""
if self.running and not self.paused:
self.paused = True
self.motion.stop_all()
self.log("Scan paused")
def resume(self):
"""Resume the scan"""
if self.running and self.paused:
self.paused = False
self.log("Scan resumed")
# =========================================================================
# Scanning Logic
# Main Scan Loop
# =========================================================================
def _scan_loop(self):
"""Main scanning loop"""
try:
self.log("Starting scan loop")
self.log(f"Max dimensions: {self.config.max_mosaic_width}x{self.config.max_mosaic_height}")
self.motion.set_speed(self.config.scan_speed_index)
time.sleep(0.1)
@ -377,35 +484,30 @@ class StitchingScanner:
frame = self._capture_frame()
self._init_mosaic(frame)
row = 0
while self.running:
for row in range(self.config.rows):
if not self.running:
break
with self._state_lock:
self.state.current_row = row
self.state.total_rows = row + 1
self.log(f"=== Row {row + 1} ===")
self.log(f"=== Row {row + 1}/{self.config.rows} ===")
# Serpentine: even rows right, odd rows left
h_direction = ScanDirection.RIGHT if row % 2 == 0 else ScanDirection.LEFT
# Serpentine pattern
if row % 2 == 0:
h_direction = ScanDirection.RIGHT
else:
h_direction = ScanDirection.LEFT
stop_reason = self._scan_direction(h_direction)
self._scan_horizontal(h_direction)
if not self.running:
break
# Check max height
if self.state.mosaic_height >= self.config.max_mosaic_height:
self.log(f"Max height reached ({self.state.mosaic_height}px)")
break
if row < self.config.rows - 1:
self._move_to_next_row()
# Move to next row using same stitching approach
if not self._move_to_next_row():
self.log("Failed to move to next row")
break
row += 1
self.log(f"Scan complete! Final: {self.state.mosaic_width}x{self.state.mosaic_height}")
self.log("Scan complete!")
except Exception as e:
self.log(f"Scan error: {e}")
@ -417,8 +519,8 @@ class StitchingScanner:
with self._state_lock:
self.state.is_scanning = False
def _scan_direction(self, direction: ScanDirection) -> str:
"""Scan in a direction until edge or max dimension reached."""
def _scan_horizontal(self, direction: ScanDirection):
"""Scan horizontally with fixed displacement tracking"""
self.log(f"Scanning {direction.value}...")
with self._state_lock:
@ -426,51 +528,44 @@ class StitchingScanner:
frame = self._capture_frame()
h, w = frame.shape[:2]
# Setup based on direction
if direction in [ScanDirection.RIGHT, ScanDirection.LEFT]:
threshold_pixels = w * self.config.displacement_threshold
max_dim = self.config.max_mosaic_width
current_dim = lambda: self.state.mosaic_width
start_cmd = 'E' if direction == ScanDirection.RIGHT else 'W'
stop_cmd = 'e' if direction == ScanDirection.RIGHT else 'w'
else:
threshold_pixels = h * self.config.displacement_threshold
max_dim = self.config.max_mosaic_height
current_dim = lambda: self.state.mosaic_height
start_cmd = 'S' if direction == ScanDirection.DOWN else 'N'
stop_cmd = 's' if direction == ScanDirection.DOWN else 'n'
# Initialize tracking
self._prev_frame = frame.copy()
self._append_ref_frame = frame.copy()
self._displacement_since_append_x = 0.0
self._displacement_since_append_y = 0.0
start_time = time.time()
no_movement_count = 0
max_no_movement = 50
stop_reason = 'stopped'
while self.running and not self.paused:
if time.time() - start_time > self.config.max_scan_time:
self.log("Scan timeout")
stop_reason = 'timeout'
break
if current_dim() >= max_dim:
self.log(f"Max dimension reached ({current_dim()}px)")
stop_reason = 'max_dim'
break
# Pulse the motor
if direction == ScanDirection.RIGHT:
self.motion.send_command('E')
else:
self.motion.send_command('W')
# Pulse motor
self.motion.send_command(start_cmd)
time.sleep(self.config.movement_interval)
self.motion.send_command(stop_cmd)
if direction == ScanDirection.RIGHT:
self.motion.send_command('e')
else:
self.motion.send_command('w')
# Wait for settle
time.sleep(self.config.frame_interval)
# Capture and measure
curr_frame = self._capture_frame()
dx, dy = self._detect_displacement_robust(self._prev_frame, curr_frame)
# Accumulate displacement SINCE LAST APPEND
self._displacement_since_append_x += dx
self._displacement_since_append_y += dy
@ -480,119 +575,78 @@ class StitchingScanner:
self.state.last_displacement = (dx, dy)
self.state.frame_count += 1
# Edge detection
movement = abs(dx) if direction in [ScanDirection.RIGHT, ScanDirection.LEFT] else abs(dy)
if movement < 1.0:
# Check for no movement
if abs(dx) < 1.0 and abs(dy) < 1.0:
no_movement_count += 1
if no_movement_count >= max_no_movement:
self.log(f"Edge detected (no movement)")
stop_reason = 'edge'
self.log(f"Edge detected (no movement for {no_movement_count} frames)")
break
else:
no_movement_count = 0
# Append when threshold reached
disp = abs(self._displacement_since_append_x) if direction in [ScanDirection.RIGHT, ScanDirection.LEFT] else abs(self._displacement_since_append_y)
if disp >= threshold_pixels:
self._append_strip(curr_frame, direction)
self.log(f"Appended {disp:.1f}px, mosaic: {self.state.mosaic_width}x{self.state.mosaic_height}")
# Check threshold and append
if abs(self._displacement_since_append_x) >= threshold_pixels:
self._append_to_mosaic_fixed(curr_frame, direction)
self.log(f"Appended {abs(self._displacement_since_append_x):.1f}px strip, "
f"mosaic: {self.state.mosaic_width}x{self.state.mosaic_height}")
# Update prev_frame for next displacement calculation
self._prev_frame = curr_frame.copy()
if self.on_progress:
self.on_progress(self.state.append_count, 0)
self.motion.send_command(stop_cmd)
time.sleep(self.config.settle_time)
self.log(f"Direction finished: {stop_reason}")
return stop_reason
# Stop
if direction == ScanDirection.RIGHT:
self.motion.send_command('e')
else:
self.motion.send_command('w')
def _move_to_next_row(self) -> bool:
"""
Move down to next row using displacement-based stitching.
Same approach as horizontal scanning.
"""
time.sleep(self.config.settle_time)
def _move_to_next_row(self):
"""Move down to next row"""
self.log("Moving to next row...")
frame = self._capture_frame()
h, w = frame.shape[:2]
# Target: move (1 - overlap) * frame_height
target_displacement = h * (1 - self.config.row_overlap)
threshold_pixels = h * self.config.displacement_threshold
self.log(f"Target Y: {target_displacement:.0f}px, threshold: {threshold_pixels:.0f}px")
move_distance = h * (1 - self.config.row_overlap)
with self._state_lock:
self.state.direction = 'down'
self.state.cumulative_y = 0.0
self.motion.send_command('N')
self._prev_frame = frame.copy()
self._displacement_since_append_x = 0.0
self._displacement_since_append_y = 0.0
cumulative_y = 0.0
total_y = 0.0
no_movement_count = 0
max_no_movement = 30
# Start moving South
self.motion.send_command('S')
try:
while self.running:
time.sleep(self.config.frame_interval)
curr_frame = self._capture_frame()
dx, dy = self._detect_displacement_robust(self._prev_frame, curr_frame)
self._displacement_since_append_y += dy
total_y += dy
with self._state_lock:
self.state.cumulative_y = total_y
self.state.last_displacement = (dx, dy)
# Edge detection
if abs(dy) < 1.0:
no_movement_count += 1
if no_movement_count >= max_no_movement:
self.log("Edge detected during row transition")
self.motion.send_command('s')
time.sleep(self.config.settle_time)
return False
else:
no_movement_count = 0
# Append strip when threshold reached
if abs(self._displacement_since_append_y) >= threshold_pixels:
self._append_strip(curr_frame, ScanDirection.DOWN)
self.log(f" Row transition: appended, total Y: {abs(total_y):.1f}px")
# Done when we've moved enough
if abs(total_y) >= target_displacement:
self.log(f"Row transition complete: {abs(total_y):.1f}px")
self.motion.send_command('s')
time.sleep(self.config.settle_time)
# Reset for next horizontal row
frame = self._capture_frame()
self._prev_frame = frame.copy()
self._displacement_since_append_x = 0.0
self._displacement_since_append_y = 0.0
return True
cumulative_y += dy
self._prev_frame = curr_frame.copy()
except Exception as e:
self.log(f"Row transition error: {e}")
self.motion.send_command('s')
return False
with self._state_lock:
self.state.cumulative_y = cumulative_y
self.motion.send_command('s')
if abs(cumulative_y) >= move_distance:
break
if abs(cumulative_y) < 5 and self.state.frame_count > 50:
self.log("Warning: Minimal Y movement")
break
self.motion.send_command('n')
time.sleep(self.config.settle_time)
return False
frame = self._capture_frame()
self._start_new_row(frame, ScanDirection.DOWN)
def _capture_frame(self) -> np.ndarray:
"""Capture and rotate frame"""
frame = self.camera.capture_frame()
frame = cv2.rotate(frame, cv2.ROTATE_90_CLOCKWISE)
return frame
@ -602,6 +656,7 @@ class StitchingScanner:
# =========================================================================
def get_state(self) -> StitchState:
"""Get current scan state"""
with self._state_lock:
return StitchState(
is_scanning=self.state.is_scanning,
@ -618,12 +673,14 @@ class StitchingScanner:
)
def get_mosaic(self) -> Optional[np.ndarray]:
"""Get current mosaic (full resolution)"""
with self._mosaic_lock:
if self.mosaic is not None:
return self.mosaic.copy()
return None
def get_mosaic_preview(self, max_size: int = 600) -> Optional[np.ndarray]:
"""Get scaled mosaic for preview"""
with self._mosaic_lock:
if self.mosaic is None:
return None
@ -639,9 +696,11 @@ class StitchingScanner:
return self.mosaic.copy()
def save_mosaic(self, filepath: str) -> bool:
"""Save mosaic to file"""
with self._mosaic_lock:
if self.mosaic is None:
return False
cv2.imwrite(filepath, self.mosaic)
self.log(f"Saved mosaic to {filepath}")
return True
@ -651,106 +710,25 @@ class StitchingScanner:
# =========================================================================
def test_displacement(self, num_frames: int = 10) -> dict:
results = {'frames': [], 'total_dx': 0.0, 'total_dy': 0.0}
"""Test displacement detection"""
results = {
'frames': [],
'total_dx': 0.0,
'total_dy': 0.0
}
prev_frame = self._capture_frame()
for i in range(num_frames):
time.sleep(0.1)
curr_frame = self._capture_frame()
dx, dy = self._detect_displacement(prev_frame, curr_frame)
results['frames'].append({'frame': i, 'dx': dx, 'dy': dy})
results['total_dx'] += dx
results['total_dy'] += dy
prev_frame = curr_frame
return results
def test_row_transition(self) -> dict:
"""Test row transition using displacement stitching."""
results = {
'success': False,
'y_moved': 0.0,
'mosaic_before': (0, 0),
'mosaic_after': (0, 0),
'error': None
}
try:
self.log("Testing row transition...")
if self.mosaic is None:
frame = self._capture_frame()
self._init_mosaic(frame)
results['mosaic_before'] = (self.state.mosaic_width, self.state.mosaic_height)
with self._state_lock:
self.state.cumulative_y = 0.0
self.running = True
success = self._move_to_next_row()
self.running = False
results['success'] = success
results['y_moved'] = self.state.cumulative_y
results['mosaic_after'] = (self.state.mosaic_width, self.state.mosaic_height)
self.log(f"Row transition: {'SUCCESS' if success else 'FAILED'}, Y: {results['y_moved']:.1f}px")
except Exception as e:
results['error'] = str(e)
self.log(f"Test error: {e}")
self.running = False
return results
def test_single_row(self, direction: str = 'right') -> dict:
"""Test scanning a single row."""
results = {
'success': False,
'stop_reason': None,
'appends': 0,
'mosaic_before': (0, 0),
'mosaic_after': (0, 0),
'error': None
}
try:
self.log(f"Testing single row ({direction})...")
if self.mosaic is None:
frame = self._capture_frame()
self._init_mosaic(frame)
results['mosaic_before'] = (self.state.mosaic_width, self.state.mosaic_height)
appends_before = self.state.append_count
self.motion.set_speed(self.config.scan_speed_index)
time.sleep(0.1)
self.running = True
scan_dir = ScanDirection.RIGHT if direction == 'right' else ScanDirection.LEFT
stop_reason = self._scan_direction(scan_dir)
self.running = False
results['success'] = True
results['stop_reason'] = stop_reason
results['appends'] = self.state.append_count - appends_before
results['mosaic_after'] = (self.state.mosaic_width, self.state.mosaic_height)
except Exception as e:
results['error'] = str(e)
self.running = False
return results
def get_memory_estimate(self) -> dict:
current_bytes = self.mosaic.nbytes if self.mosaic is not None else 0
max_bytes = self.config.max_mosaic_width * self.config.max_mosaic_height * 3
return {
'current_size': (self.state.mosaic_width, self.state.mosaic_height),
'current_mb': current_bytes / (1024 * 1024),
'max_size': (self.config.max_mosaic_width, self.config.max_mosaic_height),
'max_mb': max_bytes / (1024 * 1024),
}