from typing import Any, Dict, Optional
from tqdm import tqdm
import tempfile
import os
import subprocess
from aicsimageio.writers import OmeTiffWriter
import scyjava as sj
from scyjava import jimport
import imagej
import imagej.doctor
import pandas as pd
import numpy as np
from skimage import measure
from pathlib import Path
[docs]
class TrackMate(object):
def __init__(self, FIJI_DIR="", ij=None, linking_max_distance=15.0, max_frame_gap=5, gap_closing_max_distance=15.0, pixel_size=1.0, size_min=None, size_max=None, verbose=True, use_in_memory=True) -> None:
"""
Perform TrackMate tracking with pyimagej wrapping
Args:
FIJI_APP (str): Path to Fiji.app folder
ij (imagej object): pyimagej object, if not provided a new one will be initiated from the given FIJI_APP path
linking_max_distance (float): The max distance between two consecutive spots, in physical units, allowed for creating links.
max_frame_gap (int): Gap-closing time-distance. \
The max difference in time-points between two spots to allow for linking. \
For instance a value of 2 means that the tracker will be able to make a link \
between a spot in frame t and a successor spots in frame t+2, effectively \
bridging over one missed detection in one frame.
gap_closing_max_distance (float): Gap-closing max spatial distance. The max distance between two spots, in physical units, allowed for creating links over missing detections.
size_min (int): Minimum cell size to filter out small cells
size_max (int): Maximum cell size to filter out big cells
verbose (bool): Turn on or off the processing printout
use_in_memory (bool): If True, use in-memory processing instead of file-based CLI processing
"""
self.name = "TrackMate"
self.FIJI_DIR = FIJI_DIR
self.ij = ij
self.linking_max_distance = linking_max_distance
self.max_frame_gap = max_frame_gap
self.gap_closing_max_distance = gap_closing_max_distance
self.pixel_size = pixel_size
self.size_min = size_min
self.size_max = size_max
self.verbose = verbose
self.use_in_memory = use_in_memory
def _get_fiji_executable(self):
"""
Get the correct Fiji executable path based on the operating system.
"""
import platform
if platform.system() == "Darwin": # macOS
return os.path.join(self.FIJI_DIR, "Contents", "MacOS", "ImageJ-macosx")
elif platform.system() == "Linux":
return os.path.join(self.FIJI_DIR, "ImageJ-linux64")
elif platform.system() == "Windows":
return os.path.join(self.FIJI_DIR, "ImageJ-win64.exe")
else:
# Fallback to Linux executable
return os.path.join(self.FIJI_DIR, "ImageJ-linux64")
def _check_and_install_trackmate_csv_importer(self):
"""
Check if TrackMateCSVImporter plugin is installed, and install it if not.
Uses ImageJ's CLI updater to manage plugins.
"""
if self.verbose:
print("Checking for TrackMateCSVImporter plugin...")
fiji_executable = self._get_fiji_executable()
# Command to list installed plugins and check for TrackMateCSVImporter
list_command = [
fiji_executable,
"--headless",
"--update",
"list"
]
try:
# Check if plugin is already installed
result = subprocess.run(list_command, capture_output=True, text=True, timeout=120)
if "TrackMate-CSVImporter" in result.stdout:
if self.verbose:
print("TrackMateCSVImporter plugin is already installed.")
return
except subprocess.TimeoutExpired:
if self.verbose:
print("Plugin list check timed out, proceeding with installation attempt...")
except Exception as e:
if self.verbose:
print(f"Could not check plugin status: {e}, proceeding with installation attempt...")
# Install the plugin if not found
if self.verbose:
print("Installing TrackMateCSVImporter plugin...")
install_command = [
fiji_executable,
"--headless",
"--update",
"add-update-site",
"TrackMateCSVImporter",
"https://sites.imagej.net/TrackMateCSVImporter/"
]
try:
subprocess.run(install_command, check=True, timeout=60)
if self.verbose:
print("TrackMateCSVImporter plugin installation completed.")
except subprocess.CalledProcessError as e:
print(f"Warning: Failed to install TrackMateCSVImporter plugin: {e}")
except subprocess.TimeoutExpired:
print("Warning: Plugin installation timed out.")
except Exception as e:
print(f"Warning: Unexpected error during plugin installation: {e}")
def __call__(self, data, output=False) -> Any:
# Check if we should use in-memory processing
if self.use_in_memory:
return self._run_in_memory_processing(data, output)
else:
return self._run_file_based_processing(data, output)
def _run_file_based_processing(self, data, output=False) -> Any:
"""Original file-based processing method"""
image = data["image"]
feature = data["feature"]
if self.verbose:
tqdm.write("Cell tracking with TrackMate (file-based)")
# filter by cell features
feature_filtered = feature
if self.size_min is not None:
feature_filtered = feature_filtered[feature["size"]>self.size_min]
if self.size_max is not None:
feature_filtered = feature_filtered[feature["size"]<self.size_max]
# convert csv to trackmate xml
# Create a temporary saved csv file
temp_csv_file = tempfile.NamedTemporaryFile(suffix=".csv", mode="w", delete=False)
temp_img_file = tempfile.NamedTemporaryFile(suffix=".tiff", mode="w", delete=False)
temp_xml_file = tempfile.NamedTemporaryFile(suffix=".xml", mode="w", delete=False)
print("Temporary CSV file created:", temp_csv_file.name, temp_img_file.name, temp_xml_file.name)
feature_filtered.to_csv(temp_csv_file.name, index=False)
print("Temporary image created:", temp_img_file.name)
OmeTiffWriter.save(image.T, temp_img_file.name, dim_order="TYX")
fiji_executable = self._get_fiji_executable()
conversion_command = " ".join([
fiji_executable,
"--headless",
os.path.join(self.FIJI_DIR,"scripts","CsvToTrackMate.py"),
"--csvFilePath={}".format(temp_csv_file.name),
"--imageFilePath={}".format(temp_img_file.name),
"--radius=2.5",
"--xCol={}".format(list(feature_filtered.columns).index("j")),
"--yCol={}".format(list(feature_filtered.columns).index("i")),
"--frameCol={}".format(list(feature_filtered.columns).index("frame")),
"--idCol={}".format(list(feature_filtered.columns).index("label")),
"--nameCol={}".format(list(feature_filtered.columns).index("channel")),
"--radiusCol={}".format(list(feature_filtered.columns).index("feret_radius")),
"--targetFilePath={}".format(temp_xml_file.name),
# "--targetFilePath={}".format("/app/cytotoxicity-pipeline/output/tracking/trackmate.xml")
# "--targetFilePath={}".format("/home/vpfannenstill/Projects/Cytotoxicity-Pipeline/output/tracking/trackmate.xml")
])
print("Running CsvToTrackMate script for feature CSV to TrackMate XML conversion...")
# Check and install TrackMateCSVImporter plugin if needed
self._check_and_install_trackmate_csv_importer()
os.system(conversion_command)
print("CsvToTrackMate script complete")
"""
Tracking with TrackMate
Export TrackMate file as csv
"""
if self.ij is None:
# initiate Fiji
imagej.doctor.checkup()
if not os.path.exists(self.FIJI_DIR) or self.FIJI_DIR == "":
raise Exception("Fiji.app directory not found")
print("Initializing Fiji on JVM...")
self.ij = imagej.init(self.FIJI_DIR,mode='headless')
print(self.ij.getApp().getInfo(True))
# print("Image loading on python side...")
# # image = io.imread(IMAGE_PATH)
# imp = ij.py.to_imageplus(image)
File = sj.jimport('java.io.File')
CWD = os.path.dirname(os.path.realpath(__file__))
jfile = File(os.path.join(CWD,'trackmate_script_run.py'))
# convert python side stuffs to java side
temp_out_dir = tempfile.TemporaryDirectory()
# define python side arguments
args = {
'LINKING_MAX_DISTANCE': self.linking_max_distance,
'MAX_FRAME_GAP': self.max_frame_gap,
'GAP_CLOSING_MAX_DISTANCE': self.gap_closing_max_distance,
'IMAGE_PATH': temp_img_file.name,
'XML_PATH': temp_xml_file.name,
'OUT_CSV_DIR': temp_out_dir.name,
}
jargs = self.ij.py.jargs(args)
print("Running Trackmate...")
result_future = self.ij.script().run(jfile,True,jargs)
# get the result from java future, blocking will occur here
result = result_future.get()
# if not headless:
# input("Press Enter to continue...")
# print(ij.py.from_java(result.getOutput("foo")))
# print(ij.py.from_java(result.getOutput("bar")))
# print(ij.py.from_java(result.getOutput("shape")))
# pyimagej provides java table to dataframe interface passing: https://py.imagej.net/en/latest/07-Running-Macros-Scripts-and-Plugins.html#using-scripts-ij-py-run-script
# for quick development work around we use temporary csv saving for data consistency.
# read the tracked CSV file and remap back to the unfiltered file
OUT_CSV_PATH = os.path.join(temp_out_dir.name,"trackmate_linkDist-{}_frameGap-{}_gapCloseDist-{}.csv".format(self.linking_max_distance,self.max_frame_gap,self.gap_closing_max_distance))
feature_tracked = pd.read_csv(OUT_CSV_PATH)
# Merge DataFrames using different columns
feature_out = pd.merge(feature, feature_tracked[['ID', 'TRACK_ID']], left_on='label', right_on='ID', how='left')
# Drop the redundant column
feature_out.drop('ID', axis=1, inplace=True)
# feature_out['TRACK_ID'].fillna(0, inplace=True)
feature_out = feature_out.rename(columns={"TRACK_ID": "TrackID"})
if output:
# Reading the data inside the xml
# file to a variable under the name
# data
with open(temp_xml_file.name, 'r') as f:
res_xml = f.read()
# Close the temporary files
temp_csv_file.close()
temp_img_file.close()
temp_xml_file.close()
return feature_out, res_xml
else:
# Close the temporary files
temp_csv_file.close()
temp_img_file.close()
temp_xml_file.close()
return feature_out, None
def _run_in_memory_processing(self, data: Dict[str, Any], output=False) -> Any:
"""
Perform TrackMate tracking with in-memory processing using pyimagej. If feature input is provided will skip label to sparse conversion.
Args:
data: Dictionary containing:
- "image": numpy array of the original image in TYX format (optional if features provided)
- "labels": numpy array of segmented labels in TYX format (optional if features provided)
- "feature": pandas DataFrame with features data that maps to trackmate spots (optional if labels provided)
output: Whether to return XML output as well
Returns:
Tracking results DataFrame or tuple of (DataFrame, XML) if output=True
"""
if self.verbose:
tqdm.write("Cell tracking with TrackMate (in-memory)")
# Handle different input formats for backward compatibility
features_key = "feature" if "feature" in data else "features"
# Validate input
if features_key not in data and ("labels" not in data or "image" not in data):
raise ValueError("Input must contain either 'feature'/'features' or both 'image' and 'labels'")
# Import TrackMateInMemoryProcessor
from .trackmate_in_mem import TrackMateInMemoryProcessor
# Initialize processor
if self.verbose:
tqdm.write("Initializing TrackMate processor...")
if self.ij is not None:
processor = TrackMateInMemoryProcessor(ij=self.ij)
elif self.FIJI_DIR and self.FIJI_DIR != "":
processor = TrackMateInMemoryProcessor(imagej_path=self.FIJI_DIR)
self.ij = processor.ij
else:
processor = TrackMateInMemoryProcessor()
self.ij = processor.ij
# Handle input data based on what's provided
if features_key in data:
# Convert features DataFrame to TrackMate spots format
if self.verbose:
tqdm.write("Using provided features DataFrame")
features_df = data[features_key].copy()
# Apply size filtering if specified
if self.size_min is not None:
features_df = features_df[features_df["size"] >= self.size_min]
if self.size_max is not None:
features_df = features_df[features_df["size"] <= self.size_max]
if len(features_df) == 0:
if self.verbose:
tqdm.write("No features remain after filtering")
return {"features": pd.DataFrame()}
# Convert to TrackMate spots format
spots_df = self._features_to_trackmate_spots(features_df)
else:
# Use labels and image to extract spots
if self.verbose:
tqdm.write("Converting labels to spots...")
labels = data["labels"]
image = data["image"]
# Use the processor's method to convert labels to spots
spots_df = processor.labels_to_spots_csv(labels, image)
if len(spots_df) == 0:
if self.verbose:
tqdm.write("No spots found in the data")
return {"features": pd.DataFrame()}
# Apply size filtering
if self.size_min is not None:
spots_df = spots_df[spots_df["AREA"] >= self.size_min]
if self.size_max is not None:
spots_df = spots_df[spots_df["AREA"] <= self.size_max]
if len(spots_df) == 0:
if self.verbose:
tqdm.write("No spots remain after filtering")
return {"features": pd.DataFrame()}
# Convert spots to features for final output
features_df = self._trackmate_spots_to_features(spots_df)
# Determine image shape for TrackMate model
if "image" in data:
image_shape = data["image"].shape
else:
# Estimate shape from spots
max_frame = int(spots_df["FRAME"].max())
max_y = int(spots_df["POSITION_Y"].max()) + 50
max_x = int(spots_df["POSITION_X"].max()) + 50
image_shape = (max_frame + 1, max_y, max_x)
if self.verbose:
tqdm.write(f"Creating TrackMate model with {len(spots_df)} spots...")
# Create TrackMate model
model, settings = processor.create_trackmate_model(
spots_df=spots_df,
image_shape=image_shape,
pixel_size=self.pixel_size, # Assuming pixel coordinates
time_interval=1.0
)
# Configure tracker settings with class parameters
tracker_settings = settings.trackerSettings
tracker_settings.put('LINKING_MAX_DISTANCE', float(self.linking_max_distance))
tracker_settings.put('GAP_CLOSING_MAX_DISTANCE', float(self.gap_closing_max_distance))
tracker_settings.put('MAX_FRAME_GAP', processor.ij.py.to_java(int(self.max_frame_gap)))
if self.verbose:
tqdm.write(f"Running tracking with parameters: linking_distance={self.linking_max_distance}, gap_closing={self.gap_closing_max_distance}, frame_gap={self.max_frame_gap}")
# Run tracking
tracking_results = processor.run_tracking(model, settings)
if not tracking_results["success"]:
raise RuntimeError(f"Tracking failed: {tracking_results.get('error', 'Unknown error')}")
if self.verbose:
tqdm.write(f"Tracking successful: {tracking_results['n_tracks']} tracks from {tracking_results['n_spots']} spots")
# Extract tracking results and merge with original features
tracked_features = self._extract_trackmate_tracking_results(model, features_df)
# Return in the same format as file-based processing
if output:
# For consistency with file-based method, we don't generate XML in in-memory mode
return tracked_features, None
else:
return tracked_features
def _features_to_trackmate_spots(self, features_df: pd.DataFrame) -> pd.DataFrame:
"""
Convert features DataFrame to TrackMate spots format
Args:
features_df: Features DataFrame with cyto format
Returns:
DataFrame in TrackMate spots format
"""
spots_data = []
for _, row in features_df.iterrows():
spot_data = {
'ID': int(row['label']),
'FRAME': int(row['frame']),
'POSITION_X': float(row['x']), # x is column
'POSITION_Y': float(row['y']), # y is row
'POSITION_Z': 0.0, # 2D case
'RADIUS': float(row['feret_radius']),
'QUALITY': float(row['mean']),
'MEAN_INTENSITY': float(row['mean']),
'TOTAL_INTENSITY': float(row['mean'] * row['size']),
'AREA': float(row['size']),
}
# Add optional features if available
if 'max_intensity' in row:
spot_data['MAX_INTENSITY'] = float(row['max_intensity'])
if 'perimeter' in row:
spot_data['PERIMETER'] = float(row['perimeter'])
if 'eccentricity' in row:
spot_data['ECCENTRICITY'] = float(row['eccentricity'])
if 'solidity' in row:
spot_data['SOLIDITY'] = float(row['solidity'])
spots_data.append(spot_data)
return pd.DataFrame(spots_data)
def _trackmate_spots_to_features(self, spots_df: pd.DataFrame) -> pd.DataFrame:
"""
Convert TrackMate spots format back to features DataFrame format
Args:
spots_df: DataFrame in TrackMate spots format
Returns:
DataFrame in cyto features format
"""
features_data = []
for _, row in spots_df.iterrows():
feature_data = {
'label': int(row['ID']),
'frame': int(row['FRAME']),
'i': float(row['POSITION_Y']), # y is row
'j': float(row['POSITION_X']), # x is column
'y': float(row['POSITION_Y']),
'x': float(row['POSITION_X']),
'size': float(row['AREA']),
'mean': float(row['MEAN_INTENSITY']),
'feret_radius': float(row['RADIUS']),
'channel': 0
}
# Add optional features if available
if 'MAX_INTENSITY' in row:
feature_data['max_intensity'] = float(row['MAX_INTENSITY'])
if 'PERIMETER' in row:
feature_data['perimeter'] = float(row['PERIMETER'])
if 'ECCENTRICITY' in row:
feature_data['eccentricity'] = float(row['ECCENTRICITY'])
if 'SOLIDITY' in row:
feature_data['solidity'] = float(row['SOLIDITY'])
features_data.append(feature_data)
return pd.DataFrame(features_data)
def _extract_trackmate_tracking_results(self, model: Any, original_features: pd.DataFrame) -> pd.DataFrame:
"""
Extract tracking results from TrackMate model and merge with original features
Args:
model: TrackMate model with tracking results
original_features: Original features DataFrame
Returns:
DataFrame with tracking results merged with original features
"""
# Create mapping from spot to track ID
spot_to_track = {}
for track_id in model.getTrackModel().trackIDs(True):
track_spots = model.getTrackModel().trackSpots(track_id)
for spot in track_spots:
spot_to_track[spot.ID()] = int(track_id)
# Create tracking results DataFrame
tracking_data = []
for spot in model.getSpots().iterable(True):
track_id = int(spot_to_track.get(spot.ID(), -1)) # -1 for untracked spots
tracking_data.append({
'label': int(spot.getFeature('ORIGINAL_LABEL_ID')),
'frame': int(spot.getFeature('FRAME')),
'x': float(spot.getFeature('POSITION_X')),
'y': float(spot.getFeature('POSITION_Y')),
# 'track_id': track_id if track_id != -1 else None
'track_id': track_id
})
tracking_df = pd.DataFrame(tracking_data)
# Merge with original features
result_features = pd.merge(
original_features,
tracking_df[['label', 'frame', 'track_id']],
on=['label', 'frame'],
how='left'
)
# Ensure 'track_id' column is integer type (use -1 for missing values)
# result_features['track_id'] = result_features['track_id'].astype(int)
return result_features
[docs]
def main():
from aicsimageio import AICSImage
import pandas as pd
import numpy as np
from sklearn.model_selection import ParameterGrid
FIJI_DIR = "/home/vpfannenstill/Fiji.app"
image_ = AICSImage("/home/vpfannenstill/Projects/Cytotoxicity-Pipeline/output/preprocessing/PercentileNormalization/CancerCell.tif")
image = image_.get_image_data("XYT")
feature = pd.read_csv("/home/vpfannenstill/Projects/Cytotoxicity-Pipeline/output/tracking/Alive.csv")
feature["cell_type"] = np.nan
param_grid = {
"linking_max_distance": range(20,26,2),
"max_frame_gap": range(4,6,1),
"gap_closing_max_distance": range(10,26,2),
}
params = list(ParameterGrid(param_grid))
# initiate Fiji
imagej.doctor.checkup()
if not os.path.exists(FIJI_DIR) or FIJI_DIR == "":
raise Exception("Fiji.app directory not found")
print("Initializing Fiji on JVM...")
ij = imagej.init(FIJI_DIR,mode='headless')
print(ij.getApp().getInfo(True))
pbar = tqdm(params[0:])
for param in pbar:
tm = TrackMate(FIJI_DIR=FIJI_DIR,
ij=ij,
gap_closing_max_distance=param["gap_closing_max_distance"],
linking_max_distance=param["linking_max_distance"],
max_frame_gap=param["max_frame_gap"],
size_min=3
)
tm({"image": image, "feature":feature})
if __name__ == "__main__":
main()