import gradio as gr
from gradio_bbox_annotator import BBoxAnnotator
from PIL import Image
import numpy as np
import torch
import os
import shutil
import time
import json
import uuid
from pathlib import Path
import tempfile
import zipfile
import html
from skimage import measure
from matplotlib import cm
from glob import glob
from natsort import natsorted
from huggingface_hub import HfApi, upload_file
import spaces
from inference_seg import load_model as load_seg_model, run as run_seg
from inference_count import load_model as load_count_model, run as run_count
from inference_track import load_model as load_track_model, run as run_track
from _utils.image_io import (
standardize_image, inspect_image, image_size, array_nbytes, pixel_stats, bit_depth,
RECOMMENDED_SIZE, WARN_SIZE, MAX_SIZE, MIN_SIZE, MAX_READ_BYTES, TIFF_EXTENSIONS,
)
HF_TOKEN = os.getenv("HF_TOKEN")
DATASET_REPO = "VisionLanguageGroup/feedback"
print("===== clearing cache =====")
cache_path = os.path.expanduser("~/.cache/huggingface/gradio")
if os.path.exists(cache_path):
try:
shutil.rmtree(cache_path)
print("β Deleted ~/.cache/huggingface/gradio")
except:
pass
SEG_MODEL = None
SEG_DEVICE = torch.device("cpu")
COUNT_MODEL = None
COUNT_DEVICE = torch.device("cpu")
TRACK_MODEL = None
TRACK_DEVICE = torch.device("cpu")
def load_all_models():
global SEG_MODEL, SEG_DEVICE
global COUNT_MODEL, COUNT_DEVICE
global TRACK_MODEL, TRACK_DEVICE
print("\n" + "="*60)
print("π¦ Loading Segmentation Model")
print("="*60)
SEG_MODEL, SEG_DEVICE = load_seg_model(use_box=False)
print("\n" + "="*60)
print("π¦ Loading Counting Model")
print("="*60)
COUNT_MODEL, COUNT_DEVICE = load_count_model(use_box=False)
print("\n" + "="*60)
print("π¦ Loading Tracking Model")
print("="*60)
TRACK_MODEL, TRACK_DEVICE = load_track_model(use_box=False)
print("\n" + "="*60)
print("β All Models Loaded Successfully")
print("="*60)
load_all_models()
DATASET_DIR = Path("solver_cache")
DATASET_DIR.mkdir(parents=True, exist_ok=True)
def save_feedback_to_hf(query_id, feedback_type, feedback_text=None, img_path=None, bboxes=None):
"""Save feedback to Hugging Face Dataset"""
if not HF_TOKEN:
print("β οΈ No HF_TOKEN found, using local storage")
save_feedback(query_id, feedback_type, feedback_text, img_path, bboxes)
return
# Shared by the record and its image so the two pair up; query_id alone is not unique (a session can submit more than once).
stem = f"feedback_{query_id}_{int(time.time())}"
try:
api = HfApi()
image_in_repo = None
if img_path and os.path.exists(img_path):
try:
image_in_repo = f"images/{stem}.png"
api.upload_file(
path_or_fileobj=img_path,
path_in_repo=image_in_repo,
repo_id=DATASET_REPO,
repo_type="dataset",
token=HF_TOKEN
)
except Exception as e:
print(f"β οΈ Failed to upload image: {e}")
image_in_repo = None
feedback_data = {
"query_id": query_id,
"feedback_type": feedback_type,
"feedback_text": feedback_text,
"image_path": image_in_repo,
"bboxes": str(bboxes),
"datetime": time.strftime("%Y-%m-%d %H:%M:%S"),
"timestamp": time.time()
}
filename = f"{stem}.json"
with open(filename, 'w', encoding='utf-8') as f:
json.dump(feedback_data, f, indent=2, ensure_ascii=False)
api.upload_file(
path_or_fileobj=filename,
path_in_repo=f"data/{filename}",
repo_id=DATASET_REPO,
repo_type="dataset",
token=HF_TOKEN
)
os.remove(filename)
print(f"β Feedback saved to HF Dataset: {DATASET_REPO} ({stem})")
except Exception as e:
print(f"β οΈ Failed to save to HF Dataset: {e}")
save_feedback(query_id, feedback_type, feedback_text, img_path, bboxes)
def save_feedback(query_id, feedback_type, feedback_text=None, img_path=None, bboxes=None):
"""Save feedback to local JSON file"""
feedback_data = {
"query_id": query_id,
"feedback_type": feedback_type,
"feedback_text": feedback_text,
"image": img_path,
"bboxes": bboxes,
"datetime": time.strftime("%Y%m%d_%H%M%S")
}
feedback_file = DATASET_DIR / query_id / "feedback.json"
feedback_file.parent.mkdir(parents=True, exist_ok=True)
if feedback_file.exists():
with feedback_file.open("r") as f:
existing = json.load(f)
if not isinstance(existing, list):
existing = [existing]
existing.append(feedback_data)
feedback_data = existing
else:
feedback_data = [feedback_data]
with feedback_file.open("w") as f:
json.dump(feedback_data, f, indent=4, ensure_ascii=False)
def parse_first_bbox(bboxes):
"""Parse the first bounding box from the annotation input, supports dict or list format"""
if not bboxes:
return None
b = bboxes[0]
if isinstance(b, dict):
x, y = float(b.get("x", 0)), float(b.get("y", 0))
w, h = float(b.get("width", 0)), float(b.get("height", 0))
return x, y, x + w, y + h
if isinstance(b, (list, tuple)) and len(b) >= 4:
return float(b[0]), float(b[1]), float(b[2]), float(b[3])
return None
def parse_bboxes(bboxes):
"""Parse all bounding boxes from the annotation input"""
if not bboxes:
return None
result = []
for b in bboxes:
if isinstance(b, dict):
x, y = float(b.get("x", 0)), float(b.get("y", 0))
w, h = float(b.get("width", 0)), float(b.get("height", 0))
result.append([x, y, x + w, y + h])
elif isinstance(b, (list, tuple)) and len(b) >= 4:
result.append([float(b[0]), float(b[1]), float(b[2]), float(b[3])])
return result
def colorize_mask(mask: np.ndarray, num_colors: int = 512) -> np.ndarray:
"""Convert a 2D mask of instance IDs to a color image for visualization."""
def hsv_to_rgb(h, s, v):
i = int(h * 6.0)
f = h * 6.0 - i
i = i % 6
p = v * (1 - s)
q = v * (1 - f * s)
t = v * (1 - (1 - f) * s)
if i == 0: r, g, b = v, t, p
elif i == 1: r, g, b = q, v, p
elif i == 2: r, g, b = p, v, t
elif i == 3: r, g, b = p, q, v
elif i == 4: r, g, b = t, p, v
else: r, g, b = v, p, q
return int(r * 255), int(g * 255), int(b * 255)
palette = [(0, 0, 0)]
for i in range(1, num_colors):
h = (i % num_colors) / float(num_colors)
palette.append(hsv_to_rgb(h, 1.0, 0.95))
palette_arr = np.array(palette, dtype=np.uint8)
color_idx = mask % num_colors
return palette_arr[color_idx]
def render_seg_overlay(img_np, inst_mask, overlay_alpha):
"""Render segmentation overlay from cached image/mask."""
if img_np is None or inst_mask is None:
return None
overlay = img_np.copy()
alpha = float(np.clip(overlay_alpha, 0.0, 1.0))
for inst_id in np.unique(inst_mask):
if inst_id == 0:
continue
binary_mask = (inst_mask == inst_id).astype(np.uint8)
color = get_well_spaced_color(inst_id)
overlay[binary_mask == 1] = (1 - alpha) * overlay[binary_mask == 1] + alpha * color
contours = measure.find_contours(binary_mask, 0.5)
for contour in contours:
contour = contour.astype(np.int32)
valid_y = np.clip(contour[:, 0], 0, overlay.shape[0] - 1)
valid_x = np.clip(contour[:, 1], 0, overlay.shape[1] - 1)
overlay[valid_y, valid_x] = [1.0, 1.0, 0.0]
overlay = np.clip(overlay * 255.0, 0, 255).astype(np.uint8)
return Image.fromarray(overlay)
def render_count_overlay(img_np, density_normalized, overlay_alpha):
"""Render counting heatmap overlay from cached image/density."""
if img_np is None or density_normalized is None:
return None
alpha = float(np.clip(overlay_alpha, 0.0, 1.0))
cmap = cm.get_cmap("jet")
density_colored = cmap(density_normalized)[:, :, :3]
overlay = img_np.copy()
threshold = 0.01
significant_mask = density_normalized > threshold
overlay[significant_mask] = (1 - alpha) * overlay[significant_mask] + alpha * density_colored[significant_mask]
overlay = np.clip(overlay * 255.0, 0, 255).astype(np.uint8)
return Image.fromarray(overlay)
def update_seg_overlay_alpha(overlay_alpha, seg_vis_cache):
"""Live update segmentation visualization without rerunning inference."""
if not seg_vis_cache:
return None
return render_seg_overlay(seg_vis_cache.get("img_np"), seg_vis_cache.get("inst_mask"), overlay_alpha)
def update_count_overlay_alpha(overlay_alpha, count_vis_cache):
"""Live update counting visualization without rerunning inference."""
if not count_vis_cache:
return None
return render_count_overlay(count_vis_cache.get("img_np"), count_vis_cache.get("density_normalized"), overlay_alpha)
def update_tracking_overlay_alpha(overlay_alpha, track_vis_cache):
"""Regenerate tracking visualization at new opacity using cached outputs."""
if not track_vis_cache:
return None
tif_dir = track_vis_cache.get("tif_dir")
output_dir = track_vis_cache.get("output_dir")
valid_tif_files = track_vis_cache.get("valid_tif_files")
if not tif_dir or not output_dir or not valid_tif_files:
return None
try:
return create_tracking_visualization(
tif_dir=tif_dir,
output_dir=output_dir,
valid_tif_files=valid_tif_files,
overlay_alpha=overlay_alpha
)
except Exception as e:
print(f"β οΈ Failed to update tracking opacity: {e}")
return None
def cleanup_tracking_cache(track_vis_cache):
"""Delete cached tracking temp directories from the previous run."""
if not track_vis_cache:
return
for key in ["input_temp_dir", "output_dir"]:
path = track_vis_cache.get(key)
if path and os.path.isdir(path):
try:
shutil.rmtree(path)
except Exception:
pass
# Per session. Refused rather than evicting the oldest, since there is no way to
# remove a single entry.
MAX_GALLERY_UPLOADS = 10
_GALLERY_RAW_SOURCE = {}
_RAW_SOURCE_CAP = 500
def _remember_gallery_source(thumb_path, raw_path):
"""Map a gallery thumbnail back to the original file it was made from.
"""
_GALLERY_RAW_SOURCE[thumb_path] = raw_path
while len(_GALLERY_RAW_SOURCE) > _RAW_SOURCE_CAP: # dicts keep insertion order
_GALLERY_RAW_SOURCE.pop(next(iter(_GALLERY_RAW_SOURCE)))
def _annot_path(annot_value):
"""Extract the image path from a BBoxAnnotator value (path or (path, boxes))."""
if not annot_value:
return None
if isinstance(annot_value, (list, tuple)):
return annot_value[0] if len(annot_value) > 0 else None
return annot_value
def _human_bytes(n):
"""Format a byte count as MB or GB, whichever reads better."""
gb = n / 1024 ** 3
return f"{gb:.1f} GB" if gb >= 1 else f"{n / 1024 ** 2:.0f} MB"
def check_image(img_path):
"""Validate an uploaded file, cheapest checks first.
Returns a rejection message if the file cannot be used, otherwise None.
Reject: empty file; unreadable / unsupported format; dimensions over
MAX_SIZE; full array over MAX_READ_BYTES (a small time-lapse can still be
huge); corrupt pixels; non-finite (NaN/inf) pixels. Advisory ``gr.Warning``
(does not reject): larger than WARN_SIZE; smaller than MIN_SIZE; blank
(uniform) image; very narrow dynamic range.
Only the header is touched until the size guards pass; pixel checks decode
image data afterwards, when doing so is bounded by the memory guard.
"""
if os.path.exists(img_path) and os.path.getsize(img_path) == 0:
return "This file is empty (0 bytes). Please upload a valid image file."
w, h = image_size(img_path)
if w <= 0 or h <= 0:
# Neither reader could parse the header, so say so rather than failing
# silently further down.
ext = os.path.splitext(img_path)[1].lower() or "(no extension)"
return (f"Could not read {ext} as an image. Supported formats: "
f"TIFF / OME-TIFF (8/16/32-bit, stacks, multi-channel), PNG, JPG, and other formats supported by tifffile. "
f"Please export to TIFF/PNG/JPG (e.g. from ImageJ/Fiji) first.")
if w > MAX_SIZE or h > MAX_SIZE:
return (f"Image is {w}Γ{h}, larger than the {MAX_SIZE}Γ{MAX_SIZE} limit. "
f"Please crop or downsample it before uploading.")
nbytes = array_nbytes(img_path)
if nbytes > MAX_READ_BYTES:
info = inspect_image(img_path)
detail = f"{w}Γ{h}"
if info.frames > 1:
detail += f" Γ {info.frames} frames"
if info.channels > 1:
detail += f" Γ {info.channels} channels"
return (f"This file needs {_human_bytes(nbytes)} of memory to open "
f"({detail}), over the {_human_bytes(MAX_READ_BYTES)} limit. "
f"Please crop it, or split out the frames you need.")
if w > WARN_SIZE or h > WARN_SIZE:
gr.Warning(f"Image is {w}Γ{h} and will be resized to "
f"{RECOMMENDED_SIZE}Γ{RECOMMENDED_SIZE}, so fine detail could be lost. "
f"You can try cropping the image for better results.",
duration=None, title="β οΈ Large image")
if 0 < min(w, h) < MIN_SIZE:
gr.Warning(f"Image is only {w}Γ{h} and will be upscaled to "
f"{RECOMMENDED_SIZE}Γ{RECOMMENDED_SIZE}, so results may be unreliable. "
f"A larger image may work better.",
duration=None, title="β οΈ Very small image")
# Size guards passed, so pixel decoding is bounded. Check problems that
# the header cannot reveal.
rep = pixel_stats(img_path)
if not rep.decoded:
return ("This image appears to be corrupted or truncated and could not be read. "
"Please re-export it (e.g. from ImageJ/Fiji) and try again.")
if not rep.finite:
return ("This image contains invalid pixel values (NaN or infinity). "
"Please clean or re-export it before uploading.")
if rep.vmin == rep.vmax:
if rep.vmax == 0:
kind = "completely black"
elif rep.dtype_max and rep.vmax >= rep.dtype_max:
kind = "completely white"
else:
kind = "a single uniform value"
gr.Warning(f"This image is {kind}, it has no visible content, so results will not be meaningful.",
duration=None, title="β οΈ Blank image")
elif rep.low_range:
gr.Warning("This image may use a very narrow intensity range (low contrast). Signal may be too weak for reliable results; consider adjusting acquisition or contrast before uploading.",
duration=None, title="β οΈ Low dynamic range")
return None
def _describe_read(info):
"""Plain-language summary of how a file's dimensions were interpreted.
Deliberately avoids array jargon ("axis", "size-4"): the reader thinks in
channels / z-slices / timepoints, not numpy axes.
"""
# lines = [f"Detected shape: {info.shape}"]
if info.frames >1:
lines = [f"Detected {info.channels}-channel {info.frames}-frame image stack of size {info.width}Γ{info.height}."]
else:
lines = [f"Detected {info.channels}-channel image of size {info.width}Γ{info.height}."]
if info.channels >= 3:
lines.append(f"Loaded as 3-channel RGB image.")
elif info.channels == 2:
lines.append(f"Loaded as 2-channel image (red, green).")
else:
lines.append(f"Loaded as single-channel grayscale image.")
if info.frames > 1:
lines.append(f"Showing first frame by default (you can use the stack frame slider to pick another).")
if info.sub_frames > 1:
lines.append(f"Showing the first of {info.sub_frames} {info.sub_label}s by default (use the {info.sub_label} slider to pick another).")
lines.append(
"If wrong, please convert your image to a 8-bit RGB/Grayscale image file before uploading (there are available tools such as ImageJ/Fiji)."
)
return " ".join(lines)
def prepare_uploaded_image(annot_value):
"""On annotator upload: standardize frame 0 for preview and configure the
stack sliders.
Browsers cannot render 16-bit / float / multi-page TIFFs, so we standardize
to an 8-bit RGB PNG. If the file is a stack, notify the user and reveal a
frame slider (default frame 1); a file with a second stack axis (e.g. Z in a
time+Z series) also gets a z-plane slider (0 = max-intensity projection).
Sliders that don't apply stay hidden.
Returns (annotator_value, raw_path, frame_slider_update, zplane_slider_update).
"""
hidden = gr.update(visible=False, value=1)
img_path = _annot_path(annot_value)
if not img_path:
return annot_value, None, hidden, hidden
# Clear the annotator on refusal, so a rejected file cannot reach inference.
rejected = check_image(img_path)
if rejected:
gr.Warning(rejected, duration=None, title="β Cannot use this file")
return None, None, hidden, hidden
info = inspect_image(img_path)
display = standardize_image(img_path, frame=0,
sub_frame=0 if info.sub_frames > 1 else None)
# Stay quiet unless something non-obvious happened; a plain RGB or grayscale
# image needs no explanation.
if info.guessed or info.frames > 1 or info.channels > 3:
gr.Info(_describe_read(info), duration=None, title="π Image Loading Info")
slider = (gr.update(visible=True, maximum=info.frames, value=1) if info.frames > 1
else hidden)
zslider = (gr.update(visible=True, maximum=info.sub_frames, value=1,
label=f"π¬ {info.sub_label.capitalize()} (choose plane to use)")
if info.sub_frames > 1 else hidden)
return display, img_path, slider, zslider
def select_frame(raw_path, frame_num, zplane=1):
"""Re-render the annotator preview for the chosen frame and z-plane (both
1-based; z-plane is ignored for files with only one stack axis)."""
if not raw_path:
return gr.update()
return standardize_image(raw_path, frame=int(frame_num) - 1, sub_frame=int(zplane) - 1)
@spaces.GPU
def segment_with_choice(use_box_choice, annot_value, overlay_alpha):
"""Segmentation handler - supports bounding box, returns colorized overlay and original mask path"""
if annot_value is None or len(annot_value) < 1:
print("β No annotation input")
return None, None, {}
img_path = annot_value[0]
bboxes = annot_value[1] if len(annot_value) > 1 else []
print(f"πΌοΈ Image path: {img_path}")
# One standardized image for both the model and the overlay (WYSIWYG).
img_path = standardize_image(img_path)
print(f"π§ͺ Standardized image: {img_path}")
box_array = None
if use_box_choice == "Yes" and bboxes:
box = parse_bboxes(bboxes)
if box:
box_array = box
print(f"π¦ Using bounding boxes: {box_array}")
try:
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
mask = run_seg(SEG_MODEL, img_path, box=box_array, device=device)
print("π mask shape:", mask.shape, "dtype:", mask.dtype)
except Exception as e:
print(f"β Inference failed: {str(e)}")
return None, None, {}
temp_mask_file = tempfile.NamedTemporaryFile(delete=False, suffix=".tif")
mask_img = Image.fromarray(mask.astype(np.uint16))
mask_img.save(temp_mask_file.name)
print(f"πΎ Original mask saved to: {temp_mask_file.name}")
try:
img = Image.open(img_path)
print("π· Image mode:", img.mode, "size:", img.size)
except Exception as e:
print(f"β Failed to open image: {e}")
return None, None, {}
try:
img_rgb = img.convert("RGB").resize(mask.shape[::-1], resample=Image.BILINEAR)
img_np = np.array(img_rgb, dtype=np.float32)
if img_np.max() > 1.5:
img_np = img_np / 255.0
except Exception as e:
print(f"β Error in image conversion/resizing: {e}")
return None, None, {}
mask_np = np.array(mask)
inst_mask = mask_np.astype(np.int32)
unique_ids = np.unique(inst_mask)
num_instances = len(unique_ids[unique_ids != 0])
if num_instances == 0:
print("β οΈ No instance found, returning dummy red image")
return Image.new("RGB", mask.shape[::-1], (255, 0, 0)), None, {}
overlay_img = render_seg_overlay(img_np, inst_mask, overlay_alpha)
seg_vis_cache = {"img_np": img_np, "inst_mask": inst_mask}
return overlay_img, temp_mask_file.name, seg_vis_cache
@spaces.GPU
def count_cells_handler(use_box_choice, annot_value, overlay_alpha):
"""Counting handler - supports bounding box, returns only density map"""
if annot_value is None or len(annot_value) < 1:
return None, None, "β οΈ Please provide an image.", {}
image_path = annot_value[0]
bboxes = annot_value[1] if len(annot_value) > 1 else []
print(f"πΌοΈ Image path: {image_path}")
# One standardized image for both the model and the overlay (WYSIWYG).
image_path = standardize_image(image_path)
print(f"π§ͺ Standardized image: {image_path}")
box_array = None
if use_box_choice == "Yes" and bboxes:
box = parse_bboxes(bboxes)
if box:
box_array = box
print(f"π¦ Using bounding boxes: {box_array}")
try:
print(f"π’ Counting - Image: {image_path}")
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
result = run_count(
COUNT_MODEL,
image_path,
box=box_array,
device=device,
visualize=True
)
if 'error' in result:
return None, None, f"β Counting failed: {result['error']}", {}
count = result['count']
density_map = result['density_map']
temp_density_file = tempfile.NamedTemporaryFile(delete=False, suffix=".npy")
np.save(temp_density_file.name, density_map)
print(f"πΎ Density map saved to {temp_density_file.name}")
try:
img = Image.open(image_path)
print("π· Image mode:", img.mode, "size:", img.size)
except Exception as e:
print(f"β Failed to open image: {e}")
return None, None, f"β Failed to open image: {str(e)}", {}
try:
img_rgb = img.convert("RGB").resize(density_map.shape[::-1], resample=Image.BILINEAR)
img_np = np.array(img_rgb, dtype=np.float32)
img_np = (img_np - img_np.min()) / (img_np.max() - img_np.min() + 1e-8)
if img_np.max() > 1.5:
img_np = img_np / 255.0
except Exception as e:
print(f"β Error in image conversion/resizing: {e}")
return None, None, f"β Error in image conversion/resizing: {str(e)}", {}
density_normalized = density_map.copy()
if density_normalized.max() > 0:
density_normalized = (density_normalized - density_normalized.min()) / (density_normalized.max() - density_normalized.min())
overlay_img = render_count_overlay(img_np, density_normalized, overlay_alpha)
result_text = f"β Detected {round(count)} objects"
if use_box_choice == "Yes" and box_array:
result_text += f"\nπ¦ Using bounding box: {box_array}"
print(f"β Counting done - Count: {count:.1f}")
count_vis_cache = {"img_np": img_np, "density_normalized": density_normalized}
return overlay_img, temp_density_file.name, result_text, count_vis_cache
except Exception as e:
print(f"β Counting error: {e}")
import traceback
traceback.print_exc()
return None, None, f"β Counting failed: {str(e)}", {}
def find_tif_dir(root_dir):
"""Recursively find the first directory containing .tif files"""
for dirpath, _, filenames in os.walk(root_dir):
if '__MACOSX' in dirpath:
continue
if any(f.lower().endswith('.tif') for f in filenames):
return dirpath
return None
def is_valid_tiff(filepath):
"""Check if a file is a valid TIFF image"""
try:
with Image.open(filepath) as img:
img.verify()
return True
except Exception as e:
return False
def find_valid_tif_dir(root_dir):
"""Recursively find the first directory containing valid .tif files"""
for dirpath, dirnames, filenames in os.walk(root_dir):
if '__MACOSX' in dirpath:
continue
potential_tifs = [
os.path.join(dirpath, f)
for f in filenames
if f.lower().endswith(('.tif', '.tiff')) and not f.startswith('._')
]
if not potential_tifs:
continue
valid_tifs = [f for f in potential_tifs if is_valid_tiff(f)]
if valid_tifs:
print(f"β Found {len(valid_tifs)} valid TIFF files in: {dirpath}")
return dirpath
return None
def create_ctc_results_zip(output_dir):
"""
Create a ZIP file with CTC format results
Parameters:
-----------
output_dir : str
Directory containing tracking results (res_track.txt, etc.)
Returns:
--------
zip_path : str
Path to created ZIP file
"""
# Create temp directory for ZIP
temp_zip_dir = tempfile.mkdtemp()
zip_filename = f"tracking_results_{time.strftime('%Y%m%d_%H%M%S')}.zip"
zip_path = os.path.join(temp_zip_dir, zip_filename)
print(f"π¦ Creating results ZIP: {zip_path}")
# Create ZIP with all tracking results
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
# Add all files from output directory
for root, dirs, files in os.walk(output_dir):
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.relpath(file_path, output_dir)
zipf.write(file_path, arcname)
print(f" π Added: {arcname}")
# Add a README with summary
readme_content = f"""Tracking Results Summary
========================
Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}
Files:
------
- res_track.txt: CTC format tracking data
Format: track_id start_frame end_frame parent_id
- Segmentation masks
For more information on CTC format:
http://celltrackingchallenge.net/
"""
zipf.writestr("README.txt", readme_content)
print(f"β ZIP created: {zip_path} ({os.path.getsize(zip_path) / 1024:.1f} KB)")
return zip_path
def get_well_spaced_color(track_id, num_colors=256):
"""Generate well-spaced colors, using contrasting colors for adjacent IDs"""
golden_ratio = 0.618033988749895
hue = (track_id * golden_ratio) % 1.0
import colorsys
rgb = colorsys.hsv_to_rgb(hue, 0.9, 0.95)
return np.array(rgb)
def extract_first_frame(tif_dir):
"""
Extract the first frame from a directory of TIF files
Returns:
--------
first_frame_path : str
Path to the first TIF frame
"""
tif_files = natsorted(glob(os.path.join(tif_dir, "*.tif")) +
glob(os.path.join(tif_dir, "*.tiff")))
valid_tif_files = [f for f in tif_files
if not os.path.basename(f).startswith('._') and is_valid_tiff(f)]
if valid_tif_files:
return valid_tif_files[0]
return None
def create_tracking_visualization(tif_dir, output_dir, valid_tif_files, overlay_alpha=0.3):
"""
Create an animated GIF/video showing tracked objects with consistent colors
Parameters:
-----------
tif_dir : str
Directory containing input TIF frames
output_dir : str
Directory containing tracking results (masks)
valid_tif_files : list
List of valid TIF file paths
Returns:
--------
video_path : str
Path to generated visualization (GIF or first frame)
"""
import numpy as np
from matplotlib import colormaps
from skimage import measure
import tifffile
# Look for tracking mask files in output directory
# Common CTC formats: man_track*.tif, mask*.tif, or numbered masks
mask_files = natsorted(glob(os.path.join(output_dir, "mask*.tif")) +
glob(os.path.join(output_dir, "man_track*.tif")) +
glob(os.path.join(output_dir, "*.tif")))
if not mask_files:
print("β οΈ No mask files found in output directory")
# Return first frame as fallback
return valid_tif_files[0]
print(f"π Found {len(mask_files)} mask files")
frames = []
alpha = float(np.clip(overlay_alpha, 0.0, 1.0)) # Transparency for overlay
# Process each frame
num_frames = min(len(valid_tif_files), len(mask_files))
for i in range(num_frames):
try:
# Load original image using tifffile (handles ZSTD compression)
try:
img_np = tifffile.imread(valid_tif_files[i])
# Normalize to [0, 1] range based on actual data type and values
if img_np.dtype == np.uint8:
img_np = img_np.astype(np.float32) / 255.0
elif img_np.dtype == np.uint16:
# Normalize uint16 to [0, 1] using actual min/max
img_min, img_max = img_np.min(), img_np.max()
if img_max > img_min:
img_np = (img_np.astype(np.float32) - img_min) / (img_max - img_min)
else:
img_np = img_np.astype(np.float32) / 65535.0
else:
# For float or other types, normalize based on actual range
img_np = img_np.astype(np.float32)
img_min, img_max = img_np.min(), img_np.max()
if img_max > img_min:
img_np = (img_np - img_min) / (img_max - img_min)
else:
img_np = np.clip(img_np, 0, 1)
# Convert to RGB if grayscale
if img_np.ndim == 2:
img_np = np.stack([img_np]*3, axis=-1)
img_np = img_np.astype(np.float32)
if img_np.max() > 1.5:
img_np = img_np / 255.0
except Exception as e:
print(f"β οΈ Error loading image frame {i}: {e}")
# Fallback to PIL
img = Image.open(valid_tif_files[i]).convert("RGB")
img_np = np.array(img, dtype=np.float32) / 255.0
# Load tracking mask using tifffile (handles ZSTD compression)
try:
mask = tifffile.imread(mask_files[i])
except Exception as e:
print(f"β οΈ Error loading mask frame {i}: {e}")
# Fallback to PIL
mask = np.array(Image.open(mask_files[i]))
# Resize mask to match image if needed
if mask.shape[:2] != img_np.shape[:2]:
from scipy.ndimage import zoom
zoom_factors = [img_np.shape[0] / mask.shape[0], img_np.shape[1] / mask.shape[1]]
mask = zoom(mask, zoom_factors, order=0).astype(mask.dtype)
# Create overlay
overlay = img_np.copy()
# Get unique track IDs (excluding background 0)
track_ids = np.unique(mask)
track_ids = track_ids[track_ids != 0]
# Color each tracked object
for track_id in track_ids:
# Create binary mask for this track
binary_mask = (mask == track_id)
# Get consistent color for this track ID
# color = np.array(cmap(int(track_id) % 256)[:3])
color = get_well_spaced_color(int(track_id))
# Blend color onto image
overlay[binary_mask] = (1 - alpha) * overlay[binary_mask] + alpha * color
# Draw contours (optional, adds yellow boundaries)
try:
contours = measure.find_contours(binary_mask.astype(np.uint8), 0.5)
for contour in contours:
contour = contour.astype(np.int32)
valid_y = np.clip(contour[:, 0], 0, overlay.shape[0] - 1)
valid_x = np.clip(contour[:, 1], 0, overlay.shape[1] - 1)
overlay[valid_y, valid_x] = [1.0, 1.0, 0.0] # Yellow contour
except:
pass # Skip contours if they fail
# Convert to uint8
overlay_uint8 = np.clip(overlay * 255.0, 0, 255).astype(np.uint8)
frames.append(Image.fromarray(overlay_uint8))
if i % 10 == 0 or i == num_frames - 1:
print(f" πΈ Processed frame {i+1}/{num_frames}")
except Exception as e:
print(f"β οΈ Error processing frame {i}: {e}")
import traceback
traceback.print_exc()
continue
if not frames:
print("β οΈ No frames were processed successfully")
return valid_tif_files[0]
# Save as animated GIF
try:
temp_gif = tempfile.NamedTemporaryFile(delete=False, suffix=".gif")
frames[0].save(
temp_gif.name,
save_all=True,
append_images=frames[1:],
duration=200, # 200ms per frame = 5fps
loop=0
)
temp_gif.close() # Close the file handle
print(f"β Created tracking visualization GIF: {temp_gif.name}")
print(f" Size: {os.path.getsize(temp_gif.name)} bytes, Frames: {len(frames)}")
return temp_gif.name
except Exception as e:
print(f"β οΈ Failed to create GIF: {e}")
import traceback
traceback.print_exc()
# Return first frame as static image fallback
try:
temp_img = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
frames[0].save(temp_img.name)
temp_img.close()
return temp_img.name
except:
return valid_tif_files[0]
@spaces.GPU
def track_video_handler(use_box_choice, first_frame_annot, zip_file_obj, overlay_alpha, prev_track_vis_cache):
"""
Tracking handler - processes a ZIP of TIF frames, supports bounding box, returns visualization and results ZIP
Parameters:
-----------
use_box_choice : str
"Yes" or "No" - whether to use bounding box annotation for tracking
first_frame_annot : tuple or None
(image_path, bboxes) from BBoxAnnotator, only used if user annotated first frame
zip_file_obj : File
Uploaded ZIP file containing TIF sequence
"""
if zip_file_obj is None:
return None, "β οΈ Please upload a ZIP file containing video frames (.zip)", None, None, {}
cleanup_tracking_cache(prev_track_vis_cache)
temp_dir = None
output_temp_dir = None
try:
# Parse bounding box if provided
box_array = None
if use_box_choice == "Yes" and first_frame_annot is not None:
if isinstance(first_frame_annot, (list, tuple)) and len(first_frame_annot) > 1:
bboxes = first_frame_annot[1]
if bboxes:
box = parse_bboxes(bboxes)
if box:
box_array = box
print(f"π¦ Using bounding boxes: {box_array}")
# Extract input ZIP
temp_dir = tempfile.mkdtemp()
print(f"\nπ¦ Extracting to temporary directory: {temp_dir}")
with zipfile.ZipFile(zip_file_obj.name, 'r') as zip_ref:
extracted_count = 0
skipped_count = 0
for member in zip_ref.namelist():
basename = os.path.basename(member)
if ('__MACOSX' in member or
basename.startswith('._') or
basename.startswith('.DS_Store') or
member.endswith('/')):
skipped_count += 1
continue
try:
zip_ref.extract(member, temp_dir)
extracted_count += 1
if basename.lower().endswith(('.tif', '.tiff')):
print(f"π Extracted TIFF: {basename}")
except Exception as e:
print(f"β οΈ Failed to extract {member}: {e}")
print(f"\nπ Extracted: {extracted_count} files, Skipped: {skipped_count} files")
# Find valid TIFF directory
tif_dir = find_valid_tif_dir(temp_dir)
if tif_dir is None:
return None, "β Did not find valid TIF directory", None, None, {}
# Validate TIFF files
tif_files = natsorted(glob(os.path.join(tif_dir, "*.tif")) +
glob(os.path.join(tif_dir, "*.tiff")))
valid_tif_files = [f for f in tif_files
if not os.path.basename(f).startswith('._') and is_valid_tiff(f)]
if len(valid_tif_files) == 0:
return None, "β Did not find valid TIF files", None, None, {}
print(f"π Using {len(valid_tif_files)} TIF files")
# Store paths for later visualization
first_frame_path = valid_tif_files[0]
# Create temporary output directory for CTC results
output_temp_dir = tempfile.mkdtemp()
print(f"πΎ CTC-format results will be saved to: {output_temp_dir}")
# Run tracking with optional bounding box
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
result = run_track(
TRACK_MODEL,
video_dir=tif_dir,
box=box_array, # Pass bounding box if specified
device=device,
output_dir=output_temp_dir
)
if 'error' in result:
return None, f"β Tracking failed: {result['error']}", None, None, {}
# Create visualization video of tracked objects
print("\n㪠Creating tracking visualization...")
try:
tracking_video = create_tracking_visualization(
tif_dir,
output_temp_dir,
valid_tif_files,
overlay_alpha=overlay_alpha
)
except Exception as e:
print(f"β οΈ Failed to create visualization: {e}")
import traceback
traceback.print_exc()
# Fallback to first frame if visualization fails
try:
tracking_video = Image.open(first_frame_path)
except:
tracking_video = None
# Create downloadable ZIP with results
try:
results_zip = create_ctc_results_zip(output_temp_dir)
except Exception as e:
print(f"β οΈ Failed to create ZIP: {e}")
results_zip = None
bbox_info = ""
if box_array:
bbox_info = f"\nπ² Using bounding box: [{box_array[0][0]}, {box_array[0][1]}, {box_array[0][2]}, {box_array[0][3]}]"
result_text = (
f"β Tracking completed!\n"
f"\n"
f"πΌοΈ Processed frames: {len(valid_tif_files)}{bbox_info}\n"
f"\n"
f"π₯ Use the \"Download (.zip)\" button above to get the results, which include:\n"
f"- res_track.txt (CTC-format tracking data)\n"
f"- Other tracking-related files\n"
f"- README.txt (Results description)"
)
if use_box_choice == "Yes" and box_array:
result_text += f"\nπ¦ Using bounding box: {box_array}"
print(f"\nβ Tracking completed")
track_vis_cache = {
"tif_dir": tif_dir,
"valid_tif_files": valid_tif_files,
"output_dir": output_temp_dir,
"input_temp_dir": temp_dir,
}
return results_zip, result_text, gr.update(visible=True), tracking_video, track_vis_cache
except zipfile.BadZipFile:
return None, "β Not a valid ZIP file", None, None, {}
except Exception as e:
import traceback
traceback.print_exc()
# Clean up on error
for d in [temp_dir, output_temp_dir]:
if d:
try:
shutil.rmtree(d)
except:
pass
return None, f"β Tracking failed: {str(e)}", None, None, {}
# ===== Example Images =====
example_images_seg = [f for f in glob("example_imgs/seg/*")]
example_images_cnt = [f for f in glob("example_imgs/cnt/*")]
example_tracking_zips = [f for f in glob("example_imgs/tra/*.zip")]
_CHANNEL_NOTE = {1: "grayscale", 3: "RGB", 4: "RGBA"}
def describe_image_info(raw_path, frame=1):
"""Metadata line under the result: file, size, channels, bit depth, frames.
Reads the *original* upload (not the standardized preview) so the numbers
describe what the user actually provided. Metadata only - no pixel decode.
"""
if not raw_path or not os.path.exists(raw_path):
return gr.update(value="", visible=False)
info = inspect_image(raw_path)
bits = bit_depth(raw_path)
fields = [f"File: {html.escape(os.path.basename(raw_path))}"]
if info.width and info.height:
fields.append(f"Size: {info.width} Γ {info.height} px")
note = _CHANNEL_NOTE.get(info.channels)
fields.append(f"Channels: {info.channels}"
+ (f" ({note})" if note else ""))
if bits:
fields.append(f"Bit depth: {bits}-bit")
if info.frames > 1:
fields.append(f"Frames: {int(frame)} / {info.frames}")
if info.sub_frames > 1:
fields.append(f"{info.sub_label.capitalize()}s: {info.sub_frames}")
body = "".join(f"{f}" for f in fields)
return gr.update(
value=f"