ByoTrack fundamental features

import cv2
import numpy as np
import matplotlib as mpl
import matplotlib.pyplot as plt
import torch

import byotrack
import byotrack.example_data
import byotrack.visualize

Loading a video

# Open an example video
video = byotrack.example_data.hydra_neurons()[130:]  # Let's start at frame 130 where the animal is contracting

# Or provide a path to one of your video
# video = byotrack.Video("path/to/video.ext")

# Or load manually a video as a numpy array
# video = np.random.randn(50, 500, 500, 3)  # T, H, W, C

TEST = True  # Set to False to analyze a whole video

if TEST:
    video = video[:50]  # Temporal slicing to analyze only the first 50 frames

# For video only (With numpy arrays, your are responsible for channels aggregation and normalization)
# A transform can be added to normalize and aggregate channels

transform_config = byotrack.VideoTransformConfig(
    aggregate=True, normalize=True, q_min=0.01, q_max=0.999, smooth_clip=1.0
)
video.set_transform(transform_config)

# Show the min max value used to clip and normalize
print(video._normalizer.mini, video._normalizer.maxi)

[0.] [248.]

# Display the first frame

plt.figure(figsize=(24, 16), dpi=100)
plt.imshow(video[0])
plt.show()

# Display the video with opencv
# Use w/x to move forward in time (or space to run/pause the video)
# Use v to switch on/off0 the display of the video

byotrack.visualize.InteractiveVisualizer(video).run()

Detections on a video: Example of WaveletDetector

# Create the detector object with its hyper parameters
from byotrack.implementation.detector.wavelet import WaveletDetector

detector = WaveletDetector(scale=1, k=3.0, min_area=5.0, batch_size=20, device=torch.device("cpu"))

# Run the detection process on the current video

detections_sequence = detector.run(video)

# Display the first detections

segmentation = detections_sequence[0].segmentation.clone()
segmentation[segmentation!=0] += 50  # Improve visibility of firsts labels

plt.figure(figsize=(24, 16), dpi=100)
plt.imshow(segmentation)
plt.show()

# Display the detections with opencv
# Use w/x to move forward in time (or space to run/pause the video)
# Use v to switch on/off the display of the video
# Use d to switch detection display mode (None, mask, segmentation)

byotrack.visualize.InteractiveVisualizer(video, detections_sequence).run()

# Set hyperparameters manually on the video
# Use w/x to move backward/forward in the video
# Use c/v to update k (the main hyperparameter)
# You can restard with another scale/min_area

K_SPEED = 0.01

i = 0
detector = WaveletDetector(scale=1, k=3.0, min_area=5.0, device=torch.device("cpu"))

while True:
    frame = video[i]
    detections = detector.detect(frame[None, ...])[0]
    mask = (detections.segmentation.numpy() != 0).astype(np.uint8) * 255

    # Display the resulting frame
    cv2.imshow('Frame', mask)
    cv2.setWindowTitle('Frame', f'Frame {i} / {len(video)} - k={detector.k} - Num detections: {detections.length}')

    # Press Q on keyboard to  exit
    key = cv2.waitKey() & 0xFF

    if key == ord('q'):
        break

    if cv2.getWindowProperty("Frame", cv2.WND_PROP_VISIBLE) <1:
        break

    if key == ord("w"):
        i = (i - 1) % len(video)

    if key == ord("x"):
        i = (i + 1) % len(video)

    if key == ord("c"):
        detector.k = detector.k * (1 - K_SPEED)

    if key == ord("v"):
        detector.k = detector.k * (1 + K_SPEED)


cv2.destroyAllWindows()

Link detections: Example of KOFTLinker (Kalman and Optical Flow Tracking)

# KOFT requires Optical Flow. We give here the example of Farneback from Open-CV.

from byotrack.implementation.linker.frame_by_frame.koft import KOFTLinker, KOFTLinkerParameters
from byotrack.implementation.optical_flow.opencv import OpenCVOpticalFlow

# Prepare the optical flow algorithm
optflow = OpenCVOpticalFlow(cv2.FarnebackOpticalFlow_create(winSize=20), downscale=4)

# Create the linker
# You can look at the documentation (KOFTLinkerParameters?) to see what parameters are available

specs = KOFTLinkerParameters(
    association_threshold=2e-3,  # Most important parameter: don't link if the association likelihood is smaller than 1e-3.
    detection_std=3.0,  # Detections location are precise up to 3.0 pixels (Usually ~ size of spots)
    process_std=1.5,  # Kalman filter predictions are precise up to 1.5 pixels
    flow_std=1.0,  # Optical flow predictions are precise up to 1.0 pixels/frame
    kalman_order=1,  # Order of the kalman filter (Model up to 1st derivatives of positions)
    n_gap=5,  # Allow to link after 5 consecutive missed detections
    cost="likelihood",  # See koft.Cost? to see which other cost are available, by default it uses Euclidean distance (And association threshod should be express in pixels)
)

linker = KOFTLinker(specs, optflow)

# Before linking, let's check visually that the optical flow algorithm works
# We sample a grid of points that are moved by the flow computed.
# The computed flows are good if the points roughly follows the video motion

# Use w/x to move forward in time (or space to run/pause the video)
# Use g to reset the grid of points

byotrack.visualize.InteractiveFlowVisualizer(video, optflow).run()

# Run the linker given a video and detections

tracks = linker.run(video, detections_sequence)

# Visualize track lifetime

# Each track is in white when it alive. (Track on x-axis, time on y-axis)

byotrack.visualize.display_lifetime(tracks)

# Project tracks onto a single image and color by time

# Create a list of colors for each time frame
# From cyan (start of the video) to red (end of the video)

hsv = mpl.colormaps["hsv"]
colors = [tuple(int(c * 255) for c in hsv(0.5 + 0.5 * k / (len(detections_sequence) - 1))[:3]) for k in range(len(detections_sequence))]

visu = byotrack.visualize.temporal_projection(
    byotrack.Track.tensorize(tracks),
    colors=colors,
    background=video[0],
    color_by_time=True
)

plt.figure(figsize=(24, 16))
plt.imshow(visu)
plt.show()

# Display the tracks with opencv
# Use w/x to move forward in time (or space to run/pause the video)
# Use v (resp. t) to switch on/off the display of video (resp. tracks)
# Use d to switch detection display mode (None, mask, segmentation)

byotrack.visualize.InteractiveVisualizer(video, detections_sequence, tracks).run()

Tracks refinement: Example of Cleaner, followed by EMC2 Stitcher

from byotrack.implementation.refiner.cleaner import Cleaner
from byotrack.implementation.refiner.interpolater import ForwardBackwardInterpolater
from byotrack.implementation.refiner.stitching import EMC2Stitcher

# Split tracks with consecutive dist > 5. Splitting may be counterproductive, check the tracks before applying such cleaner.
# Drop tracks with length < 5

cleaner = Cleaner(min_length=5, max_dist=5.)
tracks = cleaner.run(video, tracks)

Cleaning: Split 68 tracks and dropped 88 resulting ones
Cleaning: From 871 to 851 tracks

# Visualize track lifetime

byotrack.visualize.display_lifetime(tracks)

# Stitch tracks together in order to produce coherent track on all the video

stitcher = EMC2Stitcher(eta=5.0)  # Don't link tracks if they are too far (EMC dist > 5 (pixels))
tracks = stitcher.run(video, tracks)

Merging 851 tracks into 808 resulting tracks

# Visualize track lifetime

byotrack.visualize.display_lifetime(tracks)

# After EMC2 stitching, NaN values can be inside merged tracks.
# It can be filled with interpolation between known positions

method = "tps"  # tps / constant / flow (You need to provided a valid byotrack.OpticalFlow then)
full = False  # Extrapolate position of the tracks on the all frame range and not just for the track lifespan

interpolater = ForwardBackwardInterpolater(method, full)
tracks = interpolater.run(video, tracks)

# Visualize track lifetime

byotrack.visualize.display_lifetime(tracks)

# Project tracks onto a single image and color by time

# Create a list of colors for each time frame
# From cyan (start of the video) to red (end of the video)

hsv = mpl.colormaps["hsv"]
colors = [tuple(int(c * 255) for c in hsv(0.5 + 0.5 * k / (len(detections_sequence) - 1))[:3]) for k in range(len(detections_sequence))]

visu = byotrack.visualize.temporal_projection(
    byotrack.Track.tensorize(tracks),
    colors=colors,
    background=video[0],
    color_by_time=True
)

plt.figure(figsize=(24, 16))
plt.imshow(visu)
plt.show()

End-to-end tracking - Online or Offline tracking

from byotrack import BatchMultiStepTracker, MultiStepTracker

# Create a full tracking pipeline from detector, linker and refiners
# If you have a BatchDetector and a OnlineLinker (True for WaveletDetector and KOFTLinker)
# You may use BatchMultiStepTracker that will process online the video (never keeping all the segmentations in RAM)
# Otherwise, use MultiStepTracker (Run Detections, then linking)

tracker = BatchMultiStepTracker(detector, linker, (cleaner, stitcher, interpolater))
# tracker = MultiStepTracker(detector, linker, (cleaner, stitcher, interpolater))

tracks = tracker.run(video)

Cleaning: Split 68 tracks and dropped 88 resulting ones
Cleaning: From 871 to 851 tracks

Merging 851 tracks into 808 resulting tracks

# Visualize track lifetime

byotrack.visualize.display_lifetime(tracks)

# Project tracks onto a single image and color by track

# Create a list of colors for each track (if more tracks than colors, it will cycle through it)

hsv = mpl.colormaps["hsv"]
colors = [tuple(int(c * 255) for c in hsv(k / 199)[:3]) for k in range(200)]

visu = byotrack.visualize.temporal_projection(
    byotrack.Track.tensorize(tracks),
    colors=colors,
)

plt.figure(figsize=(24, 16))
plt.imshow(visu)
plt.show()

# Display the tracks with opencv
# Use w/x to move forward in time (or space to run/pause the video)
# Use v (resp. t) to switch on/off the display of video (resp. tracks)
# Use d to switch detection display mode (None, mask, segmentation)

byotrack.visualize.InteractiveVisualizer(video, detections_sequence, tracks).run()

Load or save tracks to files

# Save tracks in ByoTrack format (compressed in a torch tensor)

byotrack.Track.save(tracks, "tracks.pt")

# Can be reload with
tracks = byotrack.Track.load("tracks.pt")

# We also provide IO with Icy software

from byotrack import icy


icy.save_tracks(tracks, "tracks.xml")  # Note that holes should should be filled first with the ForwardBackwardInterpolater

# You can (re)load tracks from icy with
tracks = icy.load_tracks("tracks.xml")